AIPod

A governed software construction agent and compositional runtime for AI-built Python applications.

AIPod lets AI build ordinary Python applications inside a small, explicit architecture. The model generates one bounded artifact at a time; local code controls ordering, Contracts, validation, freezing, retries, and repair scope.

Model → Provider → Service → Pipeline → Interface
 data    capability    business    composition    delivery

The result is not an opaque AI session. It is a resumable project with typed boundaries, registered routes, runtime evidence, and generated source code that can be inspected and maintained with normal tools.

AIPod is currently alpha software. Review generated code and platform installers before production use.

Quick start

Python 3.10 or newer is required.

pip install -U AIPodCli

mkdir todo-app
cd todo-app
aipod init

Configure an OpenAI-compatible model endpoint once:

aipod config set OPENAI_API_KEY sk-your-key
aipod config set OPENAI_BASE_URL https://api.openai.com/v1
aipod config set OPENAI_MODEL your-model

Build an application:

aipod pod --yes \
  "Create a local todo application with persistent tasks, add/list/complete routes, and a CLI Adapter."

Inspect and test it:

aipod inspect --summary --json
aipod inspect project --json
aipod interface list
aipod interface smoke <interface-name>

Run a generated Interface Adapter with a JSON event:

aipod interface run <interface-name> \
  --payload '{"action":"list"}'

Adapters that expose CLI-style arguments can receive raw arguments after --:

aipod interface run <interface-name> -- \
  --mode once \
  --payload '{"message_id":"m-1","topic":"orders","payload":{}}'

Why AIPod

Large AI-generated applications usually fail at their boundaries:

• one component produces shipment_count while another expects shipments_count;
• a Model is accidentally injected as an infrastructure dependency;
• generated code imports symbols from the wrong package;
• one Service injects and directly executes another Service, bypassing Pipeline governance;
• a downstream failure causes an upstream working file to be rewritten;
• a Pipeline exists but no verified user-facing Interface reaches it;
• syntax checks pass while imports or dependency injection fail at runtime.

AIPod addresses these failures with five rules:

1. Build in dependency order. Earlier layers are completed before downstream layers.
2. Make boundaries machine-readable. IDs, dependencies, Contracts, routes, lifecycle, permissions, and verification commands are stored as project state.
3. Freeze accepted upstream work. A downstream failure may retry or repair its own scope, but does not silently reopen stable layers.
4. Require evidence before completion. Generated artifacts must pass local structural and disposable runtime checks before a stage can freeze.
5. Keep orchestration out of Services. A Service can see its Contract, Models, and Providers, but never another Service. Composition belongs exclusively to Pipelines.

The five layers

Model

Models are shared typed data. Runtime value objects and persistent SQLModel entities are both supported.

from ai_pod_cli import Model


class Message(Model):
    message_id: str
    topic: str
    payload: dict

Models are imported as data types. They are never injected.

Provider

Providers expose infrastructure capabilities such as files, databases, HTTP clients, or message transports. They may be injected into Services.

Built-in Service-visible Providers include:

• ConfigStore
• ModelRepository

PipelineRunner is a reserved Runtime capability used behind Interface and CLI route boundaries. It is not a way for one Service to reach another Service.

Service

Services implement business transformations through execute(ctx).

from ai_pod_cli.context import PipelineContext


class MessageProcessingService:
    def execute(self, ctx: PipelineContext) -> dict:
        message_id = ctx.get("message_id")
        result = {"message_id": message_id, "status": "processed"}
        ctx.set("result", result)
        return result

A Service has a deliberately narrow capability view:

Service-to-Service dependencies are rejected independently by the Planner, Canonical Plan reducer, source validator, and DI Runtime. A Service must not become a hidden orchestrator:

# Invalid: this bypasses Contract checks, Trace, Failure, retry, and execution policy.
class GameLoopService:
    def __init__(self, physics_service, render_service):
        self.physics_service = physics_service
        self.render_service = render_service

    def execute(self, ctx):
        self.physics_service.execute(ctx)
        return self.render_service.execute(ctx)

Pipeline

Pipelines compose Services in deterministic order and are registered as named routes.

from ai_pod_cli.config import load_beans
from ai_pod_cli.container import Pod, build_container


def run(ctx):
    S = Pod(build_container(load_beans()))
    (S(ValidateMessage) | S(ProcessMessage)).execute_all(ctx)
    return ctx.summary()

Interfaces see route names and descriptions, not Service classes.

The same Pipeline Runtime supports governed asynchronous, parallel, repeated, and streaming execution. Operators are explicit:

Existing synchronous Pipelines remain compatible. AI declares these policies, while the local Runtime owns scheduling, merging, stopping, cancellation, and Trace.

from ai_pod_cli.container import parallel


async def run(ctx):
    S = Pod(build_container(load_beans()))
    flow = parallel(
        S(QueryInventory),
        S(QueryPrice),
        merge="strict",
        failure_policy="collect_all",
        concurrency=2,
    ) | S(BuildResponse)
    await flow.execute_all_async(ctx)
    return ctx.summary()

Repeated workflows such as game frames, workers, polling, and bounded retries remain visible in the Pipeline instead of being hidden inside a coordinating Service:

from ai_pod_cli.container import Pod, build_container, repeat


def run(ctx):
    S = Pod(build_container(load_beans()))
    frame = (
        S(InputHandlingService)
        | S(SceneUpdateService)
        | S(PhysicsService)
        | S(RenderService)
    )
    repeat(
        frame,
        until_field="quit_requested",
        max_iterations_field="max_frames",
        output_field="executed_frames",
        trace_limit=20,
    ).execute_all(ctx)
    return ctx.summary()

The stop condition is a named Context field, not an arbitrary AI-generated callback. Each iteration uses an isolated Context snapshot, merges successful writes deterministically, stops on Failure, and retains only a bounded number of iteration traces.

See docs/execution.md for async routes, deterministic branch merging, repetition, stream processing, failure policies, and Contract behavior.

Interface

An Interface is a delivery bundle around one AI-generated project Adapter. It can bridge any external event source to frozen Pipeline routes:

CLI arguments    \
HTTP request      \
queue message      ─→ InterfaceAdapter ─→ context.run_route() ─→ Pipeline
desktop UI event  /
file/timer event /

AIPod provides the stable SDK:

from ai_pod_cli.interface import InterfaceAdapter, InterfaceContext

AI generates project-specific glue:

class GeneratedInterfaceAdapter(InterfaceAdapter):
    def required_routes(self):
        return ["process_message"]

    def start(self, context: InterfaceContext, payload=None):
        message = receive_external_message(payload)
        return context.run_route("process_message", message)

The Adapter cannot import Models, Providers, Services, the DI container, or PipelineRunner. Its only business capability is InterfaceContext.run_route().

Multi-file Interface Adapters

Complex adapters are split into focused files and generated one file per model call:

interfaces/order-monitor/
├── adapter.py          Adapter entry class
├── queue_consumer.py   message transport
├── window.py           desktop UI
├── event_bridge.py     thread/UI bridge
├── install.ps1         platform lifecycle
└── interface.json      canonical manifest

The manifest identifies the entry source and class:

{
  "name": "order-monitor",
  "kind": "windows_desktop_queue",
  "platform": "windows",
  "adapter": {
    "entry_path": "interfaces/order-monitor/adapter.py",
    "class_name": "GeneratedInterfaceAdapter"
  },
  "artifacts": [
    {"path": "interfaces/order-monitor/adapter.py", "role": "adapter_entry"},
    {"path": "interfaces/order-monitor/queue_consumer.py", "role": "adapter_module"},
    {"path": "interfaces/order-monitor/window.py", "role": "adapter_module"},
    {"path": "interfaces/order-monitor/test_behavior.py", "role": "behavior_test", "format": "python"}
  ],
  "lifecycle": {
    "run": ["{python}", "-m", "ai_pod_cli", "interface", "run", "order-monitor"]
  },
  "permissions": ["message_queue_connect", "desktop_notification"],
  "verify": [
    {
      "name": "adapter_smoke",
      "kind": "smoke",
      "required": true,
      "command": ["{python}", "-m", "ai_pod_cli", "interface", "smoke", "order-monitor"],
      "timeout": 30
    },
    {
      "name": "order_behavior",
      "kind": "behavior",
      "required": true,
      "command": ["{python}", "-m", "ai_pod_cli.behavior_tests", "interfaces/order-monitor/test_behavior.py"],
      "cases": [{"test": "OrderBehavior.test_process_message", "requirement": "A supplied order message produces the expected order state"}],
      "timeout": 60
    }
  ]
}

All Adapter source files are staged together, loaded as a private Python package so relative imports work, and smoked in a disposable project. The complete Interface bundle is committed atomically after artifact and Adapter checks. Application completion additionally requires the declared behavior acceptance below.

The Adapter is generated during construction. Running the finished application does not call AI.

Python source output protocol

Python component generation (pod and create), Pipeline composition, Interface delivery files, and the legacy entry generator separate metadata from source:

1. Generate JSON metadata such as Contracts, dependencies, method signatures, configuration additions and extra packages, without a code or content field.
2. Freeze that metadata and request one XML-like source artifact through text mode:

<create>
  <path>modules/services/priceorder.py</path>
  <content><![CDATA[
class PriceOrder:
    def execute(self, ctx):
        return {"total_cents": 9000}
]]></content>
</create>

The source response is not forced into JSON. The planned path must match exactly; multiple actions, nested operands, extra fields and malformed XML are rejected. If a metadata response also includes unrequested root-level code or content, those fields are discarded. Source is still requested separately through XML; discarded metadata code is never executed or used as a fallback. Contract fields named code or content inside inputs/outputs are preserved. CDATA splitting supports literal ]]> in source, and the codec preserves source line endings. This uses the same strict ActUnit-compatible artifact subset as AIPod Node, implemented locally without requiring an unpublished ActUnit package or a Node process.

Source format failures retry against the frozen metadata. Existing component checks, disposable runtime verification and Interface bundle validation still apply. Planning and exact-patch repair continue to use JSON. New components normally need three model calls: metadata, test source, then implementation source. Both source files use XML; retries reuse the frozen metadata and tests.

Source requests default to a 65,536-token output budget and a 600-second SDK timeout. JSON metadata, planning and patch requests default to 32,768 tokens; the model client default timeout is 600 seconds. Internal stage-specific small limits have been removed. generate_source() accepts source_max_tokens and source_timeout_seconds overrides for controlled experiments or callers with different limits. The output budget can include model reasoning tokens; it does not represent the length of the generated source alone.

Pod Agent

aipod pod is a resumable local state machine over governed build tools:

Observe → Policy Select → Execute → Validate → Freeze → Observe

The stage order is deterministic:

generate_models
generate_providers
generate_services
compose_pipelines
generate_interfaces
verify_application
repair_current_artifact   # only after real failure evidence

The model does not choose this order. It decides the contents of the current bounded artifact.

Modifying an existing Pod

Studio and --stage auto use one focused AI call to classify the earliest layer affected by a requested change. The local scheduler then freezes upstream and rebuilds that layer plus downstream:

aipod pod --stage auto --yes \
  "Add task priority and display it in the desktop window."

An explicit stage remains available as a manual override:

aipod pod --stage interfaces --yes \
  "Replace the CLI Adapter with a desktop and message-queue Adapter."

Progressive verification

Validation happens before freezing, not only at the end:

After all layers complete, frozen component tests and every required Interface verification command run again. Optional installation checks remain visible but do not fail runtime proof.

Component planning declares a non-empty tests array before implementation:

{"tests": [{"name": "test_viewer_cannot_create", "requirement": "An active viewer is rejected and no database rows change"}]}

AIPod generates a unittest.TestCase named ComponentTests using its testing SDK. For example, a Service that only permits administrators to create tasks should have separate successful-admin and denied-viewer cases. The denied case can look like:

import unittest
from ai_pod_cli.testing import Sandbox

class ComponentTests(unittest.TestCase):
    def test_viewer_cannot_create(self):
        with Sandbox() as s:
            s.seed("UserAccount", {"username": "reader", "role": "viewer", "is_active": True})
            before = s.snapshot()
            with self.assertRaises(PermissionError):
                s.run("BugCreateService", {"current_user_username": "reader", "title": "Example"})
            self.assertEqual(s.snapshot(), before)

Use the actual project's required fields in fixtures and calls. The SDK provides seed, model, run, call_provider, rows, count, snapshot and path. Each with Sandbox(config={}, provider_overrides={}) creates fresh temporary configuration, SQLite storage, dependency injection and resource paths. It does not infer business data, roles or input values. Only declared dependency Providers may be replaced by test doubles; the target, Models and Services run their actual code. An expected exception counts as a real component invocation. Missing/skipped cases, no target invocation and obvious constant-only assertions fail verification.

Tests are saved before implementation under tests/components/<ID>_<spec-hash>.py. Their hashes and frozen metadata are recorded in .aipod/component-tests.json, and the registry records the accepted test descriptor. Source retries and resumes reuse the exact tests. Explicit requirement revisions can create a new version; previous test files remain available for review. Automatic repair cannot rewrite tests to obtain a pass. Fixture or SDK setup errors stop implementation repair and require an explicit test-plan revision. Legacy components without tests require such a revision before they can pass component reuse verification.

The SDK is available inside the framework's component test worker. The worker runs in a disposable project, with isolated test data and a timeout. This is execution and data isolation, not an operating-system security boundary for hostile code.

Rerun a component's frozen tests with the same worker:

python -m ai_pod_cli.component_tests --project-root . --component BugCreateService

Omit --component to run all recorded component tests. The command reports JSON results and exits with a nonzero code on failure; the generated test file does not need to construct a container or connect to the application's database.

Pipeline planning declares public inputs and non-empty verification_cases, each with a unique name and explicit params. A default-starting application includes an empty parameter scenario; a route needing user input provides concrete values for that input. The Pipeline sandbox never derives sample parameters or files from downstream Service requirements. Each case uses its own disposable project, and exceptions, timeouts, unhandled Runtime Failure records, and premature process exits fail the check. Cases remain frozen while source is repaired. Reusing an existing Pipeline also reruns its entry scenarios. Public inputs are saved beside the source in .contract.json and referenced by routes.toml; inferred internal requirements do not replace this boundary.

An entry scenario is a smoke check. Every Interface must also declare a required kind: "behavior" command using the framework's unittest driver, plus named cases mapping test methods to requirements. Plans missing these checks are rejected, and old smoke-only passes are invalidated. The driver is available directly:

python -m ai_pod_cli.behavior_tests tests/test_behavior.py

Each test uses a real registered route and self.assert* assertions on observed behavior. The driver rejects empty tests, missing real route execution, ignored framework failures, and tests containing only obviously constant assertions. Bare Python assert does not count as unittest evidence. Every declared case must execute and pass; a missing or skipped case cannot be replaced by another passing test. JSON proof includes per-test execution and assertion counts. Acceptance test files are excluded from automatic repair targets.

Python Pod allows up to 10 application repair attempts, including rejected patches, and verifies again after every applied repair. Attempts persist across resumes; existing plans count their already-applied repairs toward this limit. The default scheduler budget is 40 steps so all 10 rounds and the final verification can run.

These are minimum evidence checks, not a proof of complete requirements coverage or arbitrary test correctness. The planner must declare meaningful scenarios and expected outcomes; reviewers can inspect the frozen cases. Pure exception-only Interface route tests currently need a successful route invocation in the same test to satisfy the execution evidence.

Run structure-only inspection:

aipod verify --json

A structure-only result is unverified, not passed.

Run a real command:

aipod verify --json -- python -m unittest

Verification records the command, exit code, bounded stdout/stderr, project-local traceback locations, repair candidates, and a source fingerprint. A stale pass is reset to pending when relevant project files change.

Contracts

Components publish machine-readable inputs and outputs. AIPod validates:

• required fields;
• scalar and structured types;
• shared Model paths;
• nested schemas;
• Pipeline data flow;
• runtime values at component boundaries;
• Service visibility (Service → Service is always invalid);
• deterministic branch merges and bounded repeat traces.

Type, Model, missing-field, and nested-schema conflicts are errors. Similar-but-different field names are warnings because semantic similarity is heuristic.

Python Contracts normalize equivalent structured schemas and type annotations before checking them. For example, these describe the same list of shared Models:

{"type": "array", "items": {"model": "modules.models.gameentity.GameEntity"}}
"List[modules.models.gameentity.GameEntity] — entities in the current frame"

Supported annotations include List[T], list[T], typing.List[T], Dict[str, T], Optional[T], Union[A, B], and T | None, including nested combinations. Normalization preserves Model path capitalization, item schemas, and required, default, and description metadata. Static Pipeline checks, runtime validation and Model materialization, and verification samples all use this same representation.

This is a small Contract syntax, not full Python typing or JSON Schema support. Unknown legacy type names and unsupported generic tokens such as Tuple[int, int] remain compatible as type labels; retaining a label does not validate its elements. Malformed supported generics and executable type expressions are rejected.

Native Studio

Install Studio support and open a project:

pip install "AIPodCli[studio]"
aipod studio .

Studio provides:

• project switching and initialization;
• Model, Provider, Service, Pipeline, and Interface visualization;
• AI component creation and visual Pipeline composition;
• Pod build progress, cancellation, and stage evidence;
• source inspection;
• Interface Adapter, lifecycle, permission, and verification inspection;
• program output and persisted run traces.

Project structure

project/
├── aipod_plan.json          resumable Plan and public Agent state
├── beans_config.json        Bean registry and Contracts
├── config.toml              project configuration
├── routes.toml              route-to-Pipeline registry
├── requirements.txt         project-specific dependencies
├── modules/
│   ├── models/
│   ├── providers/
│   └── services/
├── pipelines/
├── interfaces/
│   └── <interface-id>/
│       ├── interface.json
│       ├── adapter.py
│       └── additional Adapter modules and lifecycle files
├── docs/aipod/              generated human-readable plans
└── .aipod/runs/             redacted execution traces

CLI reference

aipod entry remains available for legacy standalone entry generation. New Pod projects should use Interface Adapters.

Configuration

Global model configuration is stored outside individual projects. Environment variables or a local .env override saved values:

OPENAI_API_KEY
OPENAI_BASE_URL
OPENAI_MODEL
OPENAI_TIMEOUT_SECONDS

The PyPI distribution is named AIPodCli; the Python import package is ai_pod_cli.

Python and Node.js use the same global model configuration file:

~/.aipod/config.toml

[env]
OPENAI_API_KEY = "..."
OPENAI_BASE_URL = "https://api.openai.com/v1"
OPENAI_MODEL = "your-model"
OPENAI_TIMEOUT_SECONDS = "600"

They also share project-level config.toml and dot-notation ConfigStore access. Process environment variables override project .env, which overrides global [env] values.

Security and trust boundary

AIPod provides governance, not hostile-code isolation.

• Generated code is checked structurally and executed in disposable project copies before freezing.
• Adapter code can see routes but is prohibited from importing Services or runtime internals.
• Service code can see Models and Providers but is prohibited from importing, injecting, constructing, or invoking another Service.
• Generated lifecycle files must be reviewed before changing system integration.
• The final application is ordinary Python and runs with the current user's permissions.
• Third-party packages and remote model providers remain separate trust boundaries.

Do not treat generated code as safe for production without review, platform permissions, and deployment isolation appropriate to the application.

Current boundaries

• Synchronous code cannot safely force an async Pipeline inside an already-running event loop; async callers must use PipelineRunner.run_async().
• Stream processing is in-process and bounded, but durable offsets, distributed workers, and exactly-once delivery remain responsibilities of the selected queue/provider.
• Parallel execution isolates Context data, but external side effects still require idempotency and transaction design in the application Services.
• repeat is an in-process governed loop. Distributed scheduling, durable checkpoints, and process supervision remain deployment concerns.
• Contract analysis cannot prove arbitrary Python semantics.
• Synthetic smoke cannot prove access to real external databases, queues, accounts, or operating-system permissions.
• Complex platform installation may require explicit manual steps, signing, entitlements, or user approval.
• Model providers may time out or truncate large generations; Pod state remains resumable.

Development

Node.js subproject

An initial TypeScript implementation lives in aipod-node/. It provides the governed Runtime foundation—Service isolation, Contracts, sequential and parallel Pipelines, repeat, async streams, and route dispatch—plus a resumable five-stage AI construction Agent, complete generated-project semantic type checking, a persistent HTTP Broker/Worker runtime, CLI, and local browser-based Studio.

cd aipod-node
npm install
npm test

See aipod-node/README.md for its current scope and API.

git clone https://github.com/wangzhongren/ai_pod_cli.git
cd ai_pod_cli
python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install -e ".[studio]"
python -m unittest tests.test_runtime

Build and validate the package:

python -m build
python -m twine check dist/*

License

MIT