Metadata-Version: 2.4
Name: snipara-orchestrator
Version: 0.1.3
Summary: Production-first agentic orchestrator with Snipara integration for context-aware validation
Project-URL: Homepage, https://github.com/Snipara/snipara-orchestrator
Project-URL: Documentation, https://snipara.com/docs/orchestrator
Project-URL: Repository, https://github.com/Snipara/snipara-orchestrator
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

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

# 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 the next ready htasks and inspect the hierarchy
snipara-orchestrator htask-next --swarm-id swarm_abc123 --limit 5
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"
```

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

## 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(
    swarm_id=swarm["swarm_id"],
    title="Auth Overhaul",
    description="Move auth to OAuth and JWT",
    owner="codex",
    workstreams=["API", "QA"],
)

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",
)

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://snipara.com/docs/orchestrator)
- [API Reference](https://snipara.com/docs/orchestrator/api)
- [Examples](https://github.com/Snipara/snipara-orchestrator/tree/main/examples)

## License

MIT

---

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