Metadata-Version: 2.4
Name: snipara-orchestrator
Version: 1.0.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

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

# 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

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

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

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