snipara-orchestrator 1.0.2

Production-first agentic orchestrator with Snipara integration for context-aware validation

Snipara Orchestrator

Production-first agentic orchestrator with Snipara integration for context-aware validation.

Overview

Snipara Orchestrator implements the prod-first validation pattern for AI agents. It ensures that no task is marked "done" until it passes live production checks.

Key Features:

• Production-first validation - Tasks aren't done until live_check passes
• Proof-based verification - Standard proof contract (endpoint, user, result)
• Single gatekeeper - One authority for validation decisions
• Automatic cutover checklists - Generated and executed automatically
• Fail-fast on drift - Stops if route/schema drift detected
• Snipara integration - Context-aware with memory persistence
• Explicit htask coordination - Create, inspect, recommend, and complete hosted hierarchical tasks with evidence; it does not spawn Codex or Claude workers automatically
• Policy-controlled routing - Adaptive Work Routing returns an auditable policyDecision for dry_run, approval_required, and approved auto_low_risk handoffs without broadening project policy from CLI flags
• Engineering Lead Plan input - route --dry-run --lead-plan-file can derive routing requirements from Project Health or Companion Engineering Lead Contract V1 exports while keeping execution behind explicit approval and proof receipts
• Execution receipt gates - agents coordinate --lead-plan-file and agents check-receipt dry-run lead-plan coordination and verify claim, approval, proof, outcome, and Project Brain update evidence without spawning workers

Installation

pip install snipara-orchestrator

Or with all optional dependencies:

pip install snipara-orchestrator[all]

Quick Start

CLI Usage

# Initialize configuration
snipara-orchestrator init --project my-project --prod-url https://api.example.com

# Run a validation task
snipara-orchestrator run "Deploy Auth Feature" \
  --test "pnpm test" \
  --test "pnpm lint" \
  --endpoint "https://api.example.com/health" \
  --endpoint "https://api.example.com/api/auth/session" \
  --required-proofs 3

# Check for environment drift
snipara-orchestrator check-drift --route /api/users --route /api/auth

# Validate a single endpoint
snipara-orchestrator validate https://api.example.com/health

# Recall memories from previous sessions
snipara-orchestrator recall "deployment failures" --limit 5

# Store a memory
snipara-orchestrator remember "Chose Redis for rate limiting" --type decision

# Bootstrap an autonomous htask tree for workers
snipara-orchestrator htask-bootstrap "Auth Overhaul" \
  --swarm-name default-agent-orchestrator \
  --description "Move auth to OAuth and JWT" \
  --owner coordinator \
  --workstream API \
  --workstream QA \
  --custom-workstream DEPLOY_PROD_VERIFY

# Create an htask feature and workstreams
snipara-orchestrator htask-create-feature "Auth Overhaul" \
  --swarm-id swarm_abc123 \
  --description "Move auth to OAuth and JWT" \
  --owner codex \
  --workstream API \
  --workstream QA

# Create a leaf htask under a workstream
snipara-orchestrator htask-create "Add refresh endpoint" \
  --swarm-id swarm_abc123 \
  --parent-id htask_ws_api \
  --description "Implement token rotation" \
  --owner codex \
  --evidence-required '{"type":"test","description":"targeted tests passed"}'

# Pull or claim the next ready htasks and inspect the hierarchy
snipara-orchestrator htask-next \
  --swarm-name default-agent-orchestrator \
  --claim-for-agent hermes-worker-1 \
  --limit 1
snipara-orchestrator htask-tree --swarm-id swarm_abc123 --task-id htask_feature

# Complete an N3 htask with proof
snipara-orchestrator htask-complete htask_task_001 \
  --swarm-id swarm_abc123 \
  --evidence "test:pytest packages/agentic-orchestrator" \
  --result "Implemented and verified htask wrapper"

# Dry-run Adaptive Work Routing against a runtime catalog
snipara-orchestrator route --dry-run \
  --work-profile-json '{"taskType":"documentation","risk":"low"}' \
  --requirements-json '{"workerRole":"coding","plannerRetainsReasoning":true,"preferredEndpointTypes":["local"]}' \
  --catalog-file runtime-catalog.json

# Build a local LM Studio/Qwen docs and architecture runtime catalog
snipara-orchestrator local-model-catalog \
  --base-url http://127.0.0.1:1234 \
  --model qwen/qwen3-30b-a3b-2507 \
  --worker-role documentation \
  --capability documentation \
  --capability architecture_review \
  --capability planning \
  --json > .snipara/local-qwen-docs-runtime-catalog.json

# Dry-run against that local OpenAI-compatible endpoint
snipara-orchestrator route --dry-run \
  --work-profile-json '{"taskType":"documentation","risk":"low","scope":["docs/**"],"contextBudget":"small","reasoningDepth":"low"}' \
  --requirements-json '{"workerRole":"documentation","plannerRetainsReasoning":true,"preferredEndpointTypes":["local"],"allowedEndpointTypes":["local"],"writeScope":["docs/**"],"capabilities":["documentation"]}' \
  --catalog-file .snipara/local-qwen-docs-runtime-catalog.json \
  --json

# Dry-run routing from an Engineering Lead Plan V1 export
snipara-orchestrator route --dry-run \
  --lead-plan-file project-health-lead-plan.json \
  --work-package-id wp_docs \
  --catalog-file runtime-catalog.json

# Dry-run coordination and evidence gates from an Engineering Lead receipt
snipara-orchestrator agents coordinate \
  --lead-plan-file project-health-lead-plan.json \
  --work-package-id wp_docs \
  --json

snipara-orchestrator agents check-receipt \
  --receipt-file lead-execution-receipt.json \
  --evidence "proof:pytest docs passed" \
  --claim-id claim_docs \
  --approval-receipt-id approval_docs \
  --outcome-receipt-id outcome_docs \
  --brain-update-applied \
  --json

Python API

import asyncio
from snipara_orchestrator import Orchestrator, Task, ValidationCriteria
from snipara_orchestrator.models import LiveCheck, OrchestratorConfig

async def main():
    # Configuration
    config = OrchestratorConfig(
        snipara_api_key="snp-your-api-key",
        snipara_project="my-project",
        prod_url="https://api.example.com",
        repo_path="/path/to/repo",
        test_user="[e-mail]",
    )

    # Create orchestrator
    orchestrator = Orchestrator(config)
    await orchestrator.initialize()

    # Define task with validation criteria
    task = Task(
        id="deploy-auth",
        title="Deploy Authentication Feature",
        description="Deploy OAuth2 authentication to production",
        criteria=ValidationCriteria(
            local_tests=["pnpm test", "pnpm lint"],
            live_checks=[
                LiveCheck(url="https://api.example.com/health"),
                LiveCheck(
                    url="https://api.example.com/api/auth/login",
                    method="POST",
                    expected_status=200,
                    body={"email": "[e-mail]", "password": "test"},
                ),
            ],
            required_proofs=3,
        ),
    )

    # Execute the task
    result = await orchestrator.execute_task(task)

    print(f"Status: {result.status.value}")
    print(f"Proofs: {len(result.passing_proofs())}/{len(result.proofs)}")

asyncio.run(main())

Task Lifecycle

PENDING → IN_PROGRESS → LOCAL_OK → VALIDATING → PROD_OK → DONE
                             ↓           ↓
                         LOCAL_FAIL   PROD_FAIL → ENV_DRIFT?

Status	Description
PENDING	Task created, not started
IN_PROGRESS	Executing local tests
LOCAL_OK	Local tests passed
LOCAL_FAIL	Local tests failed
VALIDATING	Running production checks
PROD_OK	All production checks passed
PROD_FAIL	Production checks failed
ENV_DRIFT	Environment drift detected
DONE	Task completed successfully

Proof Contract

Every validation produces a proof with three required fields:

@dataclass
class Proof:
    endpoint: str      # URL or test identifier
    user_tested: str   # Test user email
    result: str        # "pass" or "fail"

    # Optional
    response_code: int
    response_body: dict
    error_message: str

Tasks require a minimum number of passing proofs (default: 3) to reach PROD_OK.

Adaptive Work Routing

snipara-orchestrator route --dry-run resolves provider-neutral work requirements against a runtime worker catalog. It returns a fail-closed policyDecision with one of three modes: dry_run, approval_required, or auto_low_risk when explicit policy, risk, and candidate gates allow execution. It does not spawn Codex, Claude, CI workers, local LLMs, or any other execution worker.

The resolver deliberately avoids hardcoded model-name lists. Stable inputs are:

• WorkProfile: task type, risk, scope, and context budget
• ModelRequirements: worker role, reasoning, cost, speed, endpoint type, capabilities, write scope, and fallback
• runtime candidates returned by Snipara's BYOM gateway or another trusted runtime catalog

Use this pattern when a strong planner retains deep reasoning but a scoped worker can perform the edit, test, or documentation task. Local endpoints such as Ollama, LM Studio, AnythingLLM, or other OpenAI-compatible servers must be reachable from the worker runtime. If no candidate satisfies the requirements, the resolver fails closed to main_agent.

In the Codex workflow, Codex remains the chief architect, lead orchestrator, and quality verifier. Local LM Studio models are bounded worker candidates: Codex plans the work, issues the handoff, reviews the output, and owns the final quality gate.

For LM Studio or another OpenAI-compatible local server, generate a runtime catalog first. If multiple coding models are loaded, prefer the one you want by matching the model id returned by /v1/models:

Local model	Intended use	Catalog hint
Qwen3-30B-A3B-2507	Reflection, architecture, documentation	--model qwen/qwen3-30b-a3b-2507
Devstral Small 2 24B Q8_0	Development and refactoring	--prefer-model devstral

snipara-orchestrator local-model-catalog \
  --base-url http://127.0.0.1:1234 \
  --prefer-model devstral \
  --worker-role coding \
  --capability code_edit \
  --capability refactor \
  --json > .snipara/local-devstral-runtime-catalog.json

Use an exact model id when you want to pin Qwen for documentation, architecture, or planning work:

snipara-orchestrator local-model-catalog \
  --base-url http://127.0.0.1:1234 \
  --model qwen/qwen3-30b-a3b-2507 \
  --worker-role documentation \
  --capability documentation \
  --capability architecture_review \
  --capability planning \
  --json > .snipara/local-qwen-docs-runtime-catalog.json

The catalog records the base URL, model id, and standard OpenAI-compatible routes: /v1/models, /v1/responses, /v1/chat/completions, /v1/completions, and /v1/embeddings. For one-off routing experiments, append the local endpoint directly:

snipara-orchestrator route --dry-run \
  --openai-compatible-local-url http://127.0.0.1:1234 \
  --openai-compatible-model qwen/qwen3-30b-a3b-2507 \
  --work-profile-json '{"taskType":"documentation","risk":"low","scope":["docs/**"]}' \
  --requirements-json '{"workerRole":"documentation","plannerRetainsReasoning":true,"preferredEndpointTypes":["local"],"allowedEndpointTypes":["local"],"writeScope":["docs/**"],"capabilities":[]}' \
  --json

This still resolves and proves the handoff contract only. A worker launcher must consume the selected candidate and call the local model explicitly after the required approval receipt exists.

Policy decisions keep approval receipts explicit. approval_required requires a human or project-owned approval receipt before a worker launcher may continue; auto_low_risk can only be emitted for low-risk work when approval is not required and a measured accepted candidate satisfies the budget and capability constraints.

Configuring MCP Tool Surfaces

The MCP server advertises different tool surfaces depending on the SNIPARA_EXPOSED_SURFACES environment variable. Hosted MCP defaults to inline tools plus a small companion maintenance set for index health, reindexing, and read-only memory hygiene. To expose orchestration tools directly in the advertised manifest, set SNIPARA_EXPOSED_SURFACES=inline,orchestrator on the MCP server. This is optional: orchestration tools remain discoverable via snipara_help and can be executed by direct JSON-RPC or clients/server configurations that expose those surfaces. Standard MCP agents only receive schemas for tools returned by tools/list.

Configuration

Create .snipara-orchestrator.json in your project root:

{
  "snipara_api_key": "snp-your-api-key",
  "snipara_project": "my-project",
  "prod_url": "https://api.example.com",
  "repo_path": "/path/to/repo",
  "test_user": "[e-mail]",
  "fail_fast_on_drift": true,
  "auto_remember": true,
  "verbose": true
}

Or use environment variables:

export SNIPARA_API_KEY=snp-your-api-key
export SNIPARA_PROJECT=my-project
export PROD_URL=https://api.example.com
export REPO_PATH=/path/to/repo

Snipara Integration

The orchestrator uses Snipara for:

Context Retrieval

# Get relevant documentation
context = await orchestrator.snipara.query_context(
    query="deployment checklist production",
    max_tokens=6000,
)

Memory Persistence

# Remember decisions
await orchestrator.snipara.remember(
    content="Chose Redis for rate limiting due to distributed nature",
    type="decision",
    category="architecture",
    ttl_days=30,
)

# Recall previous context
memories = await orchestrator.snipara.recall(
    query="deployment failures",
    limit=5,
)

Multi-Agent Coordination

# Create a swarm
swarm = await orchestrator.snipara.create_swarm(
    name="deployment-coordination",
    description="Multi-agent deployment workflow",
)

# Store shared state
await orchestrator.snipara.set_state(
    swarm_id=swarm["swarm_id"],
    agent_id="coordinator",
    key="deployment_status",
    value={"phase": "validating", "progress": 75},
)

Hierarchical Tasks

These methods wrap the hosted snipara_htask_* tools. They coordinate work and proofs; worker execution is still manual and explicit.

feature = await orchestrator.snipara.create_htask_feature(
    title="Auth Overhaul",
    description="Move auth to OAuth and JWT",
    owner="codex",
    auto_create_swarm=True,
    swarm_name="default-agent-orchestrator",
    create_initiative=True,
    workstreams=["API", "QA"],
    custom_workstreams=["DEPLOY_PROD_VERIFY"],
    create_actionable_tasks=True,
)

task = await orchestrator.snipara.create_htask(
    swarm_id=swarm["swarm_id"],
    parent_id=feature["workstream_ids"]["API"],
    title="Add refresh endpoint",
    description="Implement token rotation",
    owner="codex",
    evidence_required=[
        {"type": "test", "description": "targeted tests passed"},
    ],
)

ready = await orchestrator.snipara.recommend_htask_batch(
    swarm_id=swarm["swarm_id"],
    limit=5,
    owner="codex",
    claim_for_agent="hermes-worker-1",
)

tree = await orchestrator.snipara.get_htask_tree(
    swarm_id=swarm["swarm_id"],
    task_id=feature["feature_id"],
)

await orchestrator.snipara.complete_htask(
    swarm_id=swarm["swarm_id"],
    task_id=task["task_id"],
    evidence=[
        {"type": "test", "description": "pytest package suite passed"},
    ],
    result={"files_modified": ["src/auth.py"]},
)

Components

Gatekeeper

Single authority for validation decisions:

from snipara_orchestrator.gates import Gatekeeper

gatekeeper = Gatekeeper(
    snipara=snipara_client,
    validator=validator,
    drift_detector=drift_detector,
    fail_fast_on_drift=True,
)

# Gate 1: LOCAL_OK → VALIDATING
result = await gatekeeper.gate_local_to_validation(task)

# Gate 2: VALIDATING → PROD_OK
result = await gatekeeper.gate_validation_to_prod(task)

Drift Detector

Detect environment drift:

from snipara_orchestrator.drift_detector import DriftDetector

detector = DriftDetector(
    prod_url="https://api.example.com",
    repo_path="/path/to/repo",
)

report = await detector.full_drift_check(
    routes=["/api/users", "/api/auth"],
    check_schema=True,
    check_health=True,
)

if report.has_drift:
    print(f"Issues: {report.issues}")
    print(f"Recommendations: {report.recommendations}")

Validator

Collect proofs from live checks:

from snipara_orchestrator.validator import Validator
from snipara_orchestrator.models import LiveCheck

validator = Validator(test_user="[e-mail]")

proof = await validator.run_live_check(
    LiveCheck(
        url="https://api.example.com/health",
        expected_status=200,
    )
)

print(f"Result: {proof.result}")
print(f"Status: {proof.response_code}")

Executor

Run commands and tests:

from snipara_orchestrator.executor import Executor

executor = Executor(working_dir="/path/to/repo")

result = await executor.run_command("pnpm test")
print(f"Success: {result.success}")
print(f"Output: {result.stdout}")

Architecture

┌─────────────────────────────────────────────────────────────────┐
│                        ORCHESTRATOR                             │
│                                                                 │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │  SNIPARA (Context Layer)                                 │  │
│  │  • snipara_context_query → Documentation context             │  │
│  │  • snipara_remember/recall → Memory persistence              │  │
│  │  • snipara_swarm_* → Multi-agent coordination                │  │
│  └──────────────────────────────────────────────────────────┘  │
│                              │                                  │
│                              ▼                                  │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │  GATEKEEPER (Single Authority)                           │  │
│  │  • gate_local_to_validation() → LOCAL_OK → VALIDATING    │  │
│  │  • gate_validation_to_prod() → VALIDATING → PROD_OK      │  │
│  │  • generate_cutover_checklist() → Auto checklist         │  │
│  └──────────────────────────────────────────────────────────┘  │
│                              │                                  │
│        ┌─────────────────────┼─────────────────────┐           │
│        ▼                     ▼                     ▼           │
│  ┌───────────┐        ┌───────────┐        ┌───────────┐       │
│  │ EXECUTOR  │        │ VALIDATOR │        │ DRIFT     │       │
│  │ • Tests   │        │ • Live    │        │ DETECTOR  │       │
│  │ • Deploy  │        │ • UI      │        │ • Routes  │       │
│  │ • Bash    │        │ • Proofs  │        │ • Schema  │       │
│  └───────────┘        └───────────┘        └───────────┘       │
└─────────────────────────────────────────────────────────────────┘

Snipara Ecosystem

Package	Install	Purpose
snipara-mcp	pip install snipara-mcp	MCP client for context optimization
snipara-orchestrator	pip install snipara-orchestrator	Production validation orchestrator
Snipara Sandbox	pip install snipara-sandbox	Safe code execution runtime

Best Practices

1. Always define validation criteria - Don't rely on defaults
2. Use meaningful test users - Helps with debugging
3. Set appropriate proof requirements - 3 is a good minimum
4. Enable fail-fast on drift - Catch issues early
5. Use auto-remember - Preserve learnings across sessions
6. Check drift before deployment - Run snipara-orchestrator check-drift

Documentation

• Full Documentation
• API Reference
• Examples

License

MIT

---

Built with ❤️ by Snipara