snipara-orchestrator 1.0.15

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
• Gated local worker execution - agents execute-gated --plan turns an approved low-risk local handoff into an explicit dry-run or --execute local runtime call, then returns review-pending receipts without applying changes automatically
• Execution receipt ledger - agents execute-gated writes a stable gated execution receipt under .snipara/orchestrator/executions/ by default. V1 attributes the receipt to a worker/category, routing card, and stable workflow fingerprint, and fails closed when required receipt families are missing. agents review-gated persists its supervisor review under .snipara/orchestrator/reviews/ by default, and agents review-gated fails closed until receipts, scope, and source-backed proof are acceptable
• Source-backed proof review - agents verify-proof produces a separate proof review artifact from local source files, bounded scope, declared commands, and receipt fields, so agents review-gated no longer accepts self-attested proof
• Append-only execution ledger - agents ledger-append and agents ledger-verify keep a local chained JSONL ledger for receipt/proof review pairs and fail closed on tampering
• Brain-update review packs - agents brain-update-pack extracts proposed Brain-update candidates from gated execution receipts as review-pending packs with memoryWriteAllowed=false, so workers never write durable memory automatically
• Explicit Brain-update approval plans - agents brain-update-approve and agents brain-update-apply convert source-backed candidates into dry-run apply plans without writing durable memory automatically
• Companion-aware status - status --json reads local .snipara/companion/config.json project bindings and reports configuration state without exposing API keys
• Companion Session Snapshot V0 - companion-session --json reads .snipara/activity/session.json, and Team Sync gates include that snapshot while keeping hard_routing_allowed=false unless explicit policy and receipts permit routing
• Advisory Intent Detection V0 - companion-session displays the Companion snapshot intent, confidence, and suggested workflow mode as context only; it never turns intent into worker execution or hard routing by itself

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

# Show secret-safe status, including Companion workspace config when present
snipara-orchestrator status --json

# Read the local Companion Activity Timeline / Session Snapshot dogfood surface
snipara-companion workflow session --json
snipara-orchestrator companion-session --json

# 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

# Inspect a gated local worker launch contract without contacting the runtime
snipara-orchestrator agents execute-gated \
  --plan .snipara/orchestrator/handoff.json \
  --json

# Explicitly contact the selected local OpenAI-compatible runtime.
# Output is review_pending; Snipara does not auto-apply worker changes.
snipara-orchestrator agents execute-gated \
  --plan .snipara/orchestrator/handoff.json \
  --execute \
  --json

# Verify claimed proof against local source artifacts before review-gated
snipara-orchestrator agents verify-proof \
  --receipt-file .snipara/orchestrator/executions/gated-exec-abc123.json \
  --repo-root . \
  --output .snipara/orchestrator/proof/proof-review-abc123.json \
  --json

# Append and verify the receipt/proof pair in the local execution ledger
snipara-orchestrator agents ledger-append \
  --receipt-file .snipara/orchestrator/executions/gated-exec-abc123.json \
  --proof-review-file .snipara/orchestrator/proof/proof-review-abc123.json \
  --ledger-file .snipara/orchestrator/ledger/execution-ledger.jsonl \
  --json
snipara-orchestrator agents ledger-verify \
  --ledger-file .snipara/orchestrator/ledger/execution-ledger.jsonl \
  --json

# Review the persisted gated execution receipt before any follow-on action
snipara-orchestrator agents review-gated \
  --receipt-file .snipara/orchestrator/executions/gated-exec-abc123.json \
  --proof-review-file .snipara/orchestrator/proof/proof-review-abc123.json \
  --ledger-file .snipara/orchestrator/ledger/execution-ledger.jsonl \
  --json

# The accepted or blocked supervisor review is persisted under
# .snipara/orchestrator/reviews/ unless --no-review is passed.

# Extract proposed Brain updates for supervising-agent review.
# This does not write durable memory.
snipara-orchestrator agents brain-update-pack \
  --receipt-file .snipara/orchestrator/executions/gated-exec-abc123.json \
  --proof-review-file .snipara/orchestrator/proof/proof-review-abc123.json \
  --output .snipara/orchestrator/brain-updates/brain-pack-abc123.json \
  --json

# Approve source-backed candidates and emit a dry-run apply plan.
# This still does not write durable memory.
snipara-orchestrator agents brain-update-approve \
  --pack-file .snipara/orchestrator/brain-updates/brain-pack-abc123.json \
  --candidate-id brain-candidate-abc123 \
  --reviewer lead-agent \
  --output .snipara/orchestrator/brain-updates/approved-pack-abc123.json \
  --json
snipara-orchestrator agents brain-update-apply \
  --pack-file .snipara/orchestrator/brain-updates/approved-pack-abc123.json \
  --output .snipara/orchestrator/brain-updates/apply-plan-abc123.json \
  --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 the lean agent contract and routes specialist work behind simple Snipara verbs. 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, marked as opt-in specialist surfaces, 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