Building AI-Native Infrastructure with Specmatic: How I Eliminated Integration Uncertainty in a Multi-Service System

kanugu rajesh — Thu, 18 Jun 2026 07:15:29 +0000

A hands-on walkthrough of contract-first development, async event contracts, and building the feedback loops that AI coding agents actually need.

Introduction

There's a gap between what AI coding agents can generate and what production systems actually need.

An agent can write a Flask route. It can write an OpenAPI spec. It can even write tests. But when you have two services built by two different agents — or two different humans — that need to talk to each other, the integration is where everything silently breaks.

I built specmatic-taskflow to explore this exact problem. It's a multi-service task management system where a Task Service, a User Service, a Kafka event bus, and a Kanban frontend all talk to each other. But more importantly, I integrated Specmatic into every layer — turning OpenAPI and AsyncAPI specifications into living, executable tests that both humans and AI coding agents can use as a feedback signal.

This post is a walkthrough of what I built, how I designed the contract-first workflow, and what I learned about making contracts enforceable rather than aspirational.

The Problem: Integration Uncertainty at Scale

When you build a single-service application, failures are visible. When you build a multi-service system, failures are silent until production.

Here's the typical failure mode:

Service A and Service B both implement the same OpenAPI spec.
A developer changes a field name in Service A (dueDate → due_date).
The OpenAPI spec isn't updated. Tests pass because they're mocked.
Service B still expects dueDate. Everything looks fine in staging.
Production breaks.

This problem gets worse with AI coding agents. When you ask an LLM to implement a service, it generates code based on its training — not based on what the other services in your system expect. Without a feedback mechanism that can tell an AI "your implementation doesn't match the contract," you get plausible-looking code that doesn't integrate.

The solution isn't more documentation. It's executable contracts.

What I Built: specmatic-taskflow

specmatic-taskflow is an AI-native task management system with:

Task Service — Flask REST API (CRUD on tasks, publishes to Kafka)
User Service — Flask REST API (user registry, assignee lookup)
Apache Kafka — Event bus for task lifecycle events (task-created, task-updated)
Kanban Frontend — Vanilla JavaScript SPA with no build toolchain overhead
Specmatic Contract Tests — REST tests (OpenAPI 3.0.1) + Async tests (AsyncAPI 3.0.0)
Schema Resiliency Tests — Auto-generated positive variations beyond hand-written examples
Specmatic Mock Server — Serves contract-driven mock responses before real services exist
Specmatic Studio — Visual IDE for contract exploration

Everything runs via Docker Compose. Bringing up the full stack also runs contract tests — so integration verification is not a separate step, it's built into the startup.

specmatic-taskflow/
├── specs/
│   ├── openapi/
│   │   ├── task-api.yaml          # Task Service contract
│   │   └── user-api.yaml          # User Service contract
│   └── asyncapi/
│       ├── task-events.yaml       # Kafka event contract
│       └── examples/              # Before-hooks that trigger publishes via the REST API
├── services/
│   ├── task-service/              # Flask + Kafka publisher
│   └── user-service/              # Flask user registry
├── frontend/                      # Vanilla JS Kanban board
├── specmatic.yaml                 # REST test configuration
├── specmatic-async.yaml           # Async test configuration
└── docker-compose.yaml            # Full orchestration

Step 1: Write the Contract First

The first principle of this project: the spec exists before the code.

I wrote task-api.yaml before writing a single line of Flask. Every endpoint, every status code, every field name was decided in the OpenAPI document. The implementation's job was to satisfy this contract, not to define it.

Here's a simplified view of the Task API contract:

# specs/openapi/task-api.yaml
openapi: 3.0.1
info:
  title: Task Service API
  version: 1.0.0

paths:
  /tasks:
    get:
      summary: List all tasks
      parameters:
        - name: status
          in: query
          schema:
            type: string
            enum: [pending, in-progress, completed]
          examples:
            TASKS_200_OK:
              value: pending
      responses:
        '200':
          content:
            application/json:
              examples:
                TASKS_200_OK:
                  value:
                    - id: 1
                      title: "Implement Login Feature"
                      status: "in-progress"
                      priority: "high"

    post:
      summary: Create a task
      requestBody:
        content:
          application/json:
            examples:
              CREATE_TASK_201:
                value:
                  title: "Write Unit Tests"
                  priority: "medium"
              CREATE_TASK_400:
                value:
                  priority: "medium"   # Missing required 'title'
      responses:
        '201':
          content:
            application/json:
              examples:
                CREATE_TASK_201:
                  value:
                    id: 100
                    title: "Write Unit Tests"
                    status: "pending"
                    priority: "medium"
        '400':
          content:
            application/json:
              examples:
                CREATE_TASK_400:
                  value:
                    error: "Validation failed"
                    message: "title is required"

The critical detail: inline example names must match across parameters, request bodies, and responses.

When Specmatic sees CREATE_TASK_400 in both the request body and the 400 response, it pairs them into a single test scenario: "send this request body, expect this response." This is how examples become test cases without writing any test code.

The naming convention I used — TASKS_200_OK, CREATE_TASK_201, CREATE_TASK_400, GET_TASK_404 — is the entire test suite. No test files. No mocking frameworks. No setup/teardown.

Step 2: Run the Contract, Watch It Fail

Before writing any Flask code, I ran Specmatic against a stub service.

docker compose up test-task-api

Specmatic hit the (non-existent) endpoints and failed every test. But this failure is meaningful — it's a precise diff between what the contract expects and what the service provides. This is exactly the kind of signal that an AI coding agent can act on.

The specmatic.yaml configuration:

version: 3

systemUnderTest:
  service:
    definitions:
      - definition:
          source:
            filesystem:
              directory: specs/openapi
          specs:
            - task-api.yaml
    runOptions:
      openapi:
        type: test
        baseUrl: http://task-service:8080

specmatic:
  settings:
    test:
      schemaResiliencyTests: all
  governance:
    report:
      formats:
        - html
      outputDirectory: build/reports/specmatic
    successCriteria:
      minCoveragePercentage: 100
      maxMissedOperationsInSpec: 0
      enforce: true
  license:
    path: /specmatic/specmatic-license.txt

Three governance settings matter here:

minCoveragePercentage: 100 — Every single endpoint in the spec must be tested. No skipping.
maxMissedOperationsInSpec: 0 — If the service has endpoints not in the spec, it fails.
schemaResiliencyTests: all — Beyond the hand-written examples, Specmatic automatically generates every valid positive schema variation and every invalid (negative) variation — wrong types, broken enums, missing required fields. More on this in Step 4.

This makes the contract enforcing, not advisory.

Step 3: Build the Service to Satisfy the Contract

With the failing tests as my guide, I implemented the Task Service:

# services/task-service/main.py
from flask import Flask, request, jsonify
from flask_cors import CORS
import json, os

app = Flask(__name__)
CORS(app)

# Three tasks seeded with dedicated IDs — one per mutating operation
# to prevent test ordering conflicts (see Challenges section).
tasks = {
    1: {"id": 1, "title": "Implement Login Feature", "status": "in-progress",
        "assignee": "alice", "priority": "high", "createdAt": "2024-01-15T10:00:00Z"},
    2: {"id": 2, "title": "Write Unit Tests", "status": "pending",
        "assignee": "bob", "priority": "medium", "createdAt": "2024-01-15T11:00:00Z"},
    3: {"id": 3, "title": "Review Pull Request", "status": "pending",
        "assignee": "alice", "priority": "low", "createdAt": "2024-01-15T12:00:00Z"},
}
_next_id = 100  # avoids collision with seed IDs during schema resiliency POST tests

@app.route("/actuator/health", methods=["GET"])
def actuator_health():
    return jsonify({"status": "UP"})

@app.route('/tasks', methods=['GET'])
def list_tasks():
    status_filter = request.args.get('status')
    result = list(tasks.values())
    if status_filter:
        result = [t for t in result if t['status'] == status_filter]
    return jsonify(result), 200

@app.route('/tasks', methods=['POST'])
def create_task():
    global _next_id
    data = request.get_json(silent=True) or {}
    if not data.get('title'):
        return jsonify({"error": "Validation failed", "message": "title is required"}), 400
    if not data.get('priority'):
        return jsonify({"error": "Validation failed", "message": "priority is required"}), 400

    task = {
        "id": _next_id,
        "title": data['title'],
        "description": data.get('description', ''),
        "status": "pending",
        "assignee": data.get('assignee', ''),
        "priority": data['priority'],
        "createdAt": _now(),
    }
    tasks[_next_id] = task
    _next_id += 1

    _publish("task-created", {
        "taskId": task["id"],
        "title": task["title"],
        "status": task["status"],
        "assignee": task["assignee"],
        "priority": task["priority"],
        "timestamp": task["createdAt"],
    })

    return jsonify(task), 201

The implementation's shape was dictated by the contract. Field names, status codes, error response format — all copied from the spec, not invented.

After implementing all five endpoints (GET /tasks, POST /tasks, GET /tasks/{taskId}, PUT /tasks/{taskId}, DELETE /tasks/{taskId}), I ran the contract tests again:

docker compose up --exit-code-from test-task-api test-task-api
# Exit code: 0 ✓

All tests pass. Not because I wrote tests against my own implementation — but because my implementation satisfies a contract that was written independently.

Step 4: Schema Resiliency — Beyond Hand-Written Examples

One example per endpoint is a floor, not a ceiling.

With schemaResiliencyTests in specmatic.yaml, Specmatic automatically expands each example into every variation the schema allows. No extra test files. No test authoring. The schema itself defines the test space. There are three modes:

Mode	What it generates
`positiveOnly`	Every valid combination of enum values and optional-field presence
`negativeOnly`	Every invalid combination — wrong types, broken enums, missing required fields
`all`	Both — the full positive and negative test space

Here's what this means in practice for PUT /tasks/{taskId}:

UpdateTaskRequest:
  type: object
  properties:
    status:
      type: string
      enum: [pending, in-progress, completed]   # 3 values
    assignee:
      type: string
    priority:
      type: string
      enum: [low, medium, high]                 # 3 values

All three fields are optional. The hand-written example only covers {"status": "completed", "assignee": "alice"}. On the positive side, Specmatic generates every valid combination of enum values and optional field presence. On the negative side, it also throws assignee: 42, priority: "urgent", status: null, and dozens of similar mutations at the same endpoint — each one expecting a 4xx response back.

I run with schemaResiliencyTests: all. The test count jump is real:

Mode	Total tests (task-api)	Total tests (user-api)
Example-only (no resiliency)	~10	~5
`positiveOnly`	27	—
`all` (positive + negative)	135	49

This catches two different classes of bugs:

Positive resiliency: a service that handles priority: "high" but crashes on priority: "low" — something a single hand-written example would never reveal.
Negative resiliency: a service that should reject priority: "urgent" or assignee: true with a 400, but instead silently accepts it and corrupts its own state — something no hand-written example was ever written to catch, because nobody writes examples for requests they don't expect.

One important consequence: more tests means more mutations, which means test execution order matters. I cover how I solved this in the Challenges section.

Turning On `schemaResiliencyTests: all` — and Fixing What It Found

Flipping positiveOnly to all didn't just add more passing tests — it took the task-api suite from 27 green tests to 133 tests, 96 of them failing. Here's the real output from that first run:

Tests run: 133, Successes: 37, Failures: 96, WIP: 0, Errors: 0

That's not noise. It's every place where the Flask services accepted garbage instead of rejecting it. Three distinct failure patterns showed up.

Failure 1 — No type or enum validation on write endpoints

POST /tasks and PUT /tasks/{taskId} only checked that required fields were truthy — they never checked that title was a string, that priority was one of low|medium|high, or that assignee wasn't a boolean. Specmatic's negative generator throws every wrong type at every field:

-ve  Scenario: POST /tasks -> 4xx with the request from the example 'CREATE_TASK_201'
     where REQUEST.BODY contains all the keys AND the key priority is mutated
     from ("low" or "medium" or "high") to number FAILED
Reason:
    >> RESPONSE.STATUS
        Expected 4xx status, but received 201

87 of the 96 failures were exactly this shape: Expected 4xx status, but received 200/201. The fix was real input validation in both services — type checks via isinstance, enum checks against an explicit set, applied before anything gets written to the in-memory store:

# services/task-service/main.py
TASK_PRIORITIES = {"low", "medium", "high"}
TASK_STATUSES = {"pending", "in-progress", "completed"}

def create_task():
    data = request.get_json(silent=True) or {}

    title = data.get("title")
    if not isinstance(title, str) or not title:
        return _validation_error("title is required")

    priority = data.get("priority")
    if priority not in TASK_PRIORITIES:
        return _validation_error(f"priority must be one of {sorted(TASK_PRIORITIES)}")

    if "description" in data and not isinstance(data["description"], str):
        return _validation_error("description must be a string")
    if "assignee" in data and not isinstance(data["assignee"], str):
        return _validation_error("assignee must be a string")
    ...

The same pattern applied to PUT /tasks/{taskId} (status/priority enums, assignee type) and to user-service's POST /users (role enum). One subtlety that cost a second round of failures: checking value is not None and not isinstance(value, str) lets an explicit null slip through, because None is not None is False. The fix is to drop the is not None guard entirely — if the key is present, its value must satisfy the type, full stop.

Failure 2 — The contract had no `400` to validate against

After adding validation, a different error appeared for PUT /tasks/{taskId}:

Reason:
    >> RESPONSE.STATUS
        R0002: HTTP status mismatch
        Specification expected status 404 but response contained status 400

And for GET /tasks?status=<invalid>:

Reason:
    >> RESPONSE.STATUS
        Received 400, but the specification does not contain a 4xx or default
        response, hence unable to verify this response

The service was now doing the right thing — returning 400 for bad input — but the OpenAPI spec never declared a 400 response for these two operations. Only POST /tasks had one. Specmatic enforces the contract literally: if you return a status code the spec doesn't document, that's a contract violation, even if it's semantically correct. This is the contract-first principle paying for itself — the gap was in the spec, not just the code. The fix was adding the missing response definitions to task-api.yaml:

# GET /tasks
'400':
  description: Validation error — invalid status filter
  content:
    application/json:
      schema:
        $ref: '#/components/schemas/Error'
      examples:
        LIST_TASKS_400:
          value:
            error: "Validation failed"
            message: "status must be one of ['completed', 'in-progress', 'pending']"

# PUT /tasks/{taskId}
'400':
  description: Validation error — invalid field value
  content:
    application/json:
      schema:
        $ref: '#/components/schemas/Error'
      examples:
        UPDATE_TASK_400:
          value:
            error: "Validation failed"
            message: "priority must be one of ['high', 'low', 'medium']"

Each new 400 needed a matching named example on the parameter/request side too (LIST_TASKS_400 on the status query param, UPDATE_TASK_400 on both the taskId path param and the request body) — Specmatic pairs examples by name, the same rule from Challenge 1.

Failure 3 — Flask's default 404 page broke the error schema

Mutating taskId from a number to a string or boolean (e.g. DELETE /tasks/true) doesn't match Flask's <int:task_id> route converter, so Flask falls back to its own default 404 — an HTML page, not JSON:

>> RESPONSE.HEADER.Content-Type
    R1002: Value mismatch
    Specification expected application/json but response contained text/html; charset=utf-8

>> RESPONSE.BODY
    R1001: Type mismatch
    Specification expected type json object but response contained value
    "<!doctype html>\n<html lang=en>\n<title>404 Not Found</title>..."

The fix is a Flask-wide 404 handler that returns JSON matching the Error schema for any path Flask itself can't route — independent of the explicit 404 responses already returned by get_task/update_task/delete_task when a valid taskId doesn't exist:

@app.errorhandler(404)
def handle_not_found(_exc):
    return jsonify({"error": "Not Found", "message": "The requested resource was not found"}), 404

Result

After all three fixes — input validation, the missing 400 contract responses, and the JSON 404 handler — task-api went from 37/133 passing to 135/135 passing (two new explicit 400 examples added two tests), and user-api went from 30/49 to 49/49. Both numbers were stable across repeated runs, confirming the fixes weren't masking flakiness.

Tests run: 135, Successes: 135, Failures: 0, WIP: 0, Errors: 0
Tests run: 49, Successes: 49, Failures: 0, WIP: 0, Errors: 0

The headline: 96 negative-path bugs existed in services that had 100% positive-path contract coverage and a green CI pipeline. Schema resiliency in all mode is the difference between "the API works when you use it correctly" and "the API is actually safe to expose to a client you don't control" — which, for any service called by another team, another service, or an AI agent generating its own request bodies, is the only question that matters.

Step 5: Extend to Async Events with AsyncAPI

REST APIs are only half the story in an event-driven system. When the Task Service creates a task, it publishes to Kafka. Without a contract for those messages, any consumer can break silently.

I wrote an AsyncAPI 3.0.0 spec for the Kafka events:

# specs/asyncapi/task-events.yaml
asyncapi: 3.0.0
info:
  title: Task Events
  version: 1.0.0

servers:
  kafka:
    host: kafka:9092
    protocol: kafka

channels:
  task-created:
    address: task-created
    messages:
      TaskCreatedMessage:
        payload:
          type: object
          required: [taskId, title, status, timestamp]
          properties:
            taskId:
              type: integer
            title:
              type: string
            status:
              type: string
              enum: [pending, in-progress, completed]
            timestamp:
              type: string
              format: date-time
        examples:
          - payload:
              taskId: 1
              title: "New deployment task"
              status: pending
              timestamp: "2024-01-15T10:00:00Z"

The async contract test works differently from REST tests. Instead of Specmatic acting as a client hitting endpoints, the test-async process:

Subscribes to the Kafka task-created and task-updated topics
Runs a before hook that issues a real HTTP call against the Task Service (e.g. POST /tasks)
Validates the message that publish produces against the AsyncAPI schema

# specmatic-async.yaml
version: 3

systemUnderTest:
  service:
    definitions:
      - definition:
          source:
            filesystem:
              directory: specs/asyncapi
          specs:
            - task-events.yaml
    runOptions:
      asyncapi:
        type: test
        subscriberReadinessWaitTime: 5000
        servers:
          - host: kafka:9092
            protocol: kafka
    data:
      examples:
        - directories:
            - specs/asyncapi/examples

Gotcha I hit: The AsyncAPI spec initially had host: localhost:9092. Inside Docker's network, services communicate by service name, not localhost. Changing to host: kafka:9092 (matching the Docker Compose service name) fixed the connection immediately. Always verify your spec's server hosts match your runtime network topology.

Gotcha I hit (the expensive one): Inline examples on an AsyncAPI Message are documentation only — nothing actually calls the REST API to make the service publish. I initially put the example payload directly on the operations entry, which AsyncAPI 3.0 rejects (examples is only valid on Message, not Operation). Moving it to the message fixed parsing, but the test still timed out waiting for a Kafka message, because there was still nothing driving the publish. The actual mechanism is external JSON example files (specs/asyncapi/examples/, wired up via data.examples.directories) with a before array that fires the real HTTP request:
{
  "name": "TASK_CREATED_EXAMPLE",
  "before": [
    {
      "type": "http",
      "http-request": { "baseUrl": "http://task-service:8080", "path": "/tasks", "method": "POST", "body": { "title": "...", "priority": "high" } },
      "http-response": { "status": 201 }
    }
  ],
  "send": { "topic": "task-created", "payload": { "taskId": "(integer)", "title": "...", "timestamp": "(datetime)" } }
}
(integer) / (datetime) are Specmatic's type-matcher placeholders — useful here since the published taskId and timestamp are server-generated and can't be hardcoded.

Gotcha I hit (CI-only): specmatic test --config=specmatic-async.yaml quietly does nothing for AsyncAPI — --config only applies to the OpenAPI test path. Since the whole repo (including the real specmatic.yaml) is bind-mounted into every Specmatic container, the async test was silently running the REST config and reporting "no test scenarios found." Fixed by bind-mounting specmatic-async.yaml over /usr/src/app/specmatic.yaml for the test-async service specifically, so its default config lookup resolves correctly without touching the other services.

Gotcha I hit (the one that mattered most): Once the wiring was right, the test still failed — every message timed out. The task service's _publish() wraps the Kafka call in a try/except (so the service can start before Kafka is ready), and that swallowed an exception every single time: kafka-python==2.0.2 doesn't work on Python 3.12 (No module named 'kafka.vendor.six.moves'). The Kafka publish had never actually succeeded — it just looked fine because nothing was asserting on it until the async contract test existed. Switching to kafka-python-ng (a maintained drop-in fork, same import kafka) fixed it. This is exactly the failure mode contract testing is supposed to catch: a swallowed exception that made the service look healthy while silently dropping every event.

Step 6: Mock Server for Parallel Development

While building the frontend Kanban board, the Task Service wasn't fully implemented yet. Rather than hardcoding fake data or building a separate mock, I used Specmatic's mock server:

docker compose --profile mock up mock-task-api

Specmatic reads task-api.yaml, extracts the inline examples, and serves them as HTTP responses — with zero additional configuration. GET /tasks returns the TASKS_200_OK example. POST /tasks with a missing title returns the CREATE_TASK_400 example.

The frontend has a built-in API switcher in the sidebar that toggles between the mock server (:9100) and the real service (:8080). A frontend developer can work against realistic API responses before the backend exists, and switching to the real service requires changing exactly one URL.

This is the pattern that enables parallel development without integration ceremonies: frontend, backend, and mobile teams can all work simultaneously against the same source of truth (the spec), without coordinating implementation timelines.

Step 7: Governance Enforcement in CI

The final piece is making contracts enforced — not just available.

The docker-compose.yaml uses --exit-code-from to make the entire compose stack fail if contract tests fail:

# In a CI pipeline
docker compose up --exit-code-from test-task-api
echo "Exit code: $?"

With minCoveragePercentage: 100, a developer can't ship a partial implementation. If the spec defines DELETE /tasks/{taskId} and the service doesn't implement it, the entire CI run fails.

Because schemaResiliencyTests: all lives in specmatic.yaml rather than as a CLI flag, every CI run picks it up automatically — docker compose run test-task-api and docker compose run test-user-api both read the same config, so the GitHub Actions workflow (.github/workflows/ci.yml) enforces negative-path validation on every push and pull request with zero pipeline changes beyond the config file itself. There's no separate "resiliency" job to maintain or forget to run — it's the same job, just testing a larger space.

Specmatic generates HTML reports at build/reports/specmatic/test/html/index.html. These show exactly which scenarios passed, which failed, and the diff between expected and actual responses — the artifact that goes into pull request reviews.

The governance model I enforced:

Rule	Effect
Every operation in the spec = tested	Can't skip endpoints
Every response = validated against schema	Field drift caught immediately
Any drift between spec and implementation	Build failure

This turns the OpenAPI spec from passive documentation into an active quality gate.

The AI-Native Angle: Building Feedback Loops for Coding Agents

Here's why this architecture matters beyond traditional development.

When you ask an AI coding agent to implement a service, the typical workflow is:

Give the agent a task description
Agent generates code
Human manually checks if it's right

With contract-driven development, the workflow becomes:

Write (or generate) the OpenAPI spec
Give the agent the spec + the failing Specmatic output
Agent generates code
Specmatic runs and produces a precise diff
Feed the diff back to the agent
Agent iterates until all tests pass
No human verification needed at the integration layer

The contract becomes the objective ground truth that the AI optimizes against. It's not "does this code look right" — it's "does this code satisfy these concrete, machine-verifiable assertions?"

This is what "AI-native infrastructure" actually means: not using AI to write code, but building the verification substrate that lets AI code generation be reliable at scale.

Full System Architecture

┌─────────────────────────────────────────────────────────┐
│                     Docker Compose                      │
│                                                         │
│  ┌──────────────┐      ┌──────────────┐                 │
│  │ task-service │      │ user-service │                 │
│  │  Flask :8080 │      │  Flask :8081 │                 │
│  └──────┬───────┘      └──────────────┘                 │
│         │ publishes events                              │
│  ┌──────▼───────┐                                       │
│  │    Kafka     │                                       │
│  │    :9092     │                                       │
│  └──────────────┘                                       │
│                                                         │
│  ┌───────────────────┐   ┌──────────────────────────┐   │
│  │  test-task-api    │   │  test-async              │   │
│  │  Specmatic REST   │   │  Specmatic AsyncAPI      │   │
│  │  (OpenAPI 3.0.1)  │   │  (Kafka topic tests)     │   │
│  └───────────────────┘   └──────────────────────────┘   │
│                                                         │
│  ┌──────────────┐   ┌─────────────────────────────────┐ │
│  │  frontend    │   │  mock-task-api                  │ │
│  │  Kanban:3000 │   │  Specmatic Mock Server :9100    │ │
│  └──────────────┘   └─────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘

Contract Coverage at a Glance

Spec File	Endpoints / Channels	Named Examples	Tests with `schemaResiliencyTests: all`
`specs/openapi/task-api.yaml`	5 REST endpoints	12 (200, 201, 204, 400, 404 scenarios)	135
`specs/openapi/user-api.yaml`	3 REST endpoints	5 (201, 400, 200, 200, 404)	49
`specs/asyncapi/task-events.yaml`	2 Kafka channels	`task-created`, `task-updated`	—

Real Errors Specmatic Surfaced — and How I Fixed Them

To demonstrate Specmatic's feedback loop concretely, I deliberately introduced a field rename in task-service/main.py — the kind of silent drift that kills multi-service systems in production — and ran the full contract test suite to capture what Specmatic actually reports.

The break: `title` → `task_title`

A single-line change in the create_task handler:

# Before (matches the spec)
task = {
    "id": _next_id,
    "title": data["title"],
    ...
}

# After (drifted from the spec)
task = {
    "id": _next_id,
    "task_title": data["title"],   # ← renamed
    ...
}

What Specmatic reported

Running docker compose up --exit-code-from test-task-api:

Failure 1 — POST /tasks → 500 instead of 201

The service crashed immediately. The Kafka publish code still referenced task["title"], which no longer existed:

taskflow-task-service | ERROR in app: Exception on /tasks [POST]
taskflow-task-service |   File "/app/main.py", line 106, in create_task
taskflow-task-service |     "title": task["title"],
taskflow-task-service |              ~~~~^^^^^^^^^
taskflow-task-service | KeyError: 'title'
taskflow-task-service | 172.19.0.2 - - [POST /tasks] 500

taskflow-test-task-api | +ve  Scenario: POST /tasks -> 201 ... has FAILED
taskflow-test-task-api | Reason:
taskflow-test-task-api |   >> RESPONSE.STATUS
taskflow-test-task-api |       R0002: HTTP status mismatch
taskflow-test-task-api |       Specification expected status 201 but response contained status 500

All 6 schema resiliency variants of POST /tasks → 201 failed the same way.

Failure 2 — GET /tasks → schema violation (cascade)

Because the task was inserted into the in-memory store before the Kafka publish crashed, the malformed object (with task_title instead of title) persisted. When GET /tasks ran next, the list response contained those corrupt items — and Specmatic caught it at the schema layer:

taskflow-test-task-api | +ve  Scenario: GET /tasks -> 200 ... has FAILED
taskflow-test-task-api | Reason:
taskflow-test-task-api |   >> RESPONSE.BODY[3].title
taskflow-test-task-api |       R2001: Missing required property
taskflow-test-task-api |       Specification expected mandatory property "title" to be present
taskflow-test-task-api |       but was missing from the response
taskflow-test-task-api |
taskflow-test-task-api |   >> RESPONSE.BODY[3].task_title
taskflow-test-task-api |       R2003: Unknown property
taskflow-test-task-api |       Property "task_title" in the response was not in the specification

Final result:

Tests run: 27, Successes: 19, Failures: 8, WIP: 0, Errors: 0
89% API Coverage — FAILED (minCoveragePercentage = 100%)

One field rename caused 8 failures across two endpoints — 6 from POST crashing and 2 from GET returning corrupt data — before a single human reviewer saw anything.

The fix

Two things needed to be consistent:

# 1. Keep the task dict key matching the spec field name
task = {
    "id": _next_id,
    "title": data["title"],   # ← back to "title"
    ...
}

# 2. The Kafka publish code already referenced task["title"] correctly
#    — so reverting the dict key was the only change needed

After reverting: 27/27 tests passing, 100% coverage.

What this demonstrates

The contract test caught the break at the exact moment it was introduced — no integration environment, no manual testing, no waiting for another team to hit the issue. The error output told me precisely which field was wrong (RESPONSE.BODY[3].title), which rule was violated (R2001, R2003), and what the spec expected versus what the service returned.

This is the feedback loop that makes contract-first development reliable at scale.

Challenges and What I Learned

1. Example naming symmetry is non-negotiable

Specmatic matches examples across request/response pairs by name. If your parameter example is named GET_TASK_200 but your response example is GET_TASK_SUCCESS, Specmatic won't pair them into a test. I standardized on a VERB_RESOURCE_STATUSCODE convention and never had pairing issues after that.

2. Healthcheck timing in Docker Compose

The service_completed_successfully dependency type waits for a container to exit — it doesn't mean the service inside is ready. I added explicit healthchecks to every service and used service_healthy dependencies for the test containers, eliminating "connection refused" flakiness:

healthcheck:
  test: ["CMD", "python", "-c",
    "import urllib.request; urllib.request.urlopen('http://localhost:8080/health')"]
  interval: 2s
  timeout: 5s
  retries: 30
  start_period: 5s

3. Kafka publish blocking service startup

The first implementation blocked Flask startup while connecting to Kafka. If Kafka wasn't ready, the entire service failed. I wrapped the Kafka producer and every publish call in try/except:

def _publish(topic: str, payload: dict):
    try:
        from kafka import KafkaProducer
        producer = KafkaProducer(
            bootstrap_servers=KAFKA_BOOTSTRAP_SERVER,
            value_serializer=lambda v: json.dumps(v).encode("utf-8"),
        )
        producer.send(topic, payload)
        producer.flush(timeout=3)
    except Exception as exc:
        app.logger.warning("Kafka publish skipped (%s): %s", topic, exc)

The service stays healthy and responds to REST calls even if Kafka is temporarily unavailable.

4. AsyncAPI server host resolution

The AsyncAPI spec's servers[].host must match the hostname that the Specmatic container can resolve. Inside Docker Compose, that's the service name (kafka), not localhost. Always check your spec's server host against your runtime network topology.

5. Specmatic Enterprise requires `/actuator/health` to execute tests

This one cost me time. Running docker compose up test-user-api produced an HTML report showing 0% coverage, 4 tests skipped, "Actuator Not Available" — even though the Flask service was fully up and responding correctly.

Specmatic Enterprise checks for a GET /actuator/health endpoint before executing any tests. Without it, every scenario is marked "Skipped" regardless of service health. The fix is a single route on each Flask service:

@app.route("/actuator/health", methods=["GET"])
def actuator_health():
    return jsonify({"status": "UP"})

This is separate from the application's own /health endpoint. Once added, all 4 tests ran and coverage showed correctly.

6. Schema resiliency + stateful in-memory store = test ordering trap

Enabling schemaResiliencyTests: positiveOnly expanded the task-api suite from ~10 to 27 tests. More tests exposed a subtle ordering bug I hadn't hit before.

Specmatic runs operations on the same path alphabetically by HTTP method: DELETE → GET → PUT. My spec used taskId=1 for all three success scenarios. When DELETE ran first and removed task 1, the subsequent GET and all 11 PUT schema resiliency variations against taskId=1 returned 404 instead of 200 — 13 cascading failures from a single deletion.

The fix: assign each mutating operation its own dedicated seed task.

tasks = {
    1: {...},   # GET /tasks/1  — read-only, never touched by other operations
    2: {...},   # PUT /tasks/2  — mutated but never deleted
    3: {...},   # DELETE /tasks/3 — disposable
}
_next_id = 100  # POST schema resiliency tests create IDs 100+ — no collision with seeds

And in the spec, update the examples to match:

GET_TASK_200:   value: 1   # task 1
UPDATE_TASK_200: value: 2  # task 2
DELETE_TASK_204: value: 3  # task 3

Each operation now has its own data slice. Tests pass regardless of execution order.

7. `schemaResiliencyTests: all` requires the spec and the service to grow together

Switching from positiveOnly to all wasn't just a config flag — every negative test that found a missing 400 response meant the spec was incomplete, not just the code. I'd documented 400 for POST /tasks and POST /users from day one (because "what if a required field is missing" is an obvious case to write an example for), but GET /tasks?status= and PUT /tasks/{taskId} had no documented error path for invalid input, because nobody had written a negative example for them. Negative resiliency testing finds exactly the blind spots that example-driven testing structurally can't — the requests nobody thought to write down. The fix pattern is always the same: add the 4xx response to the spec, add a validation check to the service that returns it, in that order.

Running It Yourself

git clone https://github.com/kanugurajesh/Spectamic-Taskflow
cd Spectamic-Taskflow

# Copy your Specmatic Enterprise license into the project root first:
# cp /path/to/your/license.txt ./license.txt

# Full stack + REST contract tests (use --build on first run)
docker compose up --build

# Run only the contract tests
docker compose up test-task-api test-user-api --build --abort-on-container-exit

# With async event tests
docker compose --profile async up test-async --abort-on-container-exit

# Frontend against mock server (no backend needed)
docker compose --profile mock up mock-task-api frontend

# Open Specmatic Studio (visual contract editor)
docker compose --profile studio up studio
# Visit http://localhost:9000

# CI mode: fail fast on contract violations
docker compose up --exit-code-from test-task-api

# Tear down
docker compose down --remove-orphans

Test reports are generated at build/reports/specmatic/test/html/index.html.

Note: Always pass --build after changing any service code (services/*/main.py). Spec file changes (.yaml) are picked up immediately via volume mount — no rebuild needed for those.

Conclusion

The most important thing I learned building specmatic-taskflow: a contract that isn't executable isn't a contract, it's a suggestion.

OpenAPI specs are everywhere. AsyncAPI specs are increasingly common. But most of them live as YAML files in a /docs folder that nobody updates after the initial commit. They become lies — specifications that describe a system as it was, not as it is.

Specmatic changes the relationship. When your spec runs — when it becomes the test, the mock, the governance gate — it has to stay true. Every CI run verifies it. Every developer who breaks it sees it in the build log.

For AI coding agents, this matters even more. An agent that generates code against a contract and gets precise, executable feedback can iterate reliably. An agent working against documentation can only guess.

I built specmatic-taskflow to prove that a single engineer can, in a weekend, wire together a full contract-driven system across REST and async messaging — and end up with more confidence in the integration than most teams get after months of manual testing.

That's the infrastructure layer AI-native software development needs.

About the Author

Kanugurajesh — Final-year B.E. CS student at MVSR Engineering College. 4100+ GitHub contributions, hackathon winner, 2 years of internship experience at AI and SaaS companies. Building at the intersection of AI agents and reliable software systems.

Code: github.com/kanugurajesh/Spectamic-Taskflow
GitHub: github.com/kanugurajesh
Email: kanugurajesh6@gmail.com

DEV Community: kanugu rajesh