Report why a job failed to Galaxy so job resubmission works for Pulsar destinations

> **Posted by Claude (AI assistant) on jmchilton's behalf.** They did not author this text
> personally; it is the output of a research pass over `galaxyproject/galaxy@dev`,
> `galaxyproject/pulsar@master`, and the in-flight #485 / #486 branches. Treat the file/line
> references as a starting point to verify rather than as settled fact.

Design plan for the long-standing gap behind #75 and #408: Pulsar cannot tell Galaxy *why* a job
failed, so Galaxy's job-resubmission conditions never fire for Pulsar jobs. Filed as an issue
rather than a PR so it can be argued about before any code moves.

---

# Pulsar → Galaxy failure-reason plan

Refs: [pulsar#75](https://github.com/galaxyproject/pulsar/issues/75), [pulsar#408](https://github.com/galaxyproject/pulsar/issues/408), pulsar#485, pulsar#486, galaxy branch `jmchilton/galaxy:htcondor_pulsar`.

## Problem

Pulsar's wire vocabulary is exactly the eight strings in `pulsar/managers/status.py`
(`preprocessing/queued/running/complete/cancelled/failed/postprocessing/lost`). There is no
field for *why* a job ended. Consequences, precisely:

1. **Limit kills are invisible.** `pulsar/managers/base/base_drmaa.py` maps
`JobState.FAILED → status.COMPLETE` on master, so a Slurm job killed for walltime is
reported to Galaxy as a **success** (natefoo, pulsar#408). pulsar#485 already fixes this
half — see Sequencing.
2. **Even a correctly-reported failure is uncategorised.** `lib/galaxy/jobs/runners/pulsar.py`
sets no `runner_state` anywhere — it is the only major runner that doesn't. `slurm.py:141`,
`:157`, `:171`, `cli.py:240`, `htcondor.py:695/729/736`, `kubernetes.py:824/845/850`, and
`univa.py:112/120` all do.
3. Therefore `resubmit.failure()`
(`lib/galaxy/jobs/runners/state_handlers/resubmit.py`) always sees the fallback
`runner_state = UNKNOWN_ERROR`. `condition="unknown_error"` and `any_failure` *do* fire for
Pulsar today (that is what `TestJobResubmissionPulsarIntegration` covers — a connection
failure). What can never fire is `walltime_reached`, `memory_limit_reached`,
`tool_detected_failure`, `tool_timelimit_reached`. That is the actual gap.

Pulsar already owns half the answer and drops it on the floor: `pulsar/managers/util/__init__.py`
defines a `runner_states` Bunch mirrored from Galaxy, only `util/cli/job/lsf.py` ever returns one,
and nothing transports it. pulsar#486 adds real HTCondor hold/failure classification
(`queued_htcondor.__held_status`, `util/condor/htcondor.classify_hold` /
`classify_failure_event`) that likewise dead-ends inside the manager.

## Recommended approach

**Add one optional, absent-tolerant `runner_state` string (plus an optional free-text
`runner_state_message`) to the terminal-status payload, emitted from the single existing choke
point `pulsar/manager_endpoint_util.py::full_status()`, sourced from job-directory metadata
written by whichever manager detected the condition. Galaxy's `PulsarJobRunner` copies it onto
`AsynchronousJobState.runner_state` before calling `fail_job()`, which already dispatches
`resubmit.failure`. The vocabulary is the `runner_states` Bunch that is *already* mirrored into
both trees — no new vocabulary. Classification logic is shared by extending the byte-for-byte
mirrored `util/` tree that pulsar#486 established, extracting Galaxy's existing Slurm and CLI
logic into it rather than writing parallel Pulsar code.**

Why this and not the alternatives:

- **Not new `status.py` values** (`failed_walltime`, …). Those strings are consumed by
`status.is_job_done()`, `StatefulManagerProxy.__proxy_status`, `tes_state_to_pulsar_status`,
and — critically — by *old Galaxy*, which string-matches `["complete","cancelled"]` /
`["failed","lost"]` in `_update_job_state_for_status`. A new status value would fall through
every branch and leave the job watched forever. An orthogonal field degrades to a no-op in both
directions.
- **Not a free-text reason only.** Galaxy needs a controlled token to `safe_eval` resubmit
conditions against.
- **Not Galaxy-side re-derivation** from the returned `job_stderr`. It works for the Slurm
cgroup message and nothing else — Galaxy cannot run `scontrol` or read a `HoldReasonCode` on
the remote cluster — and it is exactly the duplication we're trying to remove.

### Reuse posture (this is a mature codebase; do not invent a framework)

Three abstractions already exist and cover every manager. Only one genuinely new shared module
is required (Slurm).

| Need | Existing abstraction | Work |
|---|---|---|
| Vocabulary | `runner_states` Bunch, already mirrored `galaxy/jobs/runners/util/__init__.py` ↔ `pulsar/managers/util/__init__.py` | Close the drift: Pulsar's copy is missing `TOOL_TIMELIMIT_REACHED` |
| CLI DRMs (LSF, Slurm-CLI, Torque, PBS) | `util/cli/job/*.py` `get_failure_reason()` / `parse_failure_reason()` — already mirrored, already returns `runner_states`, already consumed by `cli.py.__handle_job_failure_reasons` | `queued_cli` just has to call it |
| HTCondor | `util/condor/htcondor.py` `classify_hold()` / `classify_failure_event()` + `HOLD_MESSAGES` / `FAILURE_MESSAGES` (pulsar#486 + galaxy `9d9f636dc9`) | Add a small `→ runner_states` map inside the shared module |
| Slurm/DRMAA | *nothing shared* — logic is inline and private in `lib/galaxy/jobs/runners/slurm.py` | **Extract** (below) |

The mirrored-module trick that keeps the files byte-identical is the **relative import**:
`util/condor/htcondor.py` says `from .. import runner_states`, which resolves to
`galaxy.jobs.runners.util` in one tree and `pulsar.managers.util` in the other. `util/cli/job/lsf.py`
already uses `from ... import runner_states` for the same reason. Keep doing that.

### What to extract from `lib/galaxy/jobs/runners/slurm.py`

New `lib/galaxy/jobs/runners/util/slurm/__init__.py`, mirrored byte-for-byte to
`pulsar/managers/util/slurm/__init__.py`:

- `slurm_job_state(job_id, cluster=None, execute=None) -> str` — the `scontrol -o show job`
parse (including the `Comment=** … **`-with-spaces handling at `slurm.py:88-98`) with the
`sacct -n -o state%-32` fallback on `slurm_load_jobs error: Invalid job id specified`, and the
`"SLURM accounting storage is disabled"` special case. Returns `"NOT_FOUND"` when neither
knows. `execute` defaults to `galaxy.util.commands.execute` and is injectable for tests.
- `check_memory_limit(efile_path) -> str | None` — the 2K stderr tail scan
(`slurm.py.__check_memory_limit`) plus `SLURM_MEMORY_LIMIT_EXCEEDED_MSG` and
`SLURM_MEMORY_LIMIT_EXCEEDED_PARTIAL_WARNINGS`.
- `classify_slurm_state(slurm_state, error_file=None) -> tuple[str | None, str | None]` —
returns `(runner_state, detail)`; `TIMEOUT → WALLTIME_REACHED`,
`OUT_OF_MEMORY → MEMORY_LIMIT_REACHED`, `CANCELLED` + memory message → `MEMORY_LIMIT_REACHED`,
otherwise `(None, None)`.

Deliberately **not** extracted (they are Galaxy job-model concerns, and Pulsar wants different
answers): the `COMPLETING` exponential-backoff wait loop, the `NOT_FOUND`/`COMPLETED` →
`drmaa_job_states.DONE` reinterpretation, `NODE_FAIL` → `mark_as_resubmitted`, the
`model.Job.states.STOPPED` check, `PENDING`/`RUNNING` → return-to-monitor, and the user-facing
prose (`OUT_OF_MEMORY_MSG` etc.).

`galaxy-util>=23.0` is already a `pulsar-app` requirement (`requirements.txt:7`), and
`pulsar/managers/util/__init__.py` already does `from galaxy.util.bunch import Bunch`, so the
shared module may use `galaxy.util.commands`. The "no hard dependency on Galaxy" constraint means
the Galaxy *application*, not the `galaxy-util` PyPI package.

`slurm.py` has **zero unit tests** today. Extraction is the opportunity to add them.

Optional, separate: Galaxy repeats the same user-facing prose four times
(`slurm.py:OUT_OF_MEMORY_MSG`, `cli.py:224-234` `jobstate_map`, `htcondor.py:_MEMORY_LIMIT_HOLD_MSG`
/ `_WALLTIME_HOLD_MSG`, `kubernetes.py:844-850`). A `RUNNER_STATE_MESSAGES` dict next to
`JobState.runner_states` in `lib/galaxy/jobs/runners/__init__.py` would de-duplicate it. Keep it
out of the mirrored tree — Pulsar should not own Galaxy's user-facing copy — and out of this
plan's critical path, because the strings differ slightly and tool tests may assert on them.

---

## Wire protocol — exactly where

**Every** server→Galaxy status path funnels through one function. Verified:

| Transport | Entry point | Reaches |
|---|---|---|
| HTTP REST | `pulsar/web/routes.py:89` `@PulsarController(path="/jobs/{job_id}/status")` → `manager_endpoint_util.status_dict()` | `full_status()` |
| AMQP | `pulsar/messaging/bind_amqp.py` `bind_on_status_change()` → published to `"status_update"` | `full_status()` |
| Relay | `pulsar/messaging/bind_relay.py:138` | `full_status()` |

`full_status()` returns the rich `__job_complete_dict()` only when `status.is_job_done(job_status)`;
otherwise a three-key stub. The reason is only meaningful on terminal states, so **the single
producer edit is `__job_complete_dict()` in `pulsar/manager_endpoint_util.py`.**

Galaxy consumes the payload at:

| Transport | Galaxy entry | Has the dict today? |
|---|---|---|
| AMQP / relay | `pulsar.py:1092 __async_update(full_status)` → `_update_job_state_for_status(..., full_status=full_status)` | **yes** — zero client change needed |
| REST polling | `pulsar.py:348 check_watched_item_state()` → `client.get_status()` → `_update_job_state_for_status(job_state, status)` | **no** — `full_status` is `None` |

`JobClient.get_status()` (`pulsar/client/client.py:249`) already calls `raw_check_complete()` and
throws away everything but `["status"]`, so surfacing the dict on the polling path costs **zero
extra requests**.

---

## Phases

### Phase 0 — Galaxy-only, independent, mergeable today

No Pulsar dependency, no wire change.

**0a. `resubmit.py` KeyError.** `failure()` admits
`JobState.runner_states.JOB_OUTPUT_NOT_RETURNED_FROM_CLUSTER` through its filter, but
`MESSAGES` (`resubmit.py:19-25`) has no key for it — and `runners/__init__.py:987` sets exactly
that state. `_handle_resubmit_definitions` then raises `KeyError` inside
`_handle_runner_state`'s `try/except`, so the resubmission is silently swallowed and only shows
as *"Caught exception in runner state handler"*. Add the missing `MESSAGES` entry and the
matching `_ExpressionContext` variable, or drop the state from the filter tuple. Pick one; adding
is the safer read of intent.
*Test:* `test/unit/app/jobs/test_resubmit.py` (new) — table over every state in `failure()`'s
filter tuple asserting `MESSAGES[state]` resolves and `_ExpressionContext` exposes a variable.
Red before the fix.

**0b. Pulsar `finish_job` never consults the resubmit handler.** `AsynchronousJobRunner.finish_job`
routes through `_finish_or_resubmit_job` (`runners/__init__.py:637`), which sets
`TOOL_DETECT_ERROR` / `MEMORY_LIMIT_REACHED` from `check_output_detected_state` and calls
`_handle_runner_state("failure", …)`. `PulsarJobRunner.finish_job` (`pulsar.py:701`) overrides it
completely and calls `job_wrapper.finish(...)` directly — and `job_wrapper.finish` has no
`_handle_runner_state` call (verified: `lib/galaxy/jobs/__init__.py` has none). So
`condition="tool_detected_failure"` and detected-OOM resubmission are dead for Pulsar
*independently of any wire work*. Add the same `check_tool_output` → `runner_state` →
`_handle_runner_state` step to `PulsarJobRunner.finish_job`.
*Test:* extend `test/integration/test_job_resubmission.py` with an embedded-Pulsar variant of
`TestJobResubmissionToolDetectedErrorResubmitsIntegration` (`exit_code_from_env`) using a new
`resubmission_embedded_pulsar_job_conf.yml`. This one needs **no** new Pulsar release.

### Phase 1 — Pulsar server: record and emit

1. `pulsar/managers/util/__init__.py` — add `TOOL_TIMELIMIT_REACHED="tool_timelimit_reached"` to
close the mirror drift with Galaxy's copy (that is the only real difference; the rest is
black/isort whitespace, which should also be reconciled so future mirrors stay diff-clean).
2. `pulsar/managers/base/__init__.py` — `JOB_FILE_RUNNER_STATE = "runner_state"` and, on
`DirectoryBaseManager`:
```python
def _record_runner_state(self, job_id, runner_state, message=None):
# store a dict so a detail string can ride along without a second file
def runner_state(self, job_id):
# -> dict | None, from job-directory metadata
```
Metadata lives in the job directory (`store_metadata`/`load_metadata`,
`managers/base/__init__.py:362-375`), so it survives a Pulsar restart for free — same
mechanism as `JOB_FILE_EXTERNAL_ID` and `final_status`.
3. `pulsar/managers/__init__.py` — add **non-abstract** `ManagerInterface.runner_state(job_id)`
returning `None`, and a `ManagerProxy.runner_state` delegate. Non-abstract so out-of-tree
managers keep importing. (pulsar#486 sets this precedent with its non-abstract
`ManagerInterface.shutdown`.)
4. `pulsar/manager_endpoint_util.py::__job_complete_dict` — the one wire edit:
```python
runner_state = manager.runner_state(job_id)
if runner_state and runner_state.get("runner_state"):
as_dict["runner_state"] = runner_state["runner_state"]
if runner_state.get("message"):
as_dict["runner_state_message"] = runner_state["message"]
```
Key **omitted** when unknown — never `null` — so old Galaxy and new Galaxy both read it the
same way.
5. `pulsar/managers/stateful.py` — no change. `StatefulManagerProxy` inherits the delegate.
Note the interaction: post-#485 a failed job still runs `__handle_postprocessing`, so outputs
come back before the `failed` callback fires, and the recorded runner_state is on disk by the
time `full_status()` runs. A *postprocessing* failure that overwrites `final_status` with
`FAILED` must not fabricate a runner_state — leaving it absent is correct (Galaxy falls back
to `UNKNOWN_ERROR`).

*Tests (red first):*
- `test/manager_endpoint_util_test.py` — extend with fake managers: key present when the manager
records one, key **absent** when it doesn't.
- `test/manager_test.py` — `_record_runner_state` / `runner_state` round-trip through a real job
directory, and survival across a fresh manager instance over the same persistence directory.
- `test/routes_test.py` — `/jobs/{id}/status` JSON carries the key.
- `test/amqp_test.py` — the published `status_update` payload carries it.

### Phase 2 — Pulsar client + Galaxy consumer

Pulsar (`pulsar/client/client.py`):
- `BaseJobClient.get_full_status()` → `{"status": self.get_status()}` (default; also covers
`MessageJobClient` and `RelayJobClient`, and the polling coexecution clients that define no
`get_status` at all).
- `JobClient.get_full_status()` → `self.raw_check_complete()`.
- `K8sPollingCoexecutionJobClient` / `GcpPollingCoexecutionJobClient` / TES → `raw_check_complete()`
(they already define it).

Galaxy (`lib/galaxy/jobs/runners/pulsar.py`):
- `check_watched_item_state` (`:348`): `full_status = client.get_full_status()`;
`status = full_status.get("status")`; pass both to `_update_job_state_for_status`.
- `_update_job_state_for_status` (`:362`), in the `pulsar_status in ["failed", "lost"]` branch,
before `self.fail_job(...)`:
- read `(full_status or {}).get("runner_state")`;
- **validate against a whitelist** of `JobState.runner_states` values Galaxy actually handles —
an unrecognised token from a newer Pulsar must be logged and ignored, not assigned;
- set `job_state.runner_state` and a `job_state.fail_message` (the whitelist's prose, appended
with `runner_state_message` when present).
`fail_job` (`runners/__init__.py:600`) already calls `_handle_runner_state("failure", job_state)`,
so nothing further is needed to reach `resubmit.failure`.
- Bump `pulsar-galaxy-lib` floor in `pyproject.toml:75` and `lib/galaxy/dependencies/pinned-requirements.txt:189`.

*Tests:*
- Pulsar `test/client_test.py` — `get_full_status()` returns the whole dict for `JobClient` and
the `{"status": …}` stub for the message clients.
- Galaxy `test/unit/app/jobs/test_pulsar_runner.py` (**new file** — the `htcondor_pulsar` branch
already introduces it, reuse that): drive `_update_job_state_for_status` with a fabricated
`full_status` and a stub `fail_job`. Cases: `runner_state` absent → attribute unset;
`walltime_reached` → set; `"nonsense_state"` → unset + warning logged. Pure unit test, no server.

### Phase 3 — Populate: `queued_htcondor` (depends on #486 landing)

In `pulsar/managers/util/condor/htcondor.py` (shared, so Galaxy gets it too) add:

```python
HOLD_REASON_TO_RUNNER_STATE = {
HOLD_REASON_MEMORY_LIMIT: runner_states.MEMORY_LIMIT_REACHED,
HOLD_REASON_WALLTIME: runner_states.WALLTIME_REACHED,
HOLD_REASON_OTHER: runner_states.UNKNOWN_ERROR,
}
FAILURE_TO_RUNNER_STATE = {...} # all → UNKNOWN_ERROR except EXECUTABLE_ERROR, which stays None
```

`pulsar/managers/queued_htcondor.py`:
- `__held_status` — on the memory/walltime branch, `self._record_runner_state(job_id, …, HOLD_MESSAGES[hold_reason])` before returning `status.FAILED`.
- `__summary_to_status` — `term_signal == SIGKILL` → `MEMORY_LIMIT_REACHED` + `SIGKILL_MESSAGE`
(matching what Galaxy's own `htcondor.py:695` already concludes); `failure_event` →
`FAILURE_TO_RUNNER_STATE[...]` + `FAILURE_MESSAGES[...]`; exhausted-hold and
missing-log/status-error escalations → `UNKNOWN_ERROR`.

`lib/galaxy/jobs/runners/htcondor.py` should then consume the same two maps instead of its
inline `runner_states.…` assignments at `:695/:729/:736/:752/:879-893` — that is the de-duplication
the whole mirrored-module exercise is for.

*Tests:* extend `test/manager_htcondor_test.py` (from #486). `test/htcondor_fake/htcondor2.py`
already lets a test emit a `JOB_HELD` event with a chosen `HoldReasonCode` and a `JOB_TERMINATED`
with `TermSignal=9` — **this is how you fake a DRM limit kill with no cluster**. Assert both the
returned `status.FAILED` *and* `manager.runner_state(job_id)`.

### Phase 4 — Populate: `queued_cli` (small, depends only on Phase 1)

`pulsar/managers/queued_cli.py` currently maps `job_states.ERROR → status.FAILED` and stops.
Override `get_status(job_id)` (rather than changing `ExternalBaseManager._get_status_external`'s
signature and breaking out-of-tree managers — `queued_htcondor` sets this precedent) so that on
`ERROR` it does what `cli.py.__handle_job_failure_reasons` does:

```python
cmd_out = shell.execute(job_interface.get_failure_reason(external_id))
reported = job_interface.parse_failure_reason(cmd_out.stdout, external_id)
```

Zero new abstraction; lights up LSF `TERM_MEMLIMIT` → `MEMORY_LIMIT_REACHED` immediately
(`util/cli/job/lsf.py:97`), and any future plugin that implements `parse_failure_reason` comes
along for free. `util/cli/job/__init__.py` already declares both methods on the base interface.

*Test:* new `test/manager_cli_test.py` — a stub shell returning a canned `bjobs -l` blob
containing `TERM_MEMLIMIT`, driven through the real `Lsf` plugin. No cluster.

### Phase 5 — Slurm/DRMAA

**5a (Galaxy, refactor-only, independent):** extract the three functions above out of
`lib/galaxy/jobs/runners/slurm.py` into `lib/galaxy/jobs/runners/util/slurm/__init__.py`.
`from __future__ import annotations`, no walrus, 3.7-parseable, `execute` injectable.
`SlurmJobRunner._complete_terminal_job` keeps its control flow and calls the new functions.

*Test:* new `test/unit/app/jobs/test_slurm_util.py` — table-driven over
`classify_slurm_state`: `TIMEOUT→WALLTIME_REACHED`, `OUT_OF_MEMORY→MEMORY_LIMIT_REACHED`,
`CANCELLED` + `slurmstepd: error: Exceeded job memory limit` in the error file
`→MEMORY_LIMIT_REACHED`, `CANCELLED` + a partial cgroup warning `→MEMORY_LIMIT_REACHED`,
`CANCELLED` bare `→None`, `COMPLETED`/`NOT_FOUND`/`NODE_FAIL`→`None`. Plus `slurm_job_state`
against canned `scontrol -o` output including the `Comment=** time_limit (60m) … **` spaces case
and the `sacct` fallback. **Net-new coverage — `slurm.py` has none today.**

**5b (Pulsar):** mirror the module to `pulsar/managers/util/slurm/`, add
`pulsar/managers/queued_slurm.py` — `SlurmQueueManager(DrmaaQueueManager)`, `manager_type =
"queued_slurm"` — overriding `get_status(job_id)` so that when the DRMAA state is terminal it
consults `slurm_job_state` / `classify_slurm_state` and records the runner state. Add
`pulsar.managers.util.slurm` to `setup.py`'s `packages`.

**On flipping `JobState.UNDETERMINED`:** leave it at `status.COMPLETE`. pulsar#485 already
changed the one that mattered (`FAILED → status.FAILED`). `UNDETERMINED` means DRMAA has no
record — overwhelmingly a job that finished and aged out of the DRM's memory — and flipping it to
`FAILED` would fail every long-running successful job on DRMs with short retention. That is
precisely why Galaxy's `slurm.py:121-135` re-checks `sacct` and reinterprets `NOT_FOUND`/
`COMPLETED` as `DONE` rather than trusting DRMAA. `queued_slurm` should do the same; a generic
`queued_drmaa` has no way to and should keep the conservative mapping. Worth writing this
reasoning into pulsar#408 explicitly, since natefoo's comment there suggests the opposite.

`queued_drmaa`, `queued_drmaa_xsede`, `queued_external_drmaa`, `queued_pbs`, `queued_condor`,
`unqueued`, `queued`: leave unpopulated. `queued_condor` scrapes the text log and cannot see
`HoldReasonCode` (pulsar#486's own docstring says so) — the answer for HTCondor users is
`queued_htcondor`, not more log scraping.

### Phase 6 — Coexecution / k8s / TES (defer; scope it, don't build it yet)

Here the Pulsar server runs *inside* the container being killed and never observes its own OOM.
The observer is Galaxy-side: `LaunchesK8ContainersMixin` and `tes_state_to_pulsar_status` in
`pulsar/client/client.py`. So the fix is a client-side one — have `raw_check_complete()` put
`runner_state` into the dict it already builds (k8s: pod container
`state.terminated.reason == "OOMKilled"` → `MEMORY_LIMIT_REACHED`, `DeadlineExceeded` →
`WALLTIME_REACHED`, exactly the mapping `lib/galaxy/jobs/runners/kubernetes.py:823-855` already
performs for its own runner). TES has no equivalent vocabulary — leave absent. Note that most
coexecution runners are `use_mq=True`, so the polling path this rides on is
`PulsarKubernetesJobRunner` (`use_mq=True, poll=True`) only; the MQ path would need the value
merged into the cached status body instead. Real work, separate design.

---

## Back-compatibility

| Deployment | Behaviour |
|---|---|
| **Old Pulsar server + new Galaxy** | Key absent → `.get()` → `None` → `job_state.runner_state` never assigned → `resubmit.failure` falls back to `UNKNOWN_ERROR`. Byte-identical to today. |
| **New Pulsar server + old Galaxy** | Extra JSON key. HTTP: `JobClient.get_status()` does `.get("status")` — ignored. MQ: `__async_update` reads `["job_id"]`/`["status"]` — ignored. Staging: `PulsarOutputs.from_status_response` is `.get()`-only for every field (verified `pulsar/client/staging/__init__.py:324-334`) — ignored. Safe. |
| **New Galaxy + stale `pulsar-galaxy-lib`** | `client.get_full_status()` would `AttributeError`. Mitigated by bumping the floor in `pyproject.toml`; `pulsar-galaxy-lib` is already an unconditional Galaxy dependency, so this is a pin bump, not a new coupling. Do **not** add a `getattr` shim — it hides a real, checkable constraint. |
| **New Pulsar server, manager that can't classify** | Key absent → as row 1. |
| **Embedded Pulsar** (`PulsarEmbeddedJobRunner`) | Both halves are the same installed package, so they always move together. |
| **Unknown token from a future Pulsar** | Galaxy's whitelist rejects and logs. Without the whitelist, `resubmit.MESSAGES[runner_state]` is a latent `KeyError` (see Phase 0a). |

`status.py`'s eight strings are untouched, so nothing that string-matches on status changes.

---

## Sequencing against the in-flight work

```
Galaxy 9d9f636dc9 (htcondor_pulsar) ─┐ refactor only, no wire
├─► pulsar#486 (queued_htcondor)
pulsar#485 (terminal statuses) ─────────────────────────────┘ │
▼
Phase 0a/0b (Galaxy only, no dep) ──────────────────────────────► Phase 3 (populate htcondor)
Phase 1 (pulsar server field) ──► pulsar release ──► Phase 2 (client + Galaxy consumer + pin bump)
│
Phase 5a (Galaxy slurm extract, refactor only) ──► Phase 5b (queued_slurm)
└──► Phase 4 (queued_cli)
```

- **#485 lands first, unconditionally.** It is a prerequisite, not a nicety: without it
`base_drmaa` still reports walltime kills as `complete`, and there is nothing for a reason field
to attach to. It also already makes `status.FAILED` terminal *and* postprocessed, so a failed
job still stages its outputs back — which removes the main objection to reporting `failed`
at all.
- **Galaxy `9d9f636dc9` before pulsar#486.** The Galaxy commit is a pure refactor that creates
`galaxy/jobs/runners/util/condor/htcondor.py`; #486 mirrors it. Landing #486 first leaves the
mirror one-directional and invites drift. It needs a rebase — the branch is well behind `dev`.
- **Phase 0a and 0b are fully independent** and should go out now, ahead of everything.
- **Phase 1 and Phase 2 are not a required pair** — each degrades to today's behaviour alone.
Only the end-to-end integration test needs both, which is what forces a Pulsar release between
them.
- **Phase 5a is independent** of all Pulsar work and can be reviewed as a refactor + new tests.

---

## Testing summary

Faking a DRM limit kill with no cluster, in priority order:

1. **`test/htcondor_fake/htcondor2.py`** (pulsar#486) and its Galaxy twin
`test/integration/htcondor_fake/` — a drop-in `htcondor2` module. `import_htcondor()` picks it
up when the fake directory is on `sys.path`; `test/integration/test_htcondor_runner.py` already
does exactly this via `LIVE_FAKE_MODULE_PATH`. A test can emit `JOB_HELD` with
`HoldReasonCode=16` (periodic hold → walltime) or `34` (memory) and get a genuine end-to-end
limit kill. **This is the primary vehicle.**
2. **Canned CLI output** through the real `Lsf`/`Slurm` job plugins with a stub shell — covers
Phase 4 and, with an injectable `execute`, Phase 5a.
3. **Stub DRMAA session** — `_StubDrmaaSession` in `test/manager_drmaa_test.py` (added by #485)
already exists for the state-mapping table.
4. Real Slurm in Docker exists (`test/integration/test_cli_runners.py`, `agaveapi/slurm`) but is
too heavy for CI on this path — mention it as the manual verification story, don't build on it.

**Galaxy integration test for Pulsar-backed resubmission — feasible.** `embedded_pulsar_job_conf.yml`
already passes an inline `pulsar_app_config:` block through to the embedded Pulsar app, and
`manager_factory.build_managers` accepts a `manager:` / `managers:` key there. So a new
`test/integration/resubmission_embedded_pulsar_job_conf.yml` can declare
`manager: {type: queued_htcondor, …}`, put `test/integration/htcondor_fake` on `sys.path` the way
`test_htcondor_runner.py` does, and set ``.
The fake holds the job with code 16 → `queued_htcondor` records `walltime_reached` and returns
`status.FAILED` → `full_status()` emits it → `PulsarJobRunner` sets `job_state.runner_state` →
`resubmit.failure` resubmits to `local` → the tool passes. Add it as
`TestJobResubmissionEmbeddedPulsarWalltimeIntegration` in `test/integration/test_job_resubmission.py`,
next to the existing `TestJobResubmissionPulsarIntegration` (which stays as-is — it covers the
connection-failure/`unknown_error` case and must not be weakened).

The one honest caveat: embedded Pulsar imports the *installed* `pulsar-galaxy-lib`, so this test
only goes green after Phase 1 ships in a release (or in a dev install). Phase 0b's
tool-detected-error variant of the same harness has no such constraint and can land first, which
also proves the harness itself before the wire work depends on it.

No existing test is removed or weakened anywhere in this plan.

---

## Scope split / PR boundaries

**Small, mergeable now (each independently reviewable):**

| PR | Repo | Content |
|---|---|---|
| A | galaxy | `resubmit.py` `MESSAGES`/context gap + `test_resubmit.py` (Phase 0a) |
| B | galaxy | `PulsarJobRunner.finish_job` → `_handle_runner_state` + embedded-Pulsar tool-detected integration test (Phase 0b) |
| C | pulsar | `runner_states` mirror drift (`TOOL_TIMELIMIT_REACHED` + whitespace reconcile) |
| D | galaxy | Slurm classifier extraction to `util/slurm/` + first-ever `slurm.py` unit tests (Phase 5a) |

**Medium, ordered:**

| PR | Repo | Content | Depends on |
|---|---|---|---|
| E | pulsar | `runner_state` metadata + `ManagerInterface`/`ManagerProxy` + `__job_complete_dict` + `get_full_status()` (Phases 1–2 server/client) | #485 |
| F | galaxy | `PulsarJobRunner` consumer + whitelist + pin bump + `test_pulsar_runner.py` | E released |
| G | pulsar | `queued_cli` failure-reason wiring + `manager_cli_test.py` (Phase 4) | E, C |
| H | pulsar + galaxy | HTCondor `→ runner_states` maps in the shared module; `queued_htcondor` records; Galaxy `htcondor.py` consumes the maps (Phase 3) | #486, E |

**Larger effort, separate design:**

| PR | Repo | Content |
|---|---|---|
| I | pulsar | `queued_slurm` manager + mirrored `util/slurm/` + config docs (Phase 5b) |
| J | pulsar + galaxy | Coexecution/k8s/TES classification (Phase 6) |
| K | galaxy | `RUNNER_STATE_MESSAGES` de-duplication across `slurm`/`cli`/`htcondor`/`kubernetes` (optional cleanup) |

Close pulsar#75 on H; update pulsar#408 with the `UNDETERMINED` reasoning on I.

---

## Unresolved questions

1. Field name `runner_state` on the wire — reuses Galaxy's term, but "runner" means Galaxy-runner
in Galaxy and nothing in Pulsar. Alternative `failure_reason`. Keep `runner_state` for mirror
symmetry, or rename at the boundary?
2. Ship `runner_state_message` in Phase 1 or defer? Free-text detail is cheap (HTCondor's
`HOLD_MESSAGES` already exist) but it's a second wire field to support forever.
3. `runner_states.JOB_OUTPUT_NOT_RETURNED_FROM_CLUSTER` has a *sentence* as its value
(`"Job output not returned from cluster"`) where every sibling is snake_case. Transmit it at
all, or restrict the wire whitelist to the snake_case subset?
4. Should Pulsar map `status.LOST` → `JOB_OUTPUT_NOT_RETURNED_FROM_CLUSTER` on the Galaxy side?
Semantically right, but see Q3 and Phase 0a.
5. Galaxy imports `pulsar.client` from an unconditional dependency that also ships
`pulsar.managers.util` — so Galaxy *could* import `runner_states` from Pulsar instead of
mirroring. Keep the mirror (dependency direction), or admit the dependency already exists?
6. `queued_slurm` as a new manager type vs. a `flavor: slurm` option on `queued_drmaa`. New type
is clearer; more config surface and docs.
7. Does Phase 5a's extraction of `_get_slurm_state` change any user-visible string? Tool tests may
assert on `OUT_OF_MEMORY_MSG` — check before splitting prose from classification.
8. Pulsar release cadence — Phase 2 (Galaxy) is blocked on a Pulsar release. Cut one for Phase 1
alone, or batch 1+3+4 into a single release and hold the Galaxy PR?
9. `queued_condor` — accept that it stays blind and document `queued_htcondor` as the answer, or
is there appetite for `condor_history`-based hold detection?
10. Whitelist rejection of unknown tokens: log-and-ignore (proposed) or log-and-treat-as
`UNKNOWN_ERROR`? The latter changes behaviour for a Pulsar newer than Galaxy.
11. Mirror enforcement — nothing today detects drift between `galaxy/jobs/runners/util/` and
`pulsar/managers/util/` (they have already drifted). Worth a CI check in one repo, or out of
scope?

Start by reading the proposed Phase 0 files and tests: test/unit/app/jobs/test_resubmit.py, lib/galaxy/jobs/runners/pulsar.py, and test/integration/test_job_resubmission.py. Then trace terminal status through pulsar/manager_endpoint_util.py and the listed Pulsar and Galaxy entry points; done means the agreed design is implemented and covered by tests for failure reasons and resubmission behavior.

Written by the indexing model from the issue text.