From b70cc6b785d782079379bdde94b032cc9104c35f Mon Sep 17 00:00:00 2001 From: nw-kirill Date: Thu, 10 Sep 2026 00:04:08 +0900 Subject: [PATCH 1/2] Let users edit run.sbatch and copy Slurm logs back into the project Cluster jobs were submitting an unseen generated script and leaving slurm-*.out/.err on the compute server (including failed jobs). Prepare a draft in the project tree, pin log paths after edits, and rsync logs on COMPLETED and FAILED so the Runs panel and Jupyter can open them. --- deployment/compute_server/README.md | 41 +- .../pr-cluster-sbatch-logs-progress.log | 127 ++++ .../app/workflow/execution/__init__.py | 10 +- .../app/workflow/execution/base.py | 3 + .../app/workflow/execution/local_executor.py | 1 + .../execution/remote_slurm_executor.py | 180 +++++- .../django-project/app/workflow/models.py | 14 +- .../app/workflow/serializers.py | 31 +- .../django-project/app/workflow/urls.py | 68 +- .../django-project/app/workflow/views.py | 580 +++++++++++++----- .../django-project/config/test_settings.py | 23 + .../tests/test_cluster_sbatch_logs.py | 220 +++++++ .../src/api/workflowRunApi.ts | 92 ++- .../views/home/components/ClusterRunModal.tsx | 242 +++++++- .../views/home/components/runStatusPanel.tsx | 84 ++- .../src/views/home/homeView.tsx | 26 +- 16 files changed, 1489 insertions(+), 253 deletions(-) create mode 100644 deployment/pr-cluster-sbatch-logs-progress.log create mode 100644 gui/workflow_backend/django-project/config/test_settings.py create mode 100644 gui/workflow_backend/django-project/tests/test_cluster_sbatch_logs.py diff --git a/deployment/compute_server/README.md b/deployment/compute_server/README.md index fcf8baad..53b82dd6 100644 --- a/deployment/compute_server/README.md +++ b/deployment/compute_server/README.md @@ -1,12 +1,45 @@ # Remote Slurm execution - operator runbook (RIKEN compute server) This enables the Django backend to submit a workflow as a **Slurm batch job** on -the RIKEN compute server over SSH, poll its status, and read back results. +the RIKEN compute server over SSH, poll its status, and copy logs and results +back into the NeuroWorkflow project tree. It is **additive and OFF by default**: the live JupyterHub run path is untouched, and this path runs only when a run is submitted with `backend=slurm`. The ssh-agent in the backend container holds no key until an admin unlocks it. +## User flow (GUI) + +1. **Prepare (draft).** Opening **Run on Compute Cluster** creates a `draft` + `WorkflowRun` and writes `run.sbatch` to + `codes/projects//batch//run.sbatch` (the same tree + Jupyter mounts). No SSH and no `sbatch` yet. +2. **Edit.** The Cluster Run modal shows the script. The user can edit it + there or **Edit in Jupyter** (then **Reload from project**). Changing + partition/CPU/memory/time regenerates the script unless the textarea is + dirty (confirm before overwrite). +3. **Submit.** `POST /api/workflow//runs/submit/` with `backend: "slurm"`, + the draft `run_id`, and the (possibly edited) `sbatch` text. The executor + pins `#SBATCH --chdir` / `--output` / `--error` to the remote run dir, rsyncs + the batch dir, and runs `sbatch run.sbatch`. +4. **Copy-back.** On the first poll that sees COMPLETED, FAILED, or CANCELLED, + the backend rsyncs `slurm-*.out`, `slurm-*.err`, `stdout.log`, `stderr.log`, + `exit_code.txt`, and `manifest.json` into + `codes/projects//batch//logs/`, and `results/` into + `.../results/` when that directory exists. The Runs panel lists **Logs** and + **Results** for download. Failed jobs are copied too (that is when `.err` + matters). + +Closing the modal without submit leaves the draft in the Runs panel (delete +removes it). **Edit script & resubmit** copies the previous `run.sbatch` into a +new draft. + +The full UI/API path is `POST /api/workflow//runs/submit/` with +`{"backend": "slurm"}` (not `/run-submit/`). Drafts use +`POST /api/workflow//runs/prepare/` and +`GET`/`PUT /api/workflow//runs//sbatch/`. + + ## Facts baked into the implementation (from RIKEN) - Login node: `digitalbrain.brainminds.jp`, user `neuro-workflow`. Used only for @@ -114,8 +147,10 @@ PY ``` Success = status `COMPLETED`, exit code `0`, and stdout containing the test -summary. The full UI/API path (`POST /api/workflow//run-submit/` with -`{"backend": "slurm"}`) uses exactly this executor. +summary. The full UI/API path (`POST /api/workflow//runs/submit/` with +`{"backend": "slurm"}`) uses exactly this executor. After a terminal status, +`get_status(..., project_id=...)` also copies `slurm-*.out/.err` into +`codes/projects//batch//logs/`. --- diff --git a/deployment/pr-cluster-sbatch-logs-progress.log b/deployment/pr-cluster-sbatch-logs-progress.log new file mode 100644 index 00000000..b2280bb0 --- /dev/null +++ b/deployment/pr-cluster-sbatch-logs-progress.log @@ -0,0 +1,127 @@ +================================================================================ +PR cluster-sbatch-logs — progress log (append-only) +Worktree: /home/nw-kirill/neuro-workflow-cluster-sbatch +Branch: feat/cluster-sbatch-logs (from origin/main e1218777) +Base: origin/main (independent of #88–#95) +================================================================================ + +## Verdict + +Cluster submit on main already generates run.sbatch from a template and rsyncs +only results/ on COMPLETED. Native slurm-%j.out / .err stay on the compute +server; failed jobs are not copied back; there is no editor. This PR completes +cluster-job I/O: draft + editable run.sbatch (modal + Jupyter) and copy-back +of Slurm/Python logs into codes/projects//batch//logs/. + +Kirill’s choices +---------------- +- Study/product: Open Composer pattern inside NWF (form → edit script → submit). + Modal textarea + file on disk under Jupyter’s project tree. No OOD install. +- Copy-back into the same batch dir (logs/ + existing results/). COMPLETED and + FAILED (and CANCELLED if files exist). Pin --chdir/--output/--error after edits. +- No new dependencies, no Dockerfile.nest, no live compose/deploy/merge. + +Non-goals +--------- +sudo; write /data; nest on #88–#95; force-push; Open OnDemand; Keycloak/nginx; +expand GET .../files/ into a recursive browser; call unused get_logs() from HTTP. + +Integrity vs other PRs +----------------------- +Independent branch from origin/main. Surgical edits: executor, run views, +ClusterRunModal, Runs panel, run API client, compute-server README. +Adding WorkflowRun.Status.DRAFT is a Python choice; no DB migration. +DetailView polls only pending/running — copy-back must run on the get_status +that first sees COMPLETED/FAILED. Draft must not call sacct. + +Live UI baseline +---------------- +https://neuro-workflow.dbrain.jp/ will not show this until deploy. After +implementation, confirm the gap is still there (no sbatch editor, no .out/.err +in Runs). Prove the engine with pytest in this worktree. + +================================================================================ + +## Step 1 — Worktree and progress log + +Done. Created worktree from origin/main (e1218777, Merge pull request #87). +Branch feat/cluster-sbatch-logs. This file created. + +Verify: git status is feat/cluster-sbatch-logs; HEAD is e1218777; log exists. + +================================================================================ + +## Step 2 — Draft + editable run.sbatch API + +Done. Added WorkflowRun.Status.DRAFT (no migration). normalize_sbatch pins +--chdir/--output/--error. POST /runs/prepare/ writes run.sbatch with no SSH. +GET/PUT /runs//sbatch/. Submit accepts optional run_id (draft) + sbatch. +from_run_id copies a previous script into a new draft. PUT with +resource_requests re-renders when the modal form changes. + +Verify: prepare does not call sbatch (pytest boom on _ssh); PUT without +--chdir still has pinned directives on disk. + +## Step 3 — Copy .out/.err back + +Done. get_status on COMPLETED/FAILED/CANCELLED rsyncs slurm-*.out/.err, +stdout.log, stderr.log, exit_code.txt, manifest.json into logs/, and still +fetches results/ when present (including FAILED). artifacts JSON has files + +logs. Artifact view serves logs/, results/, run.sbatch; rejects nodes/ and +.. . DB stdout/stderr capped at 512 KiB. get_logs() still unused by HTTP. + +Verify: mocked FAILED copy-back writes logs/slurm-1.err; download 200; +path=../x is 400; nodes/foo.py is 400. + +## Step 4 — Cluster Run modal + +Done. Prepare on open; monospace textarea; Reload from project; Edit in Jupyter +(user1 tree path, same as today); dirty flag; confirm before regenerate; +submit sends run_id + sbatch after generate-code. + +## Step 5 — Runs panel + +Done. draft badge (purple). Logs + Results download lists. Open run folder in +Jupyter. Edit script & resubmit. Cancel hidden for draft (treated as terminal +for polling). + +## Step 6 — Tests + +Added django-project/tests/test_cluster_sbatch_logs.py (8 tests). No vitest +files in the frontend, so no new frontend runner. SQLite cannot apply this +repo’s Postgres migrations; pytest uses an ephemeral postgres:16 container +on docker network nw-sbatch-test (not live compose, not live DB). + +Verify: 8 passed. + +## Step 7 — Docs + +Updated deployment/compute_server/README.md: draft → edit → sbatch; copy-back +on success and failure; Jupyter path; fixed stale /run-submit/ → +/runs/submit/. deployment/DEPLOY_COMPUTE_SERVER.md is not in this git tree +(control/docs only); the in-repo operator README is the document that shipped. + +No TODOs/placeholders in the new engine paths except existing executor +NotImplemented elsewhere. + +================================================================================ + +## Step 8 — Isolated verification + +pytest django-project/tests/test_cluster_sbatch_logs.py against ephemeral +postgres:16 (docker network nw-sbatch-test, not live DB): 8 passed. + +black --check --fast --line-length 88 and isort --check --profile black on +touched Python: pass. + +Live https://neuro-workflow.dbrain.jp/ (logged in as kirill; did not submit a +cluster job, did not upload data, did not change settings): +- Toolbar still has “Run on Compute Cluster”. +- JS bundle /assets/index-CbQzE8nY.js has no run.sbatch, /runs/prepare, + “Edit in Jupyter”, or “Open run folder in Jupyter”. +- No NW_Optimization (old bundle). This is the expected baseline, not a + failed test of this PR. + +Ephemeral Postgres container nw-sbatch-pg removed after tests. + +================================================================================ diff --git a/gui/workflow_backend/django-project/app/workflow/execution/__init__.py b/gui/workflow_backend/django-project/app/workflow/execution/__init__.py index f4e312e9..ad54775a 100644 --- a/gui/workflow_backend/django-project/app/workflow/execution/__init__.py +++ b/gui/workflow_backend/django-project/app/workflow/execution/__init__.py @@ -1,6 +1,10 @@ -from .base import ExecutionBackend, ExecutionStatus, ExecutionResult +from .base import ExecutionBackend, ExecutionResult, ExecutionStatus from .local_executor import LocalExecutor -from .remote_slurm_executor import RemoteSlurmExecutor +from .remote_slurm_executor import ( + RemoteSlurmExecutor, + jupyter_sbatch_path, + normalize_sbatch, +) __all__ = [ "ExecutionBackend", @@ -8,4 +12,6 @@ "ExecutionResult", "LocalExecutor", "RemoteSlurmExecutor", + "jupyter_sbatch_path", + "normalize_sbatch", ] diff --git a/gui/workflow_backend/django-project/app/workflow/execution/base.py b/gui/workflow_backend/django-project/app/workflow/execution/base.py index 1a10effa..481a4e9d 100644 --- a/gui/workflow_backend/django-project/app/workflow/execution/base.py +++ b/gui/workflow_backend/django-project/app/workflow/execution/base.py @@ -46,12 +46,15 @@ def submit( *, run_id: Optional[str] = None, resource_requests: Optional[dict] = None, + sbatch_text: Optional[str] = None, ) -> ExecutionResult: """Submit a workflow run. Returns immediately with a pending result. ``run_id`` lets the caller pin the run identifier (e.g. the DB WorkflowRun id) so staging dirs, remote dirs and later status polls all line up. If omitted, a fresh UUID is generated. + ``sbatch_text`` is used by the Slurm backend when the user edited the + batch script; local backends ignore it. """ ... diff --git a/gui/workflow_backend/django-project/app/workflow/execution/local_executor.py b/gui/workflow_backend/django-project/app/workflow/execution/local_executor.py index ce945967..25aba004 100644 --- a/gui/workflow_backend/django-project/app/workflow/execution/local_executor.py +++ b/gui/workflow_backend/django-project/app/workflow/execution/local_executor.py @@ -34,6 +34,7 @@ def submit( *, run_id: Optional[str] = None, resource_requests: Optional[dict] = None, + sbatch_text: Optional[str] = None, ) -> ExecutionResult: result = ExecutionResult( status=ExecutionStatus.PENDING, diff --git a/gui/workflow_backend/django-project/app/workflow/execution/remote_slurm_executor.py b/gui/workflow_backend/django-project/app/workflow/execution/remote_slurm_executor.py index ac1c5601..7423886e 100644 --- a/gui/workflow_backend/django-project/app/workflow/execution/remote_slurm_executor.py +++ b/gui/workflow_backend/django-project/app/workflow/execution/remote_slurm_executor.py @@ -20,9 +20,8 @@ from pathlib import Path from typing import Optional -from django.conf import settings - from app.workflow.path_utils import batch_run_dir, projects_root +from django.conf import settings from .base import ExecutionBackend, ExecutionResult, ExecutionStatus @@ -85,6 +84,59 @@ def _rsync( subprocess.run(rsync_args, check=True, capture_output=True, text=True, timeout=300) +SBATCH_MAX_BYTES = 64 * 1024 +_PINNED_SBATCH_RE = re.compile( + r"^#SBATCH\s+(?:--(?:chdir|workdir|output|error)|-[Doe])\b.*$", + re.IGNORECASE, +) + + +def normalize_sbatch(text: str, remote_run_dir: str) -> str: + """Validate a user (or generated) sbatch script and pin log paths. + + The script body stays under the user's control. ``--chdir``, ``--output``, + and ``--error`` are rewritten so Slurm files land in the known remote run + directory where copy-back can find them. + """ + if not isinstance(text, str): + raise ValueError("sbatch script must be a string") + if "\x00" in text: + raise ValueError("sbatch script contains a NUL byte") + if len(text.encode("utf-8")) > SBATCH_MAX_BYTES: + raise ValueError("sbatch script exceeds 64 KiB") + body = text.replace("\r\n", "\n").replace("\r", "\n").strip() + if not body: + raise ValueError("sbatch script is empty") + + lines = body.split("\n") + if not lines[0].startswith("#!"): + lines.insert(0, "#!/bin/bash") + + shebang = lines[0] + kept = [line for line in lines[1:] if not _PINNED_SBATCH_RE.match(line.strip())] + remote = remote_run_dir.rstrip("/") + pinned = [ + f"#SBATCH --chdir={remote}", + f"#SBATCH --output={remote}/slurm-%j.out", + f"#SBATCH --error={remote}/slurm-%j.err", + ] + return "\n".join([shebang] + pinned + kept) + "\n" + + +def jupyter_sbatch_path(project_id, run_id) -> str: + return f"codes/projects/{project_id}/batch/{run_id}/run.sbatch" + + +_STDOUT_CAP_BYTES = 512 * 1024 + + +def _cap_text(value: str, limit: int = _STDOUT_CAP_BYTES) -> str: + raw = value.encode("utf-8") + if len(raw) <= limit: + return value + return raw[:limit].decode("utf-8", errors="ignore") + "\n...[truncated]\n" + + class RemoteSlurmExecutor(ExecutionBackend): """Submit workflow scripts to a Slurm cluster via SSH. @@ -125,9 +177,7 @@ def _sync_from_remote(self, remote: str, local: str) -> None: def _remote_run_dir(self, run_id: str) -> str: return f"{self.remote_dir}/{run_id}" - def _fetch_results( - self, run_id: str, remote_run_dir: str, project_id: str - ) -> dict: + def _fetch_results(self, run_id: str, remote_run_dir: str, project_id: str) -> dict: """Rsync the job's ``results/`` dir back to the app server and index it. Results land in ``codes/projects//batch//results/`` @@ -160,6 +210,55 @@ def _fetch_results( ) return {"files": files} + def _fetch_job_files( + self, run_id: str, remote_run_dir: str, project_id: str + ) -> dict: + """Copy Slurm/Python logs and, when present, ``results/``. + + Logs land in ``batch//logs/``. Called on the poll that first + observes a terminal Slurm state (COMPLETED, FAILED, or CANCELLED). + """ + artifacts: dict = {"files": [], "logs": []} + local_logs = batch_run_dir(project_id, run_id, create=True) / "logs" + local_logs.mkdir(parents=True, exist_ok=True) + names = ["stdout.log", "stderr.log", "exit_code.txt", "manifest.json"] + try: + listing = self._ssh( + f"ls -1 {shlex.quote(remote_run_dir)} 2>/dev/null || true" + ) + for name in listing.splitlines(): + name = name.strip() + if re.fullmatch(r"slurm-\d+\.(out|err)", name): + names.append(name) + except Exception as exc: + logger.warning("log listing failed for run %s: %s", run_id, exc) + + seen = set() + for name in names: + if not name or name in seen: + continue + seen.add(name) + try: + self._sync_from_remote( + f"{remote_run_dir}/{name}", + str(local_logs / name), + ) + except Exception as exc: + logger.debug("did not fetch %s for run %s: %s", name, run_id, exc) + + artifacts["logs"] = [ + {"path": f"logs/{f.name}", "size": f.stat().st_size} + for f in sorted(local_logs.iterdir()) + if f.is_file() + ] + + try: + results = self._fetch_results(run_id, remote_run_dir, project_id) + artifacts["files"] = results.get("files", []) if results else [] + except Exception as exc: + logger.warning("fetch_results failed for run %s: %s", run_id, exc) + return artifacts + def _build_sbatch_extras(self, rr: dict) -> str: """Translate a resource_requests dict into #SBATCH directive lines. @@ -208,6 +307,29 @@ def _render_sbatch( rendered = rendered.replace(key, value) return rendered + def write_sbatch( + self, + workflow_id: str, + run_id: str, + project_name: str, + resource_requests: Optional[dict] = None, + sbatch_text: Optional[str] = None, + ) -> str: + """Write a normalized ``run.sbatch`` into the local batch dir. No SSH.""" + remote_run_dir = self._remote_run_dir(run_id) + local_dir = batch_run_dir(workflow_id, run_id, create=True) + if sbatch_text: + script = normalize_sbatch(sbatch_text, remote_run_dir) + else: + script = normalize_sbatch( + self._render_sbatch( + run_id, project_name, remote_run_dir, resource_requests or {} + ), + remote_run_dir, + ) + (local_dir / "run.sbatch").write_text(script) + return script + # -- interface implementation ------------------------------------------- def submit( @@ -218,6 +340,7 @@ def submit( *, run_id: Optional[str] = None, resource_requests: Optional[dict] = None, + sbatch_text: Optional[str] = None, ) -> ExecutionResult: run_id = run_id or str(uuid.uuid4()) result = ExecutionResult( @@ -263,11 +386,21 @@ def submit( # Written last so the freshly generated code and sbatch are authoritative # (they override any stale copies picked up from the project dir). (local_dir / "workflow.py").write_text(code) - (local_dir / "run.sbatch").write_text( - self._render_sbatch( - run_id, project_name, remote_run_dir, resource_requests or {} - ) - ) + try: + if sbatch_text: + script = normalize_sbatch(sbatch_text, remote_run_dir) + else: + script = normalize_sbatch( + self._render_sbatch( + run_id, project_name, remote_run_dir, resource_requests or {} + ), + remote_run_dir, + ) + except ValueError as exc: + result.status = ExecutionStatus.FAILED + result.error = str(exc) + return result + (local_dir / "run.sbatch").write_text(script) try: self._ssh(f"mkdir -p {remote_run_dir}") @@ -324,25 +457,32 @@ def get_status( except Exception: pass try: - result.stdout = self._ssh(f"cat {remote_run_dir}/stdout.log") + result.stdout = _cap_text(self._ssh(f"cat {remote_run_dir}/stdout.log")) except Exception: pass try: - result.stderr = self._ssh(f"cat {remote_run_dir}/stderr.log") + result.stderr = _cap_text(self._ssh(f"cat {remote_run_dir}/stderr.log")) except Exception: pass result.finished_at = datetime.now(timezone.utc) - # On success, pull the result artifacts back to the app server. The - # DetailView only polls while the run is non-terminal, so this runs once - # (on the poll that first observes COMPLETED). - if result.status == ExecutionStatus.COMPLETED and project_id: + # DetailView only polls while the run is non-terminal, so copy-back + # runs once — on the poll that first observes COMPLETED/FAILED/CANCELLED. + if ( + result.status + in ( + ExecutionStatus.COMPLETED, + ExecutionStatus.FAILED, + ExecutionStatus.CANCELLED, + ) + and project_id + ): try: - result.artifacts = self._fetch_results( + result.artifacts = self._fetch_job_files( run_id, remote_run_dir, project_id ) except Exception as exc: - logger.warning("fetch_results failed for run %s: %s", run_id, exc) + logger.warning("fetch_job_files failed for run %s: %s", run_id, exc) return result @@ -351,9 +491,7 @@ def get_logs(self, run_id: str, *, remote_dir: Optional[str] = None) -> str: parts = [] for fname in ("stdout.log", "stderr.log"): try: - content = self._ssh( - f"cat {remote_run_dir}/{fname} 2>/dev/null || true" - ) + content = self._ssh(f"cat {remote_run_dir}/{fname} 2>/dev/null || true") if content: parts.append(content) except Exception: diff --git a/gui/workflow_backend/django-project/app/workflow/models.py b/gui/workflow_backend/django-project/app/workflow/models.py index 50b9deab..3e27c0cc 100644 --- a/gui/workflow_backend/django-project/app/workflow/models.py +++ b/gui/workflow_backend/django-project/app/workflow/models.py @@ -1,7 +1,8 @@ -from django.db import models -from django.contrib.auth.models import User import uuid +from django.contrib.auth.models import User +from django.db import models + def _default_workflow_context(): return { @@ -93,7 +94,7 @@ def get_modified_parameters(self): param_key: { "original_value": param_info.get("original_value"), "current_value": param_info.get("current_value"), - "modified_at": param_info.get("modified_at") + "modified_at": param_info.get("modified_at"), } for param_key, param_info in modifications.items() if param_info.get("is_modified", False) @@ -132,6 +133,7 @@ class WorkflowRun(models.Model): """Tracks a single execution of a workflow (local or remote).""" class Status(models.TextChoices): + DRAFT = "draft", "Draft" PENDING = "pending", "Pending" RUNNING = "running", "Running" COMPLETED = "completed", "Completed" @@ -148,7 +150,11 @@ class Backend(models.TextChoices): FlowProject, on_delete=models.CASCADE, related_name="runs" ) user = models.ForeignKey( - User, on_delete=models.SET_NULL, null=True, blank=True, related_name="workflow_runs" + User, + on_delete=models.SET_NULL, + null=True, + blank=True, + related_name="workflow_runs", ) backend = models.CharField( max_length=20, choices=Backend.choices, default=Backend.JUPYTER diff --git a/gui/workflow_backend/django-project/app/workflow/serializers.py b/gui/workflow_backend/django-project/app/workflow/serializers.py index 397c6fc7..a738389a 100644 --- a/gui/workflow_backend/django-project/app/workflow/serializers.py +++ b/gui/workflow_backend/django-project/app/workflow/serializers.py @@ -1,8 +1,8 @@ -from rest_framework import serializers -from .models import FlowProject, FlowNode, FlowEdge, WorkflowRun +from app.box.models import get_categories from django.contrib.auth.models import User +from rest_framework import serializers -from app.box.models import get_categories +from .models import FlowEdge, FlowNode, FlowProject, WorkflowRun def _valid_category_values() -> list[str]: @@ -107,9 +107,7 @@ def validate_name(self, value): request = self.context.get("request") if self.context else None owner = getattr(request, "user", None) if request else None if owner and getattr(owner, "is_authenticated", False): - qs = FlowProject.objects.filter( - owner=owner, name=name, is_active=True - ) + qs = FlowProject.objects.filter(owner=owner, name=name, is_active=True) if self.instance is not None: qs = qs.exclude(pk=self.instance.pk) if qs.exists(): @@ -169,7 +167,13 @@ class Meta: "modified_parameters", "parameter_modification_count", ] - read_only_fields = ["created_at", "updated_at", "has_parameter_modifications", "modified_parameters", "parameter_modification_count"] + read_only_fields = [ + "created_at", + "updated_at", + "has_parameter_modifications", + "modified_parameters", + "parameter_modification_count", + ] def get_has_parameter_modifications(self, obj): """Are there any parameter changes?""" @@ -340,3 +344,16 @@ class WorkflowRunSubmitSerializer(serializers.Serializer): default=WorkflowRun.Backend.JUPYTER, ) resource_requests = serializers.DictField(required=False, default=dict) + run_id = serializers.UUIDField(required=False, allow_null=True) + sbatch = serializers.CharField(required=False, allow_blank=True, default="") + + +class WorkflowRunPrepareSerializer(serializers.Serializer): + resource_requests = serializers.DictField(required=False, default=dict) + from_run_id = serializers.UUIDField(required=False, allow_null=True) + sbatch = serializers.CharField(required=False, allow_blank=True, default="") + + +class WorkflowRunSbatchSerializer(serializers.Serializer): + sbatch = serializers.CharField(required=False, allow_blank=True, default="") + resource_requests = serializers.DictField(required=False) diff --git a/gui/workflow_backend/django-project/app/workflow/urls.py b/gui/workflow_backend/django-project/app/workflow/urls.py index 012e533e..6f0e614c 100644 --- a/gui/workflow_backend/django-project/app/workflow/urls.py +++ b/gui/workflow_backend/django-project/app/workflow/urls.py @@ -1,24 +1,27 @@ from django.urls import path + from .views import ( - FlowProjectViewSet, - FlowNodeViewSet, - FlowEdgeViewSet, - SampleFlowView, BatchCodeGenerationView, BatchWorkflowRunView, - WorkflowRunStreamView, + FlowEdgeViewSet, FlowNodeInstanceNameUpdateView, FlowNodeParameterUpdateView, - WorkflowRunSubmitView, + FlowNodeViewSet, + FlowProjectViewSet, + SampleFlowView, + ViewerChatToolView, + WorkflowCodeView, + WorkflowProjectFilesView, + WorkflowReportView, + WorkflowResultsView, + WorkflowRunArtifactView, + WorkflowRunCancelView, WorkflowRunDetailView, WorkflowRunListView, - WorkflowRunCancelView, - WorkflowRunArtifactView, - WorkflowResultsView, - WorkflowReportView, - WorkflowProjectFilesView, - WorkflowCodeView, - ViewerChatToolView, + WorkflowRunPrepareView, + WorkflowRunSbatchView, + WorkflowRunStreamView, + WorkflowRunSubmitView, ) app_name = "workflow" @@ -43,7 +46,6 @@ edge_detail = FlowEdgeViewSet.as_view({"delete": "destroy"}) - urlpatterns = [ # project management path("", project_list, name="workflow-list-create"), # GET(list), POST(create) @@ -80,75 +82,83 @@ path( "/nodes//instance_name/", FlowNodeInstanceNameUpdateView.as_view(), - name="node-instance_name-update" + name="node-instance_name-update", ), # PUT(node schema.instance_name update) # Update node parameters path( "/nodes//parameters/", FlowNodeParameterUpdateView.as_view(), - name="node-parameter-update" + name="node-parameter-update", ), # PUT(node schema.parameters update) # Batch Code Generation - New Addition path( "/generate-code/", BatchCodeGenerationView.as_view(), - name="batch-code-generation" + name="batch-code-generation", ), # POST (generate code in batch from React Flow JSON) # Run Workflow (SSE streaming via Jupyter kernel) path( "/run/", WorkflowRunStreamView.as_view(), - name="workflow-run-stream" + name="workflow-run-stream", ), # POST (Run workflow on Jupyter, SSE streaming output) # Async run management path( "/runs/", WorkflowRunListView.as_view(), - name="workflow-run-list" + name="workflow-run-list", ), # GET (list runs) + path( + "/runs/prepare/", + WorkflowRunPrepareView.as_view(), + name="workflow-run-prepare", + ), # POST (create a draft + write run.sbatch, no sbatch) path( "/runs/submit/", WorkflowRunSubmitView.as_view(), - name="workflow-run-submit" + name="workflow-run-submit", ), # POST (submit a new run) + path( + "/runs//sbatch/", + WorkflowRunSbatchView.as_view(), + name="workflow-run-sbatch", + ), # GET/PUT run.sbatch path( "/runs//", WorkflowRunDetailView.as_view(), - name="workflow-run-detail" + name="workflow-run-detail", ), # GET (run status + logs) path( "/runs//cancel/", WorkflowRunCancelView.as_view(), - name="workflow-run-cancel" + name="workflow-run-cancel", ), # POST (cancel a run) path( "/runs//artifacts/", WorkflowRunArtifactView.as_view(), - name="workflow-run-artifact" + name="workflow-run-artifact", ), # GET(?path=... download one fetched result file) # Results listing path( "/results/", WorkflowResultsView.as_view(), - name="workflow-results" + name="workflow-results", ), # GET(list result files with metadata) # Generated code and notebook outputs path( - "/code/", - WorkflowCodeView.as_view(), - name="workflow-code" + "/code/", WorkflowCodeView.as_view(), name="workflow-code" ), # GET(generated .py code + notebook cell outputs) # Report save/retrieve path( "/report/", WorkflowReportView.as_view(), - name="workflow-report" + name="workflow-report", ), # GET(read report), POST(save report) # Brain-viewer chat tool dispatch (LLM Group 1-5 tools over the run's data) path( "/viewer-chat/", ViewerChatToolView.as_view(), - name="workflow-viewer-chat" + name="workflow-viewer-chat", ), # POST({tool, args, data_path}) -> tool result / action dict # Project data files (upload into codes/projects//) path( diff --git a/gui/workflow_backend/django-project/app/workflow/views.py b/gui/workflow_backend/django-project/app/workflow/views.py index 61deb5db..a1192587 100644 --- a/gui/workflow_backend/django-project/app/workflow/views.py +++ b/gui/workflow_backend/django-project/app/workflow/views.py @@ -6,6 +6,7 @@ import shutil from pathlib import Path +from app.auth.authentication import KeycloakAuthentication from django.db import transaction from django.db.models import Q from django.http import ( @@ -25,9 +26,9 @@ from rest_framework.response import Response from rest_framework.views import APIView -from app.auth.authentication import KeycloakAuthentication - from .code_generation_service import CodeGenerationService +from .execution import LocalExecutor, RemoteSlurmExecutor +from .execution.remote_slurm_executor import jupyter_sbatch_path from .jupyter_execution_service import JupyterExecutionService from .models import FlowEdge, FlowNode, FlowProject, WorkflowRun from .path_utils import ( @@ -51,10 +52,11 @@ FlowEdgeSerializer, FlowNodeSerializer, FlowProjectSerializer, + WorkflowRunPrepareSerializer, + WorkflowRunSbatchSerializer, WorkflowRunSerializer, WorkflowRunSubmitSerializer, ) -from .execution import LocalExecutor, RemoteSlurmExecutor from .services import FlowService logger = logging.getLogger(__name__) @@ -181,7 +183,9 @@ def create(self, request, *args, **kwargs): # Validate nodeType in data data_field = request.data.get("data", {}) - node_type_val = data_field.get("nodeType") if isinstance(data_field, dict) else None + node_type_val = ( + data_field.get("nodeType") if isinstance(data_field, dict) else None + ) if not node_type_val: return Response( { @@ -192,6 +196,7 @@ def create(self, request, *args, **kwargs): status=status.HTTP_400_BAD_REQUEST, ) from app.box.models import get_categories + valid_categories = [cat[0] for cat in get_categories()] if node_type_val.lower() not in valid_categories: return Response( @@ -223,7 +228,10 @@ def create(self, request, *args, **kwargs): existing_node.node_type = node_data.get("type", existing_node.node_type) new_data = node_data.get("data", existing_node.data) if isinstance(new_data, dict): - for key in ("parameter_modifications", "has_parameter_modifications"): + for key in ( + "parameter_modifications", + "has_parameter_modifications", + ): if key in existing_node.data: new_data[key] = existing_node.data[key] elif key in new_data: @@ -535,13 +543,10 @@ def get(self, request): ) - - - @method_decorator(csrf_exempt, name="dispatch") class JupyterLabView(APIView): """Views for integration with JupyterLab""" - + authentication_classes = [KeycloakAuthentication] permission_classes = [IsAuthenticated] @@ -549,25 +554,27 @@ def get(self, request, workflow_id): """Return the JupyterLab URL""" try: project = get_accessible_project(request, workflow_id, write=False) - + # JupyterLab URL generation - #jupyter_url = f"http://localhost:8000/user/user1/lab/tree/codes/projects/{workflow_id}" + # jupyter_url = f"http://localhost:8000/user/user1/lab/tree/codes/projects/{workflow_id}" jupyter_url = f"http://localhost:8000/user/user1/lab/tree/codes/projects/" - #jupyter_url = f"http://localhost:8000/user/user1/lab/workspaces/auto-E/tree/codes/nodes/{workflow_id}/{workflow_id}.py" - - - return JsonResponse({ - "status": "success", - "jupyter_url": jupyter_url, - "workflow_id": str(workflow_id), - "project_name": project.name - }) - + # jupyter_url = f"http://localhost:8000/user/user1/lab/workspaces/auto-E/tree/codes/nodes/{workflow_id}/{workflow_id}.py" + + return JsonResponse( + { + "status": "success", + "jupyter_url": jupyter_url, + "workflow_id": str(workflow_id), + "project_name": project.name, + } + ) + except Exception as e: - logger.error(f"Error generating JupyterLab URL for workflow {workflow_id}: {e}") + logger.error( + f"Error generating JupyterLab URL for workflow {workflow_id}: {e}" + ) return JsonResponse( - {"error": f"Failed to generate JupyterLab URL: {str(e)}"}, - status=500 + {"error": f"Failed to generate JupyterLab URL: {str(e)}"}, status=500 ) @@ -595,7 +602,10 @@ def put(self, request, workflow_id, node_id): if parameter_field == "value": parameter_field = "default_value" - print(f"🔍 DEBUG: Parsed - parameter_key: {parameter_key}, parameter_value: {parameter_value}, parameter_field: {parameter_field}", flush=True) + print( + f"🔍 DEBUG: Parsed - parameter_key: {parameter_key}, parameter_value: {parameter_value}, parameter_field: {parameter_field}", + flush=True, + ) if not parameter_key: return Response( @@ -609,7 +619,9 @@ def put(self, request, workflow_id, node_id): status=status.HTTP_400_BAD_REQUEST, ) - logger.info(f"Updating parameter '{parameter_key}.{parameter_field}' to {parameter_value} in node {node_id}") + logger.info( + f"Updating parameter '{parameter_key}.{parameter_field}' to {parameter_value} in node {node_id}" + ) # Check if schema.parameters exists if "schema" not in node.data: @@ -628,42 +640,75 @@ def put(self, request, workflow_id, node_id): if parameter_key not in node.data["schema"]["parameters"]: available_keys = list(node.data["schema"]["parameters"].keys()) - print(f"❌ DEBUG: Parameter '{parameter_key}' not found. Available: {available_keys}", flush=True) + print( + f"❌ DEBUG: Parameter '{parameter_key}' not found. Available: {available_keys}", + flush=True, + ) return Response( - {"error": f"Parameter '{parameter_key}' not found. Available: {available_keys}"}, + { + "error": f"Parameter '{parameter_key}' not found. Available: {available_keys}" + }, status=status.HTTP_400_BAD_REQUEST, ) # Get the value before update - old_value = node.data["schema"]["parameters"][parameter_key].get(parameter_field) - print(f"🔍 DEBUG: Updating {parameter_key}.{parameter_field} from {old_value} to {parameter_value}", flush=True) + old_value = node.data["schema"]["parameters"][parameter_key].get( + parameter_field + ) + print( + f"🔍 DEBUG: Updating {parameter_key}.{parameter_field} from {old_value} to {parameter_value}", + flush=True, + ) # Save original value (for change history) - original_value = node.data["schema"]["parameters"][parameter_key].get(parameter_field) + original_value = node.data["schema"]["parameters"][parameter_key].get( + parameter_field + ) # Directly update the field specified by parameter_field - print(f"🔍 DEBUG: Before update - schema.parameters[{parameter_key}]: {node.data['schema']['parameters'][parameter_key]}", flush=True) - node.data["schema"]["parameters"][parameter_key][parameter_field] = parameter_value - print(f"🔍 DEBUG: After update - schema.parameters[{parameter_key}]: {node.data['schema']['parameters'][parameter_key]}", flush=True) + print( + f"🔍 DEBUG: Before update - schema.parameters[{parameter_key}]: {node.data['schema']['parameters'][parameter_key]}", + flush=True, + ) + node.data["schema"]["parameters"][parameter_key][ + parameter_field + ] = parameter_value + print( + f"🔍 DEBUG: After update - schema.parameters[{parameter_key}]: {node.data['schema']['parameters'][parameter_key]}", + flush=True, + ) - print(f"🔍 DEBUG: Updated {parameter_field} from {original_value} to {parameter_value}", flush=True) + print( + f"🔍 DEBUG: Updated {parameter_field} from {original_value} to {parameter_value}", + flush=True, + ) # Track parameter changes (changes across all fields) self._update_parameter_modification_status( - node.data, parameter_key, parameter_field, + node.data, + parameter_key, + parameter_field, node.data["schema"]["parameters"][parameter_key], parameter_value, - original_value + original_value, ) # save node node.save() print(f"✅ DEBUG: Successfully saved parameter update", flush=True) - print(f"🔍 DEBUG: After save - node.data keys: {list(node.data.keys())}", flush=True) - print(f"🔍 DEBUG: After save - parameter_modifications: {node.data.get('parameter_modifications', 'NOT FOUND')}", flush=True) + print( + f"🔍 DEBUG: After save - node.data keys: {list(node.data.keys())}", + flush=True, + ) + print( + f"🔍 DEBUG: After save - parameter_modifications: {node.data.get('parameter_modifications', 'NOT FOUND')}", + flush=True, + ) - logger.info(f"Successfully updated parameter '{parameter_key}.{parameter_field}' in node {node_id}") + logger.info( + f"Successfully updated parameter '{parameter_key}.{parameter_field}' in node {node_id}" + ) return Response( { @@ -674,21 +719,35 @@ def put(self, request, workflow_id, node_id): "parameter_key": parameter_key, "parameter_field": parameter_field, "parameter_value": parameter_value, - "updated_parameter": node.data["schema"]["parameters"][parameter_key] + "updated_parameter": node.data["schema"]["parameters"][ + parameter_key + ], } ) except Exception as e: - logger.error(f"Parameter update failed for node {node_id}: {e}", exc_info=True) + logger.error( + f"Parameter update failed for node {node_id}: {e}", exc_info=True + ) return Response( {"error": f"Parameter update failed: {str(e)}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR, ) - - def _update_parameter_modification_status(self, node_data, parameter_key, parameter_field, parameter, new_value, original_value=None): + def _update_parameter_modification_status( + self, + node_data, + parameter_key, + parameter_field, + parameter, + new_value, + original_value=None, + ): """Track and update parameter changes (all fields)""" - print(f"🔍 DEBUG: Tracking modification status for {parameter_key}.{parameter_field}", flush=True) + print( + f"🔍 DEBUG: Tracking modification status for {parameter_key}.{parameter_field}", + flush=True, + ) # Ensure the structure of parameter_modifications if "parameter_modifications" not in node_data: @@ -700,7 +759,7 @@ def _update_parameter_modification_status(self, node_data, parameter_key, parame if parameter_key not in modifications: modifications[parameter_key] = { "is_modified": False, - "field_modifications": {} + "field_modifications": {}, } param_mod = modifications[parameter_key] @@ -714,11 +773,13 @@ def _update_parameter_modification_status(self, node_data, parameter_key, parame # If old data exists, it will be migrated as default_value if old_original is not None: - param_mod["field_modifications"]["default_value_original"] = old_original + param_mod["field_modifications"][ + "default_value_original" + ] = old_original param_mod["field_modifications"]["default_value"] = { "current_value": old_current, "is_modified": param_mod.get("is_modified", False), - "modified_at": param_mod.get("modified_at") + "modified_at": param_mod.get("modified_at"), } # remove old key @@ -736,13 +797,16 @@ def _update_parameter_modification_status(self, node_data, parameter_key, parame original_field_value = param_mod["field_modifications"][field_key] is_field_modified = new_value != original_field_value - print(f"🔍 DEBUG: {parameter_field} - original={original_field_value}, new={new_value}, modified={is_field_modified}", flush=True) + print( + f"🔍 DEBUG: {parameter_field} - original={original_field_value}, new={new_value}, modified={is_field_modified}", + flush=True, + ) # Update field change status param_mod["field_modifications"][parameter_field] = { "current_value": new_value, "is_modified": is_field_modified, - "modified_at": None # Assumes that the current time is set on the front end + "modified_at": None, # Assumes that the current time is set on the front end } # Update the overall parameter change status (if any field has changed) True） @@ -759,7 +823,10 @@ def _update_parameter_modification_status(self, node_data, parameter_key, parame # Update overall changes node_data["has_parameter_modifications"] = len(modifications) > 0 - print(f"✅ DEBUG: Parameter '{parameter_key}.{parameter_field}' modification status: {'modified' if is_field_modified else 'default'}", flush=True) + print( + f"✅ DEBUG: Parameter '{parameter_key}.{parameter_field}' modification status: {'modified' if is_field_modified else 'default'}", + flush=True, + ) print(f"🔍 DEBUG: Final modifications data: {modifications}", flush=True) @@ -780,18 +847,22 @@ def post(self, request, workflow_id): nodes_data = data.get("nodes", []) edges_data = data.get("edges", []) - logger.info(f"Batch code generation for project {workflow_id}: {len(nodes_data)} nodes, {len(edges_data)} edges") + logger.info( + f"Batch code generation for project {workflow_id}: {len(nodes_data)} nodes, {len(edges_data)} edges" + ) # Generate code in bulk using the code generation service code_service = CodeGenerationService() - success = code_service.generate_code_from_flow_data(str(workflow_id), project.name, nodes_data, edges_data) + success = code_service.generate_code_from_flow_data( + str(workflow_id), project.name, nodes_data, edges_data + ) response_data = { "status": "success", "message": f"Code generated from {len(nodes_data)} nodes and {len(edges_data)} edges", "workflow_id": str(workflow_id), "nodes_processed": len(nodes_data), - "edges_processed": len(edges_data) + "edges_processed": len(edges_data), } if success: @@ -805,7 +876,7 @@ def post(self, request, workflow_id): "python_file": str(code_file), "notebook_file": str(notebook_file), "python_exists": code_file.exists(), - "notebook_exists": notebook_file.exists() + "notebook_exists": notebook_file.exists(), } else: response_data["code_status"] = "Code generation failed" @@ -815,21 +886,23 @@ def post(self, request, workflow_id): except json.JSONDecodeError: return Response( - {"error": "Invalid JSON format"}, - status=status.HTTP_400_BAD_REQUEST + {"error": "Invalid JSON format"}, status=status.HTTP_400_BAD_REQUEST ) except FlowProject.DoesNotExist: return Response( {"error": f"Project {workflow_id} not found"}, - status=status.HTTP_404_NOT_FOUND + status=status.HTTP_404_NOT_FOUND, ) except Exception as e: - logger.error(f"Error in batch code generation for project {workflow_id}: {e}") + logger.error( + f"Error in batch code generation for project {workflow_id}: {e}" + ) return Response( {"error": f"Batch code generation failed: {str(e)}"}, - status=status.HTTP_500_INTERNAL_SERVER_ERROR + status=status.HTTP_500_INTERNAL_SERVER_ERROR, ) - + + def _format_sse(event_type: str, data: dict) -> str: """Format a Server-Sent Event string.""" return f"event: {event_type}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n" @@ -872,7 +945,9 @@ def post(self, request, workflow_id): if not script_path.exists(): return Response( - {"error": f"Script not found: {script_path.name}. Generate code first."}, + { + "error": f"Script not found: {script_path.name}. Generate code first." + }, status=status.HTTP_404_NOT_FOUND, ) @@ -883,7 +958,9 @@ def post(self, request, workflow_id): ast.parse(code) except SyntaxError as e: return Response( - {"error": f"Generated code has syntax error at line {e.lineno}: {e.msg}"}, + { + "error": f"Generated code has syntax error at line {e.lineno}: {e.msg}" + }, status=status.HTTP_400_BAD_REQUEST, ) @@ -895,8 +972,7 @@ def post(self, request, workflow_id): working_dir = f"{JUPYTER_HOME}/codes/projects/{project_dir.name}" code = ( f"import os\nos.makedirs({working_dir!r}, exist_ok=True)\n" - f"os.chdir({working_dir!r})\n\n" - + code + f"os.chdir({working_dir!r})\n\n" + code ) # Attribute streamed output to canvas nodes via the sidecar map @@ -926,10 +1002,13 @@ def _sync_event_generator( # on GeneratorExit when the client disconnects mid-run). final_status = "aborted" try: - yield _format_sse("run_started", { - "workflow_id": workflow_id, - "project_name": project_name, - }) + yield _format_sse( + "run_started", + { + "workflow_id": workflow_id, + "project_name": project_name, + }, + ) service = JupyterExecutionService() agen = service.execute_code(code) @@ -945,11 +1024,14 @@ def _sync_event_generator( break except Exception as e: logger.error("Jupyter execution stream error: %s", e, exc_info=True) - yield _format_sse("error", { - "ename": type(e).__name__, - "evalue": str(e), - "traceback": [], - }) + yield _format_sse( + "error", + { + "ename": type(e).__name__, + "evalue": str(e), + "traceback": [], + }, + ) final_status = "error" yield _format_sse("done", {"status": "error"}) break @@ -974,34 +1056,38 @@ def post(self, request, workflow_id): # Run Workflow Project Service run_workflow_service = RunWorkflowService() project_name = str(project.id) - result = run_workflow_service.run_workflow_code(str(workflow_id), project_name) + result = run_workflow_service.run_workflow_code( + str(workflow_id), project_name + ) response_data = { "status": "success", "message": f"Workflow project completed successfully.", "workflow_id": str(workflow_id), - "result": result + "result": result, } return Response(response_data, status=status.HTTP_200_OK) except json.JSONDecodeError: return Response( - {"error": "Invalid JSON format"}, - status=status.HTTP_400_BAD_REQUEST + {"error": "Invalid JSON format"}, status=status.HTTP_400_BAD_REQUEST ) except FlowProject.DoesNotExist: return Response( {"error": f"Project {workflow_id} not found"}, - status=status.HTTP_404_NOT_FOUND + status=status.HTTP_404_NOT_FOUND, ) except Exception as e: - logger.error(f"Error in batch code generation for project {workflow_id}: {e}") + logger.error( + f"Error in batch code generation for project {workflow_id}: {e}" + ) return Response( {"error": f"Batch code generation failed: {str(e)}"}, - status=status.HTTP_500_INTERNAL_SERVER_ERROR + status=status.HTTP_500_INTERNAL_SERVER_ERROR, ) + @method_decorator(csrf_exempt, name="dispatch") class FlowNodeInstanceNameUpdateView(APIView): """Update the instanceName of the FlowNode (do not change the base node)""" @@ -1042,7 +1128,10 @@ def put(self, request, workflow_id, node_id): # Get the value before update old_value = node.data["instanceName"] - print(f"🔍 DEBUG: Updating instanceName from {old_value} to {instance_name}", flush=True) + print( + f"🔍 DEBUG: Updating instanceName from {old_value} to {instance_name}", + flush=True, + ) # Save original value (for change history) original_value = node.data["instanceName"] @@ -1050,7 +1139,10 @@ def put(self, request, workflow_id, node_id): # Directly update the field specified by parameter_field node.data["instanceName"] = instance_name - print(f"🔍 DEBUG: Updated instance_name from {original_value} to {instance_name}", flush=True) + print( + f"🔍 DEBUG: Updated instance_name from {original_value} to {instance_name}", + flush=True, + ) # save node node.save() @@ -1063,12 +1155,14 @@ def put(self, request, workflow_id, node_id): "message": f"instance_name instance_name updated successfully", "node_id": node_id, "workflow_id": str(workflow_id), - "updated_instance_name": node.data["instanceName"] + "updated_instance_name": node.data["instanceName"], } ) except Exception as e: - logger.error(f"InstanceName update failed for node {node_id}: {e}", exc_info=True) + logger.error( + f"InstanceName update failed for node {node_id}: {e}", exc_info=True + ) return Response( {"error": f"InstanceName update failed: {str(e)}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR, @@ -1079,6 +1173,7 @@ def put(self, request, workflow_id, node_id): # Viewer file serving # --------------------------------------------------------------------------- + @csrf_exempt def viewer_file(request, project_id, subpath): """Serve a file from a project's directory, looked up by project id. @@ -1099,6 +1194,7 @@ def viewer_file(request, project_id, subpath): # Viewer chat tools (LLM tool dispatch over a run's viewer data) # --------------------------------------------------------------------------- + @method_decorator(csrf_exempt, name="dispatch") class ViewerChatToolView(APIView): """Run one brain-viewer chat tool against the active project's run data. @@ -1116,10 +1212,10 @@ class ViewerChatToolView(APIView): def post(self, request, workflow_id): # Import lazily so a viewer_tools import error never breaks other routes. - from .viewer_tools.registry import call_registered_tool, UnknownTool + from .viewer_tools.registry import UnknownTool, call_registered_tool from .viewer_tools.resolver import ( - load_project_viewer_data, ViewerDataNotFound, + load_project_viewer_data, ) project = get_accessible_project(request, workflow_id, write=False) @@ -1166,6 +1262,7 @@ def post(self, request, workflow_id): # Results listing and report saving # --------------------------------------------------------------------------- + @method_decorator(csrf_exempt, name="dispatch") class WorkflowResultsView(APIView): """List simulation result files for a workflow project.""" @@ -1179,7 +1276,9 @@ def get(self, request, workflow_id): results_dir = project_dir / "results" if not results_dir.exists(): - return JsonResponse({"status": "success", "results": [], "results_dir": str(results_dir)}) + return JsonResponse( + {"status": "success", "results": [], "results_dir": str(results_dir)} + ) files = [] for f in sorted(results_dir.iterdir()): @@ -1193,13 +1292,16 @@ def get(self, request, workflow_id): if f.suffix == ".npz": try: import numpy as np + with np.load(f, allow_pickle=False) as npz: entry["arrays"] = {k: list(npz[k].shape) for k in npz.files} except Exception as e: entry["arrays"] = {"error": str(e)} files.append(entry) - return JsonResponse({"status": "success", "results": files, "results_dir": str(results_dir)}) + return JsonResponse( + {"status": "success", "results": files, "results_dir": str(results_dir)} + ) @method_decorator(csrf_exempt, name="dispatch") @@ -1228,6 +1330,7 @@ def get(self, request, workflow_id): if notebook_path.exists(): try: import json as _json + nb = _json.loads(notebook_path.read_text(encoding="utf-8")) for cell in nb.get("cells", []): if cell.get("cell_type") != "code": @@ -1235,17 +1338,27 @@ def get(self, request, workflow_id): source = "".join(cell.get("source", [])) cell_outputs = [] for out in cell.get("outputs", []): - if out.get("output_type") in ("stream", "execute_result", "display_data"): - text = out.get("text") or out.get("data", {}).get("text/plain") or [] + if out.get("output_type") in ( + "stream", + "execute_result", + "display_data", + ): + text = ( + out.get("text") + or out.get("data", {}).get("text/plain") + or [] + ) if isinstance(text, list): text = "".join(text) if text.strip(): cell_outputs.append(text.strip()) if cell_outputs: - notebook_outputs.append({ - "source_snippet": source[:200], - "outputs": cell_outputs, - }) + notebook_outputs.append( + { + "source_snippet": source[:200], + "outputs": cell_outputs, + } + ) except Exception as e: notebook_outputs = [{"error": str(e)}] @@ -1295,7 +1408,9 @@ def post(self, request, workflow_id): request.FILES.getlist("files") ) if not uploads: - return JsonResponse({"error": "No file provided (field name: file)"}, status=400) + return JsonResponse( + {"error": "No file provided (field name: file)"}, status=400 + ) overwrite = str( request.data.get("overwrite", request.query_params.get("overwrite", "")) @@ -1317,7 +1432,10 @@ def post(self, request, workflow_id): or ".." in Path(raw_name).parts ): raise ValueError("Invalid upload filename") - if uploaded.size is not None and uploaded.size > PROJECT_UPLOAD_MAX_BYTES: + if ( + uploaded.size is not None + and uploaded.size > PROJECT_UPLOAD_MAX_BYTES + ): raise ValueError( f"File exceeds maximum size of " f"{PROJECT_UPLOAD_MAX_BYTES // (1024 * 1024)} MB" @@ -1394,9 +1512,7 @@ def post(self, request, workflow_id): def delete(self, request, workflow_id): project = get_accessible_project(request, workflow_id, write=True) filename = ( - request.query_params.get("filename") - or request.data.get("filename") - or "" + request.query_params.get("filename") or request.data.get("filename") or "" ).strip() if not filename: return JsonResponse({"error": "filename is required"}, status=400) @@ -1438,12 +1554,14 @@ def post(self, request, workflow_id): with open(report_path, "w", encoding="utf-8") as f: f.write(report_text) - return JsonResponse({ - "status": "success", - "message": f"Report saved to {filename}", - "path": str(report_path), - "size_bytes": report_path.stat().st_size, - }) + return JsonResponse( + { + "status": "success", + "message": f"Report saved to {filename}", + "path": str(report_path), + "size_bytes": report_path.stat().st_size, + } + ) def get(self, request, workflow_id): project = get_accessible_project(request, workflow_id, write=False) @@ -1456,17 +1574,20 @@ def get(self, request, workflow_id): if not report_path.exists(): return JsonResponse({"error": "Report not found"}, status=404) - return JsonResponse({ - "status": "success", - "filename": filename, - "report_text": report_path.read_text(encoding="utf-8"), - }) + return JsonResponse( + { + "status": "success", + "filename": filename, + "report_text": report_path.read_text(encoding="utf-8"), + } + ) # --------------------------------------------------------------------------- # Async run / status API (Phase 3) # --------------------------------------------------------------------------- + def _get_executor(backend_name: str): """Instantiate the appropriate execution backend.""" if backend_name == WorkflowRun.Backend.SLURM: @@ -1474,6 +1595,129 @@ def _get_executor(backend_name: str): return LocalExecutor() +def _get_accessible_run(request, workflow_id, run_id) -> WorkflowRun: + return get_object_or_404( + WorkflowRun.objects.filter( + Q(user=request.user) | Q(workflow__owner=request.user) + ), + id=run_id, + workflow_id=workflow_id, + ) + + +def _run_payload(run: WorkflowRun, *, sbatch: str = "") -> dict: + data = WorkflowRunSerializer(run).data + data["sbatch"] = sbatch + data["jupyter_path"] = jupyter_sbatch_path(run.workflow_id, run.id) + return data + + +@method_decorator(csrf_exempt, name="dispatch") +class WorkflowRunPrepareView(APIView): + """Create a draft Slurm run and write ``run.sbatch`` locally. No SSH.""" + + authentication_classes = [KeycloakAuthentication] + permission_classes = [IsAuthenticated] + + def post(self, request, workflow_id): + project = get_accessible_project(request, workflow_id, write=True) + ser = WorkflowRunPrepareSerializer(data=request.data) + ser.is_valid(raise_exception=True) + resource_reqs = ser.validated_data.get("resource_requests", {}) or {} + from_run_id = ser.validated_data.get("from_run_id") + incoming = (ser.validated_data.get("sbatch") or "").strip() + + source_text = incoming + if not source_text and from_run_id: + source_run = _get_accessible_run(request, workflow_id, from_run_id) + source_path = ( + batch_run_dir(str(workflow_id), str(source_run.id)) / "run.sbatch" + ) + if source_path.is_file(): + source_text = source_path.read_text() + + run = WorkflowRun.objects.create( + workflow=project, + user=request.user, + backend=WorkflowRun.Backend.SLURM, + status=WorkflowRun.Status.DRAFT, + resource_requests=resource_reqs, + ) + executor = RemoteSlurmExecutor() + run.remote_run_dir = executor._remote_run_dir(str(run.id)) + try: + script = executor.write_sbatch( + str(workflow_id), + str(run.id), + str(project.id), + resource_reqs, + sbatch_text=source_text or None, + ) + except ValueError as exc: + run.delete() + return Response({"error": str(exc)}, status=status.HTTP_400_BAD_REQUEST) + run.save() + return Response( + _run_payload(run, sbatch=script), status=status.HTTP_201_CREATED + ) + + +@method_decorator(csrf_exempt, name="dispatch") +class WorkflowRunSbatchView(APIView): + """Read or write the local ``run.sbatch`` for a cluster run.""" + + authentication_classes = [KeycloakAuthentication] + permission_classes = [IsAuthenticated] + + def get(self, request, workflow_id, run_id): + get_accessible_project(request, workflow_id, write=False) + run = _get_accessible_run(request, workflow_id, run_id) + path = batch_run_dir(str(workflow_id), str(run.id)) / "run.sbatch" + if not path.is_file(): + return Response( + {"error": "run.sbatch not found for this run"}, + status=status.HTTP_404_NOT_FOUND, + ) + return Response(_run_payload(run, sbatch=path.read_text())) + + def put(self, request, workflow_id, run_id): + get_accessible_project(request, workflow_id, write=True) + run = _get_accessible_run(request, workflow_id, run_id) + if run.status != WorkflowRun.Status.DRAFT: + return Response( + {"error": "run.sbatch can only be edited while the run is a draft"}, + status=status.HTTP_400_BAD_REQUEST, + ) + ser = WorkflowRunSbatchSerializer(data=request.data) + ser.is_valid(raise_exception=True) + incoming = (ser.validated_data.get("sbatch") or "").strip() + resource_reqs = ser.validated_data.get("resource_requests") + if resource_reqs: + run.resource_requests = resource_reqs + run.save(update_fields=["resource_requests"]) + if not incoming and not resource_reqs: + return Response( + {"error": "Provide sbatch text or resource_requests"}, + status=status.HTTP_400_BAD_REQUEST, + ) + executor = RemoteSlurmExecutor() + remote = run.remote_run_dir or executor._remote_run_dir(str(run.id)) + if not run.remote_run_dir: + run.remote_run_dir = remote + run.save(update_fields=["remote_run_dir"]) + try: + script = executor.write_sbatch( + str(workflow_id), + str(run.id), + str(run.workflow_id), + run.resource_requests or {}, + sbatch_text=incoming or None, + ) + except ValueError as exc: + return Response({"error": str(exc)}, status=status.HTTP_400_BAD_REQUEST) + return Response(_run_payload(run, sbatch=script)) + + @method_decorator(csrf_exempt, name="dispatch") class WorkflowRunSubmitView(APIView): """Submit a workflow run (returns immediately with run_id + status).""" @@ -1487,19 +1731,42 @@ def post(self, request, workflow_id): ser.is_valid(raise_exception=True) backend_choice = ser.validated_data["backend"] - resource_reqs = ser.validated_data.get("resource_requests", {}) + resource_reqs = ser.validated_data.get("resource_requests", {}) or {} project_name = str(project.id) + draft_id = ser.validated_data.get("run_id") + sbatch_text = ser.validated_data.get("sbatch") or "" script_path = code_file_path(project) code = script_path.read_text() if script_path.exists() else "" - run = WorkflowRun.objects.create( - workflow=project, - user=request.user, - backend=backend_choice, - status=WorkflowRun.Status.PENDING, - resource_requests=resource_reqs, - ) + if draft_id: + run = _get_accessible_run(request, workflow_id, draft_id) + if run.status != WorkflowRun.Status.DRAFT: + return Response( + {"error": "run_id must refer to a draft run"}, + status=status.HTTP_400_BAD_REQUEST, + ) + if backend_choice != WorkflowRun.Backend.SLURM: + return Response( + {"error": "draft runs can only be submitted to Slurm"}, + status=status.HTTP_400_BAD_REQUEST, + ) + if resource_reqs: + run.resource_requests = resource_reqs + run.backend = WorkflowRun.Backend.SLURM + else: + run = WorkflowRun.objects.create( + workflow=project, + user=request.user, + backend=backend_choice, + status=WorkflowRun.Status.PENDING, + resource_requests=resource_reqs, + ) + + if not sbatch_text and draft_id: + existing = batch_run_dir(str(workflow_id), str(run.id)) / "run.sbatch" + if existing.is_file(): + sbatch_text = existing.read_text() executor = _get_executor(backend_choice) try: @@ -1508,7 +1775,8 @@ def post(self, request, workflow_id): project_name=project_name, code=code, run_id=str(run.id), - resource_requests=resource_reqs, + resource_requests=resource_reqs or run.resource_requests, + sbatch_text=sbatch_text or None, ) run.status = exec_result.status.value if exec_result.remote_job_id: @@ -1601,7 +1869,9 @@ def delete(self, request, workflow_id, run_id): try: executor.cancel(str(run.id), job_id=run.slurm_job_id or None) except Exception as exc: - logger.warning("cancel before delete failed for run %s: %s", run.id, exc) + logger.warning( + "cancel before delete failed for run %s: %s", run.id, exc + ) try: executor.cleanup(str(run.id), remote_dir=run.remote_run_dir or None) except Exception as exc: @@ -1662,14 +1932,55 @@ def post(self, request, workflow_id, run_id): return Response(WorkflowRunSerializer(run).data) +def _resolve_run_artifact(workflow_id, run_id, rel: str) -> Path: + """Resolve a download path under the batch run dir. + + Allowed: ``logs/**``, ``results/**`` (including legacy paths relative to + ``results/``), and ``run.sbatch`` at the run root. ``nodes/`` is not. + """ + rel = (rel or "").replace("\\", "/").strip().lstrip("/") + if not rel or any(part == ".." for part in Path(rel).parts): + raise ValueError("Invalid path") + first = Path(rel).parts[0] + if first == "nodes": + raise ValueError("Invalid path") + batch = batch_run_dir(str(workflow_id), str(run_id)).resolve() + nodes = (batch / "nodes").resolve() + + def _under(root: Path, target: Path) -> bool: + return target == root or root in target.parents + + def _ok(target: Path) -> bool: + if not _under(batch, target): + return False + if _under(nodes, target): + return False + logs = (batch / "logs").resolve() + results = (batch / "results").resolve() + if target.name == "run.sbatch" and target.parent == batch: + return True + if _under(logs, target) or _under(results, target): + return True + return False + + direct = (batch / rel).resolve() + if _ok(direct): + return direct + if first in ("logs", "results"): + raise ValueError("Invalid path") + legacy = (batch / "results" / rel).resolve() + if _ok(legacy): + return legacy + raise ValueError("Invalid path") + + @method_decorator(csrf_exempt, name="dispatch") class WorkflowRunArtifactView(APIView): - """Download a single result artifact fetched back from a remote run. + """Download a single file fetched back from a remote run. - Files live under ``codes/projects//batch//results/`` - (populated by the executor when a run completes). The relative file path is - passed as the ``path`` query parameter and is validated against directory - traversal. + Files live under ``codes/projects//batch//``. + ``path`` may be ``logs/...``, ``results/...``, ``run.sbatch``, or a + legacy path relative to ``results/``. ``nodes/`` is rejected. """ authentication_classes = [KeycloakAuthentication] @@ -1677,13 +1988,7 @@ class WorkflowRunArtifactView(APIView): def get(self, request, workflow_id, run_id): get_accessible_project(request, workflow_id, write=False) - run = get_object_or_404( - WorkflowRun.objects.filter( - Q(user=request.user) | Q(workflow__owner=request.user) - ), - id=run_id, - workflow_id=workflow_id, - ) + run = _get_accessible_run(request, workflow_id, run_id) rel = request.query_params.get("path", "").strip() if not rel: @@ -1691,10 +1996,9 @@ def get(self, request, workflow_id, run_id): {"error": "Missing 'path' query parameter"}, status=status.HTTP_400_BAD_REQUEST, ) - - base = (batch_run_dir(str(workflow_id), str(run.id)) / "results").resolve() - target = (base / rel).resolve() - if base != target and base not in target.parents: + try: + target = _resolve_run_artifact(workflow_id, run.id, rel) + except ValueError: return Response( {"error": "Invalid path"}, status=status.HTTP_400_BAD_REQUEST ) @@ -1703,4 +2007,4 @@ def get(self, request, workflow_id, run_id): return FileResponse( open(target, "rb"), as_attachment=True, filename=target.name - ) \ No newline at end of file + ) diff --git a/gui/workflow_backend/django-project/config/test_settings.py b/gui/workflow_backend/django-project/config/test_settings.py new file mode 100644 index 00000000..dbfbd2ba --- /dev/null +++ b/gui/workflow_backend/django-project/config/test_settings.py @@ -0,0 +1,23 @@ +"""SQLite Django settings are kept for import-only experiments. + +This repository's migrations include Postgres-specific SQL, so isolated pytest +in CI/dev should use an ephemeral Postgres (see the cluster-sbatch progress +log) rather than this module. +""" + +import os + +os.environ.setdefault("DJANGO_SECRET_KEY", "test-only-cluster-sbatch-logs") +os.environ.setdefault("DB_PASSWORD", "unused") +os.environ.setdefault("DB_USER", "unused") +os.environ.setdefault("DB_NAME", "unused") +os.environ.setdefault("DB_PORT", "5432") + +from .settings import * # noqa: E402,F403 + +DATABASES = { + "default": { + "ENGINE": "django.db.backends.sqlite3", + "NAME": ":memory:", + } +} diff --git a/gui/workflow_backend/django-project/tests/test_cluster_sbatch_logs.py b/gui/workflow_backend/django-project/tests/test_cluster_sbatch_logs.py new file mode 100644 index 00000000..aad2ed99 --- /dev/null +++ b/gui/workflow_backend/django-project/tests/test_cluster_sbatch_logs.py @@ -0,0 +1,220 @@ +"""Tests for draft run.sbatch editing, pin, and Slurm log copy-back.""" + +from pathlib import Path + +import pytest +from app.workflow.execution.base import ExecutionStatus +from app.workflow.execution.remote_slurm_executor import ( + RemoteSlurmExecutor, + normalize_sbatch, +) +from app.workflow.models import FlowProject, WorkflowRun +from app.workflow.path_utils import batch_run_dir +from app.workflow.views import _resolve_run_artifact +from django.urls import reverse + + +def _make_project(owner, *, name="ClusterProj"): + return FlowProject.objects.create(name=name, owner=owner, visibility="private") + + +def test_normalize_sbatch_injects_pinned_directives(): + remote = "/data/neuro-workflow/runs/abc" + out = normalize_sbatch("echo hi", remote) + assert out.startswith("#!/bin/bash\n") + assert f"#SBATCH --chdir={remote}" in out + assert f"#SBATCH --output={remote}/slurm-%j.out" in out + assert f"#SBATCH --error={remote}/slurm-%j.err" in out + assert "echo hi" in out + + +def test_normalize_sbatch_replaces_existing_output_paths(): + remote = "/data/neuro-workflow/runs/abc" + raw = "\n".join( + [ + "#!/bin/bash", + "#SBATCH --output=/tmp/evil.out", + "#SBATCH --error=/tmp/evil.err", + "#SBATCH --chdir=/tmp", + "python workflow.py", + ] + ) + out = normalize_sbatch(raw, remote) + assert "/tmp/evil" not in out + assert out.count("#SBATCH --chdir=") == 1 + assert f"#SBATCH --output={remote}/slurm-%j.out" in out + assert "python workflow.py" in out + + +def test_normalize_sbatch_rejects_empty_and_oversize(): + with pytest.raises(ValueError, match="empty"): + normalize_sbatch(" ", "/r") + with pytest.raises(ValueError, match="64 KiB"): + normalize_sbatch("x" * (64 * 1024 + 1), "/r") + + +@pytest.mark.django_db +def test_prepare_creates_draft_without_sbatch( + auth_client, user_alice, tmp_path, settings, monkeypatch +): + settings.BASE_DIR = tmp_path + ssh_calls = [] + + def boom(self, cmd): + ssh_calls.append(cmd) + raise AssertionError(f"prepare must not SSH: {cmd}") + + monkeypatch.setattr(RemoteSlurmExecutor, "_ssh", boom) + project = _make_project(user_alice) + client = auth_client(user_alice) + url = reverse("workflow:workflow-run-prepare", args=[project.id]) + resp = client.post( + url, + {"resource_requests": {"partition": "ccalc", "time": "00:05:00"}}, + format="json", + ) + assert resp.status_code == 201, resp.content + body = resp.json() + assert body["status"] == "draft" + assert body["backend"] == "slurm" + assert ssh_calls == [] + path = batch_run_dir(project.id, body["id"]) / "run.sbatch" + assert path.is_file() + text = path.read_text() + assert "#SBATCH --chdir=" in text + assert "#SBATCH --output=" in text + assert "sbatch" not in ssh_calls + + +@pytest.mark.django_db +def test_put_sbatch_without_chdir_is_pinned( + auth_client, user_alice, tmp_path, settings +): + settings.BASE_DIR = tmp_path + project = _make_project(user_alice) + client = auth_client(user_alice) + prep = client.post( + reverse("workflow:workflow-run-prepare", args=[project.id]), + {"resource_requests": {"partition": "ccalc"}}, + format="json", + ) + run_id = prep.json()["id"] + url = reverse("workflow:workflow-run-sbatch", args=[project.id, run_id]) + resp = client.put( + url, + {"sbatch": "#!/bin/bash\necho CUSTOM_BODY\n"}, + format="json", + ) + assert resp.status_code == 200, resp.content + text = (batch_run_dir(project.id, run_id) / "run.sbatch").read_text() + assert "CUSTOM_BODY" in text + assert "#SBATCH --chdir=" in text + assert "#SBATCH --output=" in text + assert "#SBATCH --error=" in text + + +def test_submit_custom_sbatch_is_not_silently_replaced(tmp_path, settings, monkeypatch): + settings.BASE_DIR = tmp_path + settings.MEDIA_ROOT = str(tmp_path / "no-nodes") + ex = RemoteSlurmExecutor() + monkeypatch.setattr(ex, "_ssh", lambda cmd: "Submitted batch job 4242") + monkeypatch.setattr(ex, "_sync_to_remote", lambda *a, **k: None) + marker = "echo USER_EDITED_SBATCH" + result = ex.submit( + "proj-id", + "proj-id", + "print(1)\n", + run_id="run-custom", + resource_requests={"partition": "ccalc"}, + sbatch_text=f"#!/bin/bash\n{marker}\n", + ) + assert result.status == ExecutionStatus.PENDING + assert result.remote_job_id == "4242" + text = (batch_run_dir("proj-id", "run-custom") / "run.sbatch").read_text() + assert marker in text + assert "#SBATCH --job-name=" not in text or marker in text + assert "USER_EDITED_SBATCH" in text + + +@pytest.mark.django_db +def test_failed_copy_back_writes_err_and_artifact_download( + auth_client, user_alice, tmp_path, settings, monkeypatch +): + settings.BASE_DIR = tmp_path + project = _make_project(user_alice) + run = WorkflowRun.objects.create( + workflow=project, + user=user_alice, + backend=WorkflowRun.Backend.SLURM, + status=WorkflowRun.Status.PENDING, + slurm_job_id="1", + remote_run_dir="/data/neuro-workflow/runs/run-1", + ) + + def fake_ssh(self, cmd): + if "sacct" in cmd: + return "FAILED" + if cmd.startswith("ls "): + return "slurm-1.err\nstdout.log\nstderr.log" + if "exit_code.txt" in cmd: + return "137" + if "stdout.log" in cmd: + return "" + if "stderr.log" in cmd: + return "oom" + return "" + + def fake_sync_from(self, remote, local): + Path(local).parent.mkdir(parents=True, exist_ok=True) + name = Path(remote).name + Path(local).write_text(f"copied:{name}") + + monkeypatch.setattr(RemoteSlurmExecutor, "_ssh", fake_ssh) + monkeypatch.setattr(RemoteSlurmExecutor, "_sync_from_remote", fake_sync_from) + monkeypatch.setattr( + RemoteSlurmExecutor, + "_fetch_results", + lambda self, *a, **k: {"files": []}, + ) + + client = auth_client(user_alice) + detail = reverse("workflow:workflow-run-detail", args=[project.id, run.id]) + resp = client.get(detail) + assert resp.status_code == 200, resp.content + body = resp.json() + assert body["status"] == "failed" + err_path = batch_run_dir(project.id, run.id) / "logs" / "slurm-1.err" + assert err_path.is_file() + assert "slurm-1.err" in { + item["path"].split("/")[-1] for item in body["artifacts"]["logs"] + } + + art = reverse("workflow:workflow-run-artifact", args=[project.id, run.id]) + got = client.get(art, {"path": "logs/slurm-1.err"}) + assert got.status_code == 200 + assert b"copied:slurm-1.err" in b"".join(got.streaming_content) + + bad = client.get(art, {"path": "../x"}) + assert bad.status_code == 400 + + +@pytest.mark.django_db +def test_nodes_package_is_not_downloadable(auth_client, user_alice, tmp_path, settings): + settings.BASE_DIR = tmp_path + project = _make_project(user_alice) + run = WorkflowRun.objects.create( + workflow=project, + user=user_alice, + backend=WorkflowRun.Backend.SLURM, + status=WorkflowRun.Status.COMPLETED, + ) + nodes = batch_run_dir(project.id, run.id, create=True) / "nodes" + nodes.mkdir(parents=True, exist_ok=True) + secret = nodes / "foo.py" + secret.write_text("print('nope')\n") + client = auth_client(user_alice) + art = reverse("workflow:workflow-run-artifact", args=[project.id, run.id]) + resp = client.get(art, {"path": "nodes/foo.py"}) + assert resp.status_code == 400 + with pytest.raises(ValueError): + _resolve_run_artifact(project.id, run.id, "nodes/foo.py") diff --git a/gui/workflow_frontend/src/api/workflowRunApi.ts b/gui/workflow_frontend/src/api/workflowRunApi.ts index ffe56c00..fe30de75 100644 --- a/gui/workflow_frontend/src/api/workflowRunApi.ts +++ b/gui/workflow_frontend/src/api/workflowRunApi.ts @@ -114,36 +114,113 @@ export const fetchRunFigureManifest = async ( // Async run management API (Phase 3) // --------------------------------------------------------------------------- +export interface ArtifactFile { + path: string; + size: number; +} + export interface WorkflowRunRecord { id: string; workflow: string; user: string | null; backend: "local" | "slurm" | "jupyter"; - status: "pending" | "running" | "completed" | "failed" | "cancelled"; + status: "draft" | "pending" | "running" | "completed" | "failed" | "cancelled"; slurm_job_id: string; exit_code: number | null; stdout: string; stderr: string; error_message: string; resource_requests: Record; - artifacts: Record; + artifacts: { + files?: ArtifactFile[]; + logs?: ArtifactFile[]; + }; submitted_at: string; started_at: string | null; finished_at: string | null; + sbatch?: string; + jupyter_path?: string; } +async function parseError(res: Response, fallback: string): Promise { + const body = await res.text(); + try { + const parsed = JSON.parse(body); + return parsed.error || parsed.detail || body || fallback; + } catch { + return body || fallback; + } +} + +export const prepareClusterRun = async ( + workflowId: string, + resourceRequests: Record = {}, + fromRunId?: string +): Promise => { + const headers = await createAuthHeaders(); + const body: Record = { + resource_requests: resourceRequests, + }; + if (fromRunId) body.from_run_id = fromRunId; + const res = await fetch(`${API_PREFIX}/workflow/${workflowId}/runs/prepare/`, { + method: "POST", + headers, + body: JSON.stringify(body), + }); + if (!res.ok) throw new Error(await parseError(res, `Prepare failed: ${res.status}`)); + return res.json(); +}; + +export const getClusterSbatch = async ( + workflowId: string, + runId: string +): Promise => { + const headers = await createAuthHeaders(); + const res = await fetch( + `${API_PREFIX}/workflow/${workflowId}/runs/${runId}/sbatch/`, + { headers } + ); + if (!res.ok) throw new Error(await parseError(res, `Fetch sbatch failed: ${res.status}`)); + return res.json(); +}; + +export const putClusterSbatch = async ( + workflowId: string, + runId: string, + payload: { sbatch?: string; resource_requests?: Record } +): Promise => { + const headers = await createAuthHeaders(); + const res = await fetch( + `${API_PREFIX}/workflow/${workflowId}/runs/${runId}/sbatch/`, + { + method: "PUT", + headers, + body: JSON.stringify(payload), + } + ); + if (!res.ok) throw new Error(await parseError(res, `Save sbatch failed: ${res.status}`)); + return res.json(); +}; + export const submitWorkflowRun = async ( workflowId: string, backend: "local" | "slurm" | "jupyter" = "jupyter", - resourceRequests: Record = {} + resourceRequests: Record = {}, + opts: { runId?: string; sbatch?: string } = {} ): Promise => { const headers = await createAuthHeaders(); + const body: Record = { + backend, + resource_requests: resourceRequests, + }; + if (opts.runId) body.run_id = opts.runId; + if (opts.sbatch) body.sbatch = opts.sbatch; const res = await fetch(`${API_PREFIX}/workflow/${workflowId}/runs/submit/`, { method: "POST", headers, - body: JSON.stringify({ backend, resource_requests: resourceRequests }), + body: JSON.stringify(body), }); - if (!res.ok) throw new Error(`Submit failed: ${res.status}`); + if (!res.ok) throw new Error(await parseError(res, `Submit failed: ${res.status}`)); return res.json(); }; @@ -171,11 +248,6 @@ export const listWorkflowRuns = async ( return res.json(); }; -export interface ArtifactFile { - path: string; - size: number; -} - /** * Download a single result artifact fetched back from a remote run. * diff --git a/gui/workflow_frontend/src/views/home/components/ClusterRunModal.tsx b/gui/workflow_frontend/src/views/home/components/ClusterRunModal.tsx index a2d523da..eafb756b 100644 --- a/gui/workflow_frontend/src/views/home/components/ClusterRunModal.tsx +++ b/gui/workflow_frontend/src/views/home/components/ClusterRunModal.tsx @@ -1,4 +1,4 @@ -import React, { useEffect, useState } from "react"; +import React, { useCallback, useEffect, useRef, useState } from "react"; import { Modal, ModalOverlay, @@ -15,20 +15,33 @@ import { VStack, HStack, Text, + Textarea, + Link, + Spinner, } from "@chakra-ui/react"; +import { + prepareClusterRun, + putClusterSbatch, + getClusterSbatch, +} from "../../../api/workflowRunApi"; +import { JUPYTER_BASE_URL } from "../../../config/urls"; + +export interface ClusterSubmitPayload { + resourceRequests: Record; + runId: string; + sbatch: string; +} interface ClusterRunModalProps { isOpen: boolean; onClose: () => void; - onSubmit: (resourceRequests: Record) => void; + onSubmit: (payload: ClusterSubmitPayload) => void; isSubmitting: boolean; - // Project-level defaults from FlowProject.workflow_context.resource_requirements - // (cpus / memory_gb / gpus / walltime_hours / queue). Used to prefill the - // form; the user can still override for this particular run. + workflowId?: string | null; + fromRunId?: string | null; contextResources?: Record; } -// Partition -> GPU model, per the RIKEN compute server (gcalc1: L40, gcalc2: H100). const GPU_PARTITIONS: Record = { gcalc1: "L40", gcalc2: "H100", @@ -36,8 +49,6 @@ const GPU_PARTITIONS: Record = { const KNOWN_PARTITIONS = new Set(["ccalc", "gcalc1", "gcalc2"]); -// WorkflowContextEditor stores wall time as a number of hours; the sbatch -// --time directive wants HH:MM:SS. const hoursToHHMMSS = (h: unknown): string | undefined => { const n = typeof h === "number" ? h : Number(h); if (!n || n <= 0 || Number.isNaN(n)) return undefined; @@ -48,11 +59,16 @@ const hoursToHHMMSS = (h: unknown): string | undefined => { return [hh, mm, ss].map((x) => String(x).padStart(2, "0")).join(":"); }; +const jupyterFileUrl = (jupyterPath: string) => + `${JUPYTER_BASE_URL}/user/user1/lab/workspaces/auto-E/tree/${jupyterPath}`; + const ClusterRunModal: React.FC = ({ isOpen, onClose, onSubmit, isSubmitting, + workflowId, + fromRunId, contextResources, }) => { const [partition, setPartition] = useState("ccalc"); @@ -60,43 +76,157 @@ const ClusterRunModal: React.FC = ({ const [cpus, setCpus] = useState("2"); const [memGb, setMemGb] = useState("4"); const [gpus, setGpus] = useState("1"); - - // Prefill from the project's resource defaults each time the dialog opens. - useEffect(() => { - if (!isOpen) return; - const r = (contextResources || {}) as Record; - const q = typeof r.queue === "string" ? r.queue : ""; - setPartition(KNOWN_PARTITIONS.has(q) ? q : "ccalc"); - setCpus(r.cpus != null ? String(r.cpus) : "2"); - setMemGb(r.memory_gb != null ? String(r.memory_gb) : "4"); - setWalltime(hoursToHHMMSS(r.walltime_hours) ?? "00:30:00"); - setGpus(r.gpus != null ? String(r.gpus) : "1"); - }, [isOpen, contextResources]); + const [sbatch, setSbatch] = useState(""); + const [draftId, setDraftId] = useState(null); + const [jupyterPath, setJupyterPath] = useState(""); + const [dirty, setDirty] = useState(false); + const [preparing, setPreparing] = useState(false); + const [prepareError, setPrepareError] = useState(""); + const rrSignature = useRef(""); const isGpu = partition in GPU_PARTITIONS; - const handleSubmit = () => { + const buildResourceRequests = useCallback(() => { const rr: Record = { partition, time: walltime }; if (cpus.trim()) rr.cpus_per_task = Number(cpus); if (memGb.trim()) rr.mem = `${Number(memGb)}G`; if (isGpu && gpus.trim()) { rr.gres = `gpu:${GPU_PARTITIONS[partition]}:${Number(gpus)}`; } - onSubmit(rr); + return rr; + }, [partition, walltime, cpus, memGb, gpus, isGpu]); + + useEffect(() => { + if (!isOpen) { + setSbatch(""); + setDraftId(null); + setJupyterPath(""); + setDirty(false); + setPrepareError(""); + setPreparing(false); + rrSignature.current = ""; + return; + } + const r = (contextResources || {}) as Record; + const q = typeof r.queue === "string" ? r.queue : ""; + const nextPartition = KNOWN_PARTITIONS.has(q) ? q : "ccalc"; + const nextCpus = r.cpus != null ? String(r.cpus) : "2"; + const nextMem = r.memory_gb != null ? String(r.memory_gb) : "4"; + const nextWall = hoursToHHMMSS(r.walltime_hours) ?? "00:30:00"; + const nextGpus = r.gpus != null ? String(r.gpus) : "1"; + setPartition(nextPartition); + setCpus(nextCpus); + setMemGb(nextMem); + setWalltime(nextWall); + setGpus(nextGpus); + setDirty(false); + + if (!workflowId) return; + const rr: Record = { + partition: nextPartition, + time: nextWall, + }; + if (nextCpus.trim()) rr.cpus_per_task = Number(nextCpus); + if (nextMem.trim()) rr.mem = `${Number(nextMem)}G`; + if (nextPartition in GPU_PARTITIONS && nextGpus.trim()) { + rr.gres = `gpu:${GPU_PARTITIONS[nextPartition]}:${Number(nextGpus)}`; + } + let cancelled = false; + setPreparing(true); + setPrepareError(""); + prepareClusterRun(workflowId, rr, fromRunId || undefined) + .then((run) => { + if (cancelled) return; + setDraftId(run.id); + setSbatch(run.sbatch || ""); + setJupyterPath(run.jupyter_path || ""); + rrSignature.current = JSON.stringify(rr); + }) + .catch((err: unknown) => { + if (cancelled) return; + setPrepareError( + err instanceof Error ? err.message : "Failed to prepare run.sbatch" + ); + }) + .finally(() => { + if (!cancelled) setPreparing(false); + }); + return () => { + cancelled = true; + }; + }, [isOpen, workflowId, fromRunId, contextResources]); + + useEffect(() => { + if (!isOpen || !workflowId || !draftId || preparing) return; + const rr = buildResourceRequests(); + const sig = JSON.stringify(rr); + if (sig === rrSignature.current) return; + if (dirty) { + const replace = window.confirm( + "Resource fields changed. Replace the edited run.sbatch with a newly generated script?" + ); + if (!replace) { + rrSignature.current = sig; + return; + } + setDirty(false); + } + let cancelled = false; + putClusterSbatch(workflowId, draftId, { resource_requests: rr }) + .then((run) => { + if (cancelled) return; + setSbatch(run.sbatch || ""); + rrSignature.current = sig; + }) + .catch(() => { + /* keep the last script; user can still submit or reload */ + }); + return () => { + cancelled = true; + }; + }, [ + isOpen, + workflowId, + draftId, + dirty, + preparing, + buildResourceRequests, + ]); + + const handleReload = async () => { + if (!workflowId || !draftId) return; + try { + const run = await getClusterSbatch(workflowId, draftId); + setSbatch(run.sbatch || ""); + setDirty(false); + } catch (err: unknown) { + setPrepareError( + err instanceof Error ? err.message : "Failed to reload run.sbatch" + ); + } + }; + + const handleSubmit = () => { + if (!draftId) return; + onSubmit({ + resourceRequests: buildResourceRequests(), + runId: draftId, + sbatch, + }); }; return ( - + - + Run on compute cluster - The workflow code is regenerated and submitted as a Slurm batch job - on the RIKEN compute server. Progress and results appear in the Runs - panel (bottom-right). Values are prefilled from the project's - resource settings — adjust to override for this run. + Resource fields generate a Slurm script. Edit it here or in + Jupyter, then submit. Closing without submit keeps a draft in the + Runs panel. Progress and copied logs appear there after the job + finishes. @@ -158,11 +288,64 @@ const ClusterRunModal: React.FC = ({ )} + + + + + run.sbatch + + + + {jupyterPath && ( + + Edit in Jupyter + + )} + + + {preparing ? ( + + + + Preparing run.sbatch… + + + ) : ( +