Merge pull request #1029 from PolicyEngine/codex/pipeline-run-status-index

Add structured pipeline run index endpoint

Author: anth-volk

changelog.d/1028.added

@@ -0,0 +1 @@
+Add structured pipeline run index endpoints for dashboard status discovery.

changelog.d/1028.changed

@@ -0,0 +1 @@
+Update the pinned PolicyEngine US dependency to 1.696.0.

docs/engineering/skills/pipeline_operations.md

@@ -20,12 +20,16 @@ The status system reports:
 ## Status Surfaces
 
 The structured status payload is canonical. The pipeline status sub-app exposes
-three Modal functions:
+run-level and run-index Modal functions:
 
 - `get_pipeline_status`: Python-callable structured JSON for agents, scripts,
   dashboards, and tests. Prefer this for diagnosis and automation.
 - `pipeline_status_endpoint`: protected HTTP endpoint returning the same
   structured JSON for non-Python clients. Use Modal proxy auth headers.
+- `list_pipeline_runs`: Python-callable structured JSON index of recent runs.
+  Use this for dashboards that need to discover candidate run IDs.
+- `pipeline_runs_endpoint`: protected HTTP endpoint returning the same
+  structured recent-run index for non-Python clients.
 - `pipeline_status_snippet`: human-readable text used by
   `modal run modal_app/pipeline.py::main --action status`. This is for quick
   terminal inspection only and must not be treated as a schema.

modal_app/pipeline_status.py

@@ -16,7 +16,10 @@
 
 from modal_app.images import cpu_image as image  # noqa: E402
 from modal_app.step_manifests.state import PIPELINE_MOUNT, RUNS_DIR  # noqa: E402
-from modal_app.step_manifests.status import build_pipeline_status_payload  # noqa: E402
+from modal_app.step_manifests.status import (  # noqa: E402
+    build_pipeline_runs_payload,
+    build_pipeline_status_payload,
+)
 
 app = modal.App(
     os.environ.get("US_DATA_PIPELINE_STATUS_APP_NAME")
@@ -46,6 +49,22 @@ def get_pipeline_status(
     return build_pipeline_status_payload(run_id)
 
 
+@app.function(
+    image=image,
+    timeout=60,
+    volumes={PIPELINE_MOUNT: pipeline_volume},
+)
+def list_pipeline_runs(
+    limit: int = 25,
+    status: str = "",
+    branch: str = "",
+) -> dict:
+    """Get a structured index of recent pipeline runs."""
+
+    pipeline_volume.reload()
+    return build_pipeline_runs_payload(limit=limit, status=status, branch=branch)
+
+
 @app.function(
     image=status_image,
     timeout=60,
@@ -65,6 +84,27 @@ def pipeline_status_endpoint(
     return build_pipeline_status_payload(run_id)
 
 
+@app.function(
+    image=status_image,
+    timeout=60,
+    volumes={PIPELINE_MOUNT: pipeline_volume},
+)
+@modal.fastapi_endpoint(
+    method="GET",
+    docs=False,
+    requires_proxy_auth=True,
+)
+def pipeline_runs_endpoint(
+    limit: int = 25,
+    status: str = "",
+    branch: str = "",
+) -> dict:
+    """Protected HTTP endpoint for a structured pipeline run index."""
+
+    pipeline_volume.reload()
+    return build_pipeline_runs_payload(limit=limit, status=status, branch=branch)
+
+
 @app.function(
     image=image,
     timeout=60,

modal_app/step_manifests/status.py

@@ -27,6 +27,8 @@
 from modal_app.step_manifests.specs import RUN_MANIFEST_STEP_IDS, step_title
 
 PIPELINE_STATUS_SCHEMA_VERSION = "1"
+DEFAULT_RUNS_LIMIT = 25
+MAX_RUNS_LIMIT = 100
 
 
 def _run_dir(run_id: str, runs_dir: str | Path | None = None) -> Path:
@@ -149,6 +151,176 @@ def _manifest_payload(manifest) -> dict[str, Any]:
     }
 
 
+def _bounded_limit(limit: int | str | None) -> int:
+    try:
+        parsed = int(limit if limit is not None else DEFAULT_RUNS_LIMIT)
+    except (TypeError, ValueError):
+        parsed = DEFAULT_RUNS_LIMIT
+    return max(0, min(parsed, MAX_RUNS_LIMIT))
+
+
+def _index_error_payload(error: dict[str, Any] | None) -> dict[str, Any] | None:
+    if error is None:
+        return None
+    allowed = (
+        "stage_id",
+        "substage_id",
+        "surface",
+        "error_type",
+        "message",
+        "message_truncated",
+        "record_path",
+        "latest_path",
+        "traceback_available",
+    )
+    return {key: error[key] for key in allowed if key in error}
+
+
+def _latest_manifest_payload(
+    stage_manifests: list[dict[str, Any]],
+) -> dict[str, Any] | None:
+    if not stage_manifests:
+        return None
+    item = stage_manifests[-1]
+    manifest = item["manifest"]
+    return {
+        "step_id": item["step_id"],
+        "stage_id": item["stage_id"],
+        "substage_id": item["substage_id"],
+        "title": item["title"],
+        "status": item["status"],
+        "started_at": manifest.get("started_at"),
+        "completed_at": manifest.get("completed_at"),
+        "duration_s": manifest.get("duration_s"),
+        "reuse_decision": manifest.get("reuse_decision", "not_applicable"),
+    }
+
+
+def _run_index_item(
+    run_id: str,
+    *,
+    runs_dir: str | Path | None = None,
+) -> dict[str, Any]:
+    payload = build_pipeline_status_payload(run_id, runs_dir=runs_dir)
+    run_manifest = payload.get("run_manifest") or {}
+    stage_manifests = payload.get("stage_manifests") or []
+    missing = payload.get("missing_expected_manifest_ids") or []
+    expected = list(run_manifest.get("known_step_ids") or RUN_MANIFEST_STEP_IDS)
+    return {
+        "run_id": payload["run_id"],
+        "status": payload["status"],
+        "message": payload["message"],
+        "branch": run_manifest.get("branch"),
+        "sha": run_manifest.get("sha"),
+        "candidate_version": run_manifest.get("candidate_version"),
+        "release_version": run_manifest.get("release_version"),
+        "started_at": run_manifest.get("started_at"),
+        "updated_at": payload.get("updated_at"),
+        "completed_at": run_manifest.get("completed_at"),
+        "modal_app_name": payload.get("modal_app_name"),
+        "modal_environment": payload.get("modal_environment"),
+        "hf_staging_prefix": run_manifest.get("hf_staging_prefix"),
+        "github_run_url": (run_manifest.get("run_context") or {}).get("github_run_url"),
+        "latest_manifest": _latest_manifest_payload(stage_manifests),
+        "progress": {
+            "expected_manifests": len(expected),
+            "present_manifests": len(stage_manifests),
+            "missing_manifests": len(missing),
+        },
+        "error": _index_error_payload(payload.get("error")),
+    }
+
+
+def _unreadable_run_index_item(run_id: str, exc: BaseException) -> dict[str, Any]:
+    message = redacted_bounded_error_text(
+        f"{type(exc).__name__}: {exc}",
+        max_chars=DEFAULT_ERROR_MESSAGE_MAX_CHARS,
+    ).text
+    return {
+        "run_id": run_id,
+        "status": "unreadable",
+        "message": message,
+        "branch": None,
+        "sha": None,
+        "candidate_version": None,
+        "release_version": None,
+        "started_at": None,
+        "updated_at": None,
+        "completed_at": None,
+        "modal_app_name": None,
+        "modal_environment": None,
+        "hf_staging_prefix": None,
+        "github_run_url": None,
+        "latest_manifest": None,
+        "progress": {
+            "expected_manifests": 0,
+            "present_manifests": 0,
+            "missing_manifests": 0,
+        },
+        "error": {
+            "error_type": type(exc).__name__,
+            "message": message,
+            "traceback_available": False,
+        },
+    }
+
+
+def _run_sort_key(item: dict[str, Any]) -> tuple[str, str]:
+    return (
+        str(item.get("updated_at") or item.get("started_at") or ""),
+        str(item.get("run_id") or ""),
+    )
+
+
+def build_pipeline_runs_payload(
+    *,
+    limit: int | str | None = DEFAULT_RUNS_LIMIT,
+    status: str = "",
+    branch: str = "",
+    runs_dir: str | Path | None = None,
+) -> dict[str, Any]:
+    """Build a JSON-serializable index of recent pipeline runs."""
+
+    bounded_limit = _bounded_limit(limit)
+    root = Path(runs_dir) if runs_dir is not None else Path(pipeline_state.RUNS_DIR)
+    filters = {"status": status or "", "branch": branch or ""}
+    if not root.exists():
+        return {
+            "schema_version": PIPELINE_STATUS_SCHEMA_VERSION,
+            "count": 0,
+            "limit": bounded_limit,
+            "filters": filters,
+            "runs": [],
+        }
+
+    items = []
+    for entry in root.iterdir():
+        if not entry.is_dir():
+            continue
+        manifest_path = run_manifest_path(entry)
+        if not manifest_path.exists():
+            continue
+        try:
+            item = _run_index_item(entry.name, runs_dir=root)
+        except Exception as exc:
+            item = _unreadable_run_index_item(entry.name, exc)
+        if filters["status"] and item.get("status") != filters["status"]:
+            continue
+        if filters["branch"] and item.get("branch") != filters["branch"]:
+            continue
+        items.append(item)
+
+    items.sort(key=_run_sort_key, reverse=True)
+    runs = items[:bounded_limit]
+    return {
+        "schema_version": PIPELINE_STATUS_SCHEMA_VERSION,
+        "count": len(runs),
+        "limit": bounded_limit,
+        "filters": filters,
+        "runs": runs,
+    }
+
+
 def build_pipeline_status_payload(
     run_id: str,
     *,

pyproject.toml

@@ -22,7 +22,7 @@ classifiers = [
     "Programming Language :: Python :: 3.14",
 ]
 dependencies = [
-    "policyengine-us==1.694.0",
+    "policyengine-us==1.696.0",
     # policyengine-core 3.26.1 is the current 3.26.x runtime and includes the fix for
     # PolicyEngine/policyengine-core#482 (user-set ETERNITY inputs lost
     # after _invalidate_all_caches) and is required by policyengine-us 1.682.1+.

tests/integration/test_modal_pipeline_seams.py

@@ -128,6 +128,23 @@ def test_pipeline_status_callable_reports_missing_run():
     assert result["stage_manifests"] == []
 
 
+def test_pipeline_runs_callable_reports_structured_index():
+    _require_modal_tokens()
+
+    fn = modal.Function.from_name(
+        APP_NAME,
+        "list_pipeline_runs",
+        environment_name=MODAL_ENVIRONMENT,
+    )
+    result = fn.remote(limit=1)
+
+    assert result["schema_version"] == "1"
+    assert result["limit"] == 1
+    assert result["count"] <= 1
+    assert isinstance(result["runs"], list)
+    assert result["filters"] == {"status": "", "branch": ""}
+
+
 def test_pipeline_status_http_endpoint_reports_missing_run():
     _require_modal_tokens()
     headers = _modal_proxy_auth_headers()
@@ -155,6 +172,34 @@ def test_pipeline_status_http_endpoint_reports_missing_run():
     assert result["error"] is None
 
 
+def test_pipeline_runs_http_endpoint_reports_structured_index():
+    _require_modal_tokens()
+    headers = _modal_proxy_auth_headers()
+
+    fn = modal.Function.from_name(
+        APP_NAME,
+        "pipeline_runs_endpoint",
+        environment_name=MODAL_ENVIRONMENT,
+    )
+    endpoint = fn.get_web_url()
+    assert endpoint
+
+    response = requests.get(
+        endpoint,
+        params={"limit": "1"},
+        headers=headers,
+        timeout=30,
+    )
+
+    assert response.status_code == 200, response.text[:500]
+    result = response.json()
+    assert result["schema_version"] == "1"
+    assert result["limit"] == 1
+    assert result["count"] <= 1
+    assert isinstance(result["runs"], list)
+    assert result["filters"] == {"status": "", "branch": ""}
+
+
 def test_pipeline_status_cli_snippet_reports_missing_run():
     _require_modal_tokens()