Metadata-Version: 2.4
Name: snipara-orchestrator
Version: 1.2.1
Summary: Production-first agentic orchestrator with Snipara integration for context-aware validation
Project-URL: Homepage, https://www.snipara.com/docs/integration/orchestrator
Project-URL: Documentation, https://www.snipara.com/docs/integration/orchestrator
Project-URL: Packages, https://www.snipara.com/docs/integration/packages
Author-email: Snipara <hello@snipara.com>
License: MIT
Keywords: agents,ai,llm,mcp,orchestration,snipara,validation
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Requires-Dist: anyio>=4.0.0
Requires-Dist: httpx>=0.25.0
Requires-Dist: pydantic>=2.0.0
Requires-Dist: rich>=13.0.0
Requires-Dist: typer>=0.9.0
Provides-Extra: all
Requires-Dist: mypy>=1.0.0; extra == 'all'
Requires-Dist: playwright>=1.40.0; extra == 'all'
Requires-Dist: pytest-asyncio>=0.21.0; extra == 'all'
Requires-Dist: pytest-cov>=4.0.0; extra == 'all'
Requires-Dist: pytest>=7.0.0; extra == 'all'
Requires-Dist: ruff>=0.1.0; extra == 'all'
Provides-Extra: dev
Requires-Dist: mypy>=1.0.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.21.0; extra == 'dev'
Requires-Dist: pytest-cov>=4.0.0; extra == 'dev'
Requires-Dist: pytest>=7.0.0; extra == 'dev'
Requires-Dist: ruff>=0.1.0; extra == 'dev'
Provides-Extra: playwright
Requires-Dist: playwright>=1.40.0; extra == 'playwright'
Description-Content-Type: text/markdown

# Snipara Orchestrator

[![PyPI version](https://badge.fury.io/py/snipara-orchestrator.svg)](https://badge.fury.io/py/snipara-orchestrator)
[![Python 3.10+](https://img.shields.io/badge/python-3.10+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

Production-first agentic orchestrator with [Snipara](https://snipara.com) 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
- **Host-native orchestration** - `host discover/run/status/steer/cancel`
  supervises Codex App Server, Claude Code CLI, and loopback
  OpenAI-compatible inference through their native protocols, with explicit
  execution, bounded scope, redacted receipts, resume identifiers, and
  fail-closed cancellation

## Installation

```bash
pip install snipara-orchestrator
```

Or with all optional dependencies:

```bash
pip install snipara-orchestrator[all]
```

## Quick Start

### CLI Usage

```bash
# 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

# Discover native hosts without executing work
snipara-orchestrator host discover --adapter all

# Plan a bounded Codex App Server run; add --execute only after review
snipara-orchestrator host run \
  --adapter codex_app_server \
  --task "Update one bounded documentation file" \
  --workspace . \
  --write-scope docs/README.md \
  --acceptance "documentation matches shipped behavior" \
  --proof "git diff --check"

# 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

```python
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="test@example.com",
    )

    # 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": "test@test.com", "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:

```python
@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.
Routing itself does not spawn Codex, Claude, CI workers, local LLMs, or any
other execution worker. The separate `host` command family can explicitly
dispatch a reviewed work package through a native adapter.

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`         |

```bash
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:

```bash
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:

```bash
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. Use `host run` to
dispatch explicitly after the required review boundary exists.

## Host-Native Orchestration

The host runtime normalizes lifecycle state without replacing the underlying
agent host:

- Codex uses the stable App Server JSON-RPC protocol over local stdio, including
  thread start/resume, turn start, in-flight steer when supported, and native
  interrupt.
- Claude Code uses structured print mode and its native session resume id.
  In-flight steering remains unavailable on this CLI path; use the Agent SDK
  when a long-lived embedded Claude lifecycle is required.
- LM Studio and other OpenAI-compatible runtimes are inference-only workers.
  Loopback URLs are the default; credential-bearing URLs, embedding/reranker
  models, and remote endpoints without an explicit opt-in are rejected.
- Managed Agents discovery is feature-flagged and reports unavailable instead
  of pretending to execute an unvalidated beta contract.

Every real run requires `--execute` and at least one `--proof`, stores a private
receipt under `.snipara/orchestrator/host-runs/`, captures provider usage when
the host exposes it, and compares actual Git changes with the declared write
scope. Scope violations fail the run even when the host reports success. Use
repeatable `--output-fragment` values when a response must retain exact lines
(for example both sides of a unified diff); missing fragments fail the run and
are persisted in `evidence.outputValidation`.

Declared proof commands run automatically after dispatch with `shell=False`, an
executable allowlist, bounded output, and a per-command timeout. Use
`--no-run-proof` only to hand the receipt to a reviewer; that run is recorded as
`verification_required`, not completed. For unattended execution, add
`--require-approval --approval-receipt-id <id> --approval-receipt-file <json>`.
The local supervisor requires the full `adaptive-routing-approval-v1` JSON and
rechecks its endpoint, write scope, decision, and expiry before contacting the
host; an id by itself is never treated as approval.

```bash
snipara-orchestrator host discover --adapter all

snipara-orchestrator host run \
  --adapter openai_compatible \
  --base-url http://127.0.0.1:1234 \
  --model qwen/qwen3-coder-next \
  --task "Return a bounded patch for parser.py" \
  --workspace . \
  --write-scope parser.py \
  --acceptance "targeted parser test passes" \
  --proof "pytest tests/test_parser.py" \
  --output-fragment "return old_value" \
  --output-fragment "return new_value" \
  --execute
```

Provider credentials are declared by environment-variable name, never by value:

```bash
snipara-orchestrator host run \
  --adapter openai_compatible \
  --base-url https://provider.example.com \
  --allow-remote \
  --api-key-env PROVIDER_API_KEY \
  --api-key-header authorization \
  --model provider/coder \
  --task "Run one bounded task" --workspace . \
  --write-scope docs/** --proof "git diff --check" --execute
```

For the Companion worker registry, use
`snipara-companion workers local add --base-url <url> --api-key-env PROVIDER_API_KEY`
(`--api-key-header x-api-key` is also supported). Native Codex and Claude
profiles use their host-managed authentication; a generic CLI declaration is
rejected unless its command maps to the supported Codex or Claude adapter.

The 2026-07-17 live smoke did not justify unreviewed automatic local delegation:
strong-only verified 3/3 tasks, while Codex planning plus LM Studio verified
2/3 and was slower at the median. Host-native infrastructure is available for
explicit supervised runs; automatic routing remains opt-in while broader paired
samples validate the routing-quality gate.

The comparative ten-repetition study (30 samples per condition) separates local
model quality from orchestration handoff quality. LM Studio local-only matched
Codex strong-only at 30/30 semantic completion and was 5,362.5 ms faster at the
median (2,418 ms vs 7,780.5 ms). The raw Codex-plan-plus-LM-Studio path passed
28/30 and added 3,935 ms of median latency; a strong-host repair pass recovered
both failures (30/30, with 2/30 supervisor recoveries). The planner stage alone
had a 9,542 ms median, so the observed loss is a handoff problem rather than
evidence that the local model is intrinsically incapable. Semantic validators
and exact output-fragment contracts are reported separately, scope violations
were zero, and these samples remain ineligible for Worker Trust promotion with
frontier cost data still unavailable. In this setup, LM Studio runs on an
existing Mac and the strong cloud host is subscription-covered, so neither path
adds an incremental per-call charge; hardware, electricity, subscription
allocation, and quota are outside the benchmark rather than counted as zero.

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:

```json
{
  "snipara_api_key": "snp-your-api-key",
  "snipara_project": "my-project",
  "prod_url": "https://api.example.com",
  "repo_path": "/path/to/repo",
  "test_user": "test@example.com",
  "fail_fast_on_drift": true,
  "auto_remember": true,
  "verbose": true
}
```

Or use environment variables:

```bash
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](https://snipara.com) for:

### Context Retrieval

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

### Memory Persistence

```python
# 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

```python
# 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.

```python
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:

```python
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:

```python
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:

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

validator = Validator(test_user="test@example.com")

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:

```python
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](https://pypi.org/project/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](https://www.snipara.com/docs/integration/orchestrator)
- [API Reference](https://www.snipara.com/docs/integration/orchestrator#api-reference)
- [Examples](https://www.snipara.com/docs/integration/orchestrator#examples)

## License

MIT

---

Built with ❤️ by [Snipara](https://snipara.com)
