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 | '' |