Gravadora Asset Core Decision Gate
Docs
Decision Gate Docs

Avaluació de portes determinista, reproduïble amb decisions auditable.

Asset Core docs

Manual d’OpenAPI Tipat

A Simple Vista

Què: Importar OpenAPI en artefactes de proveïdor tipats i gestionar les versions del cicle de vida. Per què: Obtenir artefactes de capacitat/runtime estrictes sense construir un servidor MCP personalitzat. Qui: Operadors que integren APIs basades en OpenAPI a Decision Gate. Requisits previs: provider_schema_authoring.md, condition_authoring.md, openapi_reference_library.md

Flux d’Eines

Eines de cicle de vida tipades:

  1. proveïdors_tipats_import
  2. typed_providers_register
  3. typed_providers_list
  4. typed_providers_get
  5. typed_providers_activate
  6. typed_providers_deprecate

typed_providers_import és el camí OSS principal per a fluxos de treball impulsats per OpenAPI. Compila artefactes de contracte/runtime deterministes i registra una versió del cicle de vida en una sola operació.

Tots els eines de cicle de vida tipades requereixen camps d’abast explícits:

  • tenant_id
  • namespace_id

Importar Knobs i Modes de Conformitat

typed_providers_import admet controls limitats per a una importació determinista:

  • credential_bindings: map of OpenAPI security scheme id -> structured binding (locator + value_render requerits; display_name opcional)
  • openapi_semantics_mode: automàtic | oas30 | oas31
  • media_support_mode: json_only | all_media
  • external_ref_mode: fitxer_local_sola | llista_permesa_de_xarxa
  • openapi_conformance_mode: estricte | auditoria

strict rebutja construccions d’OpenAPI no suportades. audit permet el registre mentre emet troballes no suportades deterministes en conformance_summary. credential_bindings és sempre requerit pel payload de l’eina; els camps omesos fallen en la decodificació/validació. Proporcioneu un mapa buit per a importacions no autenticades i mapeigs explícits per a operacions segures. Els esquemes de localització permesos són només secret://... i env://...; la resolució env://... està desactivada per defecte i requereix dev.allow_dev_env_credentials=true. value_render és explícit i falla tancat:

  • {"mode":"identity"} utilitza el secret en brut tal com és.
  • {"mode":"prefix","prefix":"Token "} prepends a deterministic prefix. Notes de provisió secreta:
  • Store raw provider credentials in the encrypted secret store and reference by secret://... localitzador.
  • If keyring is unavailable/headless, set DECISION_GATE_SECRETS_PASSPHRASE before running secrets commands so the store can be unlocked. JSON object and array-complex response schemas must declare x-decision-gate.projections; missing projection metadata fails import. Legacy response_projection_mode input is removed and rejected. Projection metadata is evaluated on normalized/resolved schemas, so component-level metadata via $ref is first-class. No duplicar esquemes de resposta en línia només per portar projeccions.

El comportament semàntic s’aplica en el moment de la importació:

  • auto: inferir semàntica de l’encapçalament openapi (3.0.x o 3.1.x).
  • oas30: l’encapçalament ha de ser 3.0.x; els arrays del tipus JSON Schema són rebutjats.
  • oas31: l’encapçalament ha de ser 3.1.x; la paraula clau nullable llegat és rebutjada.

Perfil d’execució i Semàntica de Desviació

Els registres de cicle de vida tipats i els perfils d’execució porten metadades de resum:

  • source_digest: digest de l’entrada de font importada
  • profile_digest: digest del perfil d’execució generat
  • contract_hash: hash canònic del contracte del proveïdor generat

typed_providers_get pot calcular la deriva amb observed_source_digest + observed_profile_digest i retorna drift_status quan es detecten discrepàncies.

Configuració del Proveïdor Tipat

Els proveïdors tipats es configuren amb artefactes preconstruïts:

[[providers]]
name = "typed_asset_api"
type = "typed"
capabilities_path = "contracts/typed_asset_api.json"
runtime_profile_path = "profiles/typed_asset_api_runtime.json"
typed_protocol = "openapi_http"

Exemple de validació:

[server]
transport = "http"
bind = "127.0.0.1:4000"
mode = "strict"

[server.auth]
mode = "local_only"

[namespace]
allow_default = true
default_tenants = [1]

[trust]
default_policy = "audit"
min_lane = "verified"

[evidence]
allow_raw_values = false
require_provider_opt_in = true

[schema_registry]
type = "sqlite"
path = "decision-gate-registry.db"

[schema_registry.acl]
allow_local_only = true
require_signing = false

[run_state_store]
type = "sqlite"
path = "decision-gate.db"
journal_mode = "wal"
sync_mode = "full"
busy_timeout_ms = 5000

[[providers]]
name = "time"
type = "builtin"

[[providers]]
name = "env"
type = "builtin"

[[providers]]
name = "json"
type = "builtin"
config = { root = "./evidence", root_id = "evidence-root", max_bytes = 1048576, allow_yaml = true }

[[providers]]
name = "http"
type = "builtin"

[[providers]]
name = "rest"
type = "builtin"

[[providers]]
name = "typed_asset_api"
type = "typed"
capabilities_path = "contracts/typed_asset_api.json"
runtime_profile_path = "profiles/typed_asset_api_runtime.json"
typed_protocol = "openapi_http"

Càrrega d’Importació OpenAPI

Forma mínima de càrrega d’importació:

{
  "provider_id": "typed_asset_api",
  "version": "v1",
  "openapi": {
    "openapi": "3.1.0",
    "paths": {
      "/assets": {
        "get": {
          "operationId": "listAssets",
          "responses": {
            "200": {
              "description": "ok",
              "content": {
                "application/json": {
                  "schema": {
                    "type": "object",
                    "properties": {
                      "next_cursor": { "type": "string" }
                    },
                    "required": ["next_cursor"],
                    "additionalProperties": false,
                    "x-decision-gate": {
                      "projections": [
                        {
                          "id": "next_cursor",
                          "pointer": "/next_cursor",
                          "schema": { "type": "string" }
                        }
                      ]
                    }
                  }
                }
              }
            }
          }
        }
      }
    }
  },
  "operation_allowlist": ["listAssets"],
  "credential_bindings": {},
  "allow_unsafe_methods": false,
  "timeout_ms": 5000,
  "max_response_bytes": 1048576,
  "activate": true
}

Comprovació del Cicle de Vida Executable

Aquest bloc importa un document OpenAPI mínim, valida la visibilitat de la llista/obtenció i afirma els camps de metadades de contracte/execució.

import json
import os
from urllib import request


def call_tool(endpoint: str, tool_name: str, arguments: dict) -> dict:
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "tools/call",
        "params": {"name": tool_name, "arguments": arguments},
    }
    req = request.Request(
        endpoint,
        data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with request.urlopen(req, timeout=20) as resp:
        body = json.loads(resp.read().decode("utf-8"))
    if "error" in body:
        raise RuntimeError(f"tool call failed: {body['error']}")
    content = body.get("result", {}).get("content", [])
    if not content or "json" not in content[0]:
        raise RuntimeError(f"unexpected tool result envelope: {body}")
    return content[0]["json"]


endpoint = os.environ.get("DG_ENDPOINT", "http://127.0.0.1:8080/rpc")
provider_id = "docs_typed_asset_api"
version = "v1"

openapi_doc = {
    "openapi": "3.1.0",
    "components": {
        "schemas": {
            "AssetListResponse": {
                "type": "object",
                "properties": {"next_cursor": {"type": "string"}},
                "required": ["next_cursor"],
                "additionalProperties": False,
                "x-decision-gate": {
                    "projections": [
                        {
                            "id": "next_cursor",
                            "pointer": "/next_cursor",
                            "schema": {"type": "string"},
                        }
                    ]
                },
            }
        }
    },
    "paths": {
        "/assets": {
            "get": {
                "operationId": "listAssets",
                "responses": {
                    "200": {
                        "description": "ok",
                        "content": {
                            "application/json": {
                                "schema": {
                                    "$ref": "#/components/schemas/AssetListResponse"
                                }
                            }
                        },
                    }
                },
            }
        }
    },
}

import_response = call_tool(
    endpoint,
    "typed_providers_import",
    {
        "tenant_id": 1,
        "namespace_id": 1,
        "provider_id": provider_id,
        "version": version,
        "openapi": openapi_doc,
        "operation_allowlist": ["listAssets"],
        "credential_bindings": {},
        "allow_unsafe_methods": False,
        "timeout_ms": 5000,
        "max_response_bytes": 1048576,
        "openapi_conformance_mode": "strict",
        "activate": True,
    },
)
assert import_response["provider_id"] == provider_id, import_response
assert import_response["version"] == version, import_response
assert import_response["register_outcome"] in {"registered", "updated"}, import_response
assert import_response["active_version"] == version, import_response
assert import_response["source_digest"]["algorithm"] == "sha256", import_response
assert isinstance(import_response["source_digest"]["value"], str), import_response
assert import_response["source_digest"]["value"], import_response
assert import_response["profile_digest"]["algorithm"] == "sha256", import_response
assert isinstance(import_response["profile_digest"]["value"], str), import_response
assert import_response["profile_digest"]["value"], import_response
assert import_response["contract_hash"]["algorithm"] == "sha256", import_response
assert isinstance(import_response["contract_hash"]["value"], str), import_response
assert import_response["contract_hash"]["value"], import_response
assert import_response["operation_count"] >= 1, import_response

list_response = call_tool(endpoint, "typed_providers_list", {"tenant_id": 1, "namespace_id": 1})
items = list_response.get("items", [])
assert any(item.get("provider_id") == provider_id for item in items), list_response

get_response = call_tool(
    endpoint,
    "typed_providers_get",
    {"tenant_id": 1, "namespace_id": 1, "provider_id": provider_id, "version": version},
)
assert get_response["provider_id"] == provider_id, get_response
assert get_response["selected_version"] == version, get_response
assert get_response["contract"]["transport"] == "typed", get_response
assert get_response["runtime_profile"]["provider_id"] == provider_id, get_response
assert get_response["record"]["source_digest"] == import_response["source_digest"], get_response
assert get_response["record"]["profile_digest"] == import_response["profile_digest"], get_response
assert get_response["runtime_profile"]["source_digest"] == import_response["source_digest"], get_response
assert get_response["runtime_profile"]["profile_digest"] == import_response["profile_digest"], get_response
assert get_response["drift_status"] is None, get_response
operations = get_response["runtime_profile"].get("operations", [])
assert isinstance(operations, list) and len(operations) > 0, get_response
assert isinstance(operations[0].get("check_id"), str), get_response

Notes

  • Mantingueu operation_allowlist explícit i estable per preservar els IDs de comprovació deterministes.
  • Utilitzeu els camps de resum typed_providers_get per al seguiment de desviacions en els fluxos de treball de desplegament.
  • Utilitzeu typed_providers_deprecate amb semàntica de retrocés per a canvis de cicle de vida controlats.