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:
- Infrastructure -- TLS, authentication, rate limiting, container hardening
- Framework -- CSF safety checks with configured hazard bounds
- 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:
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_metaaccepts arepo_pathparameter forGitRollbackManagersnapshots 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.