Scope boundary

Owns deterministic conformance of model output to a predefined structured schema and verifies the bounded parser that extracts tool calls, reasoning fields, and visible content from raw generation before downstream use.

AID-H-006.002: AID-H-006.001 proves structural schema conformance; AID-H-006.002 applies sink-specific encoding, URL policy, redaction, remote-subresource removal, streaming holdback, and final text or markup release. Schema-valid content can still be unsafe for its destination.

Implementation guidance

• AID-H-006.001-G001: Enforce schema-bound structured outputs using either native function-calling or response-model wrappers.
• AID-H-006.001-G002: High-impact JSON Schema + Pydantic enforcement profile with fail-closed validation, bounded normalization, and audit persistence.
• AID-H-006.001-G003: Verify the deployed tool-call and reasoning parsers against a signed positive and negative fixture population before their extracted fields reach downstream code.

AID-H-006.001-G001

Concept:

Schema-bound output is one control even when teams implement it through different SDKs. The core requirement is the same: do not let downstream systems parse free-form model text when the response is supposed to conform to a contract. Use either a wrapper such as Instructor/Pydantic or a vendor-native function-calling API, but treat both as equivalent implementations of one structured-output boundary.

Step 1: Define one canonical response schema

# File: schemas/output_schemas.py
from pydantic import BaseModel, ConfigDict, Field
from typing import Literal

# aidefend.structured-output.v1 requires non-empty reasoning; this is a wire-schema invariant.
NON_EMPTY_REASONING_LENGTH = 1

class UserSentiment(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)

    sentiment: Literal["positive", "negative", "neutral"]
    confidence: float = Field(ge=0.0, le=1.0)
    reasoning: str = Field(min_length=NON_EMPTY_REASONING_LENGTH)

Step 2: Wrapper path using Instructor or an equivalent response-model library

# File: analysis/sentiment_analyzer.py
import instructor
from openai import OpenAI
from schemas.output_schemas import UserSentiment

client = instructor.patch(OpenAI())

def get_structured_sentiment(text: str) -> UserSentiment:
    return client.chat.completions.create(
        model="gpt-4.1",
        response_model=UserSentiment,
        messages=[
            {"role": "user", "content": f"Analyze the sentiment of this text: {text}"}
        ],
    )

Step 3: Native function-calling path with the same schema semantics

# File: tools/schemas.py
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_user_sentiment",
            "description": "Return structured sentiment analysis",
            "parameters": {
                "type": "object",
                "properties": {
                    "sentiment": {"type": "string", "enum": ["positive", "negative", "neutral"]},
                    "confidence": {"type": "number", "minimum": 0.0, "maximum": 1.0},
                    "reasoning": {"type": "string", "minLength": 1}
                },
                "required": ["sentiment", "confidence", "reasoning"],
                "additionalProperties": False
            }
        }
    }
]

# File: analysis/tool_call_sentiment.py
from openai import OpenAI
import json
from schemas.output_schemas import UserSentiment
from tools.schemas import tools

client = OpenAI()

def _unique_object(pairs):
    value = {}
    for key, item in pairs:
        if key in value:
            raise ValueError(f"duplicate JSON key: {key}")
        value[key] = item
    return value

def _reject_constant(value):
    raise ValueError(f"non-finite JSON number: {value}")

def get_sentiment_via_tool_call(text: str) -> UserSentiment:
    response = client.chat.completions.create(
        model="gpt-4.1",
        messages=[{"role": "user", "content": f"Analyze the sentiment of this text: {text}"}],
        tools=tools,
        tool_choice={"type": "function", "function": {"name": "get_user_sentiment"}},
    )
    tool_call = response.choices[0].message.tool_calls[0]
    payload = json.loads(
        tool_call.function.arguments,
        object_pairs_hook=_unique_object,
        parse_constant=_reject_constant,
    )
    return UserSentiment.model_validate(payload, strict=True)

Operational notes

• Pick one primary implementation path per service so testing, evidence, and failure handling stay consistent.
• Validate the final payload with the same schema regardless of whether the model response came from a wrapper or native tool calling.
• Store schema versions and structured-output rejection logs with the service release record so evidence can be tied to the exact contract in force.

Action: Make structured-output enforcement a single recommendation unit. The implementation may vary by provider, but the security boundary is the shared schema contract.

Verify safely

A separately credentialed verifier independently reads back or replays the exact model/endpoint/schema version and complete response population requiring typed structured output, exercises positive, negative, boundary and dependency-failure fixtures, and reconciles the complete eligible population.

AID-H-006.001-G002

Concept:

For high-impact actions, the schema check must be a real execution gate. Define one canonical contract, allow only bounded normalization, persist the raw candidate and normalized payload for evidence, and fail closed before any tool or side effect runs.

Step 1: Define one strict high-impact contract

# File: output_contracts/high_impact.py
from __future__ import annotations
import math
import os

from typing import Any, Literal

from pydantic import BaseModel, ConfigDict, Field, model_validator

def _required_positive_int(name: str) -> int:
    value = int(os.environ[name])
    if value <= 0:
        raise RuntimeError(f"{name} must be a positive integer")
    return value

JUSTIFICATION_MIN_LENGTH = _required_positive_int("VERIFIED_H006_JUSTIFICATION_MIN_LENGTH")
JUSTIFICATION_MAX_LENGTH = _required_positive_int("VERIFIED_H006_JUSTIFICATION_MAX_LENGTH")
PROVENANCE_MIN_ITEMS = _required_positive_int("VERIFIED_H006_PROVENANCE_MIN_ITEMS")
PROVENANCE_MAX_ITEMS = _required_positive_int("VERIFIED_H006_PROVENANCE_MAX_ITEMS")



if JUSTIFICATION_MAX_LENGTH < JUSTIFICATION_MIN_LENGTH or PROVENANCE_MAX_ITEMS < PROVENANCE_MIN_ITEMS:
    raise RuntimeError("verified high-impact schema bounds are invalid")


ACTION_PARAM_FIELDS = {
    "pay_vendor_invoice": {"amount_usd", "vendor_id", "invoice_id"},
    "rotate_credential": {"credential_id"},
    "update_k8s_config": {"cluster_id", "namespace", "manifest_sha256"},
}

class HighImpactRequest(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)

    action: Literal["pay_vendor_invoice", "rotate_credential", "update_k8s_config"]
    riskTier: Literal["low", "medium", "high"]
    justification: str = Field(min_length=JUSTIFICATION_MIN_LENGTH, max_length=JUSTIFICATION_MAX_LENGTH)
    provenance: list[str] = Field(min_length=PROVENANCE_MIN_ITEMS, max_length=PROVENANCE_MAX_ITEMS)
    params: dict[str, Any]

    @model_validator(mode="after")
    def validate_action_bounds(self) -> "HighImpactRequest":
        expected = ACTION_PARAM_FIELDS[self.action]
        if set(self.params) != expected:
            raise ValueError("action_parameter_schema_mismatch")
        if any(not isinstance(key, str) for key in self.params):
            raise ValueError("parameter_key_type_invalid")
        if self.action == "pay_vendor_invoice":
            amount = self.params.get("amount_usd")
            if (isinstance(amount, bool) or not isinstance(amount, (int, float)) or
                    not math.isfinite(amount) or amount <= 0 or amount > 10_000):
                raise ValueError("amount_out_of_policy_bounds")
        if self.action == "update_k8s_config":
            digest = self.params.get("manifest_sha256")
            if not isinstance(digest, str) or __import__("re").fullmatch(r"[0-9a-f]{64}", digest) is None:
                raise ValueError("manifest_digest_invalid")
        for key, value in self.params.items():
            if key != "amount_usd" and (not isinstance(value, str) or not value.strip()):
                raise ValueError("action_parameter_value_invalid")
        return self

Step 2: Normalize only the fields you can repair deterministically

Do not invent missing actions, provenance, or parameters. Bounded normalization is limited to things like trimming strings, lowercasing enumerated values, and dropping empty provenance entries.

# File: output_contracts/enforcer.py
from __future__ import annotations

import hashlib
import json
from pathlib import Path

from pydantic import ValidationError

from output_contracts.high_impact import HighImpactRequest


class ContractViolation(RuntimeError):
    pass


def normalize_candidate_payload(payload: dict) -> dict:
    normalized = dict(payload)
    if isinstance(normalized.get("riskTier"), str):
        normalized["riskTier"] = normalized["riskTier"].strip().lower()
    if isinstance(normalized.get("justification"), str):
        normalized["justification"] = normalized["justification"].strip()
    if (isinstance(normalized.get("provenance"), list) and
            all(isinstance(item, str) for item in normalized["provenance"])):
        normalized["provenance"] = [
            item.strip() for item in normalized["provenance"] if item.strip()
        ]
    return normalized


def append_contract_audit(audit_path: str, record: dict) -> None:
    path = Path(audit_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(record, sort_keys=True) + "\n")


def validate_high_impact_payload(raw_payload: dict, *, audit_path: str) -> HighImpactRequest:
    normalized = normalize_candidate_payload(raw_payload)
    raw_sha256 = hashlib.sha256(
        json.dumps(raw_payload, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8")
    ).hexdigest()

    try:
        parsed = HighImpactRequest.model_validate(normalized)
    except ValidationError as exc:
        append_contract_audit(
            audit_path,
            {
                "status": "rejected",
                "reason": "schema_validation_failed",
                "raw_payload_sha256": raw_sha256,
                "normalized_payload_sha256": hashlib.sha256(
                    json.dumps(normalized, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8")
                ).hexdigest(),
                "errors": json.loads(exc.json()),
            },
        )
        raise ContractViolation("high_impact_schema_rejected") from exc

    append_contract_audit(
        audit_path,
        {
            "status": "accepted",
            "raw_payload_sha256": raw_sha256,
            "normalized_payload_sha256": hashlib.sha256(
                parsed.model_dump_json().encode("utf-8")
            ).hexdigest(),
        },
    )
    return parsed

Step 3: Gate the dispatcher on the validator result

# File: output_contracts/dispatch.py
from output_contracts.enforcer import ContractViolation, validate_high_impact_payload


def dispatch_high_impact(raw_payload: dict, *, audit_path: str, tool_handlers: dict, authorize):
    validated = validate_high_impact_payload(raw_payload, audit_path=audit_path)
    decision = authorize(validated.model_dump())
    if not isinstance(decision, dict) or decision.get("outcome") != "ALLOW":
        raise PermissionError("high_impact_authorization_denied")
    handler = tool_handlers.get(validated.action)
    if handler is None:
        raise ContractViolation("approved_action_has_no_handler")
    return handler(validated.params)

Operational notes

• Persist raw candidates only in an explicitly approved encrypted sensitive-evidence store. The default audit path records digests, schema errors, and the normalized decision without duplicating prompts, credentials, or payment data.
• Reject unknown fields with extra="forbid" so the model cannot smuggle undeclared control data into the side-effect path.
• Run this validator in the same process path that dispatches the action; do not validate in one service and execute in another without binding them.

Action: Put one fail-closed schema enforcer in front of every high-impact action path. If the payload cannot be deterministically normalized into the approved contract, or the separate authorization owner does not return an exact ALLOW decision for the validated action and parameters, block execution and retain a minimized rejection record.

Verify safely

A separately credentialed verifier independently reads back or replays the exact validated output schema plus action-specific semantic, authorization and policy constraints for every proposed action, exercises positive, negative, boundary and dependency-failure fixtures, and reconciles the complete eligible population.

AID-H-006.001-G003

Use a parser contract, not parser-name trust

Expose a release-test adapter with the contract parse(chunks: list[str]) -> object; it must drive the deployed parser through the same incremental callbacks and finalization path used in production. Bind fixtures to the exact model, tokenizer, chat template, runtime image, parser implementation, and parser configuration recorded by AID-M-001.002. The parser must preserve visible content, tool-call order and arguments, and reasoning boundaries; ambiguous, duplicated, truncated, nested, or mixed marker forms must fail closed.

# File: tests/verify_serving_parser.py
from __future__ import annotations

import hashlib
import importlib
import json
import os
from pathlib import Path
from typing import Any

MAX_FIXTURE_BYTES = int(os.environ["VERIFIED_H006_PARSER_FIXTURE_MAX_BYTES"])
ENTRYPOINT = os.environ["VERIFIED_H006_PARSER_ENTRYPOINT"]
EXPECTED_FIXTURE_SHA256 = os.environ["VERIFIED_H006_PARSER_FIXTURE_SHA256"].lower()
INTERFACE_TUPLE_SHA256 = os.environ["VERIFIED_H006_INTERFACE_TUPLE_SHA256"].lower()
for name, value in {"fixture": EXPECTED_FIXTURE_SHA256,
                    "interface tuple": INTERFACE_TUPLE_SHA256}.items():
    if len(value) != 64 or set(value) - set("0123456789abcdef"):
        raise RuntimeError(name + " digest is invalid")


def strict_json(raw: bytes) -> Any:
    def unique(pairs):
        result = {}
        for key, value in pairs:
            if key in result:
                raise ValueError("duplicate JSON key: " + key)
            result[key] = value
        return result
    return json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique,
        parse_constant=lambda value: (_ for _ in ()).throw(
            ValueError("non-finite JSON number: " + value)
        ),
    )


def load_parser():
    module_name, separator, function_name = ENTRYPOINT.partition(":")
    if not separator or not module_name or not function_name:
        raise ValueError("parser entry point must be module:function")
    parser = getattr(importlib.import_module(module_name), function_name)
    if not callable(parser):
        raise TypeError("parser entry point is not callable")
    return parser


def main() -> None:
    fixture_path = Path("tests/serving-parser-fixtures.json")
    raw = fixture_path.read_bytes()
    if not raw or MAX_FIXTURE_BYTES < 1 or len(raw) > MAX_FIXTURE_BYTES:
        raise ValueError("fixture population is empty or exceeds policy")
    fixture_sha256 = hashlib.sha256(raw).hexdigest()
    if fixture_sha256 != EXPECTED_FIXTURE_SHA256:
        raise PermissionError("fixture differs from the signature-verified release digest")
    document = strict_json(raw)
    if not isinstance(document, dict) or set(document) != {"schema_version", "cases"}:
        raise ValueError("fixture schema differs")
    cases = document["cases"]
    if document["schema_version"] != "aidefend.serving-parser-fixtures.v1" or not isinstance(cases, list) or not cases:
        raise ValueError("fixture population is invalid")
    parser = load_parser()
    seen = set()
    results = []
    for case in cases:
        if not isinstance(case, dict) or set(case) != {"id", "chunks", "expected", "must_reject"}:
            raise ValueError("case schema differs")
        case_id = case["id"]
        if not isinstance(case_id, str) or not case_id or case_id in seen:
            raise ValueError("case ID is empty or duplicated")
        seen.add(case_id)
        must_reject = case["must_reject"]
        chunks = case["chunks"]
        if (type(must_reject) is not bool or not isinstance(chunks, list) or not chunks
                or any(not isinstance(chunk, str) for chunk in chunks)
                or sum(len(chunk.encode("utf-8")) for chunk in chunks) > MAX_FIXTURE_BYTES):
            raise ValueError("case types or streaming bounds differ")
        try:
            observed = parser(chunks)
        except (ValueError, TypeError, PermissionError):
            if not must_reject:
                raise AssertionError("valid parser fixture was rejected: " + case_id)
            results.append({"id": case_id, "result": "REJECTED_AS_EXPECTED"})
            continue
        if must_reject:
            raise AssertionError("malformed parser fixture was accepted: " + case_id)
        if observed != case["expected"]:
            raise AssertionError("parser output differs: " + case_id)
        results.append({"id": case_id, "result": "MATCH"})
    evidence = {
        "schema_version": "aidefend.serving-parser-evidence.v1",
        "fixture_sha256": fixture_sha256,
        "interface_tuple_sha256": INTERFACE_TUPLE_SHA256,
        "entrypoint": ENTRYPOINT,
        "case_count": len(cases),
        "results": results,
        "outcome": "PASS",
    }
    Path("evidence/serving-parser-evidence.json").write_text(
        json.dumps(evidence, sort_keys=True, separators=(",", ":")) + "\n",
        encoding="utf-8",
    )


if __name__ == "__main__":
    main()

Run the verifier inside the release image after admission injects the digest of the exact signature-verified fixture and model-interface tuple; the code rejects any later substitution before parsing. Include valid no-tool, single-tool, multi-tool, reasoning-only, and reasoning-plus-tool cases plus malformed markers, duplicated tool IDs, invalid JSON, truncation, unknown fields, nesting, Unicode boundary, and streaming-split cases. Sign the evidence and independently replay the same complete population. A missing case, parser exception on a valid case, acceptance of a negative case, or any field mismatch blocks promotion.

Action: Gate every parser release on the signed fixture replay and block downstream delivery when any positive, negative, split-stream, or malformed case fails.

Open-source tools

• Instructor
• Pydantic
• Guardrails AI
• NVIDIA NeMo Guardrails
• Outlines
• TypeChat
• LangChain
• Microsoft Semantic Kernel
• XGrammar
• lm-format-enforcer

Commercial / hosted tools

• OpenAI API
• Google Vertex AI
• Anthropic API (Tool Use)

Threat mappings

MITRE ATLAS

• AML.T0050 Command and Scripting Interpreter
• AML.T0067 LLM Trusted Output Components Manipulation (structured schemas prevent format manipulation)
• AML.T0063 Discover AI Model Outputs (structured schemas prevent discovery of non-required output fields)

MAESTRO

• Integration Risks (L7) (structured outputs protect downstream integrations)

OWASP LLM Top 10 2026

• LLM10:2026 Improper Output Handling

OWASP ML Top 10 2023

• ML09:2023 Output Integrity Attack

OWASP Top 10 for Agentic Applications 2026

• N/A

NIST Adversarial Machine Learning 2025

• NISTAML.039 Compromising connected resources (structured output protects downstream systems)

Cisco Integrated AI Security and Safety Framework

• AISubtech-1.1.4 Token Exploitation
• AISubtech-12.1.1 Parameter Manipulation
• AISubtech-4.3.1 Schema Inconsistencies
• AITech-1.1 Direct Prompt Injection and Jailbreak (Prompt-based Manipulation)
• AITech-12.1 Tool Exploitation
• AITech-12.2 Insecure Output Handling
• AITech-4.3 Protocol Manipulation

Google Secure AI Framework 2.0 - Risks

• IMO: Insecure Model Output

Databricks AI Security Framework 3.0

• N/A