Planet J Digital • Platform API Reference & Specifications

Deterministic SRE & Autonomous Agent Primitives

Complete developer reference for the 6 core Planet J Digital engines: VEB, AWOS, Aegis, Optima, Sentinel, and Flux. Built for high-scale agent swarms, zero session crashes, and strict enterprise Zero Data Retention (ZDR).

The SRE for AI Manifesto

As AI development matures from simple single-turn chatbots into multi-agent autonomous swarms, the failure modes have fundamentally shifted. Traditional software crashes deterministically when a syntax error or null pointer is hit. Autonomous agents fail stochastically: they reason in circles, subtly modify prompt arguments to avoid basic step limits, and trigger cascading microservice deadlocks.

1. NEVER HARD CRASH

Throwing an unhandled 500 error when an agent hits a spend cap is unacceptable in production. SRE dictates graceful degradation.

2. DETECT SEMANTIC DRIFT

Integer counters are commoditized. Real protection requires tracking semantic vector oscillations across reasoning chains.

3. CROSS-SERVICE SRE

Multi-agent systems require distributed cycle detection. Protect the swarm, not just the isolated function.

1. Installation & Python Quickstart

Install the open-core package via pip and wrap any function executing autonomous tool chains or agent handoffs with graceful auto-recovery.

bash

pip install aegis-guard

python

from aegis import guard, RecoveryMode

# Wrap any agent invocation with deterministic limits & graceful state recovery
@guard(
    max_budget_usd=2.50,                          # Hard limit on model token spend per execution
    detect_semantic_oscillations=True,           # Embeddings cosine check catches rephrased dead-ends
    oscillation_threshold=0.88,                  # Sensitivity for semantic drift
    on_trip=RecoveryMode.DOWNSCALE_AND_SUMMARIZE # Auto-summarize & downscale; never crash user session
)
async def run_autonomous_pipeline(prompt_input: str):
    # Compatible with Anthropic Claude, Google Gemini, OpenAI & DeepSeek
    return await my_agent_swarm.run(prompt_input)

# Execute safely: even if an infinite loop begins, the user receives a clean synthesized result
result = await run_autonomous_pipeline("Analyze market competitor pricing vectors")
print("Safe Agent Result:", result.output)
print("Telemetry Status:", result.sre_telemetry["circuit_status"])

Aegis: Semantic Loop Detection & State Recovery

FLAGSHIP SRE

Aegis is the deterministic runtime shield for autonomous agents. It performs vector cosine similarity on tool arguments and reasoning traces to detect infinite loops before step limits trip. If recursion is caught, it triggers graceful downscaling and returns a coherent response rather than an unhandled 500 error.

python

from aegis import Guard, RecoveryPolicy

guard = Guard(
    semantic_threshold=0.92,
    max_loops=4,
    recovery=RecoveryPolicy.SUMMARIZE_AND_DOWNGRADE
)

@guard.protect
def run_autonomous_research(prompt: str):
    return agent.kickoff(prompt)

VEB: Virtual Executive Board (Adversarial Consensus)

STRATEGY ENGINE

Connects Google Gemini, Anthropic Claude, OpenAI, and DeepSeek into a multi-model cross-examination panel. Synthesizes conflicting model viewpoints into deterministic, structured work orders with formal dissenting opinions.

python

from planetj.veb import BoardCouncil

council = BoardCouncil(
    members=["gemini-2.5-pro", "claude-3-7-sonnet"],
    adversarial_rounds=2
)
decision = council.debate(
    topic="Cloud Run cluster autoscaling policy vs GKE",
    constraints={"p95_latency_ms": 250, "budget_cap_usd": 5000}
)
print("Consensus Score (1-10):", decision.consensus_score)

AWOS: Agentic Workforce Orchestration

CLOUD RUN RUNTIME

Event-driven autonomous task runners deployed on serverless Google Cloud Run. Workers cold-start in sub-second latency, execute isolated tasks from VEB work orders, checkpoint state in real-time, and spin down to zero.

typescript

import { AWOSWorkerSwarm } from '@planetj/awos';

const swarm = new AWOSWorkerSwarm({
  project: 'agentorbit',
  region: 'us-central1',
  concurrency: 16
});

await swarm.dispatch({
  taskId: 'TASK-9041-BENCHMARK',
  containerImage: 'gcr.io/agentorbit/worker-sre:latest'
});

Optima: Dynamic LLM Routing & Cost Optimizer

ROUTING GATEWAY

Dynamically evaluates prompt semantic complexity at ingress. Routes simple categorization tasks to high-speed, cost-efficient edge models (Gemini 2.5 Flash, local Ollama) and complex synthesis tasks to reasoning engines (Gemini 2.5 Pro, Claude Sonnet), slashing token expenditure up to 84%.

python

from planetj.optima import SmartRouter

router = SmartRouter(strategy="latency_and_cost")
response = router.complete(
    messages=[{"role": "user", "content": "Extract SKU numbers from shipping label"}],
    primary="google/gemini-2.5-flash",
    fallback="anthropic/claude-3-7-sonnet"
)

Sentinel: Real-Time Webhook & State Ingestion

STATE PIPELINE

High-throughput reactive state engine with cryptographic signature verification and sub-150ms ingestion latency. Guarantees deterministic state mutation and prevents race conditions across multi-tenant agent triggers.

bash

curl -X POST https://api.planetjdigital.com/v1/webhook \
  -H "Authorization: Bearer aegis_live_..." \
  -H "X-PlanetJ-Signature: sha256=8f4a..." \
  -H "Content-Type: application/json" \
  -d '{"event": "agent.state_mutated", "session_id": "sess_994"}'

Flux: Spend Governance & Rate Caps

FINOPS ENGINE

Enforce hard budget ceilings and per-agent token quotas. If cloud provider budgets reach defined thresholds, Flux automatically throttles burst traffic or seamlessly fails over to zero-cost local GPU clusters.

json

{
  "flux_policy": {
    "tenant_id": "org_enterprise_01",
    "daily_spend_limit_usd": 50.00,
    "on_breach": "FAILOVER_TO_LOCAL_OLLAMA",
    "notify_webhook": "https://alerts.yourcompany.com/finops"
  }
}

2. LangChain & LangGraph Integration

Use the native AegisCallbackHandler to monitor LangChain agent tool loops and halt execution if an agent oscillates between repetitive tools without convergence.

python

from aegis.integrations.langchain import AegisCallbackHandler
from langchain.agents import initialize_agent, AgentType
from langchain_openai import ChatOpenAI

# Initialize Aegis guardrail callback
aegis_handler = AegisCallbackHandler(
    max_step_cycles=4,
    spend_limit_usd=1.00,
    enable_otel=True
)

llm = ChatOpenAI(model="gpt-4o", callbacks=[aegis_handler])
agent = initialize_agent(tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION)

# Execute safely - recursion cycles are halted in-memory
response = agent.run("Perform complex multi-tool research task")

3. CrewAI Multi-Agent Handoff Guard

Prevent inter-agent ping-pong loops when multiple CrewAI agents delegate tasks back and forth.

python

from aegis.integrations.crewai import guarded_crew
from crewai import Crew, Agent, Task

# Create your standard CrewAI agents
researcher = Agent(role="Analyst", goal="Conduct market sweeps", backstory="Senior SRE")
task = Task(description="Map cloud provider costs", agent=researcher)

crew = Crew(agents=[researcher], tasks=[task])

# Wrap Crew execution with deterministic circuit breaker
safe_crew = guarded_crew(
    crew,
    max_inter_agent_handoffs=8,
    cost_ceiling_usd=3.00,
    alert_webhook="https://hooks.slack.com/services/..."
)

safe_crew.kickoff()

4. Raw REST API & LiteLLM Proxy Ingress

If running in Node.js, Go, or using LiteLLM as an API gateway, route completion payloads through the Aegis proxy with BYOK credentials.

python (requests)

import requests

# Send payload through the Aegis zero-retention gateway proxy
response = requests.post(
    "https://api.planetjdigital.com/v1/guard/inspect",
    headers={
        "Authorization": "Bearer aegis_live_your_key",
        "X-Provider-Key": "sk-ant-api03-...",   # BYOK: client key passed through
        "Content-Type": "application/json"
    },
    json={
        "session_id": "sess_8941a",
        "model": "claude-3-7-sonnet",
        "messages": [{"role": "user", "content": "Execute workflow loop"}],
        "circuit_policy": {
            "max_loops": 5,
            "timeout_seconds": 15
        }
    }
)

data = response.json()
print("Guard Status:", data["status"])  # "PASSED" or "CIRCUIT_OPEN"

5. Circuit Breaker Policies & Parameters

Parameter	Default	Description
max_loops	5	Maximum allowed recursive cyclic steps before tripping
spend_ceiling_usd	2.50	Hard token expenditure cap per task invocation
on_trip	"raise"	Action on breach: "raise", "fallback_to_human", or "return_last_state"
hash_tolerance	0.92	Cosine similarity threshold for repetitive tool input states

6. OpenTelemetry & Cloud Tracing

Aegis exports standard W3C trace contexts and OTel spans for deep observability without locking you into proprietary dashboards.

json (OTel Span Export)

{
  "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736",
  "span_id": "00f067aa0ba902b7",
  "name": "aegis.circuit_breaker.evaluate",
  "attributes": {
    "aegis.loop_depth": 2,
    "aegis.verdict": "CLEAR",
    "aegis.cost_accrued_usd": 0.042,
    "aegis.models": ["anthropic.claude-3-7-sonnet", "google.gemini-2.5-flash"]
  }
}