Merge branch 'release_26.1' into dev

2026-09-01 15:37:32 +08:00 · 2026-07-23 12:36:49 +02:00
parent de59688c84 84d19ac1af
commit 0e4beaa04f
17 changed files with 1399 additions and 65 deletions
@@ -0,0 +1,42 @@
+import { getLocalVue } from "@tests/vitest/helpers";
+import { shallowMount } from "@vue/test-utils";
+import axios from "axios";
+import { describe, expect, it, vi } from "vitest";
+
+import TabularChunkedView from "./TabularChunkedView.vue";
+import GTable from "@/components/Common/GTable.vue";
+
+vi.mock("axios");
+vi.mock("@/onload/loadConfig", () => ({
+    getAppRoot: () => "/",
+}));
+
+const localVue = getLocalVue();
+
+function mountChunkedView(fileExt: string) {
+    vi.mocked(axios.get).mockResolvedValue({ data: { ck_data: "", offset: 0, data_line_offset: 0 } });
+    return shallowMount(TabularChunkedView as object, {
+        localVue,
+        propsData: {
+            options: {
+                id: "dataset-id",
+                file_ext: fileExt,
+                metadata_columns: 2,
+            },
+        },
+    });
+}
+
+describe("TabularChunkedView", () => {
+    it("hides the table header for generic tabular datasets", () => {
+        const wrapper = mountChunkedView("tabular");
+
+        expect(wrapper.findComponent(GTable).props("hideHeader")).toBe(true);
+    });
+
+    it("keeps the table header for CSV datasets", () => {
+        const wrapper = mountChunkedView("csv");
+
+        expect(wrapper.findComponent(GTable).props("hideHeader")).toBe(false);
+    });
+});
@@ -200,6 +200,7 @@ onMounted(() => {
            striped
            head-variant="dark"
            :fields="fields"
+            :hide-header="props.options.file_ext === 'tabular'"
            :items="tableRows"
            :load-more-loading="loading" />
    </div>
@@ -177,34 +177,47 @@ defineExpose({
    .container-root {
        container-type: inline-size;
        width: 100%;
+        height: 100%;
+        min-height: 0;
        overflow: auto;
    }

    .published-workflow {
        display: grid;
        gap: 0.5rem 1rem;
-        grid-template-columns: auto auto 30%;
+        grid-template-columns: minmax(0, 1fr) minmax(18rem, 30%);
+        grid-template-rows: auto minmax(0, 1fr);

        height: 100%;
+        min-height: 0;

        .workflow-header {
-            grid-column: 1 / span 3;
+            grid-column: 1 / -1;

            display: flex;
+            align-items: center;
+            gap: 1rem;
            justify-content: flex-end;
        }

        .workflow-preview {
-            grid-column: 1 / span 2;
+            grid-column: 1;
+            min-height: 0;

            &.only-preview {
-                grid-column: 1 / span 3;
+                grid-column: 1 / -1;
+            }
+
+            &:deep(.card-body) {
+                height: 100%;
+                min-height: 0;
            }
        }

        &:deep(.workflow-information-container) {
            height: 100%;
            max-width: 500px;
+            align-self: stretch;
            overflow: auto;
        }
    }
@@ -212,10 +225,11 @@ defineExpose({
    @container (max-width: 900px) {
        .published-workflow {
            height: unset;
-            grid-template-columns: auto;
+            grid-template-columns: minmax(0, 1fr);
+            grid-template-rows: auto auto auto;

            .workflow-preview {
-                grid-column: 1 / span 3;
+                grid-column: 1;
                height: 450px;
            }

@@ -225,7 +239,7 @@ defineExpose({
            }

            .workflow-information-container {
-                grid-column: 1 / span 3;
+                grid-column: 1;
            }
        }
    }
@@ -1,8 +1,8 @@
 ---
 myst:
-    substitutions:
-        GA4GH_DRS: GA4GH Data Repository Service (DRS)
-        GA4GH_WES: GA4GH Workflow Execution Service (WES)
+  substitutions:
+    GA4GH_DRS: GA4GH Data Repository Service (DRS)
+    GA4GH_WES: GA4GH Workflow Execution Service (WES)
 ---

 # GA4GH API Support
@@ -73,6 +73,7 @@ You should see output like:
 ```

 Verify that:
+
 - `organization.name` and `organization.url` match your configured values
 - `environment` is set appropriately for your deployment
 - `id` reflects your `ga4gh_service_id` setting (or sensible defaults if not configured)
@@ -82,6 +83,8 @@ Verify that:
 The {{ GA4GH_WES }} enables external systems to submit and monitor Galaxy workflow executions.

 For detailed API specifications, see the [GA4GH WES specification](https://ga4gh.github.io/workflow-execution-service-schemas/).
+For a client/developer walkthrough of submitting and monitoring runs against this API, see
+[Developing Against the GA4GH WES API](../dev/ga4gh_wes.md).

 ### Workflow Types

@@ -140,6 +143,7 @@ You should see output like:
 ```

 Verify that:
+
 - `organization.name` and `organization.url` match your configured values
 - `environment` is set appropriately for your deployment
 - `id` reflects your `ga4gh_service_id` setting (or sensible defaults if not configured)
@@ -150,12 +154,12 @@ All GA4GH configuration is optional and falls back to sensible defaults based on

 ### Settings

-| Setting | Default | Purpose |
-|---------|---------|---------|
-| `organization_name` | Reversed hostname | Organization name in service responses |
-| `organization_url` | Scheme + hostname from request | Organization website URL |
-| `ga4gh_service_id` | Reversed hostname | Service ID in reverse domain format (e.g., `org.example`) |
-| `ga4gh_service_environment` | (none) | Environment classifier (e.g., "test", "staging", "production") |
+| Setting                     | Default                        | Purpose                                                        |
+| --------------------------- | ------------------------------ | -------------------------------------------------------------- |
+| `organization_name`         | Reversed hostname              | Organization name in service responses                         |
+| `organization_url`          | Scheme + hostname from request | Organization website URL                                       |
+| `ga4gh_service_id`          | Reversed hostname              | Service ID in reverse domain format (e.g., `org.example`)      |
+| `ga4gh_service_environment` | (none)                         | Environment classifier (e.g., "test", "staging", "production") |

 ### Complete Configuration Example

@@ -0,0 +1,447 @@
+---
+myst:
+  substitutions:
+    GA4GH_WES: GA4GH Workflow Execution Service (WES)
+---
+
+# Developing Against the GA4GH WES API
+
+This is a developer-facing companion to the [admin GA4GH docs](../admin/ga4gh.md). The
+admin docs cover configuring and verifying the service; this guide walks an API client
+through actually submitting and monitoring Galaxy workflow runs over the {{ GA4GH_WES }}
+v1.0.0 wire protocol.
+
+Every request below targets a Galaxy server and uses Galaxy's workflow formats. The curl
+examples were exercised against a live development server; the responses shown are real
+(ids shortened/anonymized).
+
+## Example clients
+
+If you would rather drive the flow below from a runnable tool than from raw curl, two small
+reference clients walk exactly this sequence (discover → stage → submit → poll → outputs →
+tasks). Both expose a subcommand-per-endpoint CLI plus a `demo` subcommand that runs the
+whole example end to end, and both are published to PyPI so they run straight under
+[uv](https://docs.astral.sh/uv/) without installing anything:
+
+```bash
+export GXY_WES_API_KEY=...
+uvx gxy-wes demo --galaxy-url http://localhost:8080            # requests-only
+uvx gxy-wes-bioblend demo --galaxy-url http://localhost:8080   # BioBlend-backed
+```
+
+- **[gxy-wes](https://github.com/jmchilton/gxy-wes)** — a hand-rolled `requests` wrapper
+  with no dependency beyond `requests`, so it stays dependency-light and reads as a literal
+  transcript of the wire protocol.
+- **[gxy-wes-bioblend](https://github.com/jmchilton/gxy-wes-bioblend)** — the same CLI
+  surface, but every call is routed through a [BioBlend](https://bioblend.readthedocs.io/)
+  `GalaxyInstance`, showing the shape a BioBlend-based application would use.
+
+```{important}
+These two packages are **illustrative examples, not production-supported projects** — they
+exist to make this document executable and to show the API two different ways. Do not build
+a product on them. For real workflow-execution tooling use
+[Planemo](https://planemo.readthedocs.io/) or talk to the Galaxy API through
+[BioBlend](https://bioblend.readthedocs.io/) directly.
+```
+
+## How Galaxy maps onto WES
+
+WES is workflow-engine agnostic. Galaxy implements it on top of its existing workflow
+invocation machinery, so the WES nouns map onto Galaxy nouns:
+
+| WES concept        | Galaxy concept                                                       |
+| ------------------ | -------------------------------------------------------------------- |
+| `run` / `run_id`   | `WorkflowInvocation` (the `run_id` is the encoded invocation id)     |
+| `task` / `task_id` | An invocation step, or a single job within a collection-mapping step |
+| run `outputs`      | Invocation outputs (HDAs / HDCAs), decorated with DRS URIs           |
+| `workflow_type`    | A Galaxy workflow format (see below) — **not** CWL/Nextflow/WDL      |
+
+`workflow_type` is one of Galaxy's own two workflow formats:
+
+- **`gx_workflow_format2`** — the gxformat2 ("Format2") YAML format. CWL-like, hand-authorable.
+- **`gx_workflow_ga`** — the native `.ga` JSON format produced by Galaxy's workflow editor.
+
+`workflow_type_version` is currently accepted as a free-form string (e.g. `"1.0.0"`); the
+type itself is what matters, and Galaxy will also auto-detect it from the workflow body
+(`class: GalaxyWorkflow` ⇒ format2; a top-level `steps`/`workflow` key ⇒ `.ga`).
+
+## Base URL and authentication
+
+All endpoints live under `/ga4gh/wes/v1/`. With a local dev server that is:
+
+```
+http://localhost:8080/ga4gh/wes/v1/
+```
+
+Every endpoint except `service-info` requires authentication. Galaxy uses its normal API
+key auth — pass `x-api-key: <key>` (the WES spec's bearer-token scheme is not used). Get a
+key for an existing account with HTTP basic auth:
+
+```bash
+curl -s -u "you@example.com:password" \
+  http://localhost:8080/api/authenticate/baseauth
+# {"api_key": "317ab17d..."}
+```
+
+Anonymous and inactive accounts are rejected on submission and on every run-scoped read.
+
+## 1. Discover the service
+
+`service-info` is public and advertises the supported formats, engine parameters, and
+filesystem protocols.
+
+```bash
+curl -s http://localhost:8080/ga4gh/wes/v1/service-info | jq .
+```
+
+Trimmed to the WES-relevant fields (the real response also carries
+`system_state_counts`, `auth_instructions_url`, `tags`, and the standard GA4GH
+`organization`/`version`/`environment` keys):
+
+```json
+{
+  "id": "localhost.wes",
+  "name": "Galaxy WES API",
+  "type": { "group": "org.ga4gh", "artifact": "wes", "version": "1.0.0" },
+  "workflow_type_versions": {
+    "gx_workflow_ga": { "workflow_type_version": ["1.0.0"] },
+    "gx_workflow_format2": { "workflow_type_version": ["1.0.0"] }
+  },
+  "supported_wes_versions": ["1.0.0"],
+  "supported_filesystem_protocols": ["http", "https", "file", "s3", "gs"],
+  "workflow_engine_versions": {
+    "galaxy": { "workflow_engine_version": ["1.0.0"] }
+  },
+  "default_workflow_engine_parameters": [
+    { "name": "history_name", "type": "string", "default_value": "" },
+    { "name": "history_id", "type": "string", "default_value": "" },
+    {
+      "name": "preferred_object_store_id",
+      "type": "string",
+      "default_value": ""
+    },
+    { "name": "use_cached_job", "type": "boolean", "default_value": "false" }
+  ]
+}
+```
+
+## 2. Provide the workflow and its inputs
+
+A WES `RunRequest` is `multipart/form-data`. The fields Galaxy honors:
+
+| Field                                         | Required              | Notes                                                                                                                           |
+| --------------------------------------------- | --------------------- | ------------------------------------------------------------------------------------------------------------------------------- |
+| `workflow_type`                               | yes                   | `gx_workflow_format2` or `gx_workflow_ga`. Must match the auto-detected type or you get a 400.                                  |
+| `workflow_type_version`                       | yes                   | Free-form string, e.g. `"1.0.0"`.                                                                                               |
+| `workflow_url`                                | one of url/attachment | A URL Galaxy can fetch (`http(s)://`, `s3://`, `gs://`, `file://`, `base64://`) **or** a `gxworkflow://` reference (see below). |
+| `workflow_attachment`                         | one of url/attachment | The workflow file uploaded inline.                                                                                              |
+| `workflow_params`                             | no                    | JSON object of workflow inputs (see below).                                                                                     |
+| `workflow_engine_parameters`                  | no                    | JSON object of Galaxy-specific run options (see below).                                                                         |
+| `tags`                                        | no                    | Accepted but currently not persisted onto the invocation.                                                                       |
+| `workflow_engine` / `workflow_engine_version` | no                    | Accepted; informational.                                                                                                        |
+
+### `workflow_params` — wiring up inputs
+
+Galaxy workflow inputs are keyed by the workflow's **input label**, and dataset inputs are
+references to data that already exists in Galaxy. Galaxy invokes with `inputs_by="name"`,
+so the keys are the workflow input names.
+
+A dataset input is a `{"src": ..., "id": ...}` reference:
+
+```json
+{ "input1": { "src": "hda", "id": "1e8ab44153008be8" } }
+```
+
+`src` is `hda` for a single dataset or `hdca` for a dataset collection; `id` is the encoded
+content id. Simple parameter inputs (integers, text, booleans) are passed as plain JSON
+values under their input name.
+
+WES has **no data-staging endpoint** of its own. You stage inputs through Galaxy's normal
+API first (create a history, upload/fetch datasets) and then reference them here.
+
+### `workflow_engine_parameters` — Galaxy run options
+
+A JSON object. Recognized keys (also advertised in `service-info`):
+
+| Key                         | Effect                                                                           |
+| --------------------------- | -------------------------------------------------------------------------------- |
+| `history_id`                | Run into this existing history (encoded id). Otherwise a new history is created. |
+| `history_name`              | Name for the auto-created history (default `"WES Run"`).                         |
+| `preferred_object_store_id` | Object store for the run's outputs.                                              |
+| `use_cached_job`            | `"true"`/`"false"` string — reuse equivalent prior job results.                  |
+
+### Referencing a workflow already in Galaxy: `gxworkflow://`
+
+Galaxy adds a non-standard URL scheme so you don't have to ship the workflow body on every
+submission. Use it as `workflow_url`:
+
+```
+gxworkflow://<encoded_workflow_id>            # the StoredWorkflow (latest version)
+gxworkflow://<encoded_workflow_id>?instance=true  # a specific Workflow instance
+```
+
+The workflow only has to be accessible to the caller — the same rule a normal invocation
+uses: owned by them, shared with them, published/importable, or the caller is an admin.
+Anything else is a 403.
+
+With `gxworkflow://`, Galaxy skips the import step and invokes the referenced workflow
+directly. `workflow_type` is still required by the form but is **not** validated against
+the stored workflow in this case — the "must match the auto-detected type or 400" check
+only applies to inline `workflow_attachment` / fetched `workflow_url` submissions.
+
+## 3. Submit the run
+
+Staging step (Galaxy-native, abbreviated) — create a history and upload one dataset:
+
+```bash
+KEY=317ab17d...; BASE=http://localhost:8080
+HIST=$(curl -s -X POST "$BASE/api/histories" -H "x-api-key: $KEY" \
+  -H 'Content-Type: application/json' -d '{"name":"WES Tutorial"}' | jq -r .id)
+HDA=$(curl -s -X POST "$BASE/api/tools/fetch" -H "x-api-key: $KEY" \
+  -F "history_id=$HIST" \
+  -F 'targets=[{"destination":{"type":"hdas"},"elements":[{"src":"pasted","paste_content":"line one\nline two\n","ext":"txt","name":"wes_input"}]}]' \
+  | jq -r '.outputs[0].id')
+```
+
+A minimal Format2 workflow (`simple.gxwf.yml`):
+
+```yaml
+class: GalaxyWorkflow
+name: Simple Workflow
+inputs:
+  input1: data
+outputs:
+  wf_output_1:
+    outputSource: first_cat/out_file1
+steps:
+  first_cat:
+    tool_id: cat1
+    in:
+      input1: input1
+```
+
+Submit it (the WES `POST /runs`):
+
+```bash
+curl -s -X POST "$BASE/ga4gh/wes/v1/runs" -H "x-api-key: $KEY" \
+  -F "workflow_type=gx_workflow_format2" \
+  -F "workflow_type_version=1.0.0" \
+  -F "workflow_params={\"input1\":{\"src\":\"hda\",\"id\":\"$HDA\"}}" \
+  -F "workflow_engine_parameters={\"history_id\":\"$HIST\"}" \
+  -F "workflow_attachment=@simple.gxwf.yml;type=application/x-yaml"
+```
+
+```json
+{ "run_id": "33b43b4e7093c91f" }
+```
+
+The `run_id` is the encoded Galaxy invocation id. Batch invocations (multiple invocations
+from one request) are rejected by WES.
+
+## 4. Monitor the run
+
+Abbreviated status:
+
+```bash
+curl -s "$BASE/ga4gh/wes/v1/runs/33b43b4e7093c91f/status" -H "x-api-key: $KEY"
+# {"run_id":"33b43b4e7093c91f","state":"RUNNING"}
+```
+
+Full run log — outputs are decorated with DRS URIs so a client can hand them straight to
+the DRS API:
+
+```bash
+curl -s "$BASE/ga4gh/wes/v1/runs/33b43b4e7093c91f" -H "x-api-key: $KEY" | jq .
+```
+
+```json
+{
+  "run_id": "33b43b4e7093c91f",
+  "request": null,
+  "state": "COMPLETE",
+  "run_log": null,
+  "task_logs_url": "/ga4gh/wes/v1/runs/33b43b4e7093c91f/tasks",
+  "task_logs": null,
+  "outputs": {
+    "wf_output_1": {
+      "src": "hda",
+      "id": "417e33144b294c21",
+      "workflow_step_id": 30,
+      "drs_uri": "drs://drs.localhost:8080/hda-4dcf52aeee371f21"
+    }
+  }
+}
+```
+
+Notes on the run log:
+
+- `request` is always `null` — Galaxy does not persist the original `RunRequest`, so it
+  cannot be reconstructed.
+- `task_logs` is intentionally `null` (deprecated in the WES spec). Use `task_logs_url`.
+- Only HDA (single-dataset) outputs get a `drs_uri`; collection (HDCA) outputs carry an
+  encoded `id` but no DRS URI.
+
+### State mapping
+
+WES has a single flat `State` enum (`QUEUED`, `INITIALIZING`, `RUNNING`, `COMPLETE`,
+`EXECUTOR_ERROR`, `CANCELING`, `CANCELED`, ...). A Galaxy run is a `WorkflowInvocation`,
+whose state is mapped onto that enum (`GALAXY_TO_WES_STATE` in `services/wes.py`):
+
+| Galaxy invocation state    | WES state        | Meaning                                                 |
+| -------------------------- | ---------------- | ------------------------------------------------------- |
+| `new`                      | `QUEUED`         | Invocation created, not yet scheduling.                 |
+| `requires_materialization` | `INITIALIZING`   | Deferred inputs are being materialized.                 |
+| `ready`                    | `INITIALIZING`   | Ready for a scheduling iteration.                       |
+| `scheduled`                | `RUNNING`        | Fully scheduled; jobs are running/queued.               |
+| `completed`                | `COMPLETE`       | Scheduling done and every job reached a terminal state. |
+| `failed`                   | `EXECUTOR_ERROR` | The invocation itself failed to schedule.               |
+| `cancelled`                | `CANCELED`       | Invocation was cancelled.                               |
+| `cancelling`               | `CANCELING`      | Cancellation requested, in progress.                    |
+| (anything unmapped)        | `UNKNOWN`        | Fallback; should not occur for current states.          |
+
+Poll `status` (or the run log `state`) until the state is terminal: `COMPLETE` (success),
+`EXECUTOR_ERROR` (failure), or `CANCELED` (cancelled). Keep polling while it is any of the
+non-terminal states (`QUEUED`, `INITIALIZING`, `RUNNING`, `CANCELING`).
+
+#### Why `completed` ⇒ `COMPLETE`, even when a job failed
+
+This is intentional and important: **a failed job does not make the run unsuccessful.**
+WES `state` describes the run (the invocation), not individual jobs. Galaxy's `completed`
+means scheduling finished and all jobs reached a terminal state (`ok`, `error`, `skipped`,
+`paused`, ...) — and a job ending in `error` is frequently part of a perfectly normal,
+valid execution path. For example, a workflow can route a step through a _filter-failed_
+collection operation precisely so that some jobs are allowed to fail and the failures are
+filtered out downstream; the invocation still completes successfully.
+
+So:
+
+- The only invocation state that maps to `EXECUTOR_ERROR` is Galaxy's own `failed` — i.e.
+  the invocation could not be scheduled — **not** "some job errored."
+- Clients should **not** infer run failure from a non-zero per-task `exit_code`. Per-task
+  `exit_code`s (from the tasks endpoint) are for inspecting individual steps, not for
+  deciding whether the run as a whole succeeded.
+- Determine run success from the run `state` (`COMPLETE`); inspect outputs and per-task
+  detail for finer-grained results.
+
+## 5. Tasks (per-step / per-job detail)
+
+`task_logs_url` points at a paginated task list. A `task_id` is the step `order_index`
+(e.g. `"0"`, `"1"`), or `order_index.job_index` (e.g. `"1.0"`, `"1.2"`) for the individual
+jobs of a collection-mapping step.
+
+```bash
+curl -s "$BASE/ga4gh/wes/v1/runs/33b43b4e7093c91f/tasks" -H "x-api-key: $KEY" | jq .
+```
+
+```json
+{
+  "task_logs": [
+    {
+      "id": "0",
+      "name": "input1",
+      "exit_code": null,
+      "stdout": null,
+      "stderr": null
+    },
+    {
+      "id": "1",
+      "name": "first_cat",
+      "exit_code": 0,
+      "stdout": "/api/jobs/417e33144b294c21/stdout",
+      "stderr": "/api/jobs/417e33144b294c21/stderr"
+    }
+  ],
+  "next_page_token": null
+}
+```
+
+`stdout` / `stderr` are URLs to plain-text job-output endpoints
+(`GET /api/jobs/{job_id}/stdout` and `/stderr`). Steps with no underlying job — input
+steps, for example — have `null` `exit_code`, `stdout`, and `stderr` (as `id: "0"` shows
+above). Fetch a single task with `GET /ga4gh/wes/v1/runs/{run_id}/tasks/{task_id}`.
+
+### Pagination
+
+Both `GET /runs` and `GET /runs/{id}/tasks` use opaque keyset (cursor) tokens, not
+offsets. Pass `page_size` (1–100, default 10); if more results exist the response includes
+`next_page_token`, which you echo back as `page_token` on the next request. Do not try to
+decode or construct tokens yourself.
+
+```bash
+curl -s "$BASE/ga4gh/wes/v1/runs?page_size=2" -H "x-api-key: $KEY"
+curl -s "$BASE/ga4gh/wes/v1/runs?page_size=2&page_token=<next_page_token>" -H "x-api-key: $KEY"
+```
+
+`GET /runs` only ever lists the authenticated user's own runs.
+
+## 6. Cancel a run
+
+```bash
+curl -s -X POST "$BASE/ga4gh/wes/v1/runs/33b43b4e7093c91f/cancel" -H "x-api-key: $KEY"
+# {"run_id": "33b43b4e7093c91f"}
+```
+
+This requests cancellation of the underlying invocation; the run then transitions through
+`CANCELING` to `CANCELED`.
+
+## Errors
+
+Errors come back as Galaxy's standard JSON error body (`{"err_msg": ..., "err_code": ...}`)
+with a matching HTTP status. The ones a client is most likely to hit:
+
+| Situation                                                                        | Status |
+| -------------------------------------------------------------------------------- | ------ |
+| Missing/invalid required form field (e.g. no `workflow_type`)                    | 400    |
+| `workflow_type` disagrees with the auto-detected type (attachment / fetched URL) | 400    |
+| Malformed `gxworkflow://` URI                                                    | 400    |
+| Submit while anonymous                                                           | 403    |
+| Submit while logged in but inactive (unactivated account)                        | 403    |
+| Run / task id does not exist                                                     | 404    |
+| Invalid `task_id` format                                                         | 404    |
+
+Access to resources you do not own is **not** reported uniformly — watch for this:
+
+- Reading a **run** you don't own → `403` (`AuthenticationRequired`).
+- Submitting against a **history** you don't own → `404` (`ObjectNotFound`).
+- A `gxworkflow://` reference to a **workflow** you cannot access (not owned, not shared
+  with you, not published) → `403` (`ItemAccessibilityException`).
+
+So a `404` on submission can mean "your history id is wrong" _or_ "that history belongs to
+someone else"; don't assume `404` always means the object is absent.
+
+## Endpoint summary
+
+| Method | Path                                          | Auth     | Purpose                                 |
+| ------ | --------------------------------------------- | -------- | --------------------------------------- |
+| GET    | `/ga4gh/wes/v1/service-info`                  | public   | Capabilities and supported formats      |
+| POST   | `/ga4gh/wes/v1/runs`                          | required | Submit a run (`multipart/form-data`)    |
+| GET    | `/ga4gh/wes/v1/runs`                          | required | List the user's runs (keyset paginated) |
+| GET    | `/ga4gh/wes/v1/runs/{run_id}`                 | required | Full run log + DRS-decorated outputs    |
+| GET    | `/ga4gh/wes/v1/runs/{run_id}/status`          | required | Abbreviated `{run_id, state}`           |
+| POST   | `/ga4gh/wes/v1/runs/{run_id}/cancel`          | required | Request cancellation                    |
+| GET    | `/ga4gh/wes/v1/runs/{run_id}/tasks`           | required | Paginated task list                     |
+| GET    | `/ga4gh/wes/v1/runs/{run_id}/tasks/{task_id}` | required | Single task detail                      |
+
+## Known limitations and gotchas
+
+- **Run `COMPLETE` reflects the invocation, not individual jobs** — by design, a failed
+  job does not make the run fail (a failed job can be a normal, valid path, e.g.
+  filter-failed). Only an invocation that fails to schedule maps to `EXECUTOR_ERROR`. Do
+  not infer run failure from a per-task `exit_code`. See §4.
+- **Inputs must be pre-staged** — there is no WES upload endpoint; stage datasets via the
+  Galaxy API and reference them by `{src, id}` in `workflow_params`.
+- **`request` is never returned** in the run log — the original `RunRequest` is not stored.
+- **DRS URIs only on HDA outputs**, not on collection outputs.
+- **`tags` are not persisted** to the invocation, and run summaries return `tags: {}`.
+- **`service-info` `auth_instructions_url` is a placeholder** (`"TODO"`).
+- The generated Pydantic models under `lib/galaxy/schema/wes/` are produced by
+  `gen.sh` from the upstream GA4GH OpenAPI spec with no Galaxy post-processing — treat
+  them as regenerable and avoid hand edits.
+
+## Source pointers
+
+- Router: `lib/galaxy/webapps/galaxy/api/wes.py`
+- Service: `lib/galaxy/webapps/galaxy/services/wes.py`
+- Generated models: `lib/galaxy/schema/wes/` (regenerate via `gen.sh`)
+- Shared GA4GH service-info builder: `lib/galaxy/webapps/galaxy/services/ga4gh.py`
+- API tests (worked examples for every endpoint): `lib/galaxy_test/api/test_wes.py`
@@ -21,6 +21,7 @@ A multi-hour long video playlist covering these slides can be found at
  data_source
  data_types
  tool_source_storage
+  ga4gh_wes
  faq
  writing_tests
  debugging_tests
@@ -15,7 +15,7 @@
  # for accessing passwords stored in a vault:
  # password: ${user.user_vault.read_secret('preferences/owncloud/password')}

-  # By default, the plugin will use temp files to avoid loading entire files into memory. 
+  # By default, the plugin will use temp files to avoid loading entire files into memory.
  # You can change the directory here or omit to use the default temp directory.
  temp_path: /your/temp/path
  # Set writable to true if you have write access to this source
@@ -84,8 +84,8 @@
  enable_resume: true  # Enable resume for interrupted transfers
  # SSH private key for authentication (download ENA key from: https://www.ebi.ac.uk/ena/browser/about/ascp)
  # Embed the key content directly in the configuration (required - not file paths)
-  # Note: SSH key content is required because Galaxy jobs often run on clusters that don't 
-  # mount Galaxy's root or configuration directories. The configuration block is copied 
+  # Note: SSH key content is required because Galaxy jobs often run on clusters that don't
+  # mount Galaxy's root or configuration directories. The configuration block is copied
  # to the job's directory, but referenced key paths wouldn't be accessible.
  ssh_key_content: |
    -----BEGIN RSA PRIVATE KEY-----
@@ -259,6 +259,12 @@
  # token: ${user.preferences['invenio_sandbox|token']} # Alternatively use this for retrieving the token from user preferences instead of the Vault
  public_name: ${user.preferences['invenio_sandbox|public_name']}
  writable: true
+  # Enable multipart upload for files of size >= threshold (values in MB, optional, disabled by default)
+  # multipart_threshold: 100
+  # Part size for multipart uploads (values in MB, optional, defaults to 5 MB minimum)
+  # multipart_chunk_size: 50
+  # Default resource type for new records (optional, defaults to "dataset")
+  # default_resource_type:

 - type: zenodo
  id: zenodo
@@ -390,6 +390,12 @@ class Tabular(TabularData):
    def get_column_names(self, first_line: str) -> list[str] | None:
        return None

+    def make_html_peek_header(self, dataset: DatasetProtocol, *args, **kwargs) -> str:
+        # Only generic tabular loses generated headers; subclasses may define real column labels.
+        if type(self) is Tabular:
+            return ""
+        return super().make_html_peek_header(dataset, *args, **kwargs)
+
    def set_meta(
        self,
        dataset: DatasetProtocol,
@@ -23,11 +23,17 @@ log = logging.getLogger(__name__)
 class RDMFileSourceTemplateConfiguration(BaseFileSourceTemplateConfiguration):
    token: str | TemplateExpansion | None = None
    public_name: str | TemplateExpansion | None = None
+    multipart_threshold: int | TemplateExpansion | None = None  # MB
+    multipart_chunk_size: int | TemplateExpansion | None = None  # MB
+    default_resource_type: str | TemplateExpansion | None = None


 class RDMFileSourceConfiguration(BaseFileSourceConfiguration):
    token: str | None = None
    public_name: str | None = None
+    multipart_threshold: int | None = None  # MB
+    multipart_chunk_size: int | None = None  # MB
+    default_resource_type: str | None = None


 class ContainerAndFileIdentifier(NamedTuple):
@@ -1,6 +1,13 @@
 import datetime
 import json
+import logging
+import math
+import os
 import re
+from concurrent.futures import (
+    as_completed,
+    ThreadPoolExecutor,
+)
 from typing import (
    Any,
    cast,
@@ -12,6 +19,8 @@ from typing_extensions import (
    TypedDict,
 )

+log = logging.getLogger(__name__)
+
 from galaxy.exceptions import (
    AuthenticationRequired,
    MessageException,
@@ -101,6 +110,86 @@ class RecordLinks(TypedDict):
    reserve_doi: str


+# AWS S3 multipart default limits (used by Invenio RDM)
+MIN_UPLOAD_PART_SIZE = 5 * 1024 * 1024  # 5 MiB
+MAX_UPLOAD_PART_SIZE = 5 * 1024**3  # 5 GiB
+MAX_UPLOAD_PARTS = 10_000
+
+
+def calculate_multipart_params(file_size: int, preferred_part_size: int | None = None) -> tuple[int, int]:
+    """Calculate parts count and part size for multipart upload.
+
+    Args:
+        file_size: Total file size in bytes
+        preferred_part_size: Preferred part size in bytes (optional)
+
+    Returns:
+        Tuple of (parts_count, part_size)
+
+    Raises:
+        ValueError: If the file is larger than MAX_UPLOAD_PARTS * MAX_UPLOAD_PART_SIZE
+            (~48.8 TiB), which exceeds the maximum uploadable size.
+
+    Note:
+        Maximum uploadable file size is MAX_UPLOAD_PARTS * MAX_UPLOAD_PART_SIZE (~48.8 TiB).
+        Files larger than this cannot be uploaded via multipart and raise ValueError.
+    """
+    if file_size == 0:
+        return 1, MIN_UPLOAD_PART_SIZE
+
+    # Start with preferred or minimum part size, clamped to [min, max]
+    part_size = preferred_part_size or MIN_UPLOAD_PART_SIZE
+    part_size = max(part_size, MIN_UPLOAD_PART_SIZE)
+    part_size = min(part_size, MAX_UPLOAD_PART_SIZE)
+
+    # Grow part_size to the minimum that keeps the part count within MAX_UPLOAD_PARTS.
+    part_size = max(part_size, math.ceil(file_size / MAX_UPLOAD_PARTS))
+    part_size = min(part_size, MAX_UPLOAD_PART_SIZE)
+
+    max_upload_size = MAX_UPLOAD_PARTS * MAX_UPLOAD_PART_SIZE
+    if file_size > max_upload_size:
+        raise ValueError(
+            f"File size {file_size} bytes exceeds the maximum multipart upload size "
+            f"of {max_upload_size} bytes ({MAX_UPLOAD_PARTS} parts x {MAX_UPLOAD_PART_SIZE} bytes)."
+        )
+
+    parts = math.ceil(file_size / part_size)
+    return parts, part_size
+
+
+class _LimitedFileReader:
+    """File-like wrapper that limits reads to a specified number of bytes.
+
+    Enables streaming a slice of a file for upload without loading the
+    entire slice into memory. The __len__ method lets requests determine
+    the correct Content-Length for the upload.
+    """
+
+    def __init__(self, file_obj, length: int):
+        self._file = file_obj
+        self._length = length
+
+    def read(self, size: int = -1) -> bytes:
+        if size is None or size < 0:
+            size = self._length
+        else:
+            size = min(size, self._length)
+        data = self._file.read(size)
+        self._length -= len(data)
+        return data
+
+    def __len__(self) -> int:
+        return self._length
+
+
+class InvenioRequestError(Exception):
+    """Raised when an Invenio API request returns an unexpected status code."""
+
+    def __init__(self, message: str, status_code: int):
+        super().__init__(message)
+        self.status_code = status_code
+
+
 class InvenioRecord(TypedDict):
    id: str
    title: str
@@ -304,12 +393,13 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor):
    ) -> dict[str, Any]:
        today = datetime.date.today().isoformat()
        creator = self._get_creator_from_public_name(public_name)
+        resource_type_id = context.config.default_resource_type or "dataset"
        create_record_request = {
            "files": {"enabled": True},
            "metadata": {
                "title": title,
                "publication_date": today,
-                "resource_type": {"id": "dataset"},
+                "resource_type": {"id": resource_type_id},
                "creators": [
                    creator,
                ],
@@ -330,6 +420,35 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor):
        file_path: str,
        context: FilesSourceRuntimeContext[RDMFileSourceConfiguration],
    ):
+        file_size = os.path.getsize(file_path)
+
+        threshold_mb = context.config.multipart_threshold
+        # Convert threshold from MB to bytes (config value is always in MB)
+        threshold_bytes = threshold_mb * 1024 * 1024 if threshold_mb else None
+        use_multipart = file_size >= threshold_bytes if threshold_bytes else False
+        if use_multipart:
+            self._upload_file_multipart(record_id, filename, file_path, file_size, context)
+        else:
+            try:
+                self._upload_file_single(record_id, filename, file_path, context)
+            except InvenioRequestError as e:
+                if e.status_code == 413:
+                    raise Exception(
+                        f"Failed to upload file '{filename}' ({file_size} bytes): HTTP 413 Payload Too Large. "
+                        f"The server rejected the upload because the file is too large for a single request. "
+                        f"Please configure 'multipart_threshold' in the file source configuration to enable multipart upload for files of this size."
+                    ) from e
+                raise
+
+    def _upload_file_single(
+        self,
+        record_id: str,
+        filename: str,
+        file_path: str,
+        context: FilesSourceRuntimeContext[RDMFileSourceConfiguration],
+    ):
+        """Upload a file using single PUT request."""
+
        record = self._get_draft_record(record_id, context)
        upload_file_url = record["links"]["files"]
        headers = self._get_request_headers(context, auth_required=True)
@@ -351,6 +470,132 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor):
        response = requests.post(commit_file_upload_url, headers=headers)
        self._ensure_response_has_expected_status_code(response, 200)

+    def _upload_file_multipart(
+        self,
+        record_id: str,
+        filename: str,
+        file_path: str,
+        file_size: int,
+        context: FilesSourceRuntimeContext[RDMFileSourceConfiguration],
+    ):
+        """Upload a file using multipart upload.
+
+        Flow:
+        1. Calculate parts/part_size
+        2. POST with transfer metadata
+        3. Server returns links.parts[] with URL for each part
+        4. Upload parts (parallel for > 2 parts)
+        5. POST to commit URL
+        """
+        preferred_part_size_mb = context.config.multipart_chunk_size
+        # Convert chunk size from MB to bytes (config value is always in MB)
+        preferred_part_size = preferred_part_size_mb * 1024 * 1024 if preferred_part_size_mb else None
+        num_parts, part_size = calculate_multipart_params(file_size, preferred_part_size)
+
+        log.info(f"Multipart upload: {num_parts} parts of {part_size} bytes each for '{filename}'")
+
+        record = self._get_draft_record(record_id, context)
+        upload_file_url = record["links"]["files"]
+        headers = self._get_request_headers(context, auth_required=True)
+
+        file_metadata = {
+            "key": filename,
+            "size": file_size,
+            "transfer": {
+                "type": "M",
+                "parts": num_parts,
+                "part_size": part_size,
+            },
+        }
+        response = requests.post(upload_file_url, json=[file_metadata], headers=headers)
+        self._ensure_response_has_expected_status_code(response, 201)
+
+        # Get part upload URLs from response
+        entries = response.json()["entries"]
+        file_entry = next(entry for entry in entries if entry["key"] == filename)
+        commit_url = file_entry["links"]["commit"]
+        part_links = file_entry.get("links", {}).get("parts", [])
+
+        if len(part_links) != num_parts:
+            raise Exception(
+                f"Server returned {len(part_links)} part URLs but expected {num_parts} for file '{filename}'"
+            )
+
+        # Sort part links by part number to ensure correct ordering
+        part_links = sorted(part_links, key=lambda p: p.get("part", 0))
+        self._upload_parts(file_path, file_size, part_size, part_links, headers)
+        response = requests.post(commit_url, json={}, headers=headers)
+        self._ensure_response_has_expected_status_code(response, 200)
+        log.info(f"Multipart upload completed for '{filename}'")
+
+    def _upload_parts(
+        self,
+        file_path: str,
+        file_size: int,
+        part_size: int,
+        part_links: list[dict],
+        headers: dict,
+    ):
+        """Upload all parts in parallel using a thread pool."""
+        num_parts = len(part_links)
+        max_workers = min(4, num_parts)
+
+        with ThreadPoolExecutor(max_workers=max_workers) as executor:
+            futures = {}
+            for part_index, part_info in enumerate(part_links):
+                future = executor.submit(
+                    self._upload_single_part,
+                    file_path,
+                    file_size,
+                    part_size,
+                    part_index,
+                    part_info,
+                    headers,
+                )
+                futures[future] = part_index
+
+            for future in as_completed(futures):
+                part_index = futures[future]
+                try:
+                    future.result()
+                except Exception as e:
+                    log.error(f"Failed to upload part {part_index}: {e}")
+                    raise
+
+    def _upload_single_part(
+        self,
+        file_path: str,
+        file_size: int,
+        part_size: int,
+        part_index: int,
+        part_info: dict,
+        headers: dict,
+    ):
+        """Upload a single part of a multipart upload."""
+        part_url = part_info.get("url")
+        if not part_url:
+            raise Exception(f"No URL provided for part {part_index}")
+
+        start_byte = part_index * part_size
+        end_byte = min(start_byte + part_size, file_size)
+        part_content_length = end_byte - start_byte
+
+        log.debug(f"Uploading part {part_index}: bytes {start_byte}-{end_byte - 1} ({part_content_length} bytes)")
+
+        # Presigned S3 URLs are authenticated via query parameters; adding
+        # Authorization or other headers would invalidate the signature.
+        # Invenio API proxy URLs require the Authorization header.
+        # Check for both X-Amz-Signature (current) and Signature (legacy v2) query params.
+        is_presigned = "X-Amz-Signature" in part_url or "Signature=" in part_url
+        part_headers = None if is_presigned else headers
+
+        # Stream the file slice without loading the entire part into memory
+        with open(file_path, "rb") as f:
+            f.seek(start_byte)
+            reader = _LimitedFileReader(f, part_content_length)
+            response = requests.put(part_url, data=reader, headers=part_headers)
+            self._ensure_response_has_expected_status_code(response, 200)
+
    def download_file_from_container(
        self,
        container_id: str,
@@ -551,17 +796,23 @@ class InvenioRepositoryInteractor(RDMRepositoryInteractor):
            if response.status_code == 403:
                self._raise_auth_required()
            error_message = self._get_response_error_message(response)
-            raise Exception(
-                f"Request to {response.url} failed with status code {response.status_code}: {error_message}"
+            raise InvenioRequestError(
+                f"Request to {response.url} failed with status code {response.status_code}: {error_message}",
+                status_code=response.status_code,
            )

    def _raise_auth_required(self):
        raise AuthenticationRequired(
-            f"Please provide a personal access token in your user's preferences for '{self.plugin.label}'"
+            f"Access denied. Please make sure you have provided a personal access token in your user's preferences for '{self.plugin.label}'"
        )

    def _get_response_error_message(self, response):
-        response_json = response.json()
+        try:
+            response_json = response.json()
+        except Exception:
+            # Response is not JSON, return raw text or status info
+            return response.text or f"HTTP {response.status_code} error"
+
        error_message = response_json.get("message") if response.status_code == 400 else response.text
        errors = response_json.get("errors", [])
        for error in errors:
@@ -264,8 +264,15 @@ class HistoriesService(ServiceBase, ConsumesModelStores, ServesExportStores):
        from URL or File depending on the provided parameters in the payload.
        """
        copy_this_history_id = payload.history_id
-        if trans.anonymous and not copy_this_history_id:  # Copying/Importing histories is allowed for anonymous users
-            raise glx_exceptions.AuthenticationRequired("You need to be logged in to create histories.")
+        if trans.anonymous:
+            # Copying/Importing histories is allowed for anonymous users, but only when there is a
+            # galaxy_session to own the result. A request with neither an API key nor a galaxysession
+            # cookie (e.g. a raw API call) has nowhere to put the new history, so reject it instead of
+            # doing the (potentially expensive) copy and creating an unreachable, orphaned history.
+            if not copy_this_history_id:
+                raise glx_exceptions.AuthenticationRequired("You need to be logged in to create histories.")
+            if not trans.galaxy_session:
+                raise glx_exceptions.AuthenticationRequired("You need an active session to copy histories.")
        if trans.user and trans.user.bootstrap_admin_user:
            raise glx_exceptions.RealUserRequiredException("Only real users can create histories.")
        hist_name = None
@@ -70,8 +70,10 @@ log = logging.getLogger(__name__)
 # Map Galaxy workflow invocation states to WES states
 GALAXY_TO_WES_STATE = {
    "new": State.QUEUED,
+    "requires_materialization": State.INITIALIZING,
    "ready": State.INITIALIZING,
    "scheduled": State.RUNNING,
+    "completed": State.COMPLETE,
    "failed": State.EXECUTOR_ERROR,
    "cancelled": State.CANCELED,
    "cancelling": State.CANCELING,
@@ -480,25 +482,21 @@ class WesService(ServiceBase):
            workflow_uri = workflow_dict["workflow_uri"]
            encoded_workflow_id, instance = _parse_gxworkflow_uri(workflow_uri)

-            # Load the workflow from the database
+            # Load the workflow from the database, applying the same accessibility
+            # rules as a normal invocation (owned, shared, published, or admin).
            # by_stored_id=not instance means:
            # - False (instance=False) -> load StoredWorkflow (by_stored_id=True)
            # - True (instance=True) -> load Workflow (by_stored_id=False)
-            try:
-                stored_workflow = self._workflows_service._workflows_manager.get_stored_workflow(
-                    trans, encoded_workflow_id, by_stored_id=not instance
-                )
-            except Exception as e:
-                raise exceptions.ObjectNotFound(
-                    f"Workflow '{encoded_workflow_id}' not found or not accessible: {str(e)}"
-                )
-
-            # Validate user has access to this workflow
-            if stored_workflow.user_id != trans.user.id and not trans.user_is_admin:
-                raise exceptions.ItemAccessibilityException("You do not have access to this workflow")
+            stored_workflow = self._workflows_service._workflows_manager.get_stored_accessible_workflow(
+                trans, encoded_workflow_id, by_stored_id=not instance
+            )

            # Use the existing workflow directly - no need to create a new one
-            # Skip to step 5 (engine parameters and history)
+            if instance:
+                # The URI named a specific version - invoke that one, not the latest.
+                invoke_workflow_id = trans.security.decode_id(encoded_workflow_id)
+            else:
+                invoke_workflow_id = stored_workflow.id
        else:
            # Step 2: Determine/validate workflow type
            detected_type = _determine_workflow_type(workflow_dict)
@@ -525,6 +523,8 @@ class WesService(ServiceBase):
                source="WES API",
            )
            stored_workflow = created_workflow.stored_workflow
+            invoke_workflow_id = stored_workflow.id
+            instance = False

        # Step 5: Parse engine parameters and create/select history
        engine_params = {}
@@ -537,9 +537,10 @@ class WesService(ServiceBase):
        history = _get_or_create_history(trans, engine_params)

        # Step 6: Parse workflow parameters
-        invoke_params = {
+        invoke_params: dict[str, Any] = {
            "history_id": trans.security.encode_id(history.id),
            "inputs_by": "name",
+            "instance": instance,
        }

        if workflow_params:
@@ -559,7 +560,7 @@ class WesService(ServiceBase):
        invoke_payload = InvokeWorkflowPayload(**invoke_params)
        workflow_invocation_response = self._workflows_service.invoke_workflow(
            trans,
-            trans.security.encode_id(stored_workflow.id),
+            invoke_workflow_id,
            invoke_payload,
        )

@@ -180,10 +180,11 @@ class SessionRequestContext(WorkRequestContext):
        return self.galaxy_session

    def set_history(self, history):
-        if history and not history.deleted and self.galaxy_session:
-            self.galaxy_session.current_history = history
-        self.sa_session.add(self.galaxy_session)
-        self.sa_session.commit()
+        if self.galaxy_session:
+            if history and not history.deleted:
+                self.galaxy_session.current_history = history
+            self.sa_session.add(self.galaxy_session)
+            self.sa_session.commit()


 def proxy_work_context_for_history(
@@ -1,10 +1,14 @@
 import time
 from typing import ClassVar
 from unittest import SkipTest
+from urllib.parse import urljoin
 from uuid import uuid4

 import pytest
-from requests import put
+from requests import (
+    post,
+    put,
+)

 from galaxy.model.unittest_utils.store_fixtures import (
    history_model_store_dict,
@@ -638,6 +642,19 @@ class TestHistoriesApi(ApiTestCase, BaseHistories):
            }
            self.dataset_populator.import_history(import_data)

+    def test_anonymous_without_session_cannot_copy_published(self):
+        history_id = self.dataset_populator.new_history(name=f"for_copying_without_session_{uuid4()}")
+        self.dataset_populator.make_public(history_id)
+
+        # A request with neither an API key nor a galaxysession cookie has no session to own the copy,
+        # so copying a published history is rejected rather than doing the copy and orphaning it.
+        # Regression test for https://github.com/galaxyproject/galaxy/issues/23148.
+        copy_response = post(
+            urljoin(self.url, "api/histories"),
+            json={"history_id": history_id, "name": f"copied_without_session_{uuid4()}"},
+        )
+        self._assert_status_code_is(copy_response, 403)
+
    def test_publish_non_alphanumeric(self):
        history_name = "تاریخچه"
        history_id = self.dataset_populator.new_history(name=history_name)
@@ -385,22 +385,14 @@ steps:
                history_id=history_id,
            )

-            # Get status via WES API
-            status = self._get_run_status_validated(invocation_id)
+            # A successfully finished invocation (Galaxy state "completed") must map to
+            # the terminal WES state COMPLETE - not UNKNOWN - so clients can poll for done.
+            self.workflow_populator.wait_for_invocation_and_completion(invocation_id, assert_ok=True)
+            invocation = self.workflow_populator.get_invocation(invocation_id)
+            assert invocation["state"] == "completed"

-            # Verify state is properly mapped
-            # After waiting for completion, should be COMPLETE or similar
-            assert status["state"] in [
-                "QUEUED",
-                "INITIALIZING",
-                "RUNNING",
-                "PAUSED",
-                "COMPLETE",
-                "EXECUTOR_ERROR",
-                "SYSTEM_ERROR",
-                "CANCELED",
-                "CANCELING",
-            ]
+            status = self._get_run_status_validated(invocation_id)
+            assert status["state"] == "COMPLETE"

    def test_wes_error_handling_missing_workflow_type(self):
        """Test error handling when workflow_type is missing."""
@@ -527,16 +519,23 @@ steps:
            self._assert_status_code_is(response, 200)

    def test_wes_submit_run_with_gxworkflow_uri_with_instance_param(self):
-        """Test gxworkflow:// URI with instance=true parameter."""
+        """Test gxworkflow:// URI with instance=true runs that version, not the latest."""
        with self.dataset_populator.test_history() as history_id:
            dataset_id = self._get_test_dataset_id(history_id)

            # Upload a workflow to get its ID
            workflow_id = self._upload_yaml_workflow(WORKFLOW_SIMPLE)
-            latest_instance_id = self._latest_instance_id(workflow_id, history_id)
+            first_instance_id = self._latest_instance_id(workflow_id, history_id)
+
+            # Create a second version so the referenced instance is no longer the latest
+            workflow_object = self._download_workflow(workflow_id)
+            workflow_object["steps"]["1"]["annotation"] = "second version"
+            update_response = self.workflow_populator.update_workflow(workflow_id, workflow_object)
+            self._assert_status_code_is(update_response, 200)
+            assert self._latest_instance_id(workflow_id, history_id) != first_instance_id

            # Construct gxworkflow:// URI with instance=true
-            workflow_uri = f"gxworkflow://{latest_instance_id}?instance=true"
+            workflow_uri = f"gxworkflow://{first_instance_id}?instance=true"

            # Submit workflow using the gxworkflow:// URI
            data = {
@@ -552,6 +551,46 @@ steps:

            # Validate response
            assert run_id is not None
+            invocation = self.workflow_populator.get_invocation(run_id)
+            assert invocation["workflow_id"] == first_instance_id
+
+    def test_wes_submit_run_with_gxworkflow_uri_published_workflow(self):
+        """A published workflow owned by another user can be run via gxworkflow://."""
+        with self._different_user():
+            workflow_id = self._upload_yaml_workflow(WORKFLOW_SIMPLE)
+            self.workflow_populator.make_public(workflow_id)
+
+        with self.dataset_populator.test_history() as history_id:
+            dataset_id = self._get_test_dataset_id(history_id)
+
+            data = {
+                "workflow_type": "gx_workflow_ga",
+                "workflow_type_version": "v1",
+                "workflow_params": json.dumps({"input1": dataset_id}),
+                "workflow_url": f"gxworkflow://{workflow_id}",
+            }
+
+            response = self._wes_post("ga4gh/wes/v1/runs", data=data)
+            self._assert_status_code_is(response, 200)
+            assert response.json()["run_id"] is not None
+
+    def test_wes_submit_run_with_gxworkflow_uri_inaccessible_workflow(self):
+        """An unshared workflow owned by another user cannot be run via gxworkflow://."""
+        with self._different_user():
+            workflow_id = self._upload_yaml_workflow(WORKFLOW_SIMPLE)
+
+        with self.dataset_populator.test_history() as history_id:
+            dataset_id = self._get_test_dataset_id(history_id)
+
+            data = {
+                "workflow_type": "gx_workflow_ga",
+                "workflow_type_version": "v1",
+                "workflow_params": json.dumps({"input1": dataset_id}),
+                "workflow_url": f"gxworkflow://{workflow_id}",
+            }
+
+            response = self._wes_post("ga4gh/wes/v1/runs", data=data)
+            self._assert_status_code_is(response, 403)

    def test_wes_job_stdout_endpoint(self):
        """Test /api/jobs/{job_id}/stdout endpoint returns job stdout."""
@@ -1,4 +1,8 @@
 import tempfile
+from typing import (
+    Any,
+    cast,
+)

 from galaxy.datatypes.tabular import (
    MAX_DATA_LINES,
@@ -112,3 +116,18 @@ def test_tabular_column_types_override():
        assert dataset.metadata.columns == 6
        assert dataset.metadata.delimiter == "\t"
        assert not hasattr(dataset.metadata, "column_names")
+
+
+def test_tabular_display_peek_does_not_render_table_header():
+    dataset = MockDataset(id=1)
+    dataset_for_peek = cast(Any, dataset)
+    metadata = cast(Any, dataset.metadata)
+    dataset_for_peek.peek = "question_id\tcurator_name\nensembl-grab-q1\tLG\n"
+    metadata.columns = 2
+    metadata.delimiter = "\t"
+
+    html = Tabular().display_peek(dataset)  # type: ignore [arg-type]
+
+    assert "<th>" not in html
+    assert "<td>question_id</td>" in html
+    assert "<td>curator_name</td>" in html
@@ -0,0 +1,472 @@
+"""Unit tests for Invenio multipart upload functionality."""
+
+from unittest.mock import (
+    MagicMock,
+    patch,
+)
+
+import pytest
+
+from galaxy.files.sources.invenio import (
+    _LimitedFileReader,
+    calculate_multipart_params,
+    InvenioRepositoryInteractor,
+    InvenioRequestError,
+    MAX_UPLOAD_PART_SIZE,
+    MAX_UPLOAD_PARTS,
+    MIN_UPLOAD_PART_SIZE,
+)
+
+
+class TestCalculateMultipartParams:
+    """Tests for calculate_multipart_params function."""
+
+    def test_calculate_multipart_params_zero_byte(self):
+        """Zero-byte files return minimum part size."""
+        parts, part_size = calculate_multipart_params(0)
+        assert parts == 1
+        assert part_size == MIN_UPLOAD_PART_SIZE
+
+    def test_calculate_multipart_params_small_file(self):
+        """Files under 5 MiB should use minimum part size."""
+        # 2 MiB file
+        file_size = 2 * 1024 * 1024
+        parts, part_size = calculate_multipart_params(file_size)
+        assert parts == 1
+        assert part_size == MIN_UPLOAD_PART_SIZE
+
+    def test_calculate_multipart_params_medium_file(self):
+        """Files between 5 MiB and 10 MiB."""
+        # 7.5 MiB file
+        file_size = 7 * 1024 * 1024 + 512 * 1024
+        parts, part_size = calculate_multipart_params(file_size)
+        assert parts == 2
+        assert part_size == MIN_UPLOAD_PART_SIZE
+
+    def test_calculate_multipart_params_large_file(self):
+        """Large files requiring multiple parts."""
+        # 25 MiB file
+        file_size = 25 * 1024 * 1024
+        parts, part_size = calculate_multipart_params(file_size)
+        assert parts == 5
+        assert part_size == MIN_UPLOAD_PART_SIZE
+
+    def test_calculate_multipart_params_respects_max_parts(self):
+        """Very large files should not exceed MAX_UPLOAD_PARTS."""
+        # File larger than MAX_UPLOAD_PARTS * MIN_UPLOAD_PART_SIZE
+        file_size = MAX_UPLOAD_PARTS * MIN_UPLOAD_PART_SIZE + MIN_UPLOAD_PART_SIZE
+        parts, part_size = calculate_multipart_params(file_size)
+        assert parts <= MAX_UPLOAD_PARTS
+        assert part_size >= MIN_UPLOAD_PART_SIZE
+
+    def test_calculate_multipart_params_extremely_large_file(self):
+        """Files larger than the maximum uploadable size raise ValueError instead of truncating."""
+        # 100 TiB file - exceeds theoretical s3 maximum (~48.8 TiB)
+        file_size = 100 * 1024**4
+        with pytest.raises(ValueError, match="exceeds the maximum multipart upload size"):
+            calculate_multipart_params(file_size)
+
+    def test_calculate_multipart_params_raises_at_max_boundary(self):
+        """One byte over the maximum uploadable size raises; exactly at the max does not."""
+        max_upload_size = MAX_UPLOAD_PARTS * MAX_UPLOAD_PART_SIZE
+        # Exactly at the boundary is allowed (parts == MAX_UPLOAD_PARTS at MAX part size).
+        parts, part_size = calculate_multipart_params(max_upload_size)
+        assert parts == MAX_UPLOAD_PARTS
+        assert part_size == MAX_UPLOAD_PART_SIZE
+        # One byte over raises.
+        with pytest.raises(ValueError):
+            calculate_multipart_params(max_upload_size + 1)
+
+    def test_calculate_multipart_params_respects_preferred_part_size(self):
+        """Should use preferred part size when provided and valid."""
+        # 150 MiB file with 100 MiB preferred part size
+        file_size = 150 * 1024 * 1024
+        preferred_part_size = 100 * 1024 * 1024  # 100 MiB
+        parts, part_size = calculate_multipart_params(file_size, preferred_part_size)
+        assert parts == 2
+        assert part_size == preferred_part_size
+
+    def test_calculate_multipart_params_preferred_too_small(self):
+        """Should use minimum part size if preferred is too small."""
+        # 100 MiB file with 1 MiB preferred part size (too small)
+        file_size = 100 * 1024 * 1024
+        preferred_part_size = 1 * 1024 * 1024  # 1 MiB - too small
+        parts, part_size = calculate_multipart_params(file_size, preferred_part_size)
+        assert part_size == MIN_UPLOAD_PART_SIZE  # Should be bumped to minimum
+
+    def test_calculate_multipart_params_preferred_exceeds_max(self):
+        """Should cap at MAX_UPLOAD_PART_SIZE if preferred exceeds it."""
+        # Small file with huge preferred part size
+        file_size = 100 * 1024 * 1024
+        preferred_part_size = MAX_UPLOAD_PART_SIZE * 2  # Exceeds max
+        parts, part_size = calculate_multipart_params(file_size, preferred_part_size)
+        assert parts == 1
+        assert part_size == MAX_UPLOAD_PART_SIZE
+
+    def test_calculate_multipart_params_exact_multiple(self):
+        """File size that's an exact multiple of part size."""
+        # Exactly 3 * MIN_UPLOAD_PART_SIZE
+        file_size = 3 * MIN_UPLOAD_PART_SIZE
+        parts, part_size = calculate_multipart_params(file_size)
+        assert parts == 3
+        assert part_size == MIN_UPLOAD_PART_SIZE
+
+    def test_calculate_multipart_params_one_byte_over(self):
+        """File size one byte over an exact multiple."""
+        # 3 * MIN_UPLOAD_PART_SIZE + 1 byte
+        file_size = 3 * MIN_UPLOAD_PART_SIZE + 1
+        parts, part_size = calculate_multipart_params(file_size)
+        assert parts == 4  # Need 4 parts for 3 full + 1 byte
+        assert part_size == MIN_UPLOAD_PART_SIZE
+
+    def test_calculate_multipart_params_at_boundary(self):
+        """Test file at MAX_UPLOAD_PARTS boundary."""
+        # Exactly at the boundary where we need to increase part size
+        file_size = (MAX_UPLOAD_PARTS + 1) * MIN_UPLOAD_PART_SIZE
+        parts, part_size = calculate_multipart_params(file_size)
+        assert parts <= MAX_UPLOAD_PARTS
+        # Part size should have increased
+        assert part_size > MIN_UPLOAD_PART_SIZE
+
+
+def _mock_response(status_code=200, json_data=None, text=""):
+    """Create a mock requests.Response object."""
+    response = MagicMock()
+    response.status_code = status_code
+    response.text = text
+    if json_data is not None:
+        response.json.return_value = json_data
+    return response
+
+
+def _make_interactor():
+    """Create an InvenioRepositoryInteractor with a minimal mock plugin."""
+    plugin = MagicMock()
+    plugin.label = "test-invenio"
+    plugin.get_uri_root.return_value = "inveniordm://test"
+    return InvenioRepositoryInteractor("https://invenio.example.org", plugin)
+
+
+def _make_context(config=None):
+    """Create a mock runtime context with the given config."""
+    context = MagicMock()
+    context.config = config or MagicMock(
+        multipart_threshold=None,
+        multipart_chunk_size=None,
+        default_resource_type=None,
+        token="test-token",
+        public_name="Doe, Jane",
+    )
+    return context
+
+
+def _make_draft_record():
+    """Standard draft record response with files link."""
+    return {"links": {"files": "https://invenio.example.org/api/records/abc/files"}}
+
+
+def _make_single_upload_entry():
+    """File entry returned after POST for single upload."""
+    return {
+        "key": "test.txt",
+        "links": {
+            "content": "https://invenio.example.org/api/records/abc/files/content",
+            "commit": "https://invenio.example.org/api/records/abc/files/commit",
+        },
+    }
+
+
+def _make_multipart_entries(num_parts):
+    """File entry returned after POST for multipart upload, with part links."""
+    parts = [{"part": i, "url": f"https://invenio.example.org/parts/{i}"} for i in range(num_parts)]
+    return {
+        "key": "test.txt",
+        "links": {
+            "commit": "https://invenio.example.org/api/records/abc/files/commit",
+            "self": "https://invenio.example.org/api/records/abc/files/test.txt",
+            "parts": parts,
+        },
+    }
+
+
+class TestUploadFileSingle:
+    """Tests for _upload_file_single."""
+
+    def test_upload_file_single_success(self, tmp_path):
+        """Verify POST (metadata), PUT (content), POST (commit) sequence."""
+        file_path = tmp_path / "test.txt"
+        file_path.write_bytes(b"hello world")
+
+        interactor = _make_interactor()
+        context = _make_context()
+
+        with (
+            patch.object(interactor, "_get_draft_record", return_value=_make_draft_record()),
+            patch.object(interactor, "_get_request_headers", return_value={"Authorization": "Bearer x"}),
+            patch("galaxy.files.sources.invenio.requests") as mock_requests,
+        ):
+            mock_requests.post.side_effect = [
+                _mock_response(201, {"entries": [_make_single_upload_entry()]}),  # metadata
+                _mock_response(200),  # commit
+            ]
+            mock_requests.put.return_value = _mock_response(200)
+
+            interactor._upload_file_single("abc", "test.txt", str(file_path), context)
+
+            assert mock_requests.post.call_count == 2
+            assert mock_requests.put.call_count == 1
+
+    def test_upload_file_single_propagates_413_as_typed_error(self, tmp_path):
+        """A 413 from the content PUT surfaces as an InvenioRequestError carrying the status code."""
+        file_path = tmp_path / "test.txt"
+        file_path.write_bytes(b"hello world")
+
+        interactor = _make_interactor()
+        context = _make_context()
+
+        with (
+            patch.object(interactor, "_get_draft_record", return_value=_make_draft_record()),
+            patch.object(interactor, "_get_request_headers", return_value={"Authorization": "Bearer x"}),
+            patch("galaxy.files.sources.invenio.requests") as mock_requests,
+        ):
+            mock_requests.post.return_value = _mock_response(201, {"entries": [_make_single_upload_entry()]})
+            mock_requests.put.return_value = _mock_response(413)
+
+            with pytest.raises(InvenioRequestError) as exc_info:
+                interactor._upload_file_single("abc", "test.txt", str(file_path), context)
+
+            assert exc_info.value.status_code == 413
+
+
+class TestUploadFileMultipart:
+    """Tests for _upload_file_multipart."""
+
+    def test_upload_file_multipart_success(self, tmp_path):
+        """Verify transfer metadata, part uploads, and commit call."""
+        file_size = 3 * MIN_UPLOAD_PART_SIZE
+        file_path = tmp_path / "test.txt"
+        file_path.write_bytes(b"x" * file_size)
+
+        interactor = _make_interactor()
+        context = _make_context(
+            config=MagicMock(
+                multipart_threshold=1,
+                multipart_chunk_size=None,
+                default_resource_type=None,
+                token="test-token",
+                public_name="Doe, Jane",
+            )
+        )
+
+        entries = _make_multipart_entries(3)
+
+        with (
+            patch.object(interactor, "_get_draft_record", return_value=_make_draft_record()),
+            patch.object(interactor, "_get_request_headers", return_value={"Authorization": "Bearer x"}),
+            patch.object(interactor, "_upload_single_part") as mock_upload_part,
+            patch("galaxy.files.sources.invenio.requests") as mock_requests,
+        ):
+            mock_requests.post.side_effect = [
+                _mock_response(201, {"entries": [entries]}),  # initial POST
+                _mock_response(200),  # commit POST
+            ]
+
+            interactor._upload_file_multipart("abc", "test.txt", str(file_path), file_size, context)
+
+            # Verify transfer metadata in the initial POST
+            initial_post_args = mock_requests.post.call_args_list[0]
+            sent_metadata = initial_post_args.kwargs["json"][0]
+            assert sent_metadata["key"] == "test.txt"
+            assert sent_metadata["size"] == file_size
+            assert sent_metadata["transfer"]["type"] == "M"
+            assert sent_metadata["transfer"]["parts"] == 3
+            assert sent_metadata["transfer"]["part_size"] == MIN_UPLOAD_PART_SIZE
+
+            # All 3 parts uploaded
+            assert mock_upload_part.call_count == 3
+            # Commit called (second POST)
+            assert mock_requests.post.call_count == 2
+
+    def test_upload_file_multipart_wrong_part_count_from_server(self, tmp_path):
+        """Server returning fewer part links than expected should raise."""
+        file_size = 3 * MIN_UPLOAD_PART_SIZE
+        file_path = tmp_path / "test.txt"
+        file_path.write_bytes(b"x" * file_size)
+
+        interactor = _make_interactor()
+        context = _make_context(
+            config=MagicMock(
+                multipart_threshold=1,
+                multipart_chunk_size=None,
+                default_resource_type=None,
+                token="test-token",
+                public_name="Doe, Jane",
+            )
+        )
+
+        # Server returns only 2 part links instead of 3
+        entries = _make_multipart_entries(2)
+
+        with (
+            patch.object(interactor, "_get_draft_record", return_value=_make_draft_record()),
+            patch.object(interactor, "_get_request_headers", return_value={"Authorization": "Bearer x"}),
+            patch("galaxy.files.sources.invenio.requests") as mock_requests,
+        ):
+            mock_requests.post.return_value = _mock_response(201, {"entries": [entries]})
+
+            with pytest.raises(Exception, match="2 part URLs"):
+                interactor._upload_file_multipart("abc", "test.txt", str(file_path), file_size, context)
+
+
+class TestUploadSinglePart:
+    """Tests for _upload_single_part byte-range streaming."""
+
+    def test_upload_single_part_streams_correct_byte_range(self, tmp_path):
+        """Verify the uploaded data matches the expected byte slice."""
+        data = b"0123456789ABCDEF" * 4  # 64 bytes
+        file_path = tmp_path / "test.txt"
+        file_path.write_bytes(data)
+
+        interactor = _make_interactor()
+        part_size = 16
+        part_info = {"part": 2, "url": "https://invenio.example.org/parts/2"}
+
+        captured_data = bytearray()
+
+        def fake_put(url, data=None, **kwargs):
+            captured_data.extend(data.read())
+            return _mock_response(200)
+
+        with patch("galaxy.files.sources.invenio.requests") as mock_requests:
+            mock_requests.put.side_effect = fake_put
+
+            interactor._upload_single_part(str(file_path), len(data), part_size, 2, part_info, {})
+
+            # Part 2 = bytes 32-48
+            assert bytes(captured_data) == data[32:48]
+
+
+class TestLimitedFileReader:
+    """Tests for _LimitedFileReader streaming wrapper."""
+
+    def test_read_limited_bytes(self, tmp_path):
+        """Reader should only return the specified number of bytes."""
+        file_path = tmp_path / "test.txt"
+        file_path.write_bytes(b"0123456789")
+
+        with open(file_path, "rb") as f:
+            f.seek(3)
+            reader = _LimitedFileReader(f, 4)
+            assert reader.read() == b"3456"
+
+    def test_read_in_chunks(self, tmp_path):
+        """Reader should serve data in small chunks without over-reading."""
+        file_path = tmp_path / "test.txt"
+        file_path.write_bytes(b"0123456789")
+
+        with open(file_path, "rb") as f:
+            f.seek(2)
+            reader = _LimitedFileReader(f, 5)
+            assert reader.read(2) == b"23"
+            assert reader.read(2) == b"45"
+            assert reader.read(2) == b"6"  # only 1 byte remaining
+            assert reader.read(2) == b""  # now exhausted
+
+    def test_len_reports_remaining(self, tmp_path):
+        """__len__ should report remaining bytes."""
+        file_path = tmp_path / "test.txt"
+        file_path.write_bytes(b"0123456789")
+
+        with open(file_path, "rb") as f:
+            f.seek(1)
+            reader = _LimitedFileReader(f, 5)
+            assert len(reader) == 5
+            reader.read(2)
+            assert len(reader) == 3
+
+
+class TestUploadFileToDraftContainerRouting:
+    """Tests for upload_file_to_draft_container threshold routing."""
+
+    def test_routes_to_single_when_below_threshold(self, tmp_path):
+        """Files below threshold should use single upload."""
+        file_path = tmp_path / "test.txt"
+        file_path.write_bytes(b"small")
+
+        interactor = _make_interactor()
+        context = _make_context(
+            config=MagicMock(
+                multipart_threshold=100,
+                multipart_chunk_size=None,
+                default_resource_type=None,
+                token="test-token",
+                public_name="Doe, Jane",
+            )
+        )
+
+        with patch.object(interactor, "_upload_file_single") as mock_single:
+            interactor.upload_file_to_draft_container("abc", "test.txt", str(file_path), context)
+            mock_single.assert_called_once()
+
+    def test_routes_to_multipart_when_above_threshold(self, tmp_path):
+        """Files at or above threshold should use multipart upload."""
+        file_size = 2 * 1024 * 1024  # 2 MiB
+        file_path = tmp_path / "test.txt"
+        file_path.write_bytes(b"x" * file_size)
+
+        interactor = _make_interactor()
+        context = _make_context(
+            config=MagicMock(
+                multipart_threshold=1,  # 1 MB threshold, file is 2 MB
+                multipart_chunk_size=None,
+                default_resource_type=None,
+                token="test-token",
+                public_name="Doe, Jane",
+            )
+        )
+
+        with patch.object(interactor, "_upload_file_multipart") as mock_multipart:
+            interactor.upload_file_to_draft_container("abc", "test.txt", str(file_path), context)
+            mock_multipart.assert_called_once()
+            # Verify file_size is passed through
+            assert mock_multipart.call_args[0][3] == file_size
+
+    def test_routes_to_single_when_no_threshold(self, tmp_path):
+        """No threshold configured should always use single upload."""
+        file_size = 100 * 1024 * 1024  # 100 MiB
+        file_path = tmp_path / "test.txt"
+        file_path.write_bytes(b"x" * file_size)
+
+        interactor = _make_interactor()
+        context = _make_context()
+
+        with patch.object(interactor, "_upload_file_single") as mock_single:
+            interactor.upload_file_to_draft_container("abc", "test.txt", str(file_path), context)
+            mock_single.assert_called_once()
+
+    def test_upload_file_single_413_router_raises_helpful_error(self, tmp_path):
+        """Router should wrap 413 with actionable multipart_threshold message."""
+        file_path = tmp_path / "test.txt"
+        file_path.write_bytes(b"hello world")
+
+        interactor = _make_interactor()
+        context = _make_context(
+            config=MagicMock(
+                multipart_threshold=None,  # No threshold configured
+                multipart_chunk_size=None,
+                default_resource_type=None,
+                token="test-token",
+                public_name="Doe, Jane",
+            )
+        )
+
+        with (
+            patch.object(interactor, "_get_draft_record", return_value=_make_draft_record()),
+            patch.object(interactor, "_get_request_headers", return_value={"Authorization": "Bearer x"}),
+            patch("galaxy.files.sources.invenio.requests") as mock_requests,
+        ):
+            mock_requests.post.return_value = _mock_response(201, {"entries": [_make_single_upload_entry()]})
+            mock_requests.put.return_value = _mock_response(413)
+
+            with pytest.raises(Exception, match="multipart_threshold"):
+                interactor.upload_file_to_draft_container("abc", "test.txt", str(file_path), context)