Contract-First Agent Design: Structured Outputs, Scratch Pads, and the Governance Layer Between Agents

The architecture that holds up in production is the same one you'd use between microservices: define the schema first, treat it as the source of truth, and validate at every boundary crossing. In an agentic pipeline, this means each agent node in your graph has an explicit input schema and an explicit output schema — and the runtime enforces both, not the prompt. When you're handling sensitive data like email content or user behavioral signals, those schemas also become your first line of PII governance: a field tagged `pii: true` in your contract can trigger automatic redaction in your trace exporter before anything hits your logging backend. Source: Building A Generative AI Platform — Huyen Chip (https://huyenchip.com//2024/07/25/genai-platform.html)

AIThe pattern emerging across production deployments is that contract failures are almost never total — they're partial. An agent returns 9 out of 10 expected fields, and the missing field either defaults to null and corrupts downstream logic, or the consuming agent hallucinates a value to fill the gap. Catching this requires schema validation with strict mode enabled and a rejection path that routes malformed outputs to a human review queue rather than silently forwarding them.

import { z } from 'zod';

// Define the contract at the boundary, not inside the prompt
const EmailThreatAssessmentSchema = z.object({
  threatLevel: z.enum(['none', 'low', 'medium', 'high', 'critical']),
  confidence: z.number().min(0).max(1),
  indicators: z.array(z.string()).min(1),
  senderReputation: z.object({
    domain: z.string(),
    score: z.number(),
  }),
  // PII annotation — strip before exporting to observability layer
  recipientEmail: z.string().email().meta({ pii: true }),
});

type EmailThreatAssessment = z.infer<typeof EmailThreatAssessmentSchema>;

async function validateAgentOutput(
  raw: unknown,
  agentId: string
): Promise<EmailThreatAssessment> {
  const result = EmailThreatAssessmentSchema.safeParse(raw);

  if (!result.success) {
    // Route to human review, never swallow the error
    await auditQueue.push({
      agentId,
      violation: result.error.flatten(),
      timestamp: new Date().toISOString(),
    });
    throw new ContractViolationError(`Agent ${agentId} output failed schema validation`);
  }

  await auditQueue.push({
    agentId,
    schemaVersion: 'v2.1',
    // Redact PII before logging
    sanitized: omitPiiFields(result.data, EmailThreatAssessmentSchema),
  });

  return result.data;
}

Abnormal AI's deployment illustrates why the scratch pad is architecturally distinct from in-context reasoning. When you're analyzing billions of signals per day for email threats, the model's context window isn't a safe or scalable place to hold intermediate aggregation state — you need a real compute surface where code runs, data structures get built, and results are materialized before being handed back to the reasoning layer. Critically, for a security product handling email content from Fortune 500 companies, isolation between runs isn't optional: any state persistence across invocations is a data boundary violation waiting to happen. Source: Abnormal AI: Amazon Bedrock AgentCore for agentic email security at scale — AWS (https://aws.amazon.com/blogs/machine-learning/abnormal-ai-amazon-bedrock-agentcore-for-agentic-email-security-at-scale/)

AIThe contract design implication is that the scratch pad should have its own output schema — the structured result that gets returned from the execution environment back to the agent — rather than letting agents consume raw execution output. This makes the scratch pad a first-class, testable component in your pipeline rather than an opaque black box, and it gives your observability layer a consistent place to capture what computation actually happened during a run.

// Typed contract for what comes back from the scratch pad
interface ScratchPadResult<T> {
  runId: string;
  agentId: string;
  executionMs: number;
  output: T;
  executionTrace: {
    stepsExecuted: number;
    dataSourcesAccessed: string[];
    // No raw data here — summary only for audit
  };
}

const EmailAggregationResultSchema = z.object({
  signalCount: z.number(),
  anomalyScore: z.number().min(0).max(1),
  correlatedCampaignId: z.string().nullable(),
  aggregationWindowMs: z.number(),
});

async function invokeScratchPad(
  agentId: string,
  task: EmailAggregationTask
): Promise<ScratchPadResult<z.infer<typeof EmailAggregationResultSchema>>> {
  const raw = await bedrockAgentCore.codeInterpreter.execute({
    sessionId: crypto.randomUUID(), // ephemeral per invocation
    task: task.payload,
  });

  const validatedOutput = EmailAggregationResultSchema.parse(raw.output);

  return {
    runId: raw.runId,
    agentId,
    executionMs: raw.durationMs,
    output: validatedOutput,
    executionTrace: {
      stepsExecuted: raw.steps,
      dataSourcesAccessed: raw.accessLog.map(e => e.source),
    },
  };
}

The software factory framing is useful precisely because it forces the governance question into concrete engineering terms. You're not designing 'how much AI involvement' — you're designing a loop, and the question is where in that loop a human reads an output before it becomes an input to the next stage. For agentic pipelines handling sensitive data, the answer should be encoded in your schema contracts: certain output types are flagged as requiring human acknowledgment, and your orchestrator routes them to a review queue rather than forwarding them automatically. Source: Software Factories, Light and Dark — Addy Osmani (https://addyosmani.com/blog/software-factories/)

AIThe longitudinal pattern from last week — immutable audit logs, confidence-gated retrieval, semantic integrity checks — maps directly onto the factory governance problem. A confidence score below a threshold isn't just a signal to the consuming agent; it should be a contract condition that gates whether the output is forwarded automatically or held for review. Building that threshold into the schema itself, as a first-class field with explicit routing logic, means your governance rules are version-controlled and inspectable rather than buried in prompt instructions.

// Confidence-gated routing as a first-class contract concern
const GovernedOutputSchema = z.object({
  payload: z.unknown(),
  confidence: z.number().min(0).max(1),
  requiresHumanReview: z.boolean(),
  sensitivityLevel: z.enum(['standard', 'pii', 'regulated', 'critical']),
});

async function routeAgentOutput(
  output: z.infer<typeof GovernedOutputSchema>,
  pipelineConfig: PipelineGovernanceConfig
): Promise<'forwarded' | 'queued_for_review'> {
  const { confidence, requiresHumanReview, sensitivityLevel } = output;

  const threshold = pipelineConfig.thresholds[sensitivityLevel] ?? 0.85;
  const needsReview = requiresHumanReview || confidence < threshold;

  await auditLog.append({
    event: needsReview ? 'HELD_FOR_REVIEW' : 'AUTO_FORWARDED',
    confidence,
    sensitivityLevel,
    timestamp: new Date().toISOString(),
    // Immutable — append-only store
  });

  if (needsReview) {
    await humanReviewQueue.enqueue(output);
    return 'queued_for_review';
  }

  return 'forwarded';
}

The five-stage pipeline — threat model, discover, verify, triage, patch — maps naturally onto a graph where each stage has a typed input and output contract, and the contracts get progressively narrower as findings move toward action. At the discovery stage, the output schema is broad: anything that looks like a vulnerability class gets captured. By the triage stage, the schema is tight: severity, confidence, affected component, recommended action, human review flag. Each narrowing is a contract that your pipeline enforces, not a prompt that your model follows. Source: Using LLMs to Secure Source Code — Eugene Yan (https://eugeneyan.com//writing/secure-source-code/)

AIThe connection to last week's governance pattern is direct: per-phase provenance logging means each stage in the security pipeline emits a structured audit event that records not just what was found, but which model version found it, what context it was given, and what confidence it returned. When a vulnerability is later disputed or a patch is questioned, you can reconstruct the full reasoning chain from your audit log rather than relying on the model to reproduce its reasoning on demand.

// Narrowing contracts across pipeline stages
const VulnerabilityFindingSchema = z.object({
  findingId: z.string().uuid(),
  vulnerabilityClass: z.enum([
    'injection', 'broken-auth', 'sensitive-data-exposure',
    'xxe', 'broken-access-control', 'security-misconfiguration',
    'xss', 'insecure-deserialization', 'known-vulnerabilities', 'logging-monitoring'
  ]),
  severity: z.enum(['info', 'low', 'medium', 'high', 'critical']),
  confidence: z.number().min(0).max(1),
  affectedFile: z.string(),
  lineRange: z.tuple([z.number(), z.number()]),
  verified: z.boolean().default(false),
});

const VerifiedFindingSchema = VulnerabilityFindingSchema.extend({
  verified: z.literal(true),
  verificationMethod: z.enum(['static-analysis', 'dynamic-test', 'cross-model-consensus']),
  verifierAgentVersion: z.string(),
});

// Triage routing based on severity + confidence asymmetry
function shouldRequireHumanReview(
  finding: z.infer<typeof VerifiedFindingSchema>
): boolean {
  const criticalClasses = ['injection', 'broken-auth', 'sensitive-data-exposure'];
  const isCriticalClass = criticalClasses.includes(finding.vulnerabilityClass);
  // Critical class: review even at high confidence
  // Any class: review if confidence is low
  return isCriticalClass || finding.confidence < 0.80 || finding.severity === 'critical';
}

The architectural shift toward looped transformers creates a practical problem for anyone who needs to explain an agent's decision after the fact. When reasoning is distributed across multiple internal passes and the chain-of-thought is not exposed, the structured output is the only auditable artifact your pipeline produces. This means your output schema needs to carry more explanatory weight than it would if reasoning traces were available: confidence scores, decision factors, and alternative interpretations become required fields, not optional enrichment. Source: GPT-6 Astra, Looped Transformers, and Hidden Reasoning — Sebastian Raschka (https://magazine.sebastianraschka.com/p/gpt-6-astra-looped-transformers-and)

AIThe governance response to opaque model reasoning isn't to demand transparency from the model — you often can't get it. It's to design your pipeline so that the structured output contract captures enough decision context to support review without the reasoning trace. For sensitive data workflows, this means adding a `decisionFactors` field to your output schema that the model populates with the specific signals that drove its output, not a summary of its reasoning process. Those factors are then what your human reviewers assess — and what your audit log preserves.

// Output schema that compensates for opaque reasoning
const AuditableDecisionSchema = z.object({
  decision: z.string(),
  outcome: z.enum(['approved', 'rejected', 'escalated', 'deferred']),
  confidence: z.number().min(0).max(1),
  // Required even when model reasoning is not exposed
  decisionFactors: z.array(z.object({
    factor: z.string(),
    weight: z.enum(['primary', 'supporting', 'mitigating']),
    evidenceRef: z.string(), // pointer to source data, not model reasoning
  })).min(1, 'At least one decision factor is required for audit'),
  alternativesConsidered: z.array(z.string()).optional(),
  modelId: z.string(), // which model version produced this
  inputHash: z.string(), // deterministic hash of the input for replay reference
  sensitivityFlags: z.array(z.enum(['pii', 'regulated', 'financial', 'health'])),
});

// Validate that the model actually populated decision factors
// — don't let it return empty arrays for audit-critical fields
function enforceAuditCompleteness(
  output: z.infer<typeof AuditableDecisionSchema>
): void {
  if (output.sensitivityFlags.length > 0 && output.decisionFactors.length < 2) {
    throw new ContractViolationError(
      'Sensitive data decisions require at minimum 2 documented decision factors'
    );
  }
}

The CPU shortage signal is worth taking seriously as an architectural input, not just an ops concern. Tool use in agentic workflows is fundamentally different from inference: a model call is GPU-bound, but tool execution — web search, database queries, code execution, API calls — is CPU-bound and often synchronous. At scale, an agent that makes three redundant tool calls per invocation because its output contract doesn't force it to commit to a single data source before acting is a meaningful infrastructure cost. Source: The Pulse #191: a new trend of CPU shortages — Pragmatic Engineer (https://newsletter.pragmaticengineer.com/p/the-pulse-191-a-new-trend-of-cpu)

AIThe design lever here is contract specificity at the tool boundary. If your agent's input schema specifies which data sources are in scope for a given task, and your output schema requires the agent to declare which sources it actually accessed, you create a closed loop where tool use is intentional and auditable rather than opportunistic. This also gives you a concrete optimization target: if 80% of decisions reference the same two data sources, you can pre-fetch and cache them, reducing both latency and CPU load without changing the agent's reasoning logic.

// Tool use metadata as a first-class output contract field
const ToolAwareOutputSchema = z.object({
  result: z.unknown(),
  confidence: z.number(),
  toolCallsExecuted: z.array(z.object({
    toolName: z.string(),
    callCount: z.number().int().min(1),
    cacheHit: z.boolean(),
    durationMs: z.number(),
    // Was this tool call actually necessary for the decision?
    loadBearing: z.boolean(),
  })),
  // Cap tool calls per invocation as a contract constraint
  _meta: z.object({
    totalToolCalls: z.number().int().max(10), // hard ceiling per invocation
    cpuIntensiveCallsMade: z.number().int(),
  }),
});

// Use this to find agents that are over-tooling
function analyzeToolEfficiency(
  outputs: z.infer<typeof ToolAwareOutputSchema>[]
): ToolEfficiencyReport {
  const redundantCalls = outputs.flatMap(o =>
    o.toolCallsExecuted.filter(t => !t.loadBearing && t.cacheHit === false)
  );

  return {
    totalRedundantCalls: redundantCalls.length,
    topRedundantTools: redundantCalls
      .reduce((acc, t) => {
        acc[t.toolName] = (acc[t.toolName] ?? 0) + 1;
        return acc;
      }, {} as Record<string, number>),
    optimizationOpportunity: redundantCalls.length > 0
      ? 'pre-fetch and cache high-frequency non-load-bearing tool results'
      : 'none identified',
  };
}

Grafana's DEM framing — combining real user monitoring with synthetic checks to bridge the gap between metrics and user experience — maps directly onto the challenge of agent pipeline observability. A metric that says 'agent returned a valid schema' tells you nothing about whether the user got a useful outcome. What you need is the equivalent of session replay for agentic workflows: a trace that starts at the user's action, passes through each agent's structured output, and terminates at what the user actually saw or received. Source: Digital Experience Monitoring with Grafana Cloud — Grafana (https://grafana.com/blog/digital-experience-monitoring-with-grafana-cloud-session-replay-synthetic-checks-and-faster-investigations/)

AIThe engineering move here is to treat user-impact fields as required in your pipeline's final output contract, not as enrichment you add in the logging layer. If the last agent in your pipeline's output schema includes a `userImpact` field — with enum values like 'action-taken', 'recommendation-shown', 'escalated-to-human', 'no-action' — your observability layer gets a structured signal it can aggregate, alert on, and correlate with user session data. This closes the loop between what your agents decided and whether those decisions created real value for the people they were built to serve.

// Pipeline terminal output schema — user outcome as a required field
const PipelineTerminalOutputSchema = z.object({
  // Standard decision fields
  decision: z.string(),
  confidence: z.number(),

  // Observability correlation
  correlationId: z.string().uuid(), // ties to user session
  pipelineRunId: z.string().uuid(),
  agentVersions: z.record(z.string(), z.string()), // agentId -> version

  // User impact — required, not optional
  userImpact: z.object({
    outcome: z.enum([
      'action-taken',
      'recommendation-shown',
      'escalated-to-human',
      'blocked-pending-review',
      'no-action',
    ]),
    impactedUserId: z.string().meta({ pii: true }), // redact in traces
    latencyMs: z.number(),
    valueDelivered: z.boolean(), // did this produce a useful result?
  }),

  // For synthetic monitoring: expected outcome to diff against
  syntheticBaseline: z.string().optional(),
});

// Synthetic check: run known inputs and compare user impact outcomes
async function runSyntheticPipelineCheck(
  testCase: PipelineSyntheticTestCase
): Promise<SyntheticCheckResult> {
  const output = await pipeline.run(testCase.input);
  const validated = PipelineTerminalOutputSchema.parse(output);

  return {
    passed: validated.userImpact.outcome === testCase.expectedOutcome,
    actual: validated.userImpact.outcome,
    expected: testCase.expectedOutcome,
    latencyMs: validated.userImpact.latencyMs,
    correlationId: validated.correlationId,
  };
}