Schema validation

Every step boundary in a ChainWeaver flow is Pydantic-validated in both directions:

context  ─┐                      ┌─ context (merged)
          ├──► tool.input_schema  ──► fn ──►  tool.output_schema  ──┤
                  (validate)               (validate)

If validation fails, the executor raises SchemaValidationError with the failing field and the step index — no partial state is committed.

Tool boundaries

A tool's input_schema and output_schema are Pydantic BaseModel subclasses. The executor:

1. Builds the input dict from the step's input_mapping and the accumulated context.
2. Calls input_schema.model_validate(input_dict) — raises on type mismatches and missing required fields. Unknown fields follow the schema's Pydantic config: dropped by default, rejected when the schema sets extra="forbid".
3. Invokes tool.fn(validated_input); expects a dict.
4. Calls output_schema.model_validate(output_dict) — same guarantees.
5. Merges the validated output back into the context.

Flow boundaries (optional)

A Flow can also declare input_schema and output_schema:

from chainweaver import Flow

flow = Flow(
    name="etl",
    description="...",
    steps=[...],
    input_schema_ref=Flow.schema_ref_from(InitialInput),
    output_schema_ref=Flow.schema_ref_from(FinalOutput),
)

When set, the executor validates initial_input before the first step and the merged final_output after the last step. Flow-level validation surfaces in the trace as records with step_index=-1 (input) and step_index=len(steps) (output).

Schema hashing and drift

Every Tool exposes a schema_hash derived from its input schema, output schema, and schema_version:

tool.schema_hash       # SHA-256 of (input_schema, output_schema, schema_version)
tool.input_schema_hash # input only
tool.output_schema_hash # output only

The hash is stable across processes, sorted-keys-canonicalized, and short enough (16 hex chars) to log. When a tool's schema changes and its schema_hash no longer matches what a flow has snapshotted in tool_schema_hashes, FlowExecutor surfaces a DriftInfo entry via get_drift_report(). Call accept_drift(flow_name) after reviewing a flow's current tool surfaces to create or refresh that baseline.

See Cookbook recipe 5 — Schema drift in CI for the governance pattern.

Compile-time validation

compile_flow(flow, tools) performs all of the above checks statically — before any tool function runs. It accepts both a linear Flow and a DAGFlow; for a DAG each step is validated against the union of its transitive depends_on ancestors' outputs (plus the flow input) rather than list order.

• Every tool_name in the flow resolves to a registered tool.
• Every input_mapping key resolves to either a literal, a key in the initial-input schema, or a field in an upstream step's output schema.
• Type compatibility between mapped fields (basic Python types + Pydantic models).
• on_error="fallback:<tool_name>" targets are validated for input and output compatibility (see Fallback semantics).
• Optional warning for shadowed context keys.

from chainweaver import compile_flow

result = compile_flow(flow, tools={"double": double_tool, ...})
if result.errors:
    for err in result.errors:
        print(err)

The CLI exposes this as chainweaver validate flows/etl.flow.yaml.