Shared Project Intelligence for multi-agent work

Coordinate execution. Preserve understanding.

AI agents are becoming specialized. One plans. One writes code. One reviews. One tests. But every handoff can lose the decisions, active work, impact, and proof that made the previous step safe.

Hosted MCP or API gives every client the same Project Intelligence before it acts. The Project Brain belongs to the project, not to a model, chat session, or orchestration framework.

Companion is the default workflow layer for briefs, handoffs, and continuity. Sandbox and Orchestrator are optional layers for repeatable execution, proof gates, production validation, or advanced multi-agent coordination.

One execution, end to end

Orchestration is easiest to understand as a complete project loop. Snipara prepares the shared state and records what happened; your chosen agents still do the reasoning and execution.

Multi-agent coordination is not the product

Traditional multi-agent systems pass prompts between workers. Snipara keeps the durable project layer between those workers, so the handoff carries decisions and evidence rather than only text.

For normal coding work, prefer Hosted MCP plus optional snipara-companion and Snipara Sandbox first. Use orchestrator only when you need a stronger control plane around validation and release workflows.

snipara-companion may suggest this package for FULL + ORCHESTRATED work, but it does not install orchestrator or spawn workers automatically. Install it explicitly through pip install snipara-orchestrator or npx create-snipara repair --with-orchestrator.

Orchestrator also contains the Adaptive Work Routing resolver. The resolver takes a provider-neutral WorkProfile, ModelRequirements, and a sanitized runtime worker catalog, then selects a candidate or fails closed to the main agent. It does not maintain hardcoded model-name lists and does not launch hidden workers.

Philosophy

Traditional CI/CD pipelines stop at "tests pass, deploy". But production is the real test. Snipara Orchestrator implements prod-first validation:

Installation

pip install snipara-orchestrator

npx create-snipara repair --with-orchestrator

Or with all optional dependencies (Playwright for UI tests):

pip install snipara-orchestrator[all]

Quick Start

Adaptive Work Routing Resolver

Use the resolver when Snipara or companion has already produced a runtime catalog. This is the path for BYOM providers and local LLM runtimes such as Ollama, LM Studio, AnythingLLM, or OpenAI-compatible gateways reachable from the worker runtime.

snipara-orchestrator route --dry-run --catalog-file runtime-catalog.json

Routing decisions stay advisory until a worker returns an explicit receipt and Project Health has enough quality-adjusted accepted outcomes to clear the sample gate.

Host-Native Supervision

The separate host command family can explicitly supervise Codex App Server, Claude Code CLI, or a loopback OpenAI-compatible runtime such as LM Studio. The native host keeps model invocation, tools, and permissions; Snipara normalizes discovery, bounded dispatch, status, steering, cancellation, resume metadata, redacted evidence, and actual Git write-scope checks.

snipara-orchestrator host discover --adapter all

snipara-orchestrator host run \
  --adapter codex_app_server \
  --task "Update one bounded docs file" \
  --workspace . \
  --write-scope docs/README.md \
  --acceptance "docs match shipped behavior" \
  --proof "git diff --check" \
  --output-fragment "expected output line"

snipara-orchestrator host run \
  --adapter openai_compatible --base-url http://127.0.0.1:1234 --model local/coder \
  --task "Update one bounded docs file" --workspace . --write-scope docs/README.md \
  --proof "git diff --check" --strong-repair --repair-adapter codex_app_server --execute

Runs are plans until --execute is supplied, and real dispatch also requires proof. Remote or credential-bearing local-model URLs, embedding/reranker models, missing proof, and out-of-scope writes fail closed. Managed Agents discovery remains feature-flagged until its beta contract is validated end to end.

Declared proof commands run automatically after dispatch through a bounded argv-only runner. Use --no-run-proof to leave a receipt in verification_required for review, or add --require-approval and an approval receipt before unattended execution. Provider keys are declared as environment-variable names with--api-key-env; the secret value never enters the chat, worker registry, or receipt.

Comparative benchmark evidence

The live benchmark uses three bounded task families and ten repetitions per family. It compares strong-only, local-only, strong-plan plus local execution, and the same handoff with a strong repair pass only after semantic validation fails. Token and cost fields staynull when the host does not expose them; completion and latency remain directly usable. The current machine-readable run is maintained in the repository benchmark evidence and is not a Worker Trust promotion signal.

This separates model quality from orchestration quality: local-only matched Codex on these tasks and was 5.36 s faster at the median, while the planner handoff lost 6.7 points and added 3.94 s. A strong repair pass recovered both failures. Automatic local delegation remains policy-gated until this behavior is reproduced on broader real-work samples; explicit supervised runs can now request one bounded repair with the same scope and proof gates. We do not turn the existing Mac, subscription, electricity, or hardware into a fabricated per-call dollar saving; those resource costs are outside this benchmark.

Basic Usage

import asyncio

from snipara_orchestrator import Orchestrator, Task, ValidationCriteria

from snipara_orchestrator.models import LiveCheck, OrchestratorConfig

config = OrchestratorConfig(

    snipara_api_key="snp-your-key",

    snipara_project="my-project",

    prod_url="https://api.example.com",

)

orchestrator = Orchestrator(config)

await orchestrator.initialize()

task = Task(

    id="deploy-auth",

    title="Deploy Authentication",

    criteria=ValidationCriteria(

        local_tests=["pnpm test", "pnpm lint"],

        live_checks=[

            LiveCheck(url="/health", expected_status=200),

            LiveCheck(url="/auth/login", method="POST", expected_status=200),

        ],

        required_proofs=2,

    ),

)

result = await orchestrator.execute_task(task)

print(f"Status: {result.status.value}")

CLI Usage

snipara-orchestrator init

Created orchestrator.yaml with default configuration

snipara-orchestrator run "deploy-auth"

Task completed: DONE (3/3 proofs passing)

Task Lifecycle

Tasks flow through a strict state machine:

PENDING → IN_PROGRESS → LOCAL_OK → VALIDATING → PROD_OK → DONE

                                           ↘

                                        ENV_DRIFT (fail-fast)

Snipara Integration

The orchestrator uses Snipara for context-aware validation. It queries your documentation to understand expected behavior, stores decisions in memory, and coordinates htask-based multi-agent workflows when you explicitly choose that pattern.

# Query context for task understanding

context = await snipara.query("authentication API endpoints")

# Remember decisions across sessions

await snipara.remember(

    content="Used JWT for auth, not session cookies",

    type="decision",

)

# Coordinate with other agents via htasks

feature = await snipara.create_htask_feature(swarm_id, "Billing rollout", "Coordinate API, frontend, QA, and deploy validation")

Drift Detection

The orchestrator automatically detects environment drift:

Multi-Agent Workflows

Use Snipara htasks to coordinate multiple agents. The orchestrator provides wrappers for feature creation, task creation, queue recommendations, completion evidence, and tree inspection. It still leaves actual Codex, Claude, or CI worker spawning explicit.

# Coordinator creates a feature and workstreams

await snipara.create_swarm(name="deployment")

feature = await snipara.create_htask_feature(swarm_id, "Auth rollout", "Ship auth safely")

await snipara.create_htask(swarm_id, "Run unit tests", parent_id=feature["task_id"], priority=10)

# Workers request the next safe batch and report evidence

batch = await snipara.recommend_htask_batch(swarm_id, limit=3)

result = await execute_task(batch["tasks"][0])

await snipara.complete_htask(swarm_id, batch["tasks"][0]["task_id"], result=result)

snipara-orchestrator htask-next --swarm-id <swarm> --limit 3

snipara-orchestrator htask-tree --swarm-id <swarm> --task-id <feature>

API Reference

Orchestrator

Gatekeeper

Examples

Related