Skip to content

Security Architecture

G6 implements security through the Computational Safety Framework (CSF) -- a constructive approach to pre-execution safety checks with explicit risk bounds, rather than only post-hoc guardrails.

Overview

The security architecture operates at three levels:

  1. Infrastructure -- TLS, authentication, rate limiting, container hardening
  2. Framework -- CSF safety checks with configured hazard bounds
  3. Per-operation -- ResourceBounds, ResourceGuardrail, and BoundedAgent constraints

CSF Safety Framework

The CSF provides auditable safety bounds for agent operations. Rather than filtering outputs after the fact, CSF checks whether an operation stays inside a configured hazard model before execution.

What the bound means

A CSF pass is a model-relative decision: the action is within the configured hazard probabilities and epsilon budget. The default probabilities are conservative policy priors, not universal empirical failure rates. CSF should be treated as a production safety gate and audit mechanism for MVP and pilot workflows, not as standalone proof of real-world safety or regulated-industry compliance.

Formal backend availability

CSF's core safety gate does not require external theorem provers. Optional Z3, NuSMV, and Lean checks add stronger evidence when installed, but they can be unavailable or slow in production environments. A response with verified=None means the backend did not produce a proof; do not treat it as approval unless another configured gate has explicitly passed.

Core types

SafetyQuery

A frozen dataclass representing a safety verification request:

from mvp.csf import SafetyQuery

query = SafetyQuery(
    agent=bounded_agent,
    hazards={"llm_call": 0.05, "file_write": 0.01},
    epsilon=0.20,
)

SafetyVerifier

Verifies safety queries using two strategies:

Strategy Method Description
union_bound Sum of hazard probabilities Conservative bound: sum of all hazard probabilities must be < epsilon
worst_case Maximum hazard probability Even more conservative: max single hazard must be < epsilon
from mvp.csf import SafetyVerifier

verifier = SafetyVerifier()
report = verifier.verify(query, strategy="union_bound")
# SafetyDecisionReport with is_safe, bound_value, details

SafetyDecisionReport

The output of verification:

Field Type Description
is_safe bool Whether the operation passed verification
bound_value float Computed safety bound
strategy str Verification strategy used
details dict Per-hazard breakdown

G6 safety signature

The align_csf component defines G6's default hazard priors:

G6_SAFETY_SIGNATURE = {
    "llm_call": 0.05,       # 5% hazard prior per LLM call
    "file_write": 0.01,     # 1% hazard prior per file write
    "code_execute": 0.10,   # 10% hazard prior per code execution
    "external_api": 0.08,   # 8% hazard prior per external API call
    "rollback": 0.001,      # 0.1% rollback failure prior
}

G6_EPSILON = 0.20  # Overall safety threshold

make_g6_csf()

Factory function that creates a CSF verifier pre-configured with the G6 safety signature:

from mvp.align_csf import make_g6_csf

csf = make_g6_csf()
report = csf.verify(query)

csf_guarded decorator

Wraps any function with automatic CSF verification:

from mvp.align_csf import csf_guarded

@csf_guarded(hazards=["llm_call", "external_api"])
def my_agent_function(input_data):
    # This function is automatically verified before execution
    # If CSF verification fails, Result.fail is returned
    return process(input_data)

ResourceBounds and ResourceGuardrail

ResourceBounds

A frozen dataclass that defines hard limits for an operation:

from mvp.core import ResourceBounds

bounds = ResourceBounds(
    max_tokens=4096,
    max_budget_usd=0.10,
    timeout_sec=30,
    max_retries=2,
)

ResourceGuardrail

A Guardrail subclass that enforces ResourceBounds at runtime:

  • Uses >= comparison (a limit of 0 blocks all usage)
  • Tags error messages with [RESOURCE_LIMIT]
  • Tracks usage via sliding window token logs with minute/hour/day/week/month windows
from mvp.core import ResourceGuardrail, ResourceBounds

guardrail = ResourceGuardrail(
    bounds=ResourceBounds(max_tokens=4096, timeout_sec=30)
)

# Check before operation
if guardrail.check(estimated_tokens=500):
    result = block.infer(input)
else:
    # Resource limit exceeded
    pass

ResourceUsage

A mutable companion to ResourceBounds that tracks actual consumption:

from mvp.core import ResourceUsage

usage = ResourceUsage()
usage.add_tokens(150)
usage.add_cost(0.002)
# Check against bounds
usage.within(bounds)  # True/False

Distilled artifact execution

Hyperdistillation executes synthesized Python artifacts through a static safety check followed by an isolated Python subprocess with a timeout. The static check rejects imports, dynamic evaluation, dunder traversal, and common file/network access routes before runtime; the subprocess boundary prevents artifact globals from sharing the main server process and stops runaway loops through timeout enforcement.

Subprocess sandbox scope

The hyperdistillation artifact sandbox is a product reliability control, not a complete hostile-code isolation layer. It is appropriate for executing artifacts synthesized from G6-controlled traces in a single-user or trusted-workspace deployment. For multi-tenant deployments, arbitrary user-submitted artifacts, regulated data, or hostile-code threat models, run G6 behind an outer container, VM, or equivalent OS-level sandbox and keep artifact databases separated per tenant or workspace.

Verification scope

Passing artifact verification means the code passed configured tests, AST safety checks, and any available formal checks. It does not prove semantic correctness for all future inputs. Production workflows should pair artifact reuse with drift checks, held-out examples, and domain-specific acceptance tests.

BoundedAgent and SafetyMonitor

BoundedAgent

A frozen dataclass representing an agent with known safety properties:

from mvp.core import BoundedAgent

agent = BoundedAgent(
    name="research_agent",
    transition_kernel=my_kernel_fn,    # Callable defining state transitions
    resource_bounds=bounds,
)

SafetyMonitor (ABC)

Abstract base class for runtime safety monitoring:

from mvp.core import SafetyMonitor

class MySafetyMonitor(SafetyMonitor):
    def check_step(self, rationale: StepRationale) -> bool:
        # Return True if step is safe to proceed
        return rationale.risk_score < 0.5

StepRationale

Mutable dataclass capturing the reasoning behind each agent step:

Field Type Description
action str What the agent intends to do
justification str Why this action is appropriate
risk_score float Estimated risk (0.0 - 1.0)
alternatives list[str] Other options considered

GitRollbackManager

Enables safe code modification with automatic revert on failure:

from mvp.csf import GitRollbackManager

manager = GitRollbackManager(repo_path="/path/to/repo")

with manager.safe_modify() as ctx:
    # Make changes -- automatically rolled back if an exception occurs
    modify_code()
    # Commit on success
    ctx.commit("Applied safe modification")

The manager uses gitpython to create checkpoints before modifications and automatically reverts to the checkpoint if the modification fails.

Self-Optimisation CSF Integration

The opt_meta component's auto-apply path is CSF-gated: before any intervention is applied to a live pipeline, a SafetyQuery is constructed from the intervention's HAZARD_MAP entries and verified via union_bound. If verification fails or CSF is unavailable with csf_fail_closed=True, the intervention is blocked and the failure is recorded for MAB learning.

Key security properties:

  • Fail-closed mode: when csf_fail_closed=True, CSF unavailability blocks all auto-apply actions
  • Rollback integration: opt_meta accepts a repo_path parameter for GitRollbackManager snapshots before applying code-level interventions
  • 12-task hardening: schema validators (step caps, numeric bounds), NaN/inf filtering, state bounds (failed interventions cap, bandit decay), BFS iteration limits, unknown model warnings, and post-execution resource checks across all opt_ blocks

For regulated workflows, combine CSF with domain-specific specifications, evaluation sets, human-in-the-loop breakpoints, incident review, and legal/compliance assessment before making compliance claims.

Emergency Kill Switch

The global kill switch provides system-wide emergency stop capability. When activated, all autonomous operations halt immediately -- orchestrators, coding agents, SDLC workflows, and recursive architect branches.

Mechanism

A file sentinel at ~/.g6/KILL_SWITCH. When the file exists, the system is halted. This design means it works even if the server is unresponsive -- you can create the file manually as a last resort.

from mvp.emergency.kill_switch import activate, deactivate, check_or_raise

activate(reason="runaway agent detected")  # Creates sentinel, halts system
check_or_raise()  # Raises EmergencyHaltError if active
deactivate()  # Removes sentinel, resumes operation

Integration points

check_or_raise() is called at these autonomous entry points:

Entry point File
Orchestrator rate limiter autonomous_orchestrator/rate_limiter.py:acquire()
Coding session execution business_manager/coding_agent.py:execute_coding_session()
SDLC phase execution business_manager/sdlc_orchestrator.py:execute_sdlc_phase()
Recursive architect recursive_architect/executor.py:BranchExecutor.run()

Interfaces

Interface Stop Resume Status
MCP emergency_stop emergency_resume emergency_status
REST POST /emergency/stop POST /emergency/resume GET /emergency/status
GUI API POST /emergency/stop POST /emergency/resume GET /emergency/status
TUI Ctrl+X -- Status bar "HALTED" indicator

Stable Snapshots

Git-tag-based versioning with SQLite manifest. Old snapshots are never deleted -- append-only.

Snapshot types

Type Git tag format Trigger
Manual stable/manual/v{N} Human marks state as stable
Auto (tests pass) stable/auto/{ISO_DATE} Test suite passes
Pre-evolution stable/pre-evolve/{ISO_DATE} Before T3 self-modification

Usage

from mvp.emergency.snapshot_manager import SnapshotManager

mgr = SnapshotManager(workspace_root="/path/to/workspace")

# Create manual snapshot
result = mgr.create_snapshot(source="manual", description="Release v1.0")

# Auto-snapshot if tests pass
result = mgr.auto_snapshot_if_tests_pass()

# Restore to previous state
mgr.restore_snapshot(snapshot_id="abc123def456")

# List all snapshots
snapshots = mgr.list_snapshots(limit=20)

REST endpoints

Method Path Description
POST /emergency/snapshots Create manual snapshot
GET /emergency/snapshots List snapshots
POST /emergency/snapshots/{id}/restore Restore to snapshot

Infrastructure security

For cloud-assisted deployments, additional infrastructure controls are in place:

Control Implementation
API authentication Bearer token auth via hashed API keys (PostgreSQL)
TLS/HTTPS TLS 1.2+ termination at nginx, HSTS, OCSP stapling
Tool access control Tier-gated tool access per license (Free: 30, Researcher: 49, Builder: full tool access)
Rate limiting Redis sliding-window per license key (10-500 req/min by tier)
Secrets management Environment variables via .env, no hardcoded credentials
Dependency pinning Version-pinned requirements for Docker images
Container hardening Non-root user, multi-stage builds, health checks
CORS Origins restricted to known domains

Security checklist

For production deployments:

  • Bearer token API authentication (nginx)
  • TLS termination via nginx reverse proxy
  • Vault-managed secrets (no hardcoded credentials)
  • Per-key rate limiting by license tier
  • CORS origins restricted
  • Tier-based tool access control (15 categories)
  • Container images pinned to minor versions (python:3.12-slim, nginx:1.27-alpine, postgres:16-alpine)

See Configuration for CSF settings and environment variables.