Autonomy Governor¶

Autonomy Governor -- mvp.autonomy_governor

Cluster: Experience & Autonomy | Type: component | MCP Tools: 33

Overview¶

Autonomy Governor that tracks and enforces the continuous autonomy level Au(t) for the G6 Physical AI system using a Sheridan-Verplank 10-level mapping and Fitts-taxonomy operation mode classification. Provides escalate/de-escalate controls, HITL checkpoint scheduling based on operation speed and oversight capacity, and outcome-driven autonomy adaptation across skill phases.

When to use:

• Dynamically adjusting how much autonomy an agent has based on task risk, skill phase, and human capacity
• Scheduling human-in-the-loop checkpoints at the right frequency for a given operation speed
• Tracking the Sheridan authority level and operation mode to enforce deployment policy constraints

Example:

from mvp.autonomy_governor import AutonomyGovernorBlock, AutonomyGovernorInput

block = AutonomyGovernorBlock(name="governor")
block.infer(AutonomyGovernorInput(op="set_autonomy", autonomy_delta=0.3))
result = block.infer(AutonomyGovernorInput(
    op="assess_hitl_need", operation_speed=2.5, human_oversight_capacity=0.6,
))
# result.ok → True; result.value → AutonomyGovernorOutput with sheridan_level, needs_checkpoint

Works well with: autonomous_orchestrator, hat_orchestrator, experience_loop

Scope and honest limits:

• This is a governed-state MCP adapter, not a full external-system adapter. The backend it adapts is the component's own SQLite store and internal policy/state model; it does not expose native ops over external orchestration runs, safety-engine ledgers, runtime policy registries, or live agent execution traces. That is a future scope item.
• The Tier-1 block holds mutable state (it is not stateless). Autonomy out of range is clamped by phase/max caps, not rejected with ValueError.
• The MCP store uses a single SQLite connection (check_same_thread=False, WAL) with no explicit per-call locking; it is not verified thread-safe for concurrent writers.
• The optional check() LLM safety reviewer can only tighten the deterministic six-rule floor (ALLOW < REQUIRE_HITL < BLOCK); on reviewer failure the floor holds and the result is marked degraded (fail-closed). store_policy fails closed on invalid JSON or invalid AutonomyPolicy semantics before persistence.
• Read-only inspection ops (get_recent_*, get_error_log, get_outcome_history, get_escalation_log, get_hitl_events) are bounded (top_k clamped to 100), redacted, and carry no executor authority. Outputs use the canonical completion_state trio verified / qualified-draft / blocked-escalated.

Public API¶

AutonomyGovernorBlock(AIBlock[AutonomyGovernorInput, AutonomyGovernorOutput, dict])¶

Autonomy Governor for the G6 Physical AI system.

Methods:

infer(data: AutonomyGovernorInput) -> Result[AutonomyGovernorOutput]¶

check(action: GovernedAction) -> GateVerdict¶

Gate verdict for a proposed action.

set_gate_review_planner(planner) -> None¶

Inject a gate-review planner (for tests).

set_policy(policy: AutonomyPolicy) -> None¶

Replace the active policy.

record_outcome_async(action_id: str, outcome: str) -> None¶

Async wrapper for recording outcomes (used by agent pool).

AutonomyPolicy¶

AutonomyGovernorInput(BaseModel)¶

Input to AutonomyGovernorBlock.

AutonomyGovernorOutput(BaseModel)¶

Output from AutonomyGovernorBlock.

AutonomyGovernorMCPBlock(AIBlock[MCPAutonomyGovernorInput, MCPAutonomyGovernorOutput, dict])¶

25-op autonomy governor MCP block with SQLite persistence.

Methods:

close() -> None¶

Close resources owned by the MCP block.

infer(data: MCPAutonomyGovernorInput) -> Result[MCPAutonomyGovernorOutput]¶

MCPAutonomyGovernorInput(BaseModel)¶

Input to AutonomyGovernorMCPBlock -- 33 ops.

MCPAutonomyGovernorOutput(BaseModel)¶

Output from AutonomyGovernorMCPBlock.

AutonomyGovernorStore¶

SQLite-backed store for the autonomy_governor MCP sub-package.

Constructor:

Methods:

close() -> None¶

Close the backing SQLite connection.

save_snapshot(name: str, state: dict, tags: str = '') -> str¶

load_snapshot(name: str) -> dict | None¶

list_snapshots(limit: int = 50) -> list[dict]¶

store_policy(name: str, policy_json: str, tags: str = '') -> str¶

load_policy(name: str) -> dict | None¶

list_policies(limit: int = 50) -> list[dict]¶

search_policies(query: str, top_k: int = 10) -> list[dict]¶

log_escalation(old_au: float, new_au: float, reason: str = '', metadata: dict | None = None) -> str¶

get_escalation_log(limit: int = 50) -> list[dict]¶

log_hitl_event(event_type: str, operation_speed: float = 0.0, human_capacity: float = 1.0, needs_checkpoint: bool = False, reason: str = '') -> str¶

get_hitl_stats() -> dict¶

get_hitl_events(limit: int = 50) -> list[dict]¶

Read-only: recent HITL checkpoint event rows (newest first).

log_outcome(outcome: str, action_type: str = '', autonomy_at: float = 0.0, metadata: dict | None = None) -> str¶

get_outcome_stats() -> dict¶

get_outcome_history(limit: int = 50) -> list[dict]¶

Read-only: recent recorded-outcome rows (newest first).

get_autonomy_trend(limit: int = 50) -> list[dict]¶

log_error(op: str, error_message: str, params: dict | None = None) -> None¶

get_error_log(limit: int = 50) -> list[dict]¶

Read-only: recent error rows (newest first).

get_policy_decisions(limit: int = 50) -> list[dict]¶

Read-only: recent stored-policy audit rows (newest first).

count_all() -> dict[str, int]¶

Functions¶

list_patterns() -> dict[str, Any]¶

Return the autonomy_governor applied-pattern + skill surface.

MCP Tools¶