fix(handler): filesystem-resolve bundle marketplace entrypoints for input-contract and bare-manifest

.github/scripts/fetch_conformance_sarif.py

@@ -26,12 +26,17 @@
 
 Outputs written to ``$GITHUB_OUTPUT``:
 
-  * ``has_run``       — a run with live SARIF was found
-  * ``run_id``        — that run's id
-  * ``commit_sha``    — its head SHA
-  * ``branch``        — its head branch
-  * ``has_artifacts`` — at least one series actually downloaded
-  * ``series``        — space-separated series names that downloaded
+  * ``has_run``         — a run with live SARIF was found
+  * ``run_id``          — that run's id
+  * ``commit_sha``      — its head SHA
+  * ``branch``          — its head branch
+  * ``has_artifacts``   — at least one series actually downloaded
+  * ``series``          — space-separated series names that downloaded
+  * ``discovery_error`` — ``true`` when discovery *itself* failed (the
+    ``run list``/artifact-listing call errored or returned unparseable JSON),
+    as opposed to the routine "no qualifying run" case. Always ``false`` when a
+    run was found. Lets the workflow warn loudly on an operational fault
+    (auth/transport/API) without failing the best-effort publish red.
 
 Exits 0 in the "nothing to publish" cases (no candidate run, no live SARIF) —
 those are routine and gated downstream by ``has_run`` / ``has_artifacts``. A
@@ -118,11 +123,18 @@ def write_outputs(pairs: dict[str, str]) -> None:
 
 
 def discover(repo: str, workflow: str, branch: str, limit: int, gh=run_gh):
-    """Newest completed run carrying live SARIF -> ``(run_id, series)``.
-
-    Returns ``(None, [])`` when no run in the window has any. Probing newest
-    first — rather than trusting ``.[0]`` — keeps a repo publishable off an
-    older run when the newest one expired or died before uploading.
+    """Newest completed run carrying live SARIF -> ``(run_id, series, error)``.
+
+    Returns ``(None, [], False)`` when no run in the window has any — the
+    routine case. Probing newest first — rather than trusting ``.[0]`` — keeps
+    a repo publishable off an older run when the newest one expired or died
+    before uploading.
+
+    The third element flags an *operational* fault in discovery itself: the
+    ``run list`` call failing or returning unparseable JSON, or every probe of a
+    candidate run erroring. Those are auth/transport/API problems, not
+    "nothing to publish", so they surface separately (``discovery_error``) for
+    the workflow to warn on loudly, without turning the best-effort publish red.
     """
     rc, out = gh(
         [
@@ -140,34 +152,46 @@ def discover(repo: str, workflow: str, branch: str, limit: int, gh=run_gh):
     )
     if rc != 0:
         print(f"::warning::could not list {workflow} runs for {repo}")
-        return None, []
+        return None, [], True
 
     try:
         payload = json.loads(out or "[]")
     except json.JSONDecodeError:
         print(f"::warning::unparseable run list for {repo}")
-        return None, []
+        return None, [], True
 
-    for run_id in candidate_runs(payload):
+    candidates = candidate_runs(payload)
+    probed = 0
+    probe_errors = 0
+    for run_id in candidates:
         rc, out = gh(["api", f"repos/{repo}/actions/runs/{run_id}/artifacts"])
         if rc != 0:
             print(f"Run {run_id}: artifact listing failed — trying older")
+            probed += 1
+            probe_errors += 1
             continue
         try:
             artifacts = json.loads(out or "{}")
         except json.JSONDecodeError:
             print(f"Run {run_id}: unparseable artifact listing — trying older")
+            probed += 1
+            probe_errors += 1
             continue
+        probed += 1
         series = live_sarif_series(artifacts)
         if series:
             print(
                 f"Run {run_id} has {len(series)} live SARIF artifact(s) "
                 f"({', '.join(series)}) — using it"
             )
-            return run_id, series
+            return run_id, series, False
         print(f"Run {run_id} has no live SARIF artifacts — trying older")
 
-    return None, []
+    # Discovery errored only if EVERY candidate probe failed; at least one
+    # listing that parsed (even with zero live SARIF) means the API is healthy
+    # and this is the routine "nothing to publish" case.
+    error = probed > 0 and probe_errors == probed
+    return None, [], error
 
 
 def download(repo: str, run_id: int, series: list[str], dest: str, gh=run_gh):
@@ -233,13 +257,21 @@ def main(argv: list[str] | None = None, gh=run_gh) -> int:
     ap.add_argument("--dir", default="/tmp/sarif")
     args = ap.parse_args(argv)
 
-    run_id, series = discover(args.repo, args.workflow, args.branch, args.limit, gh=gh)
+    run_id, series, discovery_error = discover(
+        args.repo, args.workflow, args.branch, args.limit, gh=gh
+    )
     if run_id is None:
         print(
             f"::warning::No {args.workflow} run with live SARIF artifacts in the "
             f"last {args.limit} completed runs — skipping conformance dashboard update"
         )
-        write_outputs({"has_run": "false", "has_artifacts": "false"})
+        write_outputs(
+            {
+                "has_run": "false",
+                "has_artifacts": "false",
+                "discovery_error": "true" if discovery_error else "false",
+            }
+        )
         return 0
 
     got = download(args.repo, run_id, series, args.dir, gh=gh)

.github/scripts/tests/test_fetch_conformance_sarif.py

@@ -165,8 +165,10 @@ def test_discover_falls_back_to_an_older_run_when_newest_has_no_live_sarif():
             98: _artifacts([("conformance-ci-sarif", False)]),
         },
     )
-    run_id, series = fcs.discover("atlanhq/x", "conformance.yaml", "main", 20, gh=gh)
-    assert (run_id, series) == (98, ["ci"])
+    run_id, series, error = fcs.discover(
+        "atlanhq/x", "conformance.yaml", "main", 20, gh=gh
+    )
+    assert (run_id, series, error) == (98, ["ci"], False)
 
 
 def test_discover_returns_nothing_when_no_run_has_live_sarif():
@@ -177,6 +179,7 @@ def test_discover_returns_nothing_when_no_run_has_live_sarif():
     assert fcs.discover("atlanhq/x", "conformance.yaml", "main", 20, gh=gh) == (
         None,
         [],
+        False,
     )
 
 
@@ -189,10 +192,80 @@ def test_discover_survives_an_unlistable_run():
             ]
         ),
         artifacts_by_run={98: _artifacts([("conformance-tests-sarif", False)])},
-    )  # 99 missing -> rc 1
+    )  # 99 missing -> rc 1, but 98 lists fine -> not an error
     assert fcs.discover("atlanhq/x", "conformance.yaml", "main", 20, gh=gh) == (
         98,
         ["tests"],
+        False,
+    )
+
+
+# --------------------------------------------------------------------------
+# discover: error signal
+# --------------------------------------------------------------------------
+
+
+class _FailingGh:
+    """Every gh call fails — simulates a transport/auth outage."""
+
+    def __call__(self, args: list[str]):
+        return 1, ""
+
+
+def test_discover_flags_error_when_run_list_fails():
+    assert fcs.discover(
+        "atlanhq/x", "conformance.yaml", "main", 20, gh=_FailingGh()
+    ) == (
+        None,
+        [],
+        True,
+    )
+
+
+def test_discover_flags_error_when_run_list_is_unparseable():
+    gh = FakeGh(runs="not json")
+    assert fcs.discover("atlanhq/x", "conformance.yaml", "main", 20, gh=gh) == (
+        None,
+        [],
+        True,
+    )
+
+
+def test_discover_flags_error_when_every_probe_fails():
+    # Two candidates, neither has a retrievable artifact listing -> operational
+    # fault, not "nothing to publish".
+    gh = FakeGh(
+        runs=json.dumps(
+            [
+                {"databaseId": 99, "conclusion": "success"},
+                {"databaseId": 98, "conclusion": "success"},
+            ]
+        ),
+        artifacts_by_run={},  # both runs unlistable -> rc 1 each
+    )
+    assert fcs.discover("atlanhq/x", "conformance.yaml", "main", 20, gh=gh) == (
+        None,
+        [],
+        True,
+    )
+
+
+def test_discover_no_error_when_a_probe_succeeds_with_zero_live_sarif():
+    # Newest run unlistable, older run lists fine but has no SARIF -> the API is
+    # healthy; this is the routine empty case, not an error.
+    gh = FakeGh(
+        runs=json.dumps(
+            [
+                {"databaseId": 99, "conclusion": "success"},
+                {"databaseId": 98, "conclusion": "success"},
+            ]
+        ),
+        artifacts_by_run={98: _artifacts([("conformance-ci-sarif", True)])},  # expired
+    )
+    assert fcs.discover("atlanhq/x", "conformance.yaml", "main", 20, gh=gh) == (
+        None,
+        [],
+        False,
     )
 
 
@@ -266,7 +339,22 @@ def test_no_run_is_a_clean_skip_not_a_failure(tmp_path, monkeypatch):
     gh = FakeGh(runs="[]")
     rc, out = _run_main(gh, tmp_path, monkeypatch)
     assert rc == 0
-    assert out == {"has_run": "false", "has_artifacts": "false"}
+    assert out == {
+        "has_run": "false",
+        "has_artifacts": "false",
+        "discovery_error": "false",
+    }
+
+
+def test_discovery_failure_sets_discovery_error_output(tmp_path, monkeypatch):
+    """An operational fault (run list fails) must surface as discovery_error=true
+    while still exiting 0 — the workflow warns loudly without failing the
+    best-effort publish red."""
+    rc, out = _run_main(_FailingGh(), tmp_path, monkeypatch)
+    assert rc == 0
+    assert out["has_run"] == "false"
+    assert out["has_artifacts"] == "false"
+    assert out["discovery_error"] == "true"
 
 
 def test_all_downloads_failing_is_a_clean_skip(tmp_path, monkeypatch):

application_sdk/handler/service.py

@@ -959,6 +959,70 @@ def _published_input_contract(ep: Any) -> Any:
     return ep.input_type
 
 
+def _marketplace_entrypoint_contract(entrypoint_name: str) -> Any | None:
+    """Resolve a *marketplace* entry point's published input contract from disk.
+
+    Bundle ("uber") apps expose marketplace entry points as generated contract
+    directories — ``CONTRACT_GENERATED_DIR/<entrypoint>/manifest.json`` — whose
+    DAGs are computed per submission by ``compute_manifest``. Those entry
+    points are **not** ``@entrypoint`` workflow registrations (the registry
+    holds the DAG-node workflows, e.g. ``<entrypoint>-post``), so registry
+    resolution 404s on them and ``/v1/app`` creation breaks even though the
+    manifest and configmap routes — which are filesystem-first — serve them
+    fine. This gives the input-contract route the same filesystem authority.
+
+    The contract toolkit emits each bundle entry point's ``AppInputContract``
+    as ``_input.py``; because the kebab-named generated dir is not a Python
+    package, the bundle regen convention relocates it to
+    ``app/<entrypoint_snake>/_input.py``. Check the in-place generated path
+    first (parity with ``_published_input_contract``), then the relocation
+    path.
+
+    Returns ``None`` when the entry point has no generated contract dir or no
+    importable ``AppInputContract`` — callers fall back to their existing
+    behavior.
+    """
+    import importlib  # noqa: PLC0415 — cold path: only on /input-contract
+
+    from application_sdk.app.entrypoint import (  # noqa: PLC0415 — circular at module import time
+        entrypoint_module_segment,
+    )
+
+    # The route already validates the name, but this helper must be safe on
+    # its own: the name reaches both a filesystem path and an import path.
+    # The regex forbids path separators and dots; the containment check makes
+    # the no-traversal property locally provable (py/path-injection).
+    if not _ENTRYPOINT_NAME_RE.match(entrypoint_name):
+        return None
+    generated_root = os.path.realpath(CONTRACT_GENERATED_DIR)
+    ep_manifest = os.path.realpath(
+        os.path.join(generated_root, entrypoint_name, "manifest.json")
+    )
+    if not ep_manifest.startswith(generated_root + os.sep):
+        return None
+    if not os.path.isfile(ep_manifest):
+        return None
+    ep_module = entrypoint_module_segment(entrypoint_name)
+    for module_path in (
+        f"app.generated.{ep_module}._input",
+        f"app.{ep_module}._input",
+    ):
+        try:
+            module = importlib.import_module(module_path)
+        except ImportError:  # conformance: ignore[E008,E014] optional generated module; continue to next candidate
+            continue
+        contract = getattr(module, "AppInputContract", None)
+        if contract is not None and hasattr(contract, "model_json_schema"):
+            logger.debug(
+                "input-contract: using marketplace AppInputContract from %s "
+                "for entrypoint %s (filesystem-resolved bundle entry point)",
+                module_path,
+                entrypoint_name,
+            )
+            return contract
+    return None
+
+
 _WORKFLOW_SENSITIVE_FIELDS = {
     "username",
     "password",
@@ -2161,7 +2225,6 @@ async def _serve_manifest(
                 for ep in sorted(app_meta.entry_points.values(), key=lambda e: e.name)
                 if not ep.implicit and ep.name not in candidates
             )
-
         for cand in candidates:
             try:
                 return await _serve_entrypoint_manifest(cand, fe_inputs, deployment)
@@ -2296,9 +2359,25 @@ async def get_input_contract(entrypoint: str | None = None) -> Response:
         if entrypoint is not None and not _ENTRYPOINT_NAME_RE.match(entrypoint):
             raise HTTPException(status_code=400, detail="Invalid entrypoint name")
 
-        _, ep = _resolve_app_entrypoint(
-            _workflow_config.app_name, entrypoint, unknown_ep_status=404
-        )
+        try:
+            _, ep = _resolve_app_entrypoint(
+                _workflow_config.app_name, entrypoint, unknown_ep_status=404
+            )
+        except HTTPException as exc:
+            # Registry miss — the entry point may be a *marketplace* (bundle)
+            # entry point that exists only as a generated contract dir on disk
+            # (its registry entrypoints are the DAG-node workflows, not the
+            # marketplace name). Filesystem-resolve it like the manifest and
+            # configmap routes already do; re-raise when it isn't one.
+            if exc.status_code != 404 or not entrypoint:
+                raise
+            contract = _marketplace_entrypoint_contract(entrypoint)
+            if contract is None:
+                raise
+            return Response(
+                content=orjson.dumps(contract.model_json_schema()),
+                media_type="application/json",
+            )
 
         # Prefer the generated AppInputContract (rich, validatable, credential
         # refs) over the entry point's thin runtime input_type. See