From 732f879261c745992e0004e8f747f539b8079d0c Mon Sep 17 00:00:00 2001 From: Konrad Rokicki Date: Fri, 16 Oct 2026 07:31:53 -0400 Subject: [PATCH 1/3] LSF: key array elements by JOBINDEX so parent status aggregates correctly bjobs -json reports the bare parent ID in JOBID for every array element and puts the index only in JOBINDEX, which was not requested. All element rows collapsed into one dict entry (last row wins), and poll() then set the parent's status directly from that row instead of aggregating via compute_array_status(). Because bjobs lists DONE elements last, the parent flipped to DONE as soon as any single element finished, and array_elements was never populated. Request jobindex and re-key element rows as "parent[idx]" so the existing element-handling paths in poll() and reconnect() work. Fixes #8. Also reported as JaneliaSciComp/fileglancer#470. Co-Authored-By: Claude Fable 5.1 --- cluster_api/executors/lsf.py | 8 +++++- tests/test_lsf.py | 52 ++++++++++++++++++++++++++++++++++++ 2 files changed, 59 insertions(+), 1 deletion(-) diff --git a/cluster_api/executors/lsf.py b/cluster_api/executors/lsf.py index 679b16e..7366211 100644 --- a/cluster_api/executors/lsf.py +++ b/cluster_api/executors/lsf.py @@ -36,7 +36,7 @@ } _BJOBS_FIELDS = ( - "jobid job_name stat exit_code exec_host max_mem " + "jobid jobindex job_name stat exit_code exec_host max_mem " "submit_time start_time finish_time" ) @@ -305,6 +305,12 @@ def _parse_job_statuses( job_id = str(rec.get("JOBID", "")).strip() if not job_id: continue + # bjobs reports the bare parent ID for every array element; the + # index lives in JOBINDEX ("0" for non-array jobs). Re-key as + # "parent[idx]" so poll()/reconnect() can tell elements apart. + job_index = str(rec.get("JOBINDEX", "")).strip() + if job_index and job_index != "0" and "[" not in job_id: + job_id = f"{job_id}[{job_index}]" stat = rec.get("STAT", "").strip() status = _LSF_STATUS_MAP.get(stat) diff --git a/tests/test_lsf.py b/tests/test_lsf.py index 4458113..b718327 100644 --- a/tests/test_lsf.py +++ b/tests/test_lsf.py @@ -627,6 +627,24 @@ def test_mixed_parent_and_element(self, lsf_config): assert "12345[1]" in result + def test_real_lsf_format_uses_jobindex(self, lsf_config): + """Real bjobs -json reports the bare parent ID in JOBID and the + element index only in JOBINDEX (regression for issue #8).""" + executor = LSFExecutor(lsf_config) + output = self._make_bjobs_json([ + {"JOBID": "154541712", "JOBINDEX": "2", "STAT": "RUN", + "JOB_NAME": "wells[2]", "EXIT_CODE": "-"}, + {"JOBID": "154541712", "JOBINDEX": "1", "STAT": "DONE", + "JOB_NAME": "wells[1]", "EXIT_CODE": ""}, + {"JOBID": "99", "JOBINDEX": "0", "STAT": "RUN", + "JOB_NAME": "single", "EXIT_CODE": "-"}, + ]) + result = executor._parse_job_statuses(output) + assert set(result) == {"154541712[1]", "154541712[2]", "99"} + assert result["154541712[1]"][0] == JobStatus.DONE + assert result["154541712[2]"][0] == JobStatus.RUNNING + + class TestArrayStatusComputation: def test_no_elements(self): record = JobRecord( @@ -752,6 +770,40 @@ async def test_poll_populates_elements(self, lsf_config, work_dir): assert job.status == JobStatus.RUNNING assert statuses["12345"] == JobStatus.RUNNING + async def test_poll_real_lsf_format_one_element_done(self, lsf_config, work_dir): + """Issue #8: bjobs lists DONE elements last and reports the bare parent + ID for every element; the parent must stay RUNNING until all are done.""" + executor = LSFExecutor(lsf_config) + with patch.object( + executor, "_call", + new_callable=AsyncMock, + return_value="Job <154541712> is submitted to queue .", + ): + job = await executor.submit_array( + command="echo hello", name="wells", array_range=(1, 4), + resources=ResourceSpec(work_dir=work_dir), + ) + + def rec(idx, stat): + return {"JOBID": "154541712", "JOBINDEX": str(idx), "STAT": stat, + "JOB_NAME": f"wells[{idx}]", "EXIT_CODE": "-", + "EXEC_HOST": "-", "MAX_MEM": "-", + "SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"} + + bjobs_output = json.dumps({"RECORDS": [ + rec(2, "RUN"), rec(3, "RUN"), rec(4, "RUN"), rec(1, "DONE"), + ]}) + with patch.object(executor, "_call", new_callable=AsyncMock, return_value=bjobs_output) as m: + await executor.poll() + args = m.call_args.args[0] + assert "jobindex" in args[args.index("-o") + 1] + + assert job.status == JobStatus.RUNNING + assert not job.is_terminal + assert len(job.array_elements) == 4 + assert job.array_elements[1].status == JobStatus.DONE + assert job.array_elements[4].status == JobStatus.RUNNING + async def test_poll_all_done(self, lsf_config, work_dir): executor = LSFExecutor(lsf_config) with patch.object( From b2f187d50dc8cda8deb271b54949d28cd41bb27e Mon Sep 17 00:00:00 2001 From: Konrad Rokicki Date: Fri, 16 Oct 2026 07:33:17 -0400 Subject: [PATCH 2/3] Tests: use real bjobs -json format for array element fixtures LSF reports the bare parent ID in JOBID and the index in JOBINDEX; the fixtures previously used a "12345[1]" JOBID that LSF never emits. Co-Authored-By: Claude Fable 5.1 --- cluster_api/executors/lsf.py | 9 ++++----- tests/test_lsf.py | 22 +++++++++++----------- tests/test_reconnect.py | 7 +++++++ 3 files changed, 22 insertions(+), 16 deletions(-) diff --git a/cluster_api/executors/lsf.py b/cluster_api/executors/lsf.py index 7366211..f87fea0 100644 --- a/cluster_api/executors/lsf.py +++ b/cluster_api/executors/lsf.py @@ -129,11 +129,10 @@ def _build_submit_env(self, env: dict[str, str] | None) -> dict[str, str] | None return submit_env or None def _collect_extra_args(self, resources: ResourceSpec | None = None) -> list[str]: - """Merge config-level and per-job extra CLI args.""" - args = list(self.config.extra_args) - if resources and resources.extra_args: - args.extend(resources.extra_args) - return args + """Return per-job extra CLI args, falling back to config-level defaults.""" + if resources and resources.extra_args is not None: + return list(resources.extra_args) + return list(self.config.extra_args) @staticmethod def _env_control_args(inherit_env: bool, login_shell: bool) -> list[str]: diff --git a/tests/test_lsf.py b/tests/test_lsf.py index b718327..55b8f0c 100644 --- a/tests/test_lsf.py +++ b/tests/test_lsf.py @@ -599,10 +599,10 @@ def _make_bjobs_json(self, records): def test_element_ids_as_keys(self, lsf_config): executor = LSFExecutor(lsf_config) output = self._make_bjobs_json([ - {"JOBID": "12345[1]", "STAT": "DONE", "EXIT_CODE": "0", + {"JOBID": "12345", "JOBINDEX": "1", "STAT": "DONE", "EXIT_CODE": "0", "EXEC_HOST": "node01", "MAX_MEM": "128 MB", "SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"}, - {"JOBID": "12345[2]", "STAT": "RUN", "EXIT_CODE": "-", + {"JOBID": "12345", "JOBINDEX": "2", "STAT": "RUN", "EXIT_CODE": "-", "EXEC_HOST": "node02", "MAX_MEM": "-", "SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"}, ]) @@ -618,7 +618,7 @@ def test_mixed_parent_and_element(self, lsf_config): {"JOBID": "12345", "STAT": "RUN", "EXIT_CODE": "-", "EXEC_HOST": "-", "MAX_MEM": "-", "SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"}, - {"JOBID": "12345[1]", "STAT": "DONE", "EXIT_CODE": "0", + {"JOBID": "12345", "JOBINDEX": "1", "STAT": "DONE", "EXIT_CODE": "0", "EXEC_HOST": "node01", "MAX_MEM": "256 MB", "SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"}, ]) @@ -748,13 +748,13 @@ async def test_poll_populates_elements(self, lsf_config, work_dir): assert len(job.array_elements) == 0 bjobs_output = json.dumps({"RECORDS": [ - {"JOBID": "12345[1]", "STAT": "DONE", "EXIT_CODE": "0", + {"JOBID": "12345", "JOBINDEX": "1", "STAT": "DONE", "EXIT_CODE": "0", "EXEC_HOST": "node01", "MAX_MEM": "128 MB", "SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"}, - {"JOBID": "12345[2]", "STAT": "RUN", "EXIT_CODE": "-", + {"JOBID": "12345", "JOBINDEX": "2", "STAT": "RUN", "EXIT_CODE": "-", "EXEC_HOST": "node02", "MAX_MEM": "-", "SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"}, - {"JOBID": "12345[3]", "STAT": "PEND", "EXIT_CODE": "-", + {"JOBID": "12345", "JOBINDEX": "3", "STAT": "PEND", "EXIT_CODE": "-", "EXEC_HOST": "-", "MAX_MEM": "-", "SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"}, ]}) @@ -819,10 +819,10 @@ async def test_poll_all_done(self, lsf_config, work_dir): ) bjobs_output = json.dumps({"RECORDS": [ - {"JOBID": "12345[1]", "STAT": "DONE", "EXIT_CODE": "0", + {"JOBID": "12345", "JOBINDEX": "1", "STAT": "DONE", "EXIT_CODE": "0", "EXEC_HOST": "node01", "MAX_MEM": "128 MB", "SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"}, - {"JOBID": "12345[2]", "STAT": "DONE", "EXIT_CODE": "0", + {"JOBID": "12345", "JOBINDEX": "2", "STAT": "DONE", "EXIT_CODE": "0", "EXEC_HOST": "node02", "MAX_MEM": "256 MB", "SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"}, ]}) @@ -848,13 +848,13 @@ async def test_poll_partial_failure(self, lsf_config, work_dir): ) bjobs_output = json.dumps({"RECORDS": [ - {"JOBID": "12345[1]", "STAT": "DONE", "EXIT_CODE": "0", + {"JOBID": "12345", "JOBINDEX": "1", "STAT": "DONE", "EXIT_CODE": "0", "EXEC_HOST": "node01", "MAX_MEM": "128 MB", "SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"}, - {"JOBID": "12345[2]", "STAT": "EXIT", "EXIT_CODE": "1", + {"JOBID": "12345", "JOBINDEX": "2", "STAT": "EXIT", "EXIT_CODE": "1", "EXEC_HOST": "node02", "MAX_MEM": "-", "SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"}, - {"JOBID": "12345[3]", "STAT": "DONE", "EXIT_CODE": "0", + {"JOBID": "12345", "JOBINDEX": "3", "STAT": "DONE", "EXIT_CODE": "0", "EXEC_HOST": "node03", "MAX_MEM": "-", "SUBMIT_TIME": "-", "START_TIME": "-", "FINISH_TIME": "-"}, ]}) diff --git a/tests/test_reconnect.py b/tests/test_reconnect.py index 52adc7b..a55189c 100644 --- a/tests/test_reconnect.py +++ b/tests/test_reconnect.py @@ -27,8 +27,15 @@ def _make_record( start_time="Jan 5 10:01:00 2024", finish_time="-", ): + # Real bjobs -json: elements carry the bare parent ID in JOBID, the + # index in JOBINDEX, and "name[idx]" in JOB_NAME. + job_index = "0" + if "[" in job_id: + job_id, job_index = job_id.rstrip("]").split("[") + job_name = f"{job_name}[{job_index}]" return { "JOBID": job_id, + "JOBINDEX": job_index, "JOB_NAME": job_name, "STAT": stat, "EXIT_CODE": exit_code, From 5633e33ba29beb4c89dbe386afe7b295850e087e Mon Sep 17 00:00:00 2001 From: Konrad Rokicki Date: Fri, 16 Oct 2026 07:34:48 -0400 Subject: [PATCH 3/3] Revert unrelated _collect_extra_args change accidentally included The previous commit swept in a local WIP change to lsf.py via a mistaken `git stash pop`. Restore lsf.py to the fix commit's version. Co-Authored-By: Claude Fable 5.1 --- cluster_api/executors/lsf.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/cluster_api/executors/lsf.py b/cluster_api/executors/lsf.py index f87fea0..7366211 100644 --- a/cluster_api/executors/lsf.py +++ b/cluster_api/executors/lsf.py @@ -129,10 +129,11 @@ def _build_submit_env(self, env: dict[str, str] | None) -> dict[str, str] | None return submit_env or None def _collect_extra_args(self, resources: ResourceSpec | None = None) -> list[str]: - """Return per-job extra CLI args, falling back to config-level defaults.""" - if resources and resources.extra_args is not None: - return list(resources.extra_args) - return list(self.config.extra_args) + """Merge config-level and per-job extra CLI args.""" + args = list(self.config.extra_args) + if resources and resources.extra_args: + args.extend(resources.extra_args) + return args @staticmethod def _env_control_args(inherit_env: bool, login_shell: bool) -> list[str]: