Allow filtering job searches by history ID

Fix the following test failure in galaxyproject#19112 :

```
FAILED lib/galaxy_test/api/test_jobs.py::TestJobsApi::test_delete_job - assert 1 == 0
 +  where 1 = len([{'model_class': 'Job', 'id': 'ebdef174148ecc74', 'history_id': '4dbec071fa7bdbaa', 'tool_id': 'cat1', 'state': 'ok', 'exit_code': 0, 'create_time': '2025-11-03T22:13:05.770038', 'update_time': '2025-11-03T22:13:07.098024', 'galaxy_version': '26.0', 'external_id': None, 'handler': None, 'job_runner_name': None, 'command_line': None, 'user_email': None, 'user_id': 'adb5f5c93f827949', 'command_version': '', 'params': {'queries': '[]', 'chromInfo': '"/home/runner/work/galaxy/galaxy/galaxy root/tool-data/shared/ucsc/chrom/?.len"', 'dbkey': '"?"', '__input_ext': '"input"'}, 'inputs': {'input1': {'id': 'cda8ec5455a55990', 'src': 'hda', 'uuid': 'e7979ca3-5df9-4d8d-8654-a8db8291a6c4'}}, 'outputs': {'out_file1': {'id': '3d0ca2420d1410fd', 'src': 'hda', 'uuid': 'fbeff3c3-12f5-40bf-aa6a-ea536e482938'}}, 'copied_from_job_id': None, 'output_collections': {}}])
 +    where [{'model_class': 'Job', 'id': 'ebdef174148ecc74', 'history_id': '4dbec071fa7bdbaa', 'tool_id': 'cat1', 'state': 'ok', 'exit_code': 0, 'create_time': '2025-11-03T22:13:05.770038', 'update_time': '2025-11-03T22:13:07.098024', 'galaxy_version': '26.0', 'external_id': None, 'handler': None, 'job_runner_name': None, 'command_line': None, 'user_email': None, 'user_id': 'adb5f5c93f827949', 'command_version': '', 'params': {'queries': '[]', 'chromInfo': '"/home/runner/work/galaxy/galaxy/galaxy root/tool-data/shared/ucsc/chrom/?.len"', 'dbkey': '"?"', '__input_ext': '"input"'}, 'inputs': {'input1': {'id': 'cda8ec5455a55990', 'src': 'hda', 'uuid': 'e7979ca3-5df9-4d8d-8654-a8db8291a6c4'}}, 'outputs': {'out_file1': {'id': '3d0ca2420d1410fd', 'src': 'hda', 'uuid': 'fbeff3c3-12f5-40bf-aa6a-ea536e482938'}}, 'copied_from_job_id': None, 'output_collections': {}}] = json()
 +      where json = <Response [200]>.json
```

where searching for jobs has now started returning jobs with the same
tool_id and input hashes but from different histories.

Author: nsoranzo

client/src/api/schema/schema.ts

@@ -20056,6 +20056,11 @@ export interface components {
         };
         /** SearchJobsPayload */
         SearchJobsPayload: {
+            /**
+             * History ID
+             * @description The encoded ID of the history associated with this job.
+             */
+            history_id?: string | null;
             /**
              * Inputs
              * @description The inputs of the job.

lib/galaxy/managers/jobs.py

@@ -445,6 +445,7 @@ def by_tool_input(
         param: ToolStateJobInstancePopulatedT,
         param_dump: ToolStateDumpedToJsonInternalT,
         job_state: Optional[JobStatesT] = (Job.states.OK,),
+        history_id: Union[int, None] = None,
         require_name_match: bool = True,
     ):
         """Search for jobs producing same results using the 'inputs' part of a tool POST."""
@@ -489,6 +490,7 @@ def populate_input_data_input_id(path, key, value):
             job_state=job_state,
             param_dump=param_dump,
             wildcard_param_dump=wildcard_param_dump,
+            history_id=history_id,
             require_name_match=require_name_match,
         )
 
@@ -501,6 +503,7 @@ def __search(
         job_state: Optional[JobStatesT],
         param_dump: ToolStateDumpedToJsonInternalT,
         wildcard_param_dump=None,
+        history_id: Union[int, None] = None,
         require_name_match: bool = True,
     ):
         search_timer = ExecutionTimer()
@@ -566,7 +569,7 @@ def replace_dataset_ids(path, key, value):
                     return None
 
         stmt = stmt.where(*data_conditions).group_by(model.Job.id, *used_ids)
-        stmt = self._filter_jobs(stmt, tool_id, user.id, tool_version, job_state, wildcard_param_dump)
+        stmt = self._filter_jobs(stmt, tool_id, user.id, tool_version, job_state, wildcard_param_dump, history_id)
         stmt = self._exclude_jobs_with_deleted_outputs(stmt)
 
         for job in self.sa_session.execute(stmt):
@@ -634,8 +637,15 @@ def replace_dataset_ids(path, key, value):
         return None
 
     def _filter_jobs(
-        self, stmt, tool_id: str, user_id: int, tool_version: Optional[str], job_state, wildcard_param_dump
-    ):
+        self,
+        stmt: "Select[tuple[int]]",
+        tool_id: str,
+        user_id: int,
+        tool_version: Optional[str],
+        job_state: Union[JobStatesT, None],
+        wildcard_param_dump,
+        history_id: Union[int, None],
+    ) -> "Select[tuple[int]]":
         """Build subquery that selects a job with correct job parameters."""
         job_ids_materialized_cte = stmt.cte("job_ids_cte")
         outer_select_columns = [job_ids_materialized_cte.c[col.name] for col in stmt.selected_columns]
@@ -657,6 +667,9 @@ def _filter_jobs(
         if tool_version:
             stmt = stmt.where(Job.tool_version == str(tool_version))
 
+        if history_id is not None:
+            stmt = stmt.where(Job.history_id == history_id)
+
         if job_state is None:
             job_states: set[str] = {
                 Job.states.NEW,

lib/galaxy/schema/jobs.py

@@ -126,6 +126,11 @@ class SearchJobsPayload(Model):
         title="State",
         description="Current state of the job.",
     )
+    history_id: Union[DecodedDatabaseIdField, None] = Field(
+        default=None,
+        title="History ID",
+        description="The encoded ID of the history associated with this job.",
+    )
     model_config = ConfigDict(extra="allow")  # This is used for items named file_ and __file_
 
     @field_validator("inputs", mode="before")

lib/galaxy/webapps/galaxy/api/jobs.py

@@ -598,6 +598,7 @@ def search(
                 param=param,
                 param_dump=param_dump,
                 job_state=payload.state,
+                history_id=payload.history_id,
             )
             if job:
                 jobs.append(job)

lib/galaxy_test/api/test_jobs.py

@@ -4,6 +4,7 @@
 import time
 import urllib.parse
 from operator import itemgetter
+from typing import Union
 from unittest import SkipTest
 
 import pytest
@@ -903,9 +904,7 @@ def test_search_with_hdca_pair_input(self, history_id):
                 "f2": {"src": "hdca", "id": new_list_a},
             }
         )
-        search_payload = self._search_payload(
-            history_id=new_history_id, tool_id="multi_data_param", inputs=copied_inputs
-        )
+        search_payload = self._search_payload(tool_id="multi_data_param", inputs=copied_inputs)
         self._search(search_payload, expected_search_count=1)
         # Now we delete the original input HDCA that was used -- we should still be able to find the job
         delete_response = self._delete(f"histories/{history_id}/contents/dataset_collections/{list_id_a}")
@@ -1150,7 +1149,9 @@ def _job_search(self, tool_id, history_id, inputs):
         self._search(search_payload, expected_search_count=1)
         return tool_response
 
-    def _search_payload(self, history_id, tool_id, inputs, state="ok"):
+    def _search_payload(
+        self, tool_id: str, inputs: str, state: str = "ok", history_id: Union[str, None] = None
+    ) -> dict[str, Union[str, None]]:
         search_payload = dict(tool_id=tool_id, inputs=inputs, history_id=history_id, state=state)
         return search_payload