Find pipeline runs by id instead of scanning the latest 50

backend/app/services/pipeline_service.py

@@ -33,6 +33,12 @@
 
 PK = "PIPELINE"
 PK_RUN = "PIPELINERUN"
+
+
+def _run_pointer_key(run_id: str) -> dict:
+    """Each run also writes ``PK=PIPELINERUN#<id>, SK=META`` holding the run's
+    sort key, so a run is found by id however many runs came after it."""
+    return {"PK": f"{PK_RUN}#{run_id}", "SK": "META"}
 SOURCE = "pipeline"
 MAX_SCRIPT = 100_000
 MAX_HISTORY = 10
@@ -292,11 +298,21 @@ def pct(sorted_vals: list[float], p: int) -> float | None:
         })
 
     def get_run(self, run_id: str) -> dict | None:
+        pointer = self.table.get_item(Key=_run_pointer_key(run_id)).get("Item")
+        if pointer:
+            return self._get_run_by_sk(pointer["run_sk"])
+        # Runs created before the pointer existed: look in the recent window.
         for run in self.list_runs(limit=50):
             if run["id"] == run_id:
                 return run
         return None
 
+    def _get_run_by_sk(self, sk: str) -> dict | None:
+        # Strongly consistent: callers read a run right after writing it
+        # (start_run returns it; run_sync reads the final status to report ok).
+        item = self.table.get_item(Key={"PK": PK_RUN, "SK": sk}, ConsistentRead=True).get("Item")
+        return self._run_public(item) if item else None
+
     def _update_run(self, sk: str, **fields) -> None:
         # alias every name: status/error are DynamoDB reserved words
         expr, names, values = [], {}, {}
@@ -331,6 +347,7 @@ def _create_run(self, pipeline: str, user: str, source: str,
         if parent_run:
             item["parent_run"] = parent_run
         self.table.put_item(Item=item)
+        self.table.put_item(Item={**_run_pointer_key(run_id), "run_sk": sk})
         return sk
 
     # ------------------------------------------------------------ entrypoints
@@ -347,7 +364,7 @@ def start_run(self, *, name: str, user: str, args=None) -> dict:
         )
         self._tasks.add(task)
         task.add_done_callback(self._tasks.discard)
-        return self.get_run(sk.partition("#")[2]) or {
+        return self._get_run_by_sk(sk) or {
             "id": sk.partition("#")[2], "pipeline": name, "status": "running"}
 
     def run_sync(self, name: str, user: str = "scheduler", *, args=None,
@@ -365,7 +382,7 @@ def run_sync(self, name: str, user: str = "scheduler", *, args=None,
         sk = self._create_run(name, user, source, parent_run=parent_run)
         self._execute(sk, pipe, user, args, allow_nested=not nested,
                       timeout_s=timeout_s)
-        run = self.get_run(sk.partition("#")[2]) or {}
+        run = self._get_run_by_sk(sk) or {}
         return {"ok": run.get("status") == "completed", "run_id": run.get("id"),
                 "result": _plain(run.get("result"))}
 

backend/tests/test_pipeline_runs.py

@@ -0,0 +1,97 @@
+"""Pipeline runs are found by id, and their final status is read consistently."""
+
+import importlib
+import sys
+import types
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+import boto3
+
+
+class FakeTable:
+    """Items keyed by (PK, SK) and the Query shape list_runs uses."""
+
+    def __init__(self):
+        self.items: dict[tuple[str, str], dict] = {}
+        self.consistent_reads: list[tuple[str, str]] = []
+
+    def put_item(self, **kwargs):
+        item = kwargs["Item"]
+        self.items[(item["PK"], item["SK"])] = dict(item)
+
+    def get_item(self, **kwargs):
+        key = (kwargs["Key"]["PK"], kwargs["Key"]["SK"])
+        if kwargs.get("ConsistentRead"):
+            self.consistent_reads.append(key)
+        return {"Item": dict(self.items[key])} if key in self.items else {}
+
+    def update_item(self, **kwargs):
+        key = (kwargs["Key"]["PK"], kwargs["Key"]["SK"])
+        names = kwargs.get("ExpressionAttributeNames", {})
+        values = kwargs["ExpressionAttributeValues"]
+        item = self.items.setdefault(key, {"PK": key[0], "SK": key[1]})
+        for index in range(len(names)):
+            item[names[f"#f{index}"]] = values[f":v{index}"]
+
+    def query(self, **kwargs):
+        pk = kwargs["ExpressionAttributeValues"][":pk"]
+        rows = [dict(v) for (p, _), v in self.items.items() if p == pk]
+        rows.sort(key=lambda r: r["SK"], reverse=not kwargs.get("ScanIndexForward", True))
+        return {"Items": rows[: kwargs["Limit"]] if "Limit" in kwargs else rows}
+
+
+def load_pipeline_module():
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+    config = types.ModuleType("app.config")
+    config.settings = types.SimpleNamespace(aws_region="us-east-1", dynamo_table="test")
+    resource = types.SimpleNamespace(Table=lambda name: FakeTable())
+    with patch.dict(sys.modules, {"app.config": config}), patch.object(boto3, "resource", return_value=resource):
+        return importlib.import_module("app.services.pipeline_service")
+
+
+class PipelineRunLookupTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.module = load_pipeline_module()
+
+    @classmethod
+    def tearDownClass(cls):
+        sys.modules.pop("app.services.pipeline_service", None)
+
+    def setUp(self):
+        self.service = self.module.PipelineService.__new__(self.module.PipelineService)
+        self.service.table = FakeTable()
+        self.service._tasks = set()
+
+    def test_run_is_found_by_id_after_many_newer_runs(self):
+        sk = self.service._create_run("nightly", "tester", "api")
+        run_id = sk.partition("#")[2]
+        for i in range(60):  # newer than the 50-run window list_runs reads
+            self.service._create_run("other", "tester", "api")
+        self.assertNotIn(run_id, [r["id"] for r in self.service.list_runs(limit=50)])
+        self.assertEqual(self.service.get_run(run_id)["id"], run_id)
+
+    def test_run_sync_reports_final_status_with_a_consistent_read(self):
+        def execute(sk, pipe, user, args, **kwargs):
+            self.service._update_run(sk, status="completed", result={"ok": True})
+
+        with patch.object(self.service, "get_pipeline", return_value={"name": "nightly"}), \
+                patch.object(self.service, "_execute", side_effect=execute):
+            out = self.service.run_sync("nightly", "tester")
+        self.assertTrue(out["ok"])
+        self.assertIn(("PIPELINERUN", next(sk for (pk, sk) in self.service.table.items if pk == "PIPELINERUN")),
+                      self.service.table.consistent_reads)
+
+    def test_run_created_before_pointers_is_still_found_in_recent_window(self):
+        self.service.table.put_item(Item={
+            "PK": "PIPELINERUN", "SK": "2026-01-01T00:00:00+00:00#legacy1", "run_id": "legacy1",
+            "pipeline": "nightly", "status": "completed", "agents": [], "logs": [],
+        })
+        self.assertEqual(self.service.get_run("legacy1")["id"], "legacy1")
+        self.assertIsNone(self.service.get_run("missing"))
+
+
+if __name__ == "__main__":
+    unittest.main()