I Stopped Treating AI Pipelines Like Magic: Designing Explicit State Boundaries

I was practicing a small AI workflow recently: prepare text in Python, send it through a model step, and format the result in JavaScript.

The model was not the hardest part.

The difficult part was the boundary around it.

Normally, I would end up with:

• a temporary JSON file;
• a hidden environment variable carrying state;
• a second script that knows too much about the first script;
• and no obvious place to inspect what the model actually received.

That is how a five-line experiment quietly becomes a small integration project.

💡 The Problem: AI State Gets Lost Between Stages

An AI pipeline usually has at least three different responsibilities:

1. Preparation: normalize text, remove noise, and compute metadata.
2. Inference: send a small, well-defined input to a model.
3. Presentation: turn the model result into something a person or another program can use.

If those stages share an implicit global object, debugging becomes guesswork. If they share a temporary file, the file becomes both a coupling point and a leak surface.

The first improvement is to define the state contract before choosing a model.

{
  "request_id": "demo-001",
  "text": "Block coordinates multiple native runtimes in one file.",
  "language": "en",
  "features": {
    "characters": 60,
    "words": 9
  },
  "model_input": "Classify this developer documentation sentence."
}

The contract answers a simple question: what exactly crosses the boundary?

⚙️ Stage 1: Prepare a Stable Input in Python

The preparation stage should be deterministic. It should be possible to run it without an AI provider and inspect the result.

import json
import re
import sys


def build_state(request_id: str, text: str) -> dict:
    words = re.findall(r"[A-Za-z0-9_'-]+", text)
    return {
        "request_id": request_id,
        "text": text,
        "language": "en",
        "features": {
            "characters": len(text),
            "words": len(words),
        },
        "model_input": (
            "Classify this developer documentation sentence as "
            "technical, product, or general: " + text
        ),
    }


state = build_state(
    "demo-001",
    "Block coordinates multiple native runtimes in one file.",
)

print(json.dumps(state, ensure_ascii=False))

This stage does not call a model. That is intentional. A pipeline that cannot produce a valid input without the model is difficult to test.

💻 Stage 2: Consume the Contract in JavaScript

The next stage reads one JSON document from standard input. In production, this is where I would call a local model or a hosted inference API. For the first test, I use a deterministic classifier so the transport and validation can be checked without credentials.

let raw = "";

process.stdin.setEncoding("utf8");
process.stdin.on("data", chunk => {
  raw += chunk;
});

process.stdin.on("end", () => {
  const state = JSON.parse(raw);

  if (!state.request_id || !state.text || !state.features) {
    throw new Error("Invalid AI pipeline state");
  }

  const label = /runtime|python|javascript|code/i.test(state.text)
    ? "technical"
    : "general";

  const result = {
    request_id: state.request_id,
    label,
    confidence: label === "technical" ? 0.98 : 0.61,
    words_seen: state.features.words,
  };

  console.log(JSON.stringify(result));
});

The important detail is not the classifier. It is the explicit input and output shape.

⚡ Output You Can Actually Debug

{"request_id":"demo-001","label":"technical","confidence":0.98,"words_seen":9}

When the real model is added, the output can be compared against this same contract. A provider change should not require rewriting the preparation and presentation stages.

🚀 Key Takeaways

• State is an interface: treat model input and output like an API, not an accidental dictionary.
• Deterministic stages come first: test preparation and validation before adding model variability.
• Keep the model boundary narrow: send only the fields the model needs, and validate what comes back.
• Make failures visible: include a request ID, stage name, and safe summary in logs.

The model is only one stage in an AI program. The reliability of the whole program depends on the boundaries around it.

🧪 Test the Pipeline Without a Model

The most useful test is a boring one: send a known state through the pipeline and compare the result with a known output.

def test_state_contract():
    state = build_state("test-001", "Python sends structured state to JavaScript.")

    assert state["request_id"] == "test-001"
    assert state["features"]["words"] == 7
    assert "Classify" in state["model_input"]

The JavaScript stage should have the same kind of test. It should reject missing fields instead of silently inventing defaults:

function validateState(state) {
  const required = ["request_id", "text", "features", "model_input"];
  const missing = required.filter(key => !(key in state));
  if (missing.length > 0) {
    throw new Error(`Missing state fields: ${missing.join(", ")}`);
  }
  return state;
}

This is especially important when the model is nondeterministic. If a result changes, I want to know whether the input changed, the model changed, or the transport broke.

🔁 Adding a Real Model Later

Once the contract is stable, the deterministic classifier can be replaced with a model call. I would still keep the provider-specific code behind one function:

async function classifyWithModel(state, client) {
  const response = await client.generate({
    input: state.model_input,
    metadata: { request_id: state.request_id },
  });

  return {
    request_id: state.request_id,
    label: response.label,
    confidence: Number(response.confidence),
    words_seen: state.features.words,
  };
}

The rest of the pipeline should not need to know whether the model is local, hosted, small, or large. That decision belongs inside the inference stage.

There is another practical reason to keep the contract small: retries. If a provider times out, the pipeline can retry the inference stage with the same request_id and the same prepared input. It does not need to repeat text normalization or rebuild the entire workflow.

🔒 Do Not Put Secrets in the Shared State

State synchronization is not a reason to copy every variable between stages. API keys, access tokens, cookies, and private source documents should stay in the narrowest possible scope.

safe_state = {
    "request_id": state["request_id"],
    "text": state["text"],
    "features": state["features"],
}

The model only needs the fields required for the task. Smaller state is easier to inspect, easier to redact, and less likely to appear in a log by accident.

I am exploring the same idea in Block Language, where native runtime blocks can share a visible state pipeline in one readable file.

What is the first state boundary you make explicit in your AI projects?