diff --git a/docs/devel_doc/openapi.json b/docs/devel_doc/openapi.json index ed380a84b..c9dae94bd 100644 --- a/docs/devel_doc/openapi.json +++ b/docs/devel_doc/openapi.json @@ -11497,7 +11497,7 @@ "workflows" ], "summary": "Start Workflow Handler", - "description": "Start a new workflow execution.\n\nParameters:\n request: FastAPI request (used by middleware).\n body: Workflow definition and configuration.\n auth: Authentication tuple (used by middleware).\n\nReturns:\n Workflow ID and initial status.", + "description": "Start a new workflow execution.\n\nParameters:\n request: FastAPI request (used by middleware).\n body: Workflow definition and configuration.\n auth: Authentication tuple (used by middleware).\n\nReturns:\n Workflow ID and initial status. A response carrying ``Deprecation`` and\n ``Sunset`` headers when the caller used the deprecated\n ``credentials_secret`` field.", "operationId": "start_workflow_handler_v1_workflows_run_post", "requestBody": { "content": { @@ -11514,11 +11514,7 @@ "description": "Successful Response", "content": { "application/json": { - "schema": { - "additionalProperties": true, - "type": "object", - "title": "Response Start Workflow Handler V1 Workflows Run Post" - } + "schema": {} } } }, @@ -21760,6 +21756,123 @@ "title": "SearchRankingOptions", "description": "Options for ranking and filtering search results.\n\nThis class configures how search results are ranked and filtered. You can use algorithm-based\nrerankers (weighted, RRF) or neural rerankers. Defaults from VectorStoresConfig are\nused when parameters are not provided.\n\nExamples:\n # Weighted ranker with custom alpha\n SearchRankingOptions(ranker=\"weighted\", alpha=0.7)\n\n # RRF ranker with custom impact factor\n SearchRankingOptions(ranker=\"rrf\", impact_factor=50.0)\n\n # Use config defaults (just specify ranker type)\n SearchRankingOptions(ranker=\"weighted\") # Uses alpha from VectorStoresConfig\n\n # Score threshold filtering\n SearchRankingOptions(ranker=\"weighted\", score_threshold=0.5)\n\n:param ranker: (Optional) Name of the ranking algorithm to use. Supported values:\n - \"weighted\": Weighted combination of vector and keyword scores\n - \"rrf\": Reciprocal Rank Fusion algorithm\n - \"neural\": Neural reranking model (requires model parameter)\n Note: For OpenAI API compatibility, any string value is accepted, but only the above values are supported.\n:param score_threshold: (Optional) Minimum relevance score threshold for results. Default: 0.0\n:param alpha: (Optional) Weight factor for weighted ranker (0-1).\n - 0.0 = keyword only\n - 0.5 = equal weight (default)\n - 1.0 = vector only\n Only used when ranker=\"weighted\" and weights is not provided.\n Falls back to VectorStoresConfig.chunk_retrieval_params.weighted_search_alpha if not provided.\n:param impact_factor: (Optional) Impact factor (k) for RRF algorithm.\n Lower values emphasize higher-ranked results. Default: 60.0 (optimal from research).\n Only used when ranker=\"rrf\".\n Falls back to VectorStoresConfig.chunk_retrieval_params.rrf_impact_factor if not provided.\n:param weights: (Optional) Dictionary of weights for combining different signal types.\n Keys can be \"vector\", \"keyword\", \"neural\". Values should sum to 1.0.\n Used when combining algorithm-based reranking with neural reranking.\n Example: {\"vector\": 0.3, \"keyword\": 0.3, \"neural\": 0.4}\n:param model: (Optional) Model identifier for neural reranker (e.g., \"transformers/Qwen/Qwen3-Reranker-0.6B\").\n Required when ranker=\"neural\" or when weights contains \"neural\"." }, + "SecretBinding": { + "properties": { + "name": { + "type": "string", + "pattern": "^[a-z0-9][a-z0-9/_.-]{0,127}$", + "title": "Logical secret name", + "description": "The SecretRef.name this binding resolves." + }, + "backend": { + "type": "string", + "enum": [ + "env", + "k8s", + "file" + ], + "title": "Backend", + "description": "Where the secret value lives." + }, + "env": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Environment variable", + "description": "Used by backend env." + }, + "secret_name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Kubernetes Secret name", + "description": "Used by backend k8s." + }, + "key": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Secret key", + "description": "Key within the Secret (backend k8s)." + }, + "path": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "File path", + "description": "Used by backend file." + } + }, + "additionalProperties": false, + "type": "object", + "required": [ + "name", + "backend" + ], + "title": "SecretBinding", + "description": "Operator-only mapping of a logical secret name to a backend location.\n\nNever accepted from a request. Exactly the fields of the chosen backend\nmay be set.\n\nAttributes:\n name: Logical name this binding resolves (matches ``SecretRef.name``).\n backend: Where the value lives: ``env``, ``k8s`` or ``file``.\n env: Environment variable name (backend ``env``).\n secret_name: Kubernetes Secret name (backend ``k8s``).\n key: Key within the Kubernetes Secret (backend ``k8s``).\n path: File path (backend ``file``)." + }, + "SecretRef": { + "properties": { + "name": { + "type": "string", + "pattern": "^[a-z0-9][a-z0-9/_.-]{0,127}$", + "title": "Secret name", + "description": "Logical secret name resolved by the operator's registry." + }, + "key": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Key", + "description": "Optional key within a multi-value secret." + }, + "version": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Version", + "description": "Optional secret version." + } + }, + "additionalProperties": false, + "type": "object", + "required": [ + "name" + ], + "title": "SecretRef", + "description": "Reference to an operator-registered secret by logical name.\n\nAttributes:\n name: Logical name, e.g. ``inference/anthropic-prod``.\n key: Optional key within a multi-value secret.\n version: Optional secret version." + }, "SecurityScheme": { "anyOf": [ { @@ -23680,12 +23793,122 @@ "title": "Transcript retention days", "description": "Days to retain step execution transcripts.", "default": 30 + }, + "providers": { + "items": { + "$ref": "#/components/schemas/WorkflowInferenceProvider" + }, + "type": "array", + "title": "Workflow provider catalog", + "description": "Operator-approved inference providers for workflow runs. Empty keeps legacy mode; non-empty switches to governed mode." + }, + "secrets": { + "items": { + "$ref": "#/components/schemas/SecretBinding" + }, + "type": "array", + "title": "Secret registry", + "description": "Maps logical secret names to backend bindings. Operator-only; never accepted from a request." + }, + "default_provider": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Default provider", + "description": "Catalog entry used when a request names no provider (governed mode)." + }, + "default_model": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Default model", + "description": "Model used when a request names none (governed mode)." } }, "additionalProperties": false, "type": "object", "title": "WorkflowEngineConfiguration", - "description": "Workflow engine configuration.\n\nAttributes:\n enabled: Whether the workflow engine is active.\n max_concurrent_workflows: Maximum simultaneous workflow executions.\n transcript_retention_days: Days to retain step transcripts." + "description": "Workflow engine configuration.\n\nAttributes:\n enabled: Whether the workflow engine is active.\n max_concurrent_workflows: Maximum simultaneous workflow executions.\n transcript_retention_days: Days to retain step transcripts.\n providers: Provider catalog. Empty means legacy mode; non-empty means\n governed mode, where unknown names deny.\n secrets: Registry mapping logical secret names to backend bindings.\n default_provider: Default catalog entry on workflow paths (governed mode).\n default_model: Default model on workflow paths (governed mode)." + }, + "WorkflowInferenceProvider": { + "properties": { + "name": { + "type": "string", + "pattern": "^[a-z0-9][a-z0-9/_.-]{0,127}$", + "title": "Logical provider name", + "description": "Name callers send in the request." + }, + "executor_type": { + "type": "string", + "enum": [ + "openai", + "anthropic", + "claude", + "gemini", + "azure", + "bedrock" + ], + "title": "Executor provider type", + "description": "cloud-agents provider type; must be one the executor supports." + }, + "credential": { + "anyOf": [ + { + "$ref": "#/components/schemas/SecretRef" + }, + { + "type": "null" + } + ], + "title": "Credential", + "description": "Logical ref of the one credential this entry uses." + }, + "allowed_models": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "title": "Allowed models", + "description": "Models callers may select. Omit for any model; an empty list is rejected (remove the provider to disable it)." + }, + "base_url": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Base URL", + "description": "Operator-set endpoint for Azure and self-hosted models." + } + }, + "additionalProperties": false, + "type": "object", + "required": [ + "name", + "executor_type" + ], + "title": "WorkflowInferenceProvider", + "description": "One entry in the workflow provider catalog.\n\nAttributes:\n name: Logical name callers send.\n executor_type: cloud-agents provider type this entry runs as.\n credential: The single credential of this entry (a logical ref).\n allowed_models: Models callers may use; None means any model.\n base_url: Operator-set endpoint (Azure and self-hosted); never\n settable by a caller." } } }, diff --git a/examples/workflows/README.md b/examples/workflows/README.md index e43ebdd67..d91010dcf 100644 --- a/examples/workflows/README.md +++ b/examples/workflows/README.md @@ -5,14 +5,17 @@ This folder is a self-contained, end-to-end worked example of the design in lightspeed-stack as the sole policy/secret boundary in front of cloud-agents. > **Status: target state, being built (feature-level TDD).** This folder -> defines where issue #51 and cloud-agents#269 end up. Phase 0a is in -> review (PR #57): caller `credentials_secret` -> 400, custom images, -> `advisory` and `spawn: none/local` need admin -> 403, size and count -> caps. The `lightspeed-stack.yaml` blocks marked `PLAN(stack#51)` do not -> load yet, and the catalog/policy behavior in the workflows does not exist -> yet. `verify.py` pins the contract (see "Verification"); each phase should -> make more of it real until the whole folder runs against a deployment -> (Phases 0b–6). +> defines where issue #51 and cloud-agents#269 end up. Landed so far: +> Phase 0a (caller `credentials_secret` -> 400; custom images, `advisory` +> and `spawn: none/local` need admin -> 403; size and count caps) and +> Phase 1 (the provider catalog, secret registry and defaults load, and +> `/v1/workflows/run` resolves providers through them: the stack builds the +> credential reference, the same-entry rule holds, overrides are rewritten to +> executor types). Blocks marked `PLAN(stack#51)` that belong to later +> phases (`policy`, MCP `secret_headers` / `workflow_enabled`) do not load +> yet. `tests/unit/cloud_agents/test_workflow_contract.py` runs `cases.yaml` +> against the real handler and config for the implemented phases and skips +> the rest; `verify.py` runs all of it against the reference model. > **Auth prerequisite.** Role-based policy and the admin gates need an auth > module that resolves roles: `jwk-token` with `role_rules` (used here) or @@ -458,9 +461,13 @@ What it checks (~164 assertions): registry, RBAC `resourceNames` match the K8s-bound Secrets, pod hardening, digest-pinned images, and the dev and External Secrets sources agree. -`reference_gate.py` is a model of the design, not the implementation. As each -phase lands, run the same `cases.yaml` against the real endpoint and delete the -matching part of the reference. +`reference_gate.py` is a model of the design, not the implementation. Each case +in `cases.yaml` carries a `phase`. The unit test +`tests/unit/cloud_agents/test_workflow_contract.py` runs the cases up to +`IMPLEMENTED_PHASE` against the real handler and `WorkflowEngineConfiguration` +(status only; `reason` strings belong to the reference) and skips later ones. +Finishing a phase means raising `IMPLEMENTED_PHASE`, dropping the matching +`PENDING_ENGINE_KEYS`, and deleting that part of the reference. ## Production checklist diff --git a/examples/workflows/cases.yaml b/examples/workflows/cases.yaml index 9d75f4bac..dab0661fd 100644 --- a/examples/workflows/cases.yaml +++ b/examples/workflows/cases.yaml @@ -12,6 +12,8 @@ # expect: HTTP status, or {pre269: N, post269: N} when the stages differ # reason: substring that must appear in the returned reasons # spawner: false = stack has no spawner_configuration +# phase: stack#51 phase that makes the case real; tests/unit/cloud_agents/ +# test_workflow_contract.py runs cases up to the implemented phase # Statuses: 400 unknown to the deployment / forbidden field, 403 exists but not # for this caller, 413 too large, 422 shape or count, 202 accepted. # ============================================================================= @@ -28,27 +30,32 @@ defaults: workflow_cases: # ---- accepted ------------------------------------------------------------- - - {name: triage as agent-user, workflow: triage-github-issue, expect: 202} - - {name: kb answer as agent-support, workflow: kb-answer-with-approval, expect: 202} - - {name: admin inline MCP with registry ref, workflow: admin-inline-mcp, expect: 202} + - {phase: 1, name: triage as agent-user, workflow: triage-github-issue, expect: 202} + - {phase: 1, name: kb answer as agent-support, workflow: kb-answer-with-approval, expect: 202} + - {phase: 3, name: admin inline MCP with registry ref, workflow: admin-inline-mcp, expect: 202} - name: credential_ref equal to the entry's credential is accepted + phase: 1 workflow: triage-github-issue body_set: {provider.credential_ref: {name: inference/anthropic-prod}} expect: 202 - name: omitted provider uses workflow_engine defaults + phase: 1 workflow: triage-github-issue body_set: {provider: null} expect: 202 - name: step may pick another model on the same entry + phase: 1 workflow: kb-answer-with-approval set: {spec.steps.respond.inference_provider: {name: openai-team-b, model: gpt-4o-mini}} expect: 202 - name: admin may run spawn none (no secret_headers servers) + phase: 0 workflow: triage-github-issue as: [agent-admin] set: {spec.spawn: none, spec.mcp_servers: [], spec.allowed_skills: []} expect: 202 - name: admin may use the admin-only azure entry with its pinned model + phase: 1 workflow: triage-github-issue as: [agent-admin] body_set: {provider: {name: azure-eu, model: gpt-4o}} @@ -57,61 +64,73 @@ workflow_cases: # ---- caller cannot choose credentials (G1, G2) ---------------------------- - name: run provider credentials_secret (env var name) is rejected + phase: 0 workflow: triage-github-issue body_set: {provider.credentials_secret: DATABASE_URL} expect: 400 reason: credentials_secret - name: run provider credentials_secret is rejected even when null + phase: 0 workflow: triage-github-issue body_set: {provider.credentials_secret: null} expect: 400 - name: credentials_secret is rejected for admins too + phase: 0 workflow: admin-inline-mcp body_set: {provider.credentials_secret: ANTHROPIC_API_KEY} expect: 400 - name: definition.provider credentials_secret is rejected + phase: 0 workflow: kb-answer-with-approval set: {provider.credentials_secret: DATABASE_URL} expect: 400 - name: credential_ref for another entry's credential is rejected + phase: 1 workflow: triage-github-issue body_set: {provider.credential_ref: {name: inference/openai-team-b}} expect: 400 reason: credential_ref - name: unknown keys in the provider are rejected + phase: 1 workflow: triage-github-issue body_set: {provider.base_url: https://evil.example.com} expect: 400 # ---- catalog: unknown to the deployment (400) ------------------------------ - name: unknown provider + phase: 1 workflow: triage-github-issue body_set: {provider.name: gpt-everything} expect: 400 reason: unknown provider - name: model outside allowed_models + phase: 1 workflow: triage-github-issue body_set: {provider.model: claude-opus-4} expect: 400 reason: not allowed - name: step provider on another catalog entry (same-entry rule) + phase: 1 workflow: triage-github-issue set: {spec.steps.triage.inference_provider: {name: openai-team-b, model: gpt-4o}} as: [agent-admin] expect: 400 reason: same catalog entry - name: definition.provider on another catalog entry (same-entry rule) + phase: 1 workflow: triage-github-issue set: {provider: {name: openai-team-b, model: gpt-4o}} as: [agent-admin] expect: 400 reason: same catalog entry - name: unknown MCP server name + phase: 3 workflow: triage-github-issue set: {spec.mcp_servers: [github-readwrite]} expect: 400 reason: unknown MCP - name: inline MCP with a raw secret_name/key header (no registry ref) + phase: 3 workflow: admin-inline-mcp set: spec.steps.query.mcp_servers.0.secret_headers: @@ -119,17 +138,20 @@ workflow_cases: expect: 400 reason: registry refs - name: inline MCP with an unknown registry ref + phase: 3 workflow: admin-inline-mcp set: {spec.steps.query.mcp_servers.0.secret_headers.Authorization.name: mcp/nope} expect: 400 # ---- policy: exists, not for this caller (403) ----------------------------- - name: agent-user cannot use the openai-team-b entry + phase: 2 workflow: triage-github-issue body_set: {provider: {name: openai-team-b, model: gpt-4o}} expect: 403 reason: not granted - name: agent-support cannot use the admin-only azure entry + phase: 2 workflow: kb-answer-with-approval body_set: {provider: {name: azure-eu, model: gpt-4o}} set: @@ -137,93 +159,112 @@ workflow_cases: spec.steps.research.inference_provider: {name: azure-eu, model: gpt-4o} expect: 403 - name: agent-support cannot run the triage flow (provider, MCP, skill, tool) + phase: 2 workflow: triage-github-issue as: [agent-support] expect: 403 - name: caller without workflow_start + phase: 2 workflow: triage-github-issue as: [] expect: 403 reason: workflow_start - name: skill not granted + phase: 2 workflow: triage-github-issue set: {spec.steps.triage.allowed_skills: [kb-search]} expect: 403 reason: skill - name: tool not granted + phase: 2 workflow: triage-github-issue set: {spec.steps.triage.tools: [github.write]} expect: 403 reason: tool - name: service account not granted + phase: 2 workflow: triage-github-issue set: {spec.service_account: kb-runner} expect: 403 reason: service account - name: namespace not granted + phase: 2 workflow: triage-github-issue set: {spec.steps.triage.target_namespaces: [kube-system]} expect: 403 reason: namespace - name: step timeout over the role limit + phase: 2 workflow: triage-github-issue set: {spec.steps.triage.timeout_seconds: 3600} expect: 403 reason: timeout - name: step max_tokens over the role limit + phase: 2 workflow: triage-github-issue set: {spec.steps.triage.permissions.max_tokens: 900000} expect: 403 reason: max_tokens - name: step max_retries over the role limit + phase: 2 workflow: triage-github-issue set: {spec.steps.triage.max_retries: 9} expect: 403 reason: max_retries - name: workflow timeout over the role limit + phase: 2 workflow: triage-github-issue set: {spec.timeout_seconds: 7200} expect: 403 # ---- spawn modes, images, advisory ----------------------------------------- - name: spawn none as agent-user + phase: 0 workflow: triage-github-issue set: {spec.spawn: none, spec.mcp_servers: [], spec.allowed_skills: []} expect: 403 reason: spawn - name: spawn local on one step as agent-user + phase: 0 workflow: triage-github-issue set: {spec.steps.report.spawn: local} expect: 403 - name: ephemeral steps without spawner_configuration + phase: 0 workflow: triage-github-issue spawner: false expect: 403 reason: spawner_configuration - name: custom image on the request + phase: 0 workflow: triage-github-issue body_set: {sandbox_image: evil.example.com/sandbox:latest} expect: 403 reason: image - name: custom image on a step spawn_config + phase: 0 workflow: triage-github-issue set: {spec.steps.triage.spawn_config.sandbox_image: evil.example.com/sandbox:latest} expect: 403 - name: custom skills image + phase: 0 workflow: triage-github-issue set: {skills: {image: evil.example.com/skills:latest}} expect: 403 - name: advisory mode as agent-user + phase: 0 workflow: triage-github-issue set: {advisory: true} expect: 403 reason: advisory - name: advisory mode as admin + phase: 0 workflow: triage-github-issue as: [agent-admin] set: {advisory: true} expect: 202 - name: secret_headers MCP on spawn none (needs cloud-agents#269) + phase: 3 workflow: triage-github-issue as: [agent-admin] set: {spec.spawn: none} @@ -232,26 +273,31 @@ workflow_cases: # ---- inline MCP ------------------------------------------------------------- - name: inline MCP as agent-user + phase: 3 workflow: admin-inline-mcp as: [agent-user] expect: 403 reason: inline MCP - name: inline MCP host not on the allow-list + phase: 3 workflow: admin-inline-mcp set: {spec.steps.query.mcp_servers.0.url: "https://evil.example.com/mcp"} expect: 403 reason: host - name: inline MCP over plain http + phase: 3 workflow: admin-inline-mcp set: {spec.steps.query.mcp_servers.0.url: "http://mcp.internal.example.com/incidents"} expect: 403 reason: https - name: inline MCP with credentials in the URL + phase: 3 workflow: admin-inline-mcp set: {spec.steps.query.mcp_servers.0.url: "https://user:pw@mcp.internal.example.com/incidents"} expect: 403 reason: userinfo - name: inline MCP with a literal header value + phase: 3 workflow: admin-inline-mcp set: {spec.steps.query.mcp_servers.0.headers: {Authorization: "Bearer ghp_x"}} expect: 403 @@ -259,26 +305,32 @@ workflow_cases: # ---- limits (G9) ------------------------------------------------------------- - name: definition over the byte cap + phase: 0 workflow: triage-github-issue generate: {pad_bytes: 300000} expect: 413 - name: too many steps + phase: 0 workflow: triage-github-issue generate: {steps: 65} expect: 422 - name: too many MCP servers on the workflow default + phase: 0 workflow: triage-github-issue generate: {mcp_servers: 17} expect: 422 - name: too many secret_headers on an inline server + phase: 3 workflow: admin-inline-mcp generate: {secret_headers: 33} expect: 422 - name: missing spec.steps + phase: 0 workflow: triage-github-issue set: {spec.steps: []} expect: 422 - name: legacy step "provider" alias is a shape error + phase: 0 workflow: triage-github-issue set: {spec.steps.triage.provider: {name: claude-prod}} expect: 422 @@ -286,33 +338,43 @@ workflow_cases: # Config-load failures. `set` patches the parsed lightspeed-stack.yaml. config_cases: - name: allowed_models [] is rejected + phase: 1 set: {workflow_engine.providers.claude-prod.allowed_models: []} error: allowed_models - name: executor_type outside APPROVED_INFERENCE_PROVIDERS + phase: 1 set: {workflow_engine.providers.claude-prod.executor_type: ollama} - error: not approved + error: executor_type - name: credential without a secrets binding + phase: 1 set: {workflow_engine.providers.claude-prod.credential.name: inference/missing} error: no secrets binding - name: pre-269 inference credential must be env-backed + phase: 1 set: {workflow_engine.secrets.inference/anthropic-prod: {name: inference/anthropic-prod, backend: k8s, secret_name: s, key: k}} stage: pre269 error: must be env - name: default_provider must be in the catalog + phase: 1 set: {workflow_engine.default_provider: gone} error: default_provider - name: default_model must pass the default provider's allowed_models + phase: 1 set: {workflow_engine.default_model: claude-opus-4} error: default_model - name: workflow_enabled MCP with request-bound auth + phase: 3 set: {mcp_servers.github-readonly.authorization_headers: {Authorization: client}} error: workflow_enabled - name: workflow_enabled MCP with propagated headers + phase: 3 set: {mcp_servers.kb-search.headers: [x-rh-identity]} error: propagated - name: MCP secret ref without a binding + phase: 3 set: {mcp_servers.kb-search.secret_headers.X-API-Key.name: mcp/missing} error: no secrets binding - name: policy rule naming an undefined provider + phase: 2 set: {workflow_engine.policy.rules.0.providers: [ghost]} error: unknown provider diff --git a/examples/workflows/contract_util.py b/examples/workflows/contract_util.py new file mode 100644 index 000000000..8d573e740 --- /dev/null +++ b/examples/workflows/contract_util.py @@ -0,0 +1,91 @@ +"""Helpers shared by verify.py and the contract test for cases.yaml.""" + +import copy +from pathlib import Path +from typing import Any + +import yaml + +WF_DIR = Path(__file__).resolve().parent + + +def load(name: str) -> Any: + """Load one YAML document from this folder.""" + with open(WF_DIR / name, encoding="utf-8") as fh: + return yaml.safe_load(fh) + + +def load_all(name: str) -> list[dict]: + """Load a multi-document YAML file from this folder.""" + with open(WF_DIR / name, encoding="utf-8") as fh: + return [d for d in yaml.safe_load_all(fh) if d] + + +def deep_merge(base: Any, over: Any) -> Any: + """Maps merge; lists and scalars are replaced.""" + if isinstance(base, dict) and isinstance(over, dict): + out = dict(base) + for key, value in over.items(): + out[key] = deep_merge(base[key], value) if key in base else value + return out + return copy.deepcopy(over) + + +def _key(container: Any, token: str) -> Any: + """Resolve a path token: dict key, list index, or list item by name.""" + if isinstance(container, list): + if token.isdigit(): + return int(token) + for i, item in enumerate(container): + if isinstance(item, dict) and item.get("name") == token: + return i + raise KeyError(token) + return token + + +def patch(doc: Any, patches: dict[str, Any] | None) -> Any: + """Return a copy of ``doc`` with dotted-path ``patches`` applied.""" + doc = copy.deepcopy(doc) + for path, value in (patches or {}).items(): + tokens = path.split(".") + cur = doc + for tok in tokens[:-1]: + k = _key(cur, tok) + if isinstance(cur, dict) and k not in cur: + cur[k] = {} + cur = cur[k] + cur[_key(cur, tokens[-1])] = value + return doc + + +def generate(definition: dict, spec: dict[str, int]) -> dict: + """Build oversized inputs for the limit cases.""" + steps = definition["spec"]["steps"] + if "pad_bytes" in spec: + definition["metadata"]["padding"] = "x" * spec["pad_bytes"] + while len(steps) < spec.get("steps", 0): + clone = copy.deepcopy(steps[0]) + clone["name"] = f"gen{len(steps)}" + steps.append(clone) + if "mcp_servers" in spec: + definition["spec"]["mcp_servers"] = [ + f"srv{i}" for i in range(spec["mcp_servers"]) + ] + if "secret_headers" in spec: + server = steps[0]["mcp_servers"][0] + server["secret_headers"] = { + f"H{i}": {"name": "mcp/incidents"} for i in range(spec["secret_headers"]) + } + return definition + + +def case_body(case: dict, definitions: dict, defaults: dict) -> dict: + """Request body for a workflow case (definition, provider, overrides).""" + dflt = defaults[case["workflow"]] + definition = generate( + patch(definitions[case["workflow"]], case.get("set")), case.get("generate", {}) + ) + return patch( + {"definition": definition, "provider": copy.deepcopy(dflt["provider"])}, + case.get("body_set"), + ) diff --git a/examples/workflows/kb-answer-with-approval.yaml b/examples/workflows/kb-answer-with-approval.yaml index f6462caab..6ff8bcbd5 100644 --- a/examples/workflows/kb-answer-with-approval.yaml +++ b/examples/workflows/kb-answer-with-approval.yaml @@ -94,7 +94,7 @@ spec: agent: kb-research-agent prompt: >- Draft a customer-facing answer (under 250 words) from these approved - findings: {{ steps.research.output.findings }} + findings: {{ steps.findings.output.passages }} Cite document IDs. No tool calls needed. output_key: answer output_schema: diff --git a/examples/workflows/reference_gate.py b/examples/workflows/reference_gate.py index fcaa00df7..a50bd54f7 100644 --- a/examples/workflows/reference_gate.py +++ b/examples/workflows/reference_gate.py @@ -215,6 +215,22 @@ def _shape_copy(definition: dict) -> dict: return shaped +def executor_form(stack: dict, definition: dict) -> dict: + """Definition as cloud-agents' own validator sees it after the stack's rewrite. + + Logical provider names become executor types and inline registry refs the + executor's ``{secret_name, key}`` form (pipeline steps 4 and 6). + """ + out = _shape_copy(definition) + catalog = {p["name"]: p["executor_type"] for p in stack["workflow_engine"]["providers"]} + scopes = [out] + [s for s in out["spec"]["steps"] if s.get("inference_provider")] + for scope in scopes: + key = "provider" if scope is out else "inference_provider" + if scope.get(key): + scope[key]["name"] = catalog.get(scope[key]["name"], scope[key]["name"]) + return out + + def submit( stack: dict, body: dict, diff --git a/examples/workflows/triage-github-issue.yaml b/examples/workflows/triage-github-issue.yaml index 0bb0afea5..f988fd900 100644 --- a/examples/workflows/triage-github-issue.yaml +++ b/examples/workflows/triage-github-issue.yaml @@ -66,7 +66,9 @@ spec: agent: triage-agent prompt: >- Write a markdown triage report from this classification: - {{ steps.triage.output.classification }} + severity {{ steps.classification.output.severity }}, area + {{ steps.classification.output.area }}, summary + {{ steps.classification.output.summary }} Keep it under 300 words. No tool calls needed. output_key: report output_schema: diff --git a/examples/workflows/verify.py b/examples/workflows/verify.py index bb2ba4cb8..8f873feac 100644 --- a/examples/workflows/verify.py +++ b/examples/workflows/verify.py @@ -31,6 +31,13 @@ sys.path.insert(0, str(WF_DIR)) import reference_gate as gate # noqa: E402 +from contract_util import ( # noqa: E402 + case_body, + deep_merge, + load, + load_all, + patch, +) failures: list[str] = [] @@ -42,74 +49,6 @@ def check(cond: bool, msg: str) -> None: failures.append(msg) -def load(name: str): - """Load one YAML document.""" - with open(WF_DIR / name, encoding="utf-8") as fh: - return yaml.safe_load(fh) - - -def load_all(name: str) -> list[dict]: - """Load a multi-document YAML file.""" - with open(WF_DIR / name, encoding="utf-8") as fh: - return [d for d in yaml.safe_load_all(fh) if d] - - -def deep_merge(base, over): - """Maps merge, everything else (lists, scalars) is replaced.""" - if isinstance(base, dict) and isinstance(over, dict): - out = dict(base) - for key, value in over.items(): - out[key] = deep_merge(base[key], value) if key in base else value - return out - return copy.deepcopy(over) - - -def _key(container, token): - """Resolve a path token: dict key, list index, or list item by name.""" - if isinstance(container, list): - if token.isdigit(): - return int(token) - for i, item in enumerate(container): - if isinstance(item, dict) and item.get("name") == token: - return i - raise KeyError(token) - return token - - -def patch(doc, patches): - """Return a copy of ``doc`` with dotted-path ``patches`` applied.""" - doc = copy.deepcopy(doc) - for path, value in (patches or {}).items(): - tokens = path.split(".") - cur = doc - for tok in tokens[:-1]: - k = _key(cur, tok) - if isinstance(cur, dict) and k not in cur: - cur[k] = {} - cur = cur[k] - cur[_key(cur, tokens[-1])] = value - return doc - - -def generate(definition, spec): - """Build oversized inputs for the limit cases.""" - steps = definition["spec"]["steps"] - if "pad_bytes" in spec: - definition["metadata"]["padding"] = "x" * spec["pad_bytes"] - while len(steps) < spec.get("steps", 0): - clone = copy.deepcopy(steps[0]) - clone["name"] = f"gen{len(steps)}" - steps.append(clone) - if "mcp_servers" in spec: - definition["spec"]["mcp_servers"] = [f"srv{i}" for i in range(spec["mcp_servers"])] - if "secret_headers" in spec: - server = steps[0]["mcp_servers"][0] - server["secret_headers"] = { - f"H{i}": {"name": "mcp/incidents"} for i in range(spec["secret_headers"]) - } - return definition - - # ---------------------------------------------------------------- 1. parsing stack = load("lightspeed-stack.yaml") overrides = load("post-269-overrides.yaml") @@ -129,6 +68,12 @@ def generate(definition, spec): WorkflowDefinition.model_validate(gate._shape_copy(raw)) # pylint: disable=W0212 check(True, f"{name}: WorkflowDefinition shape valid") +from cloud_agents.workflow.core.validation import validate_definition # noqa: E402 + +for name, raw in definitions.items(): + problems = validate_definition(gate.executor_form(stacks["pre269"], raw)) + check(problems == [], f"{name}: passes cloud-agents validate_definition {problems}") + # ------------------------------------------------------------ 2. load checks for stage, cfg in stacks.items(): errors = gate.load_errors(cfg, stage) @@ -157,14 +102,9 @@ def assert_case(label, stage, got, reasons, case): for case in cases["workflow_cases"]: - base = definitions[case["workflow"]] dflt = cases["defaults"][case["workflow"]] for stage, cfg in stacks.items(): - definition = generate(patch(base, case.get("set")), case.get("generate", {})) - body = patch( - {"definition": definition, "provider": copy.deepcopy(dflt["provider"])}, - case.get("body_set"), - ) + body = case_body(case, definitions, cases["defaults"]) status, reasons = run( gate.submit, cfg, diff --git a/src/app/endpoints/workflows.py b/src/app/endpoints/workflows.py index 41efec158..ec49732ba 100644 --- a/src/app/endpoints/workflows.py +++ b/src/app/endpoints/workflows.py @@ -11,6 +11,7 @@ ) from cloud_agents.workflow.core.validation import validate_definition from fastapi import APIRouter, Depends, HTTPException, Request, status +from fastapi.responses import JSONResponse from pydantic import ValidationError from authentication import get_auth_dependency @@ -21,7 +22,13 @@ from models.api.requests.agents import ApproveWorkflowRequest, RunWorkflowRequest from models.config import Action from utils.endpoints import check_configuration_loaded -from workflow.provider_credentials import credentials_secret_for +from workflow.catalog import ( + LEGACY_DEPRECATION, + LEGACY_SUNSET, + ResolvedProvider, + normalize_definition, + resolve_run_provider, +) from workflow.spawner_factory import build_spawner from workflow.submission_guard import ( enforce_submission_hardening, @@ -112,16 +119,68 @@ def _validate_workflow_submission( ) from exc +def _prepare_submission( + body: RunWorkflowRequest, + *, + caller_is_admin: bool, + default_sandbox_image: str, + spawner_configured: bool, +) -> tuple[ResolvedProvider, dict[str, Any]]: + """Run the submission gate and return what the workflow will run with. + + Parameters: + body: Request body as sent by the caller. + caller_is_admin: Whether the caller holds the ADMIN action. + default_sandbox_image: The spawner's configured sandbox image. + spawner_configured: Whether a spawner_configuration exists. + + Returns: + The resolved run-level provider and the normalized definition. + + Raises: + HTTPException: 413, 400, 422 or 403 per the status-code rule; nothing + has been persisted when this raises. + """ + reject_oversized_definition(body.definition) + engine = configuration.workflow_engine_configuration + resolved = resolve_run_provider(engine, configuration.inference, body.provider) + definition = normalize_definition(engine, body.definition, resolved) + + # Submission-time gate (issue #55): LocalWorkflowRunner.start persists + # the raw definition and provider BEFORE build_graph/normalization + # runs, so secret-bearing, malformed, or unapproved-provider input + # must be rejected here -- otherwise it returns 202 and lands in + # workflow state first. It sees the normalized definition and the dict the + # stack built; the hardening check sees what the caller sent (minus the + # deprecated field when it was accepted as a logical name). + _validate_workflow_submission(definition, resolved.provider) + caller_provider = { + k: v + for k, v in (body.provider or {}).items() + if not (k == "credentials_secret" and resolved.legacy_credentials_secret) + } + enforce_submission_hardening( + body.definition, + caller_provider, + body.sandbox_image, + is_admin=caller_is_admin, + default_sandbox_image=default_sandbox_image, + spawner_configured=spawner_configured, + ) + return resolved, definition + + @router.post( "/workflows/run", status_code=status.HTTP_202_ACCEPTED, + response_model=None, ) @authorize(Action.WORKFLOW_START) async def start_workflow_handler( request: Request, body: RunWorkflowRequest, auth: Annotated[AuthTuple, Depends(get_auth_dependency())], -) -> dict[str, Any]: +) -> dict[str, Any] | JSONResponse: """Start a new workflow execution. Parameters: @@ -130,7 +189,9 @@ async def start_workflow_handler( auth: Authentication tuple (used by middleware). Returns: - Workflow ID and initial status. + Workflow ID and initial status. A response carrying ``Deprecation`` and + ``Sunset`` headers when the caller used the deprecated + ``credentials_secret`` field. """ user_id, username, _, _ = auth @@ -144,44 +205,20 @@ async def start_workflow_handler( else "lightspeed-agentic-sandbox:latest" ) - # Pipeline order (issue #51): byte cap (413), shape (422), then the - # stack's own rules -- forbidden caller fields (400), count caps (422) - # and privileged options (403) -- all before anything is persisted. - reject_oversized_definition(body.definition) - inference = configuration.inference - caller_provider = dict( - body.provider - or { - "name": inference.default_provider or "", - "model": inference.default_model or "", - } - ) - provider = dict(caller_provider) - if "credentials_secret" not in provider: - cred_secret = credentials_secret_for(provider.get("name") or "") - if cred_secret: - provider["credentials_secret"] = cred_secret - - # Submission-time gate (issue #55): LocalWorkflowRunner.start persists - # the raw definition and provider BEFORE build_graph/normalization - # runs, so secret-bearing, malformed, or unapproved-provider input - # must be rejected here -- otherwise it returns 202 and lands in - # workflow state first. - # `provider` carries the stack's injected credentials reference for the - # shape check; `caller_provider` keeps only the caller's keys so the - # credential-rejection check sees what the caller actually sent. - _validate_workflow_submission(body.definition, provider) - enforce_submission_hardening( - body.definition, - caller_provider, - body.sandbox_image, - is_admin=is_admin(request), + # Pipeline order (issue #51): byte cap (413), catalog resolution (400), + # shape (422), then the stack's own rules -- forbidden caller fields (400), + # count caps (422) and privileged options (403) -- all before anything is + # persisted. + resolved, definition = _prepare_submission( + body, + caller_is_admin=is_admin(request), default_sandbox_image=default_sandbox_image, spawner_configured=spawner_config is not None, ) + provider = resolved.provider workflow_input = { - "definition": body.definition, + "definition": definition, "provider": provider, "sandbox_image": body.sandbox_image or default_sandbox_image, "approval_policy": body.approval_policy, @@ -190,6 +227,8 @@ async def start_workflow_handler( "authz_context": { "user_id": user_id, "username": username, + "provider": resolved.catalog_name, + "credential_ref": resolved.credential_ref, }, } @@ -201,10 +240,17 @@ async def start_workflow_handler( username, ) - return { + result = { "workflow_id": workflow_id, "status": "running", } + if resolved.legacy_credentials_secret: + return JSONResponse( + status_code=status.HTTP_202_ACCEPTED, + content=result, + headers={"Deprecation": LEGACY_DEPRECATION, "Sunset": LEGACY_SUNSET}, + ) + return result @router.get("/workflows/{workflow_id}") diff --git a/src/models/common/secrets.py b/src/models/common/secrets.py new file mode 100644 index 000000000..1a9057460 --- /dev/null +++ b/src/models/common/secrets.py @@ -0,0 +1,41 @@ +"""Logical secret references. + +A ``SecretRef`` names an operator-approved secret; it never carries a value +or a physical identifier (env var name, K8s Secret name, namespace, path). +The operator maps each ref to a backend location in ``workflow_engine.secrets``. +""" + +from typing import Final, Optional + +from pydantic import BaseModel, ConfigDict, Field + +SECRET_NAME_PATTERN: Final[str] = r"^[a-z0-9][a-z0-9/_.-]{0,127}$" + + +class SecretRef(BaseModel): + """Reference to an operator-registered secret by logical name. + + Attributes: + name: Logical name, e.g. ``inference/anthropic-prod``. + key: Optional key within a multi-value secret. + version: Optional secret version. + """ + + model_config = ConfigDict(extra="forbid", frozen=True) + + name: str = Field( + ..., + pattern=SECRET_NAME_PATTERN, + title="Secret name", + description="Logical secret name resolved by the operator's registry.", + ) + key: Optional[str] = Field( + default=None, + title="Key", + description="Optional key within a multi-value secret.", + ) + version: Optional[str] = Field( + default=None, + title="Version", + description="Optional secret version.", + ) diff --git a/src/models/config.py b/src/models/config.py index a6eb394c8..4db03afee 100644 --- a/src/models/config.py +++ b/src/models/config.py @@ -30,6 +30,7 @@ import constants from log import get_logger +from models.common.secrets import SECRET_NAME_PATTERN, SecretRef from utils import checks from utils.mcp_auth_headers import resolve_authorization_headers from utils.types import CompiledPatterns @@ -729,6 +730,30 @@ class UnifiedInferenceProvider(ConfigurationBase): "synthesized provider's config block.", ) + @field_validator("allowed_models") + @classmethod + def validate_allowed_models(cls, value: Optional[list[str]]) -> Optional[list[str]]: + """Reject an empty allow-list. + + An empty list is ambiguous (allow nothing vs. no restriction), so it + fails at load; omit the field for no restriction or remove the provider. + + Parameters: + value: The configured allow-list, or None when omitted. + + Returns: + The allow-list unchanged. + + Raises: + ValueError: If the list is empty. + """ + if value is not None and not value: + raise ValueError( + "allowed_models must not be empty; omit it for no restriction " + "or remove the provider" + ) + return value + @field_validator("id") @classmethod def validate_id(cls, value: Optional[str]) -> Optional[str]: @@ -3117,6 +3142,156 @@ class RedactionShieldConfiguration(ConfigurationBase): """ +ExecutorType = Literal["openai", "anthropic", "claude", "gemini", "azure", "bedrock"] + + +def approved_inference_providers() -> frozenset[str]: + """Return the inference providers the installed cloud-agents executor supports. + + Returns: + The executor's APPROVED_INFERENCE_PROVIDERS. Checked against the + installed package so drift between the stack and the executor shows + up at startup. + """ + from cloud_agents.workflow.core.execution import ( # pylint: disable=import-outside-toplevel + APPROVED_INFERENCE_PROVIDERS, + ) + + return frozenset(APPROVED_INFERENCE_PROVIDERS) + + +class SecretBinding(ConfigurationBase): + """Operator-only mapping of a logical secret name to a backend location. + + Never accepted from a request. Exactly the fields of the chosen backend + may be set. + + Attributes: + name: Logical name this binding resolves (matches ``SecretRef.name``). + backend: Where the value lives: ``env``, ``k8s`` or ``file``. + env: Environment variable name (backend ``env``). + secret_name: Kubernetes Secret name (backend ``k8s``). + key: Key within the Kubernetes Secret (backend ``k8s``). + path: File path (backend ``file``). + """ + + name: str = Field( + ..., + pattern=SECRET_NAME_PATTERN, + title="Logical secret name", + description="The SecretRef.name this binding resolves.", + ) + backend: Literal["env", "k8s", "file"] = Field( + ..., + title="Backend", + description="Where the secret value lives.", + ) + env: Optional[str] = Field( + default=None, title="Environment variable", description="Used by backend env." + ) + secret_name: Optional[str] = Field( + default=None, title="Kubernetes Secret name", description="Used by backend k8s." + ) + key: Optional[str] = Field( + default=None, + title="Secret key", + description="Key within the Secret (backend k8s).", + ) + path: Optional[str] = Field( + default=None, title="File path", description="Used by backend file." + ) + + @model_validator(mode="after") + def check_backend_fields(self) -> Self: + """Require exactly the fields of the selected backend. + + Returns: + The validated binding. + + Raises: + ValueError: If a required field is missing or a field of another + backend is set. + """ + required = { + "env": {"env"}, + "k8s": {"secret_name", "key"}, + "file": {"path"}, + }[self.backend] + present = { + field + for field in ("env", "secret_name", "key", "path") + if getattr(self, field) is not None + } + if present != required: + raise ValueError( + f"backend '{self.backend}' needs exactly {sorted(required)}, " + f"got {sorted(present)}" + ) + return self + + +class WorkflowInferenceProvider(ConfigurationBase): + """One entry in the workflow provider catalog. + + Attributes: + name: Logical name callers send. + executor_type: cloud-agents provider type this entry runs as. + credential: The single credential of this entry (a logical ref). + allowed_models: Models callers may use; None means any model. + base_url: Operator-set endpoint (Azure and self-hosted); never + settable by a caller. + """ + + name: str = Field( + ..., + pattern=SECRET_NAME_PATTERN, + title="Logical provider name", + description="Name callers send in the request.", + ) + executor_type: ExecutorType = Field( + ..., + title="Executor provider type", + description="cloud-agents provider type; must be one the executor supports.", + ) + credential: Optional[SecretRef] = Field( + default=None, + title="Credential", + description="Logical ref of the one credential this entry uses.", + ) + allowed_models: Optional[list[str]] = Field( + default=None, + title="Allowed models", + description="Models callers may select. Omit for any model; an empty " + "list is rejected (remove the provider to disable it).", + ) + base_url: Optional[str] = Field( + default=None, + title="Base URL", + description="Operator-set endpoint for Azure and self-hosted models.", + ) + + @field_validator("allowed_models") + @classmethod + def validate_allowed_models(cls, value: Optional[list[str]]) -> Optional[list[str]]: + """Reject an empty allow-list. + + Parameters: + value: The configured allow-list, or None when omitted. + + Returns: + The allow-list unchanged. + + Raises: + ValueError: If the list is empty. + """ + if value is not None and not value: + raise ValueError( + "allowed_models must not be empty; omit it for any model or " + "remove the provider" + ) + return value + + class WorkflowEngineConfiguration(ConfigurationBase): """Workflow engine configuration. @@ -3124,26 +3299,149 @@ class WorkflowEngineConfiguration(ConfigurationBase): enabled: Whether the workflow engine is active. max_concurrent_workflows: Maximum simultaneous workflow executions. transcript_retention_days: Days to retain step transcripts. + providers: Provider catalog. Empty means legacy mode; non-empty means + governed mode, where unknown names deny. + secrets: Registry mapping logical secret names to backend bindings. + default_provider: Default catalog entry on workflow paths (governed mode). + default_model: Default model on workflow paths (governed mode). """ enabled: bool = Field( - False, + default=False, title="Enable workflow engine", description="Whether the /v1/workflows endpoints are active.", ) max_concurrent_workflows: PositiveInt = Field( - 10, + default=10, title="Max concurrent workflows", description="Maximum number of simultaneously running workflows.", ) transcript_retention_days: PositiveInt = Field( - 30, + default=30, title="Transcript retention days", description="Days to retain step execution transcripts.", ) + providers: list[WorkflowInferenceProvider] = Field( + default_factory=list, + title="Workflow provider catalog", + description="Operator-approved inference providers for workflow runs. " + "Empty keeps legacy mode; non-empty switches to governed mode.", + ) + + secrets: list[SecretBinding] = Field( + default_factory=list, + title="Secret registry", + description="Maps logical secret names to backend bindings. " + "Operator-only; never accepted from a request.", + ) + + default_provider: Optional[str] = Field( + default=None, + title="Default provider", + description="Catalog entry used when a request names no provider " + "(governed mode).", + ) + + default_model: Optional[str] = Field( + default=None, + title="Default model", + description="Model used when a request names none (governed mode).", + ) + + @property + def governed(self) -> bool: + """Whether the catalog is non-empty (unknown names deny).""" + return bool(self.providers) + + def provider(self, name: str) -> Optional[WorkflowInferenceProvider]: + """Look up a catalog entry by logical name. + + Parameters: + name: Logical provider name. + + Returns: + The entry, or None when the catalog has no such name. + """ + return next((p for p in self.providers if p.name == name), None) + + def binding(self, name: str) -> Optional[SecretBinding]: + """Look up a secret binding by logical name. + + Parameters: + name: Logical secret name. + + Returns: + The binding, or None when the registry has no such name. + """ + return next((b for b in self.secrets if b.name == name), None) + + @model_validator(mode="after") + def check_catalog(self) -> Self: + """Apply the load-time checks for the provider catalog and registry. + + Returns: + The validated configuration. + + Raises: + ValueError: If names repeat, an executor type is unapproved, a + credential has no env-backed binding, or the defaults do not + resolve in the catalog. + """ + errors: list[str] = [] + for label, names in ( + ("provider", [p.name for p in self.providers]), + ("secret", [b.name for b in self.secrets]), + ): + duplicates = sorted({n for n in names if names.count(n) > 1}) + errors.extend(f"duplicate {label} name '{n}'" for n in duplicates) + + approved = approved_inference_providers() if self.providers else frozenset() + for entry in self.providers: + if entry.executor_type not in approved: + errors.append( + f"provider '{entry.name}': executor_type " + f"'{entry.executor_type}' is not approved by cloud-agents" + ) + if entry.credential is None: + continue + binding = self.binding(entry.credential.name) + if binding is None: + errors.append( + f"provider '{entry.name}': credential " + f"'{entry.credential.name}' has no secrets binding" + ) + elif binding.backend != "env": + # Before cloud-agents#269 inference credentials are read from + # os.environ by key, so only env bindings can work. + errors.append( + f"provider '{entry.name}': inference credential must be " + "env-backed before cloud-agents#269" + ) + + if self.providers: + default = self.provider(self.default_provider or "") + if default is None: + errors.append( + "default_provider must name a catalog entry when providers " + "are configured" + ) + elif not self.default_model: + errors.append("default_model must be set when providers are configured") + elif ( + default.allowed_models is not None + and self.default_model not in default.allowed_models + ): + errors.append( + f"default_model '{self.default_model}' is not allowed by " + f"provider '{default.name}'" + ) + if errors: + raise ValueError("; ".join(errors)) + return self + class SpawnerConfiguration(ConfigurationBase): """Ephemeral agent spawner configuration. diff --git a/src/workflow/catalog.py b/src/workflow/catalog.py new file mode 100644 index 000000000..7cf073768 --- /dev/null +++ b/src/workflow/catalog.py @@ -0,0 +1,257 @@ +"""Resolve workflow provider selections against the operator's catalog (issue #51). + +Callers send logical names. This module maps them to the cloud-agents provider +dict the stack builds itself, so a caller never chooses an env var name, a K8s +Secret name or any other physical identifier. + +Two modes, decided by whether ``workflow_engine.providers`` is empty: + +* governed: the catalog is authoritative, unknown names deny (400), and + ``workflow_engine.default_*`` are the defaults; +* legacy: today's behaviour (``inference.default_*`` and the built-in + provider-to-env-key table), minus any caller-chosen credential. +""" + +import copy +from dataclasses import dataclass +from typing import Any, Final, Optional + +from fastapi import HTTPException, status + +from models.config import WorkflowEngineConfiguration, WorkflowInferenceProvider +from workflow.provider_credentials import credentials_secret_for + +# RFC 9745 Deprecation (a date, not "true") and RFC 8594 Sunset for the legacy +# ``credentials_secret`` field. Deprecated 2026-10-01 (stack#51 Phase 1); +# removal planned for 2027-04-01. +LEGACY_DEPRECATION: Final[str] = "@1790812800" +LEGACY_SUNSET: Final[str] = "Thu, 01 Apr 2027 00:00:00 GMT" + +_ALLOWED_KEYS: Final[frozenset[str]] = frozenset( + {"name", "model", "credential_ref", "credentials_secret"} +) + + +@dataclass(frozen=True) +class ResolvedProvider: + """The outcome of resolving a run-level provider selection. + + Attributes: + catalog_name: Logical catalog name (the provider name itself in legacy mode). + executor_type: cloud-agents provider type the run uses. + model: Resolved model identifier. + credential_ref: Logical name of the entry's credential, if it has one. + provider: Provider dict handed to cloud-agents, built by the stack. + legacy_credentials_secret: True when the caller used the deprecated + ``credentials_secret`` field (accepted only as a logical name). + """ + + catalog_name: str + executor_type: str + model: str + credential_ref: Optional[str] + provider: dict[str, str] + legacy_credentials_secret: bool = False + + +def _bad_request(detail: str) -> HTTPException: + """Build a 400 for a name the deployment does not have or a forbidden field.""" + return HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=detail) + + +def _credential_ref_name(value: Any) -> Optional[str]: + """Extract the logical name from a ``credential_ref`` value. + + Parameters: + value: ``{"name": ...}`` as sent by the caller, or None. + + Returns: + The name, or None when no ref was sent. + + Raises: + HTTPException: 400 if the value is not an object with a string name. + """ + if value is None: + return None + if not isinstance(value, dict) or set(value) - {"name", "key", "version"}: + raise _bad_request("credential_ref must be {name: }") + name = value.get("name") + if not isinstance(name, str): + raise _bad_request("credential_ref.name must be a string") + return name + + +def _lookup( + engine: WorkflowEngineConfiguration, name: str, model: Optional[str] +) -> tuple[WorkflowInferenceProvider, str]: + """Resolve a catalog name and model, enforcing allowed_models. + + Parameters: + engine: Workflow engine configuration (governed mode). + name: Logical provider name. + model: Requested model, or None to use ``default_model``. + + Returns: + The catalog entry and the model to use. + + Raises: + HTTPException: 400 for an unknown name, a missing model, or a model + outside the entry's allowed_models. + """ + if model is not None and (not isinstance(model, str) or not model.strip()): + raise _bad_request("model must be a non-empty string.") + entry = engine.provider(name) + if entry is None: + raise _bad_request(f"Unknown provider '{name}'.") + if model is None: + if name != engine.default_provider: + raise _bad_request(f"A model is required for provider '{name}'.") + model = engine.default_model + if model is None: + raise _bad_request("No model given and no default_model configured.") + if entry.allowed_models is not None and model not in entry.allowed_models: + raise _bad_request(f"Model '{model}' is not allowed for provider '{name}'.") + return entry, model + + +def _build_provider( + engine: WorkflowEngineConfiguration, entry: WorkflowInferenceProvider, model: str +) -> dict[str, str]: + """Build the provider dict for cloud-agents from operator config only.""" + provider = {"name": entry.executor_type, "model": model} + if entry.credential is not None: + binding = engine.binding(entry.credential.name) + if binding is not None and binding.env is not None: + provider["credentials_secret"] = binding.env + return provider + + +def resolve_run_provider( + engine: WorkflowEngineConfiguration, + inference: Any, + requested: Optional[dict[str, Any]], +) -> ResolvedProvider: + """Resolve the run-level provider selection. + + Parameters: + engine: Workflow engine configuration. + inference: Inference configuration (``default_provider`` / ``default_model`` + used in legacy mode only). + requested: The ``provider`` object from the request, or None. + + Returns: + The resolved provider, with a dict built by the stack. + + Raises: + HTTPException: 400 for unknown names or models, a credential the caller + may not choose, or unknown keys. + """ + requested = requested or {} + unknown = sorted(set(requested) - _ALLOWED_KEYS) + if unknown: + raise _bad_request(f"Unknown provider keys: {unknown}.") + ref = _credential_ref_name(requested.get("credential_ref")) + + if not engine.governed: + return _resolve_legacy(inference, requested, ref) + + name = requested.get("name") or engine.default_provider + if not name: + raise _bad_request("No provider given and no default_provider configured.") + entry, model = _lookup(engine, name, requested.get("model")) + own_ref = entry.credential.name if entry.credential else None + if ref is not None and ref != own_ref: + raise _bad_request("credential_ref does not match the provider's credential.") + + legacy = False + if "credentials_secret" in requested: + # Deprecated field: accepted only as the entry's logical credential name. + if own_ref is None or requested["credentials_secret"] != own_ref: + raise _bad_request( + "credentials_secret cannot be set by callers; use credential_ref " + "with the provider's logical credential name." + ) + legacy = True + return ResolvedProvider( + catalog_name=entry.name, + executor_type=entry.executor_type, + model=model, + credential_ref=own_ref, + provider=_build_provider(engine, entry, model), + legacy_credentials_secret=legacy, + ) + + +def _resolve_legacy( + inference: Any, requested: dict[str, Any], ref: Optional[str] +) -> ResolvedProvider: + """Resolve without a catalog: today's behaviour minus caller credentials.""" + if ref is not None or "credentials_secret" in requested: + raise _bad_request( + "credential_ref and credentials_secret are not accepted: no provider " + "catalog is configured, so the stack chooses the credential." + ) + name = requested.get("name") or inference.default_provider or "" + model = requested.get("model") or inference.default_model or "" + provider = {"name": name, "model": model} + cred = credentials_secret_for(name) + if cred: + provider["credentials_secret"] = cred + return ResolvedProvider( + catalog_name=name, + executor_type=name, + model=model, + credential_ref=None, + provider=provider, + ) + + +def normalize_definition( + engine: WorkflowEngineConfiguration, + definition: dict[str, Any], + resolved: ResolvedProvider, +) -> dict[str, Any]: + """Rewrite provider overrides in a definition to executor types. + + ``definition.provider`` and each step's ``inference_provider`` must resolve + to the same catalog entry as the run provider (the model may differ): + until cloud-agents#269 gives each step its own credential lease, the + credential is picked by provider name, so another entry would silently run + on the run entry's credential. + + Parameters: + engine: Workflow engine configuration. + definition: Caller's definition (never modified). + resolved: The resolved run-level provider. + + Returns: + A normalized copy; the definition unchanged in legacy mode. + + Raises: + HTTPException: 400 for an unknown override provider, a model outside + allowed_models, or an override on a different catalog entry. + """ + if not engine.governed: + return definition + normalized = copy.deepcopy(definition) + scopes = [(normalized, "provider")] + steps = normalized.get("spec", {}).get("steps", []) + scopes += [(s, "inference_provider") for s in steps if isinstance(s, dict)] + for scope, key in scopes: + override = scope.get(key) + if override is None: + continue + if not isinstance(override, dict) or not isinstance(override.get("name"), str): + # A model-only or string override would be merged onto the run + # provider by the executor and dodge allowed_models, so it must + # name a catalog entry. + raise _bad_request(f"{key} must name a catalog provider.") + entry, model = _lookup(engine, override["name"], override.get("model")) + if entry.name != resolved.catalog_name: + raise _bad_request( + f"Provider override '{entry.name}' must use the same catalog entry " + f"as the run provider '{resolved.catalog_name}'." + ) + override["name"] = entry.executor_type + override["model"] = model + return normalized diff --git a/tests/unit/cloud_agents/test_workflow_catalog.py b/tests/unit/cloud_agents/test_workflow_catalog.py new file mode 100644 index 000000000..9f6c26cd6 --- /dev/null +++ b/tests/unit/cloud_agents/test_workflow_catalog.py @@ -0,0 +1,380 @@ +"""Unit tests for workflow provider resolution against the catalog (issue #51).""" + +# pylint: disable=too-few-public-methods + +import copy +from types import SimpleNamespace +from typing import Any, Optional + +import pytest +from fastapi import HTTPException + +from models.config import WorkflowEngineConfiguration +from workflow.catalog import ( + LEGACY_DEPRECATION, + LEGACY_SUNSET, + ResolvedProvider, + normalize_definition, + resolve_run_provider, +) + +INFERENCE = SimpleNamespace(default_provider="openai", default_model="gpt-4o-mini") + + +def _governed() -> WorkflowEngineConfiguration: + """Catalog with an env-bound pinned entry, an open entry and a keyless one.""" + return WorkflowEngineConfiguration.model_validate( + { + "providers": [ + { + "name": "claude-prod", + "executor_type": "anthropic", + "credential": {"name": "inference/anthropic-prod"}, + "allowed_models": ["claude-sonnet-4-5", "claude-haiku-4-5"], + }, + { + "name": "openai-team-b", + "executor_type": "openai", + "credential": {"name": "inference/openai-team-b"}, + }, + {"name": "local-bedrock", "executor_type": "bedrock"}, + ], + "default_provider": "claude-prod", + "default_model": "claude-sonnet-4-5", + "secrets": [ + { + "name": "inference/anthropic-prod", + "backend": "env", + "env": "ANTHROPIC_API_KEY", + }, + { + "name": "inference/openai-team-b", + "backend": "env", + "env": "OPENAI_TEAM_B_KEY", + }, + ], + } + ) + + +LEGACY = WorkflowEngineConfiguration() + + +def _status(call: Any) -> Optional[int]: + """Return the HTTP status a call raises, or None when it returns.""" + try: + call() + except HTTPException as exc: + return exc.status_code + return None + + +class TestResolveGoverned: + """Governed mode: the catalog decides provider, model and credential.""" + + def test_defaults_when_no_provider_sent(self) -> None: + """No provider uses workflow_engine.default_*, ignoring inference.default_*.""" + resolved = resolve_run_provider(_governed(), INFERENCE, None) + assert resolved.catalog_name == "claude-prod" + assert resolved.provider == { + "name": "anthropic", + "model": "claude-sonnet-4-5", + "credentials_secret": "ANTHROPIC_API_KEY", + } + + def test_provider_dict_uses_executor_type_and_operator_env(self) -> None: + """The dict handed to cloud-agents is rebuilt from the catalog entry.""" + resolved = resolve_run_provider( + _governed(), INFERENCE, {"name": "openai-team-b", "model": "gpt-4o"} + ) + assert resolved.provider == { + "name": "openai", + "model": "gpt-4o", + "credentials_secret": "OPENAI_TEAM_B_KEY", + } + assert resolved.credential_ref == "inference/openai-team-b" + + def test_entry_without_credential_gets_no_credentials_secret(self) -> None: + """An entry with no credential never gets a guessed one.""" + resolved = resolve_run_provider( + _governed(), INFERENCE, {"name": "local-bedrock", "model": "m"} + ) + assert "credentials_secret" not in resolved.provider + assert resolved.credential_ref is None + + def test_model_defaults_for_the_default_provider(self) -> None: + """Naming only the default provider reuses default_model.""" + resolved = resolve_run_provider(_governed(), INFERENCE, {"name": "claude-prod"}) + assert resolved.model == "claude-sonnet-4-5" + + def test_model_required_for_a_non_default_provider(self) -> None: + """default_model belongs to the default provider only, so others need a model.""" + assert ( + _status( + lambda: resolve_run_provider( + _governed(), INFERENCE, {"name": "openai-team-b"} + ) + ) + == 400 + ) + + def test_unknown_provider_400(self) -> None: + """A name outside the catalog is 400.""" + assert ( + _status( + lambda: resolve_run_provider( + _governed(), INFERENCE, {"name": "gpt-everything", "model": "m"} + ) + ) + == 400 + ) + + def test_model_outside_allowed_models_400(self) -> None: + """A model outside allowed_models is 400.""" + assert ( + _status( + lambda: resolve_run_provider( + _governed(), INFERENCE, {"name": "claude-prod", "model": "opus"} + ) + ) + == 400 + ) + + def test_credential_ref_must_match_the_entry(self) -> None: + """credential_ref equal to the entry's credential is fine; any other is 400.""" + ok = { + "name": "claude-prod", + "model": "claude-haiku-4-5", + "credential_ref": {"name": "inference/anthropic-prod"}, + } + assert ( + resolve_run_provider(_governed(), INFERENCE, ok).model == "claude-haiku-4-5" + ) + bad = {**ok, "credential_ref": {"name": "inference/openai-team-b"}} + assert _status(lambda: resolve_run_provider(_governed(), INFERENCE, bad)) == 400 + + def test_credential_ref_on_keyless_entry_400(self) -> None: + """An entry with no credential cannot be given one by the caller.""" + body = { + "name": "local-bedrock", + "model": "m", + "credential_ref": {"name": "inference/anthropic-prod"}, + } + assert ( + _status(lambda: resolve_run_provider(_governed(), INFERENCE, body)) == 400 + ) + + @pytest.mark.parametrize("value", ["DATABASE_URL", "OPENAI_API_KEY", None, ""]) + def test_physical_credentials_secret_400(self, value: Any) -> None: + """Env var names (or null/empty) in credentials_secret are 400.""" + body = { + "name": "claude-prod", + "model": "claude-haiku-4-5", + "credentials_secret": value, + } + assert ( + _status(lambda: resolve_run_provider(_governed(), INFERENCE, body)) == 400 + ) + + def test_legacy_credentials_secret_matching_logical_name_accepted(self) -> None: + """The legacy field is accepted only as the entry's logical credential name.""" + body = { + "name": "claude-prod", + "model": "claude-haiku-4-5", + "credentials_secret": "inference/anthropic-prod", # gitleaks:allow (logical name, not a secret) + } + resolved = resolve_run_provider(_governed(), INFERENCE, body) + assert resolved.legacy_credentials_secret is True + assert resolved.provider["credentials_secret"] == "ANTHROPIC_API_KEY" + + @pytest.mark.parametrize("model", [3, ["a"], {"a": 1}, "", " "]) + def test_model_must_be_a_non_empty_string(self, model: Any) -> None: + """A non-string or blank model is 400, whatever allowed_models says.""" + body = {"name": "openai-team-b", "model": model} + assert ( + _status(lambda: resolve_run_provider(_governed(), INFERENCE, body)) == 400 + ) + + def test_unknown_provider_keys_400(self) -> None: + """Extra keys such as base_url are 400.""" + body = {"name": "claude-prod", "model": "claude-haiku-4-5", "base_url": "x"} + assert ( + _status(lambda: resolve_run_provider(_governed(), INFERENCE, body)) == 400 + ) + + def test_input_not_mutated(self) -> None: + """The caller's dict is left alone.""" + body = {"name": "openai-team-b", "model": "gpt-4o"} + before = copy.deepcopy(body) + resolve_run_provider(_governed(), INFERENCE, body) + assert body == before + + +class TestResolveLegacyMode: + """Empty catalog: behaviour is unchanged apart from the Phase 0a hardening.""" + + def test_uses_inference_defaults_and_builtin_credential_table(self) -> None: + """inference.default_* apply and the built-in env table picks the key.""" + resolved = resolve_run_provider(LEGACY, INFERENCE, None) + assert resolved.provider == { + "name": "openai", + "model": "gpt-4o-mini", + "credentials_secret": "OPENAI_API_KEY", + } + + def test_unknown_provider_gets_no_guessed_credential(self) -> None: + """A provider outside the built-in table gets no credentials_secret.""" + resolved = resolve_run_provider( + LEGACY, INFERENCE, {"name": "bedrock", "model": "m"} + ) + assert "credentials_secret" not in resolved.provider + + def test_credential_ref_and_credentials_secret_400(self) -> None: + """There are no registry names in legacy mode, so both are rejected.""" + for extra in ( + {"credential_ref": {"name": "inference/x"}}, + {"credentials_secret": "OPENAI_API_KEY"}, + ): + body = {"name": "openai", "model": "gpt-4o", **extra} + assert ( + _status(lambda b=body: resolve_run_provider(LEGACY, INFERENCE, b)) + == 400 + ) + + +def _resolved(name: str = "claude-prod") -> ResolvedProvider: + """Resolve a governed entry for the normalization tests.""" + model = "claude-sonnet-4-5" if name == "claude-prod" else "gpt-4o" + return resolve_run_provider(_governed(), INFERENCE, {"name": name, "model": model}) + + +def _definition(**top: Any) -> dict[str, Any]: + """Two-step definition with an optional workflow provider.""" + steps = top.pop("steps", [{"name": "a", "type": "agent", "output_key": "x"}]) + return { + "apiVersion": "v1", + "kind": "AgentWorkflow", + "metadata": {"name": "t"}, + "spec": {"steps": steps}, + **top, + } + + +class TestNormalizeDefinition: + """Overrides resolve through the catalog and must stay on the run's entry.""" + + def test_definition_provider_rewritten_to_executor_type(self) -> None: + """definition.provider's logical name becomes the executor type.""" + definition = _definition( + provider={"name": "claude-prod", "model": "claude-haiku-4-5"} + ) + out = normalize_definition(_governed(), definition, _resolved()) + assert out["provider"] == {"name": "anthropic", "model": "claude-haiku-4-5"} + + def test_step_inference_provider_rewritten(self) -> None: + """A step's inference_provider is rewritten the same way.""" + steps = [ + { + "name": "a", + "type": "agent", + "output_key": "x", + "inference_provider": {"name": "openai-team-b", "model": "gpt-4o-mini"}, + } + ] + out = normalize_definition( + _governed(), _definition(steps=steps), _resolved("openai-team-b") + ) + assert out["spec"]["steps"][0]["inference_provider"] == { + "name": "openai", + "model": "gpt-4o-mini", + } + + def test_override_on_another_entry_400(self) -> None: + """Same-entry rule: an override on a different catalog entry is 400.""" + definition = _definition(provider={"name": "openai-team-b", "model": "gpt-4o"}) + assert ( + _status(lambda: normalize_definition(_governed(), definition, _resolved())) + == 400 + ) + + def test_same_executor_type_different_entry_400(self) -> None: + """Two entries sharing an executor type are still different entries.""" + engine = WorkflowEngineConfiguration.model_validate( + { + "providers": [ + {"name": "a", "executor_type": "openai"}, + {"name": "b", "executor_type": "openai"}, + ], + "default_provider": "a", + "default_model": "m", + } + ) + resolved = resolve_run_provider(engine, INFERENCE, {"name": "a", "model": "m"}) + definition = _definition(provider={"name": "b", "model": "m"}) + assert ( + _status(lambda: normalize_definition(engine, definition, resolved)) == 400 + ) + + def test_unknown_override_provider_400(self) -> None: + """An override naming a provider outside the catalog is 400.""" + definition = _definition(provider={"name": "nope", "model": "m"}) + assert ( + _status(lambda: normalize_definition(_governed(), definition, _resolved())) + == 400 + ) + + def test_override_model_must_be_allowed(self) -> None: + """An override model outside allowed_models is 400.""" + definition = _definition(provider={"name": "claude-prod", "model": "opus"}) + assert ( + _status(lambda: normalize_definition(_governed(), definition, _resolved())) + == 400 + ) + + @pytest.mark.parametrize( + "override", [{"model": "gpt-evil"}, {"name": "", "model": "m"}, "gpt-evil"] + ) + def test_unnamed_or_malformed_override_400(self, override: Any) -> None: + """An override that names no catalog entry cannot dodge allowed_models.""" + for definition in ( + _definition(provider=override), + _definition( + steps=[ + { + "name": "a", + "type": "agent", + "output_key": "x", + "inference_provider": override, + } + ] + ), + ): + assert ( + _status( + lambda d=definition: normalize_definition( + _governed(), d, _resolved() + ) + ) + == 400 + ) + + def test_input_not_mutated(self) -> None: + """The caller's definition is never modified; a copy is returned.""" + definition = _definition( + provider={"name": "claude-prod", "model": "claude-haiku-4-5"} + ) + before = copy.deepcopy(definition) + normalize_definition(_governed(), definition, _resolved()) + assert definition == before + + def test_legacy_mode_is_a_no_op(self) -> None: + """Without a catalog the definition is returned unchanged.""" + resolved = resolve_run_provider(LEGACY, INFERENCE, None) + definition = _definition(provider={"name": "openai", "model": "gpt-4o"}) + assert normalize_definition(LEGACY, definition, resolved) == definition + + +def test_deprecation_headers_are_rfc_shaped() -> None: + """Deprecation is an RFC 9745 @timestamp; Sunset is an HTTP date (RFC 8594).""" + assert LEGACY_DEPRECATION.startswith("@") and LEGACY_DEPRECATION[1:].isdigit() + assert LEGACY_SUNSET.endswith("GMT") diff --git a/tests/unit/cloud_agents/test_workflow_contract.py b/tests/unit/cloud_agents/test_workflow_contract.py new file mode 100644 index 000000000..c157978e3 --- /dev/null +++ b/tests/unit/cloud_agents/test_workflow_contract.py @@ -0,0 +1,134 @@ +"""Run examples/workflows/cases.yaml against the real handler and config (issue #51). + +The cases are the acceptance contract for the #51 phases. Each case carries the +phase that makes it real; cases for later phases are skipped, so raising +``IMPLEMENTED_PHASE`` (and removing the matching part of the example's +``reference_gate.py``) is how a phase is declared done. + +Only the HTTP status is compared here: the reference gate's ``reason`` strings +are specific to the reference implementation. +""" + +# pyright: reportMissingImports=false +# pylint: disable=import-outside-toplevel,redefined-outer-name,wrong-import-position,import-error + +import sys +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +import pytest +from fastapi import HTTPException +from pydantic import ValidationError +from pytest_mock import MockerFixture + +EXAMPLES = Path(__file__).resolve().parents[3] / "examples" / "workflows" +sys.path.insert(0, str(EXAMPLES)) + +from contract_util import case_body, load, patch + +from app.endpoints.workflows import start_workflow_handler +from models.config import WorkflowEngineConfiguration + +# Highest #51 phase implemented in src/. Phase 0 is the Phase 0a hardening. +IMPLEMENTED_PHASE = 1 + +# workflow_engine keys of the example config whose phase is not built yet. +PENDING_ENGINE_KEYS = {"policy"} + +CASES = load("cases.yaml") +STACK = load("lightspeed-stack.yaml") +DEFINITIONS = { + name: load(f"{name}.yaml") + for name in ("triage-github-issue", "kb-answer-with-approval", "admin-inline-mcp") +} + + +def _engine(stack: dict[str, Any]) -> WorkflowEngineConfiguration: + """Build the real engine configuration from an example stack config.""" + engine = { + k: v + for k, v in stack["workflow_engine"].items() + if k not in PENDING_ENGINE_KEYS + } + return WorkflowEngineConfiguration.model_validate(engine) + + +def _params(section: str) -> list[Any]: + """Parametrize over a cases.yaml section, skipping later phases.""" + out = [] + for case in CASES[section]: + marks = [] + if case["phase"] > IMPLEMENTED_PHASE: + marks.append(pytest.mark.skip(reason=f"phase {case['phase']} not built")) + out.append(pytest.param(case, id=case["name"], marks=marks)) + return out + + +def _expected(case: dict[str, Any]) -> int: + """Expected status before cloud-agents#269.""" + exp = case["expect"] + return exp["pre269"] if isinstance(exp, dict) else exp + + +@pytest.mark.parametrize("case", _params("workflow_cases")) +@pytest.mark.asyncio +async def test_workflow_case(case: dict[str, Any], mocker: MockerFixture) -> None: + """The real POST /v1/workflows/run handler gives the contract's status.""" + import app.endpoints.workflows as wf_mod + + roles = set(case.get("as", CASES["defaults"][case["workflow"]]["as"])) + cfg = mocker.patch("app.endpoints.workflows.configuration") + cfg.workflow_engine_configuration = _engine(STACK) + cfg.inference = SimpleNamespace(default_provider=None, default_model=None) + cfg.spawner_configuration = ( + SimpleNamespace(sandbox_image=STACK["spawner"]["sandbox_image"]) + if case.get("spawner", True) + else None + ) + mocker.patch("app.endpoints.workflows.check_configuration_loaded") + mocker.patch( + "app.endpoints.workflows.is_admin", return_value="agent-admin" in roles + ) + executor = mocker.AsyncMock() + executor.start.return_value = "wf-contract" + wf_mod._executor = executor # pylint: disable=protected-access + body = case_body(case, DEFINITIONS, CASES["defaults"]) + from models.api.requests.agents import RunWorkflowRequest + + try: + request = RunWorkflowRequest(**body) + await start_workflow_handler.__wrapped__( + mocker.MagicMock(), request, ("uid", "user", False, "token") + ) + status = 202 + except HTTPException as exc: + status = exc.status_code + finally: + wf_mod._executor = None # pylint: disable=protected-access + + assert status == _expected(case) + assert executor.start.called == (status == 202) + + +@pytest.mark.parametrize("case", _params("config_cases")) +def test_config_case(case: dict[str, Any]) -> None: + """The real WorkflowEngineConfiguration fails to load as the contract says.""" + stack = patch(STACK, case["set"]) + with pytest.raises(ValidationError) as exc_info: + _engine(stack) + assert case["error"].lower() in str(exc_info.value).lower() + + +def test_phase_tags_are_valid() -> None: + """Every case is tagged with a known phase, so none is silently never run.""" + for section in ("workflow_cases", "config_cases"): + for case in CASES[section]: + assert case["phase"] in (0, 1, 2, 3), case["name"] + + +def test_example_config_loads_without_pending_blocks() -> None: + """The example's implemented blocks load as a real WorkflowEngineConfiguration.""" + cfg = _engine(STACK) + assert cfg.governed + assert cfg.default_provider == "claude-prod" diff --git a/tests/unit/cloud_agents/test_workflows_endpoint.py b/tests/unit/cloud_agents/test_workflows_endpoint.py index befe10540..023ac407d 100644 --- a/tests/unit/cloud_agents/test_workflows_endpoint.py +++ b/tests/unit/cloud_agents/test_workflows_endpoint.py @@ -19,6 +19,7 @@ start_workflow_handler, ) from models.api.requests.agents import ApproveWorkflowRequest, RunWorkflowRequest +from models.config import WorkflowEngineConfiguration def _request(mocker: MockerFixture, admin: bool = True) -> Any: @@ -42,8 +43,7 @@ def reset_executor() -> Generator[None, None, None]: def mock_config_fixture(mocker: MockerFixture) -> Any: """Mock the configuration singleton.""" mock_cfg = mocker.patch("app.endpoints.workflows.configuration") - mock_cfg.workflow_engine_configuration = mocker.MagicMock() - mock_cfg.workflow_engine_configuration.enabled = True + mock_cfg.workflow_engine_configuration = WorkflowEngineConfiguration(enabled=True) mock_cfg.inference = mocker.MagicMock() mock_cfg.inference.default_model = "gpt-4o" mock_cfg.inference.default_provider = "openai" @@ -325,7 +325,7 @@ async def test_secret_bearing_mcp_definition_rejected_with_422( mock_executor.start.assert_not_called() @pytest.mark.asyncio - async def test_secret_shaped_run_credentials_rejected_with_422( + async def test_secret_shaped_run_credentials_rejected_with_400( self, mocker: MockerFixture, mock_config: Any, @@ -351,7 +351,7 @@ async def test_secret_shaped_run_credentials_rejected_with_422( with pytest.raises(HTTPException) as exc_info: await start_workflow_handler.__wrapped__(request, body, auth) - assert exc_info.value.status_code == 422 + assert exc_info.value.status_code == 400 mock_executor.start.assert_not_called() @pytest.mark.asyncio @@ -975,3 +975,176 @@ async def test_returns_transcripts( assert result["workflow_id"] == "wf-1" assert "analyze" in result["transcripts"] + + +def _governed_engine() -> WorkflowEngineConfiguration: + """Catalog used by the governed-mode handler tests.""" + return WorkflowEngineConfiguration.model_validate( + { + "enabled": True, + "providers": [ + { + "name": "claude-prod", + "executor_type": "anthropic", + "credential": {"name": "inference/anthropic-prod"}, + "allowed_models": ["claude-sonnet-4-5", "claude-haiku-4-5"], + }, + { + "name": "openai-team-b", + "executor_type": "openai", + "credential": {"name": "inference/openai-team-b"}, + }, + ], + "default_provider": "claude-prod", + "default_model": "claude-sonnet-4-5", + "secrets": [ + { + "name": "inference/anthropic-prod", + "backend": "env", + "env": "ANTHROPIC_API_KEY", + }, + { + "name": "inference/openai-team-b", + "backend": "env", + "env": "OPENAI_TEAM_B_KEY", + }, + ], + } + ) + + +class TestStartWorkflowGoverned: + """start_workflow_handler against a provider catalog (issue #51, Phase 1).""" + + @pytest.fixture(autouse=True) + def _governed(self, mock_config: Any) -> None: + """Switch the shared config to governed mode with an ephemeral spawner.""" + mock_config.workflow_engine_configuration = _governed_engine() + mock_config.spawner_configuration = None + + @staticmethod + def _definition() -> dict[str, Any]: + """One-step workflow on spawn none (the tests run as admin).""" + return _valid_definition() + + @pytest.mark.asyncio + async def test_default_provider_rebuilt_from_catalog( + self, mocker: MockerFixture, mock_executor: Any + ) -> None: + """No provider sent: the run uses workflow_engine defaults, built by the stack.""" + mocker.patch("app.endpoints.workflows.check_configuration_loaded") + mock_executor.start.return_value = "wf-1" + body = RunWorkflowRequest(definition=self._definition()) + + await start_workflow_handler.__wrapped__( + _request(mocker), body, ("u", "n", False, "t") + ) + + workflow_input = mock_executor.start.call_args[0][0] + assert workflow_input["provider"] == { + "name": "anthropic", + "model": "claude-sonnet-4-5", + "credentials_secret": "ANTHROPIC_API_KEY", + } + assert workflow_input["authz_context"]["provider"] == "claude-prod" + assert ( + workflow_input["authz_context"]["credential_ref"] + == "inference/anthropic-prod" + ) + + @pytest.mark.asyncio + async def test_unknown_provider_400_nothing_started( + self, mocker: MockerFixture, mock_executor: Any + ) -> None: + """A name outside the catalog is 400 before anything is persisted.""" + from fastapi import HTTPException + + mocker.patch("app.endpoints.workflows.check_configuration_loaded") + body = RunWorkflowRequest( + definition=self._definition(), provider={"name": "gpt-x", "model": "m"} + ) + with pytest.raises(HTTPException) as exc_info: + await start_workflow_handler.__wrapped__( + _request(mocker), body, ("u", "n", False, "t") + ) + assert exc_info.value.status_code == 400 + mock_executor.start.assert_not_called() + + @pytest.mark.asyncio + async def test_overrides_rewritten_and_same_entry_enforced( + self, mocker: MockerFixture, mock_executor: Any + ) -> None: + """definition.provider is rewritten to the executor type in what is started.""" + from fastapi import HTTPException + + mocker.patch("app.endpoints.workflows.check_configuration_loaded") + mock_executor.start.return_value = "wf-1" + definition = self._definition() + definition["provider"] = {"name": "claude-prod", "model": "claude-haiku-4-5"} + body = RunWorkflowRequest(definition=definition) + + await start_workflow_handler.__wrapped__( + _request(mocker), body, ("u", "n", False, "t") + ) + started = mock_executor.start.call_args[0][0]["definition"] + assert started["provider"] == {"name": "anthropic", "model": "claude-haiku-4-5"} + assert ( + body.definition["provider"]["name"] == "claude-prod" + ) # caller copy intact + + definition["provider"] = {"name": "openai-team-b", "model": "gpt-4o"} + with pytest.raises(HTTPException) as exc_info: + await start_workflow_handler.__wrapped__( + _request(mocker), + RunWorkflowRequest(definition=definition), + ("u", "n", False, "t"), + ) + assert exc_info.value.status_code == 400 + + @pytest.mark.asyncio + async def test_legacy_credentials_secret_sets_deprecation_headers( + self, mocker: MockerFixture, mock_executor: Any + ) -> None: + """The deprecated field (logical name only) works but is flagged (RFC 9745/8594).""" + from workflow.catalog import LEGACY_DEPRECATION, LEGACY_SUNSET + + mocker.patch("app.endpoints.workflows.check_configuration_loaded") + mock_executor.start.return_value = "wf-1" + body = RunWorkflowRequest( + definition=self._definition(), + provider={ + "name": "claude-prod", + "model": "claude-haiku-4-5", + "credentials_secret": "inference/anthropic-prod", # gitleaks:allow (logical name, not a secret) + }, + ) + + response = await start_workflow_handler.__wrapped__( + _request(mocker), body, ("u", "n", False, "t") + ) + + assert response.status_code == 202 + assert response.headers["Deprecation"] == LEGACY_DEPRECATION + assert response.headers["Sunset"] == LEGACY_SUNSET + started = mock_executor.start.call_args[0][0] + assert started["provider"]["credentials_secret"] == "ANTHROPIC_API_KEY" + + @pytest.mark.asyncio + async def test_canonical_form_has_no_deprecation_headers( + self, mocker: MockerFixture, mock_executor: Any + ) -> None: + """The canonical form returns a plain dict.""" + mocker.patch("app.endpoints.workflows.check_configuration_loaded") + mock_executor.start.return_value = "wf-1" + body = RunWorkflowRequest( + definition=self._definition(), + provider={ + "name": "claude-prod", + "model": "claude-haiku-4-5", + "credential_ref": {"name": "inference/anthropic-prod"}, + }, + ) + response = await start_workflow_handler.__wrapped__( + _request(mocker), body, ("u", "n", False, "t") + ) + assert response == {"workflow_id": "wf-1", "status": "running"} diff --git a/tests/unit/models/config/test_dump_configuration.py b/tests/unit/models/config/test_dump_configuration.py index c7ee17579..b3d664336 100644 --- a/tests/unit/models/config/test_dump_configuration.py +++ b/tests/unit/models/config/test_dump_configuration.py @@ -278,6 +278,10 @@ def test_dump_configuration_minimal_cfg(tmp_path: Path) -> None: "enabled": False, "max_concurrent_workflows": 10, "transcript_retention_days": 30, + "providers": [], + "secrets": [], + "default_provider": None, + "default_model": None, }, } @@ -524,6 +528,10 @@ def test_dump_configuration_valid_values(tmp_path: Path) -> None: "enabled": False, "max_concurrent_workflows": 10, "transcript_retention_days": 30, + "providers": [], + "secrets": [], + "default_provider": None, + "default_model": None, }, } @@ -921,6 +929,10 @@ def test_dump_configuration_with_quota_limiters(tmp_path: Path) -> None: "enabled": False, "max_concurrent_workflows": 10, "transcript_retention_days": 30, + "providers": [], + "secrets": [], + "default_provider": None, + "default_model": None, }, } @@ -1202,6 +1214,10 @@ def test_dump_configuration_with_quota_limiters_different_values( "enabled": False, "max_concurrent_workflows": 10, "transcript_retention_days": 30, + "providers": [], + "secrets": [], + "default_provider": None, + "default_model": None, }, } @@ -1523,6 +1539,10 @@ def test_dump_configuration_byok(tmp_path: Path) -> None: "enabled": False, "max_concurrent_workflows": 10, "transcript_retention_days": 30, + "providers": [], + "secrets": [], + "default_provider": None, + "default_model": None, }, } @@ -1764,6 +1784,10 @@ def test_dump_configuration_pg_namespace(tmp_path: Path) -> None: "enabled": False, "max_concurrent_workflows": 10, "transcript_retention_days": 30, + "providers": [], + "secrets": [], + "default_provider": None, + "default_model": None, }, } @@ -2165,6 +2189,10 @@ def test_dump_configuration_allow_degraded_mode(tmp_path: Path) -> None: "enabled": False, "max_concurrent_workflows": 10, "transcript_retention_days": 30, + "providers": [], + "secrets": [], + "default_provider": None, + "default_model": None, }, } @@ -2412,6 +2440,10 @@ def test_dump_configuration_max_retries_settings(tmp_path: Path) -> None: "enabled": False, "max_concurrent_workflows": 10, "transcript_retention_days": 30, + "providers": [], + "secrets": [], + "default_provider": None, + "default_model": None, }, } @@ -2659,6 +2691,10 @@ def test_dump_configuration_retry_count_settings(tmp_path: Path) -> None: "enabled": False, "max_concurrent_workflows": 10, "transcript_retention_days": 30, + "providers": [], + "secrets": [], + "default_provider": None, + "default_model": None, }, } @@ -2910,5 +2946,9 @@ def test_dump_configuration_specific_compaction_values(tmp_path: Path) -> None: "enabled": False, "max_concurrent_workflows": 10, "transcript_retention_days": 30, + "providers": [], + "secrets": [], + "default_provider": None, + "default_model": None, }, } diff --git a/tests/unit/models/config/test_workflow_catalog_configuration.py b/tests/unit/models/config/test_workflow_catalog_configuration.py new file mode 100644 index 000000000..c54edcf49 --- /dev/null +++ b/tests/unit/models/config/test_workflow_catalog_configuration.py @@ -0,0 +1,250 @@ +"""Unit tests for the workflow provider catalog and secret registry (issue #51).""" + +# pylint: disable=too-few-public-methods + +from typing import Any, Optional + +import pytest +from pydantic import ValidationError + +from models.common.secrets import SecretRef +from models.config import ( + InferenceConfiguration, + SecretBinding, + UnifiedInferenceProvider, + WorkflowEngineConfiguration, + WorkflowInferenceProvider, +) + + +def _engine(**overrides: Any) -> dict[str, Any]: + """Valid governed-mode workflow_engine config; ``overrides`` replace keys.""" + config: dict[str, Any] = { + "enabled": True, + "providers": [ + { + "name": "claude-prod", + "executor_type": "anthropic", + "credential": {"name": "inference/anthropic-prod"}, + "allowed_models": ["claude-sonnet-4-5", "claude-haiku-4-5"], + }, + { + "name": "openai-team-b", + "executor_type": "openai", + "credential": {"name": "inference/openai-team-b"}, + }, + ], + "default_provider": "claude-prod", + "default_model": "claude-sonnet-4-5", + "secrets": [ + { + "name": "inference/anthropic-prod", + "backend": "env", + "env": "ANTHROPIC_API_KEY", + }, + { + "name": "inference/openai-team-b", + "backend": "env", + "env": "OPENAI_TEAM_B_KEY", + }, + ], + } + config.update(overrides) + return config + + +def _error(config: dict[str, Any]) -> str: + """Return the validation error text for an invalid config.""" + with pytest.raises(ValidationError) as exc_info: + WorkflowEngineConfiguration.model_validate(config) + return str(exc_info.value).lower() + + +class TestSecretRef: + """SecretRef holds a logical name, never a physical identifier.""" + + def test_valid_ref(self) -> None: + """A logical name with a slash is accepted and the model is frozen.""" + ref = SecretRef(name="inference/anthropic-prod") + assert ref.name == "inference/anthropic-prod" + with pytest.raises(ValidationError): + ref.name = "other" # type: ignore[misc] + + @pytest.mark.parametrize( + "name", ["", "UPPER", "has space", "-lead", "a" * 129, "../etc/passwd"] + ) + def test_invalid_names_rejected(self, name: str) -> None: + """Names outside ^[a-z0-9][a-z0-9/_.-]{0,127}$ are rejected.""" + with pytest.raises(ValidationError): + SecretRef(name=name) + + def test_extra_fields_rejected(self) -> None: + """Unknown fields (e.g. an env var name) are rejected.""" + with pytest.raises(ValidationError): + SecretRef(name="a/b", env="DATABASE_URL") # type: ignore[call-arg] + + +class TestSecretBinding: + """SecretBinding maps a logical ref to exactly one backend location.""" + + def test_env_binding(self) -> None: + """An env binding needs only ``env``.""" + binding = SecretBinding(name="a/b", backend="env", env="MY_KEY") + assert binding.env == "MY_KEY" + + def test_k8s_binding_needs_secret_name_and_key(self) -> None: + """A k8s binding needs secret_name and key.""" + SecretBinding(name="a/b", backend="k8s", secret_name="s", key="k") + with pytest.raises(ValidationError): + SecretBinding(name="a/b", backend="k8s", secret_name="s") + + def test_file_binding_needs_path(self) -> None: + """A file binding needs path.""" + SecretBinding(name="a/b", backend="file", path="/run/secrets/x") + with pytest.raises(ValidationError): + SecretBinding(name="a/b", backend="file") + + def test_fields_of_other_backends_rejected(self) -> None: + """An env binding may not also carry k8s or file fields.""" + with pytest.raises(ValidationError): + SecretBinding(name="a/b", backend="env", env="K", secret_name="s") + with pytest.raises(ValidationError): + SecretBinding(name="a/b", backend="env", env="K", path="/x") + + def test_env_binding_requires_env(self) -> None: + """An env binding without ``env`` is rejected.""" + with pytest.raises(ValidationError): + SecretBinding(name="a/b", backend="env") # type: ignore[call-arg] + + +class TestWorkflowInferenceProvider: + """A catalog entry holds one credential and an optional model allow-list.""" + + def test_defaults(self) -> None: + """Optional fields default to None.""" + entry = WorkflowInferenceProvider(name="p", executor_type="openai") + assert entry.credential is None + assert entry.allowed_models is None + assert entry.base_url is None + + def test_empty_allowed_models_rejected(self) -> None: + """allowed_models [] is rejected (G4): remove the provider instead.""" + with pytest.raises(ValidationError, match="allowed_models"): + WorkflowInferenceProvider( + name="p", executor_type="openai", allowed_models=[] + ) + + def test_unknown_executor_type_rejected(self) -> None: + """Only executor types cloud-agents knows are accepted.""" + with pytest.raises(ValidationError): + WorkflowInferenceProvider(name="p", executor_type="ollama") # type: ignore[arg-type] + + def test_credential_is_a_secret_ref(self) -> None: + """The credential is a SecretRef, so env var names are not accepted.""" + entry = WorkflowInferenceProvider( + name="p", executor_type="openai", credential={"name": "a/b"} # type: ignore[arg-type] + ) + assert isinstance(entry.credential, SecretRef) + + +class TestWorkflowEngineLoadChecks: + """Config load fails fast on the plan's load-time checks (Phase 1 subset).""" + + def test_valid_governed_config_loads(self) -> None: + """The example catalog loads and exposes governed mode.""" + cfg = WorkflowEngineConfiguration.model_validate(_engine()) + assert cfg.governed is True + assert [p.name for p in cfg.providers] == ["claude-prod", "openai-team-b"] + + def test_empty_catalog_is_legacy_mode(self) -> None: + """No providers means legacy mode and none of the catalog checks apply.""" + cfg = WorkflowEngineConfiguration.model_validate({"enabled": True}) + assert cfg.governed is False + assert cfg.providers == [] + + def test_executor_type_must_be_approved(self, mocker: Any) -> None: + """executor_type must be in cloud-agents APPROVED_INFERENCE_PROVIDERS.""" + mocker.patch( + "models.config.approved_inference_providers", + return_value=frozenset({"openai"}), + ) + assert "not approved" in _error(_engine()) + + def test_credential_needs_a_binding(self) -> None: + """A credential ref with no secrets binding fails load.""" + config = _engine() + config["providers"][0]["credential"] = {"name": "inference/missing"} + assert "no secrets binding" in _error(config) + + def test_pre_269_credential_must_be_env_backed(self) -> None: + """Before cloud-agents#269 inference credentials must be env-backed.""" + config = _engine() + config["secrets"][0] = { + "name": "inference/anthropic-prod", + "backend": "k8s", + "secret_name": "s", + "key": "k", + } + assert "must be env" in _error(config) + + def test_default_provider_must_be_in_catalog(self) -> None: + """default_provider must name a catalog entry.""" + assert "default_provider" in _error(_engine(default_provider="gone")) + + def test_default_model_must_pass_allowed_models(self) -> None: + """default_model must satisfy the default provider's allowed_models.""" + assert "default_model" in _error(_engine(default_model="claude-opus-4")) + + def test_defaults_required_in_governed_mode(self) -> None: + """A catalog without a default provider is rejected (no silent fallback).""" + config = _engine() + del config["default_provider"] + assert "default_provider" in _error(config) + + def test_default_model_required_in_governed_mode(self) -> None: + """No default_model fails at load even when the entry allows any model.""" + config = _engine(default_provider="openai-team-b") + del config["default_model"] + assert "default_model" in _error(config) + + def test_duplicate_provider_names_rejected(self) -> None: + """Provider names are unique.""" + config = _engine() + config["providers"].append(dict(config["providers"][0])) + assert "duplicate" in _error(config) + + def test_duplicate_secret_names_rejected(self) -> None: + """Secret names are unique.""" + config = _engine() + config["secrets"].append(dict(config["secrets"][0])) + assert "duplicate" in _error(config) + + def test_secrets_without_catalog_are_allowed(self) -> None: + """A registry alone (legacy mode) is harmless and loads.""" + cfg = WorkflowEngineConfiguration.model_validate( + {"secrets": _engine()["secrets"]} + ) + assert cfg.governed is False + + def test_provider_lookup_helpers(self) -> None: + """Lookup helpers return entries and bindings by logical name.""" + cfg = WorkflowEngineConfiguration.model_validate(_engine()) + assert cfg.provider("claude-prod").executor_type == "anthropic" # type: ignore[union-attr] + assert cfg.provider("nope") is None + binding: Optional[SecretBinding] = cfg.binding("inference/openai-team-b") + assert binding is not None and binding.env == "OPENAI_TEAM_B_KEY" + assert cfg.binding("nope") is None + + +class TestInferenceAllowedModels: + """The same [] rejection applies to the Llama Stack provider config (G4).""" + + def test_empty_allowed_models_rejected_for_llama_stack_providers(self) -> None: + """inference.providers rejects allowed_models [] so both consumers agree.""" + with pytest.raises(ValidationError, match="allowed_models"): + provider = UnifiedInferenceProvider( # pyright: ignore[reportCallIssue] + type="openai", allowed_models=[] + ) + InferenceConfiguration( # pyright: ignore[reportCallIssue] + providers=[provider] + )