feat(input-contract): serve generated AppInputContract over thin runtime input_type (BLDX-1354)

The /workflows/v1/input-contract endpoint returned ep.input_type — the type the
@entrypoint run() method accepts. For contract-toolkit apps that is a thin
runtime wrapper (e.g. bigquery's CrawlInput/MinerInput, extra="allow", just two
fields) rather than the rich generated AppInputContract (all configured fields +
CredentialRef-typed credential fields) at app/generated/{entrypoint}/_input.py.
Result: validation no-ops, no field/default discovery, and Heracles can't detect
credentials by $ref.

Auto-discover the generated AppInputContract by convention
(app/generated/{entrypoint}/_input.py, or app/generated/_input.py for single-ep
apps) and serve it; fall back to the runtime input_type when no generated
contract is importable, so apps not using the contract-toolkit are unchanged.

Covered by new tests: generated contract preferred when present (and per-entry
fallback for an entrypoint without one), plus a direct fallback unit test.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: OnkarVO7

application_sdk/handler/service.py

@@ -555,6 +555,50 @@ def _resolve_app_entrypoint(
     return app_meta, ep
 
 
+def _published_input_contract(ep: Any) -> Any:
+    """Resolve the entry point's *published* input contract.
+
+    Apps generated by the contract-toolkit ship a rich, declarative
+    ``AppInputContract`` — every configured field plus ``CredentialRef``-typed
+    credential fields — at ``app/generated/{entrypoint}/_input.py`` (or
+    ``app/generated/_input.py`` for a single-entry-point app). That is distinct
+    from the thin runtime wrapper the ``@entrypoint`` method accepts (which uses
+    ``extra="allow"`` just to carry the AE config dict and hydration state), and
+    it is the schema Heracles needs to validate caller inputs and discover
+    credential fields by their ``CredentialRef``/``AgentCredentialSpec`` ``$ref``.
+
+    Prefer that generated contract by convention; fall back to the entry point's
+    runtime ``input_type`` when no generated contract is importable (so apps that
+    don't use the contract-toolkit keep working unchanged).
+    """
+    import importlib  # noqa: PLC0415 — cold path: only on /input-contract
+
+    ep_module = ep.name.replace("-", "_")
+    for module_path in (
+        f"app.generated.{ep_module}._input",
+        "app.generated._input",
+    ):
+        try:
+            module = importlib.import_module(module_path)
+        except ImportError:
+            continue
+        contract = getattr(module, "AppInputContract", None)
+        if contract is not None and hasattr(contract, "model_json_schema"):
+            logger.debug(
+                "input-contract: using generated AppInputContract from %s for entrypoint %s",
+                module_path,
+                ep.name,
+            )
+            return contract
+    logger.debug(
+        "input-contract: no generated AppInputContract for entrypoint %s; "
+        "falling back to runtime input_type %s",
+        ep.name,
+        getattr(ep.input_type, "__name__", ep.input_type),
+    )
+    return ep.input_type
+
+
 _WORKFLOW_SENSITIVE_FIELDS = {
     "username",
     "password",
@@ -1548,7 +1592,10 @@ async def get_input_contract(entrypoint: str | None = None) -> Response:
             _workflow_config.app_name, entrypoint, unknown_ep_status=404
         )
 
-        input_type = ep.input_type
+        # Prefer the generated AppInputContract (rich, validatable, credential
+        # refs) over the entry point's thin runtime input_type. See
+        # _published_input_contract.
+        input_type = _published_input_contract(ep)
         if input_type is None:
             raise HTTPException(
                 status_code=500, detail="Entry point has no input type."

tests/unit/handler/test_service.py

@@ -4719,6 +4719,74 @@ async def run(self, input: _RoutingInput) -> _RoutingOutput:
         )
         assert resp.status_code == 400
 
+    @staticmethod
+    def _inject_generated_contract(ep_module: str, contract_cls: type):
+        """Register a fake app/generated/{ep}/_input.py:AppInputContract in
+        sys.modules so _published_input_contract can import it. Returns the list
+        of injected module paths for cleanup."""
+        import sys
+        import types
+
+        paths = ["app", "app.generated", f"app.generated.{ep_module}"]
+        for p in paths:
+            sys.modules.setdefault(p, types.ModuleType(p))
+        leaf = f"app.generated.{ep_module}._input"
+        mod = types.ModuleType(leaf)
+        mod.AppInputContract = contract_cls
+        sys.modules[leaf] = mod
+        return paths + [leaf]
+
+    def test_prefers_generated_app_input_contract(self) -> None:
+        """When app/generated/{ep}/_input.py:AppInputContract is importable, the
+        endpoint returns its (rich) schema, not the thin runtime input_type."""
+        import sys
+
+        from application_sdk.app.base import App
+        from application_sdk.app.entrypoint import entrypoint
+
+        class _GenContract(Input, allow_unbounded_fields=True):  # type: ignore[call-arg]
+            region: str = "region-us"
+
+        class _MultiEpGen(App):
+            @entrypoint
+            async def extract(self, input: _RoutingInput) -> _RoutingOutput:
+                return _RoutingOutput()
+
+            @entrypoint
+            async def load(self, input: _RoutingInput) -> _RoutingOutput:
+                return _RoutingOutput()
+
+        injected = self._inject_generated_contract("extract", _GenContract)
+        try:
+            resp = self._client(_MultiEpGen).get(
+                "/workflows/v1/input-contract?entrypoint=extract"
+            )
+            assert resp.status_code == 200
+            # rich generated contract, not the thin _RoutingInput
+            assert resp.json() == _GenContract.model_json_schema()
+            assert "region" in resp.json()["properties"]
+            # the entrypoint with no generated module still falls back
+            resp_load = self._client(_MultiEpGen).get(
+                "/workflows/v1/input-contract?entrypoint=load"
+            )
+            assert resp_load.json() == _RoutingInput.model_json_schema()
+        finally:
+            for p in injected:
+                sys.modules.pop(p, None)
+
+    def test_published_contract_falls_back_to_input_type(self) -> None:
+        """No generated module → _published_input_contract returns input_type."""
+        from application_sdk.app.entrypoint import EntryPointMetadata
+        from application_sdk.handler.service import _published_input_contract
+
+        ep = EntryPointMetadata(
+            name="no-generated-here",
+            input_type=_RoutingInput,
+            output_type=_RoutingOutput,
+            method_name="no_generated_here",
+        )
+        assert _published_input_contract(ep) is _RoutingInput
+
 
 class TestDefaultEntrypoint:
     """Tests for default-entrypoint resolution (@entrypoint(default=True))."""