9thlevelsoftware/legion/authority-enforcer

Input: None (uses hardcoded path)
Output: AuthorityMatrix object
 
Procedure:
1. Read `.planning/config/authority-matrix.yaml`
2. Parse YAML into structured object
3. Store in memory: agent_id → AgentConfig mapping
4. Build reverse index: domain → primary_agent_id

Input: AuthorityMatrix
Output: ValidationReport
 
Checks:
1. All agent IDs exist in agent-registry
2. No domain is assigned to more than one agent as "exclusive"
3. Flag overlapping domains for conflict resolution
4. Verify deferred_by arrays are valid

def validate_matrix(matrix):
    errors = []
    domain_owners = {}
    
    for agent_id, config in matrix.agents.items():
        # Check 1: Agent exists in registry
        if not registry.has_agent(agent_id):
            errors.append(f"Unknown agent: {agent_id}")
        
        # Check 2: No duplicate exclusive domains
        for domain in config.exclusive_domains:
            if domain in domain_owners:
                errors.append(
                    f"Domain conflict: '{domain}' assigned to both "
                    f"{domain_owners[domain]} and {agent_id}"
                )
            else:
                domain_owners[domain] = agent_id
    
    return ValidationReport(valid=len(errors) == 0, errors=errors)

Input:
  agent_id: string          # Agent requesting action
  topic: string             # Topic/domain being addressed
  active_agents: string[]   # List of currently active agents
 
Output:
  authorized: boolean       # Whether agent can proceed
  domain_owner: string|null # Owner of this domain (if any)
  reason: string            # Explanation of decision

def validate_boundary(agent_id, topic, active_agents):
    # Mode check: skip validation if authority enforcement is disabled
    if not profile.authority_enforcement:
        return {
            authorized: True,
            domain_owner: None,
            reason: "Authority enforcement disabled by control mode"
        }
    # Load agent's exclusive domains from matrix
    agent_domains = matrix.get_domains(agent_id)
    # Check 1: Agent owns this topic
    if topic in agent_domains:
        return {
            authorized: True,
            domain_owner: agent_id,
            reason: "Agent has exclusive authority over this domain"
        }
    
    # Check 2: Another agent owns this topic
    for active_agent in active_agents:
        if active_agent == agent_id:
            continue
        
        active_domains = matrix.get_domains(active_agent)
        if topic in active_domains:
            return {
                authorized: False,
                domain_owner: active_agent,
                reason: f"{active_agent} has exclusive authority; respect domain owner"
            }
    
    # Check 3: No owner for this topic
    return {
        authorized: True,
        domain_owner: None,
        reason: "Topic has no exclusive owner; general domain"
    }

Normalization Rules:
  - Case-insensitive matching
  - Hyphens and underscores equivalent ("ci-cd" == "ci_cd")
  - Substring matching for compound topics:
      "api-security" matches both "api" and "security"
  
Specificity Scoring:
  - Exact match: score 10
  - Substring match: score 5
  - Related domain: score 2
  - No match: score 0

Input:
  agent_id: string
  base_prompt: string
  active_agents: string[]
 
Output:
  enhanced_prompt: string

def inject_authority_constraints(agent_id, base_prompt, active_agents):
    constraints = []
    
    # Step 1: Add agent's own authority
    own_domains = matrix.get_domains(agent_id)
    if own_domains:
        constraints.append(
            f"## Your Authority\\n\\n"
            f"You have EXCLUSIVE AUTHORITY over these domains:\\n"
            + "\\n".join(f"- {d}" for d in own_domains)
            + "\\n\\nWhen active, other agents must respect your judgment in these areas."
        )
    
    # Step 2: Add deference rules for other active agents
    for other_agent in active_agents:
        if other_agent == agent_id:
            continue
        other_domains = matrix.get_domains(other_agent)
        if other_domains:
            agent_name = matrix.get_name(other_agent)
            constraints.append(
                f"\\n## Domain Ownership Required\\n\\n"
                f"{agent_name} ({other_agent}) has exclusive authority over:\\n"
                + "\\n".join(f"- {d}" for d in other_domains)
                + "\\n\\nDO NOT critique or override their findings in these domains."
            )
    # Step 3: Mode-specific constraints
    if profile.read_only:
        constraints.append(
            "\n## Advisory Mode Active\n\n"
            "You are in ADVISORY mode. Analyze and suggest improvements "
            "but DO NOT modify any files. Present your suggestions as a "
            "structured list of proposed changes with rationale."
        )
    if profile.file_scope_restriction:
        constraints.append(
            "\n## Surgical Mode — File Scope Restriction\n\n"
            "You may ONLY modify files explicitly listed in this plan's "
            "`files_modified` field. Do not create, edit, or delete any other files. "
            "If a task requires touching unlisted files, stop and escalate."
        )
    if not profile.human_approval_required:
        # Omit the escalation protocol section from injected constraints
        pass  # Do not append the standard "Human Approval Required" block
    else:
        # Step 3b: Inject escalation format instructions
        # When human approval is required, agents must use structured <escalation> blocks
        # for any out-of-scope decisions. Format defined in escalation-protocol.yaml.
        escalation_instructions = (
            "\n## Escalation Protocol\n\n"
            "When you encounter a decision that falls outside your autonomous scope "
            "(architecture changes, unplanned dependencies, out-of-scope files, schema changes, "
            "API contract changes, deletions, infrastructure changes, or quality gate overrides), "
            "you MUST use a structured escalation block in your output.\n\n"
            "**Format** — wrap in `<escalation>` tags with these required fields:\n\n"
            "```\n"
            "<escalation>\n"
            "severity: info | warning | blocker\n"
            "type: architecture | dependency | scope | schema | api | deletion | infrastructure | quality\n"
            "decision: What decision is needed (one sentence)\n"
            "context: Why you encountered this (2-3 sentences)\n"
            "alternatives: What you would do if authorized (optional)\n"
            "affected_files: Files that would be modified (optional)\n"
            "</escalation>\n"
            "```\n\n"
            "**Severity guide:**\n"
            "- `info` — observation only, you can continue working\n"
            "- `warning` — notable concern, you continue but it will be highlighted\n"
            "- `blocker` — you MUST stop the blocked action and work on other in-scope tasks\n\n"
            "**Do not skip this.** Ad-hoc text descriptions of out-of-scope issues are not "
            "sufficient. The wave-executor parses `<escalation>` blocks for automated routing.\n\n"
            "Reference: `.planning/config/escalation-protocol.yaml`"
        )
        constraints.append(escalation_instructions)
    # Step 3c: Add control mode context line
    control_mode_name = profile.control_mode_name if hasattr(profile, 'control_mode_name') else "guarded"
    constraints.append(
        f"\n## Active Control Mode: {control_mode_name}\n\n"
        f"Escalation behavior follows the `{control_mode_name}` profile. "
        "See `.planning/config/escalation-protocol.yaml` for severity routing rules."
    )
    # Step 4: Combine with base prompt
    if constraints:
        enhanced = "\\n\\n".join(constraints) + "\\n\\n---\\n\\n" + base_prompt
    else:
        enhanced = base_prompt
    
    return enhanced

## Your Authority
 
You have EXCLUSIVE AUTHORITY over these domains:
-security
-owasp
-vulnerability-assessment
-pentest
 
When active, other agents must respect your judgment in these areas.
 
## Domain Ownership Required
 
Backend Architect (engineering-backend-architect) has exclusive authority over:
-backend-architecture
-database-design
-api-design
 
DO NOT critique or override their findings in these domains.
 
---
 
[Original prompt content...]

Input:
  findings: Finding[]      # List of review findings
  active_agents: string[]  # Agents present in review panel
 
Output:
  filtered_findings: Finding[]
  removed_findings: Finding[]  # With reasons

Finding:
  reviewer: string        # Agent ID who made the finding
  criterion: string       # What was checked
  severity: enum          # BLOCKER | WARNING | NOTE
  message: string         # Finding description
  domain: string          # Inferred domain (auto-detected)

def filter_findings(findings, active_agents):
    # Mode check: skip filtering if domain filtering is disabled
    if not profile.domain_filtering:
        return findings, []  # Return all findings, nothing removed
    filtered = []
    removed = []