How must an instance negotiate output schema when receiver schema expectations are stricter?

**Date of Record:** 2026 October 27 **Scope:** Applies when an agentic session produces output for a receiver whose schema expectations are stricter than the session's native output schema (e.g., required fields the session does not always produce, tighter type constraints, or stricter nullability rules), and the session must negotiate what output is actually acceptable. Does not apply to sessions whose output schema is verified identical to the receiver's expectation.

Core Distinction

The separation between **output content** (what the session produces) and **output schema conformance** (what the receiver accepts). Relying on content creates the **Schema Rejection Loop**: the session produces valid content in its own schema, the receiver rejects it for schema non-conformance, the session retries with minor adjustments that still violate the schema, and the loop continues until the session gives up or the receiver escalates.

What Was Observed

In agent-to-agent output architectures, failing to negotiate schema produces three primary failure modes:

1. **Type Mismatch Cascade:** The session produces an integer, the receiver expects a string, the session converts to a string, but the receiver's validator now rejects the string for not matching a regex pattern. The session oscillates between representations.
2. **Required Field Fabrication:** The receiver requires a field the session cannot reliably produce, so the session fabricates values (nulls, empty strings, defaults) that pass schema validation but misrepresent the content.
3. **Strict Mode Rejection:** The receiver's schema validation is strict (no extra fields), but the session's output includes helpful metadata fields. The receiver rejects all outputs, and the session cannot determine which fields to remove.

Recorded Elements

1. **`receiver_schema_signature`:** The exact receiver schema (fields, types, constraints, required vs. optional, strict vs. lenient), signed by the receiver, allowing the session to compare against its native schema.
2. **`schema_delta_receipt`:** The exact differences between the session's native schema and the receiver's expected schema, allowing the session to know what transformations are needed.
3. **`negotiation_protocol_receipt`:** The agreed-upon protocol for handling schema mismatches (e.g., "session produces best-effort, receiver accepts with warning" vs. "session must transform exactly, receiver rejects otherwise").
4. **`schema_conformance_receipt`:** For each output, the cryptographic proof that the output conforms to the negotiated schema (or explicitly marked as non-conforming with reason).

Condition of Validity

* **Applies when:** The session produces output for a receiver with stricter schema, schema mismatches are possible, and rejected outputs have downstream consequences. * **Does not apply when:** The session's output schema is verified identical to the receiver's, or the receiver is tolerant to schema variance.

Decision Boundary

1. **Does the receiver have a stricter schema than the session produces?** If no, standard output applies. If yes, proceed.
2. **Is the `schema_delta_receipt` available?** If no, the session cannot know what transformations are needed. It must request the schema or produce best-effort with the delta unknown.
3. **Is the `negotiation_protocol_receipt` established?** If no, the session and receiver are in a rejection loop. They must negotiate a protocol before proceeding.
4. **Conversely, when can the session produce best-effort output?** Only when the negotiated protocol explicitly allows best-effort with receiver-side warnings, or when the receiver's schema is advisory (not strict).

Operational Effect

1. **Rejection Loop Prevention:** Explicit negotiation replaces trial-and-error retries.
2. **Fabrication Prevention:** The protocol defines what to do with missing required fields (fail vs. best-effort vs. escalation).
3. **Deterministic Conformance:** Each output includes a conformance receipt, allowing the receiver to know whether to accept, warn, or reject.

from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json

class SchemaVerdict(Enum):
    SCHEMA_COMPATIBLE = auto()
    TRANSFORMATION_REQUIRED = auto()
    NEGOTIATION_REQUIRED = auto()
    BEST_EFFORT_ALLOWED = auto()

@dataclass(frozen=True)
class SchemaDelta:
    missing_required_fields: tuple[str, ...]
    type_mismatches: tuple[str, ...]
    extra_fields: tuple[str, ...]
    nullability_conflicts: tuple[str, ...]

@dataclass(frozen=True)
class SchemaState:
    session_schema_hash: str
    receiver_schema_hash: str
    delta: SchemaDelta
    negotiation_protocol: str | None  # "TRANSFORM_ALL" | "BEST_EFFORT" | "FAIL_FAST"
    receiver_accepts_best_effort: bool

@dataclass(frozen=True)
class SchemaPolicy:
    require_negotiation: bool

@dataclass(frozen=True)
class SchemaManifest:
    verdict: SchemaVerdict
    directive: str
    manifest_digest: str

class OutputSchemaNegotiationGuard:
    @staticmethod
    def _hash(data: str) -> str:
        return hashlib.sha256(data.encode("utf-8")).hexdigest()

    @classmethod
    def evaluate_schema(
        cls, state: SchemaState, policy: SchemaPolicy
    ) -> SchemaManifest:
        
        # Gate 1: Schemas identical
        if state.session_schema_hash == state.receiver_schema_hash:
            digest = cls._hash(json.dumps({
                "schema": state.session_schema_hash,
                "status": "COMPATIBLE"
            }, sort_keys=True))
            return SchemaManifest(
                verdict=SchemaVerdict.SCHEMA_COMPATIBLE,
                directive="PROCEED: Schemas identical.",
                manifest_digest=digest
            )

        # Gate 2: No delta (compatible)
        delta_is_empty = (
            not state.delta.missing_required_fields
            and not state.delta.type_mismatches
            and not state.delta.nullability_conflicts
        )
        if delta_is_empty:
            digest = cls._hash(json.dumps({
                "session": state.session_schema_hash,
                "receiver": state.receiver_schema_hash,
                "delta": "EMPTY"
            }, sort_keys=True))
            return SchemaManifest(
                verdict=SchemaVerdict.SCHEMA_COMPATIBLE,
                directive="PROCEED: Schema delta empty, output compatible.",
                manifest_digest=digest
            )

        # Gate 3: Negotiation requirement
        if policy.require_negotiation and not state.negotiation_protocol:
            return SchemaManifest(
                verdict=SchemaVerdict.NEGOTIATION_REQUIRED,
                directive=f"HALT: Schema delta exists (missing {len(state.delta.missing_required_fields)}, mismatches {len(state.delta.type_mismatches)}), no negotiation protocol.",
                manifest_digest=cls._hash("NO_PROTOCOL")
            )

        # Gate 4: Best effort allowed
        if state.negotiation_protocol == "BEST_EFFORT" and state.receiver_accepts_best_effort:
            digest = cls._hash(json.dumps({
                "protocol": "BEST_EFFORT",
                "delta": {
                    "missing": tuple(sorted(state.delta.missing_required_fields)),
                    "mismatches": tuple(sorted(state.delta.type_mismatches))
                }
            }, sort_keys=True))
            return SchemaManifest(
                verdict=SchemaVerdict.BEST_EFFORT_ALLOWED,
                directive="PROCEED: Best effort allowed with warnings.",
                manifest_digest=digest
            )

        # Gate 5: Transform required
        digest = cls._hash(json.dumps({
            "protocol": state.negotiation_protocol or "TRANSFORM_ALL",
            "delta": {
                "missing": tuple(sorted(state.delta.missing_required_fields)),
                "mismatches": tuple(sorted(state.delta.type_mismatches)),
                "extra": tuple(sorted(state.delta.extra_fields))
            }
        }, sort_keys=True))
        
        return SchemaManifest(
            verdict=SchemaVerdict.TRANSFORMATION_REQUIRED,
            directive=f"ACTION: Transform output per {state.negotiation_protocol or 'TRANSFORM_ALL'}.",
            manifest_digest=digest
        )