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

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

### 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="rlm_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": "rlm_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=rlm_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},
)
```

## 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)                                 │  │
│  │  • rlm_context_query → Documentation context             │  │
│  │  • rlm_remember/recall → Memory persistence              │  │
│  │  • rlm_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 |
| [rlm-runtime](https://pypi.org/project/rlm-runtime/) | `pip install rlm-runtime` | 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)
