Skip to content

Hat Orchestrator

HAT Orchestrator — multi-party human-AI teaming component.

Cluster: Goal & Planning | Type: component | MCP Tools: None

Overview

Human-AI Teaming (HAT) orchestrator that manages a registry of heterogeneous agents (human, AI, robot) and coordinates them through task allocation, synchronisation points, authority resolution, and a Common Operational Picture (COP). Agents are registered with capability profiles and Sheridan authority levels; tasks are matched to the best-available agent; alerts can be broadcast to the whole team.

When to use:

  • Coordinating mixed human-AI-robot teams where authority and capability must be tracked per agent
  • Implementing synchronisation barriers that wait for all required team members before proceeding
  • Resolving authority conflicts between overlapping agents using Sheridan-level priority rules

Example:

from mvp.hat_orchestrator import HATOrchestratorBlock, HATOrchestratorInput

block = HATOrchestratorBlock(name="hat")
block.infer(HATOrchestratorInput(
    op="register_agent", agent_id="human-1", agent_type="human",
    capabilities={"review": True}, authority_level=8,
))
result = block.infer(HATOrchestratorInput(
    op="allocate_task", task_type="review", required_capabilities={"review": True},
))
# result.ok → True; result.value.allocation contains assigned agent

Works well with: autonomy_governor, autonomous_orchestrator, human_development

Operational Caveats

Pilot-ready orchestration, not a full task system

hat_orchestrator is suitable for local pilots and launch-plan workflows where a small number of agents need durable registration, task assignment, synchronisation barriers, authority resolution, and a shared Common Operational Picture. It persists state to SQLite and reserves automatic task IDs atomically, so stale block instances should not generate duplicate task_# IDs.

It now exposes complete_task and fail_task lifecycle operations, surfaces degraded allocation and persistence envelopes, and wires GuardrailConfig into registration and allocation chokepoints. Alerts remain caller-managed, but clear_alerts lets operators explicitly clear accumulated alert state. Agent availability is still manually supplied rather than derived from active workload, and authority/capability claims remain caller-supplied rather than externally identity-verified.

SQLite is appropriate for single-machine MCP/REST pilots. For high-concurrency, multi-host, or customer-critical deployments, use this component behind a single writer process or migrate the store to a server database with transactional task lifecycle operations.

Public API

HATOrchestratorBlock(AIBlock[HATOrchestratorInput, HATOrchestratorOutput, dict])

Field Type Default
name str 'hat_orchestrator'
state dict field(default_factory=dict)
resource_bounds ResourceBounds field(default_factory=ResourceBounds)
usage ResourceUsage field(default_factory=ResourceUsage)
db_path str '~/.g6/hat_orchestrator.db'

Methods:

infer(data: HATOrchestratorInput) -> Result[HATOrchestratorOutput]

health() -> dict

list_patterns() -> dict

Surface the deterministic applied-pattern + skill catalog (not an op).

HATOrchestratorInput(BaseModel)

Field Type Default
op HATOp 'list_agents'
agent_id str ''
agent_type str ''
capabilities dict Field(default_factory=dict)
availability float 1.0
authority_level int 5
communication_channels list[str] Field(default_factory=list)
role str ''
task_id str ''
task_type str ''
required_capabilities dict Field(default_factory=dict)
priority int 5
timeout_seconds float 0.0
sync_point_id str ''
required_agents list[str] Field(default_factory=list)
agent_a_id str ''
agent_b_id str ''
action str ''
alert_message str ''
alert_level str ''
metadata dict Field(default_factory=dict)

HATOrchestratorOutput(BaseModel)

Field Type Default
op str required
agent dict Field(default_factory=dict)
agents list[dict] Field(default_factory=list)
allocation dict Field(default_factory=dict)
task_status dict Field(default_factory=dict)
sync_status dict Field(default_factory=dict)
authority_result dict Field(default_factory=dict)
cop dict Field(default_factory=dict)
team_stats dict Field(default_factory=dict)
alert_sent bool False
metadata dict Field(default_factory=dict)
degraded bool False
degradation_reason str \| None None
completion_state Literal['verified', 'qualified-draft', 'blocked-escalated'] 'qualified-draft'
warning_card dict \| None None
evidence list[dict] Field(default_factory=list)
request_id str ''
task_id str ''
run_id str ''