Snipara
LoginStart

Snipara Orchestrator

Production-first agentic orchestrator that uses Snipara for context-aware validation. Tasks aren't done until they work in production.

PyPI: snipara-orchestratorGitHub

Philosophy

Traditional CI/CD pipelines stop at "tests pass, deploy". But production is the real test. Snipara Orchestrator implements prod-first validation:

Proof-Based Verification

Every check produces a proof with endpoint, user_tested, and result. No guessing.

Single Gatekeeper

One authority decides pass/fail. No conflicting validation layers.

Live Production Checks

HTTP health checks, UI tests with Playwright, schema validation against live endpoints.

Fail-Fast on Drift

Detect route changes, schema drift, or unhealthy services before wasting time.

Installation

pip install snipara-orchestrator

Or with all optional dependencies (Playwright for UI tests):

pip install snipara-orchestrator[all]

Quick Start

Basic Usage

import asyncio
from snipara_orchestrator import Orchestrator, Task, ValidationCriteria
from snipara_orchestrator.models import LiveCheck, OrchestratorConfig
config = OrchestratorConfig(
    snipara_api_key="rlm_your_key",
    snipara_project="my-project",
    prod_url="https://api.example.com",
)
orchestrator = Orchestrator(config)
await orchestrator.initialize()
task = Task(
    id="deploy-auth",
    title="Deploy Authentication",
    criteria=ValidationCriteria(
        local_tests=["pnpm test", "pnpm lint"],
        live_checks=[
            LiveCheck(url="/health", expected_status=200),
            LiveCheck(url="/auth/login", method="POST", expected_status=200),
        ],
        required_proofs=2,
    ),
)
result = await orchestrator.execute_task(task)
print(f"Status: {result.status.value}")

CLI Usage

snipara-orchestrator init
Created orchestrator.yaml with default configuration
snipara-orchestrator run --task deploy-auth
Task completed: DONE (3/3 proofs passing)

Task Lifecycle

Tasks flow through a strict state machine:

PENDING → IN_PROGRESS → LOCAL_OK → VALIDATING → PROD_OK → DONE
                                           ↘
                                        ENV_DRIFT (fail-fast)
StateDescription
PENDINGTask created, waiting to start
IN_PROGRESSLocal tests running
LOCAL_OKLocal tests passed, ready for validation
VALIDATINGRunning live checks against production
PROD_OKAll live checks passed
DONETask completed successfully
ENV_DRIFTEnvironment changed unexpectedly (routes, schema, health)
FAILEDTask failed validation

Snipara Integration

The orchestrator uses Snipara for context-aware validation. It queries your documentation to understand expected behavior, stores decisions in memory, and coordinates multi-agent workflows.

# Query context for task understanding
context = await snipara.query("authentication API endpoints")
# Remember decisions across sessions
await snipara.remember(
    content="Used JWT for auth, not session cookies",
    type="decision",
)
# Coordinate with other agents via swarms
await snipara.join_swarm(swarm_id, agent_id="validator")

Drift Detection

The orchestrator automatically detects environment drift:

Route Drift

Expected endpoints missing or returning different status codes.

Schema Drift

OpenAPI/JSON Schema mismatches between expected and actual responses.

Health Drift

Services unhealthy or returning degraded status.

Multi-Agent Workflows

Use Snipara swarms to coordinate multiple agents:

# Coordinator creates tasks
await snipara.create_swarm(name="deployment")
await snipara.create_task(swarm_id, "Run unit tests", priority=10)
await snipara.create_task(swarm_id, "Check coverage", priority=5)
# Workers claim and execute
task = await snipara.claim_task(swarm_id, agent_id="worker-1")
result = await execute_task(task)
await snipara.complete_task(swarm_id, task.id, result)

API Reference

Orchestrator

MethodDescription
initialize()Initialize orchestrator and connect to Snipara
execute_task(task)Run full task lifecycle with validation
check_drift()Check for environment drift
get_status(task_id)Get current task status

Gatekeeper

MethodDescription
gate_local_to_validation(task)Decide if local tests passed
gate_validation_to_prod(task)Decide if live checks passed
generate_cutover_checklist(task)Generate deployment checklist

Examples

Basic Usage

Simple task with local tests and live checks.

Multi-Agent Workflow

Coordinator and workers using Snipara swarms.

Related

RLM Runtime

Safe code execution for agents.

OpenClaw

Multi-agent swarm framework.

Agent Features

Memory, swarms, coordination.