The Control-
Surface Band

Two layers, one ledger

The governance engine runs at two levels. At the Arbiter level, it evaluates authority scopes and composition contracts before every agent dispatch. At the Worker level, it intercepts every tool call within a worker agent. Both layers write to the same DynamoDB governance ledger.

What makes scope conditions operational

Without runtime context, authority scopes evaluate against an empty dict. Conditions never match. Limits never fire. The build_dispatch_context() function populates DispatchRequest.context with real agent metrics and workflow state from DynamoDB.

def build_dispatch_context(agent_name, workflow_id, orchestration) -> dict:
    """
    Build the context dict for DispatchRequest.context.
    All values are deterministic facts — no LLM, no interpretation.
    This is what activates governance scope conditions and limits.
    """
    context = {}

    # Per-agent metrics from DynamoDB (atomic counters)
    metrics = load_agent_metrics(agent_name)
    if metrics:
        invocations = int(metrics.get('invocationCount', 0))
        denies = int(metrics.get('governanceDenyCount', 0))
        failures = int(metrics.get('failureCount', 0))
        successes = int(metrics.get('successCount', 0))
        total_duration = int(metrics.get('totalDurationMs', 0))

        context['agent_invocation_count'] = invocations
        context['agent_deny_rate_pct'] = (denies / max(invocations, 1)) * 100
        context['agent_failure_rate_pct'] = (failures / max(invocations, 1)) * 100
        context['agent_avg_duration_ms'] = total_duration / max(successes, 1)

    # Per-workflow state
    context['workflow_fabrication_pending'] = orchestration.get('pending_fabrication', False)
    context['workflow_agent_count'] = len(agents_used)  # distinct agents in this workflow
    context['workflow_deny_count'] = orchestration.get('_deny_count', 0)
    context['workflow_escalate_count'] = orchestration.get('_escalate_count', 0)

    return context

With this context populated, authority scopes like these become operational:

# "This agent may be invoked no more than 1000 times"
AuthorityScope(
    decision_type="invoke_agent",
    domain="*",
    conditions={},
    limits={"agent_invocation_count": 1000}
)

# "Halt if the agent's deny rate exceeds 20%"
AuthorityScope(
    decision_type="invoke_agent",
    domain="payment",
    conditions={},
    limits={"agent_deny_rate_pct": 20.0}
)

# "Do not invoke during active fabrication"
AuthorityScope(
    decision_type="invoke_agent",
    domain="*",
    conditions={"workflow_fabrication_pending": False},
    limits={}
)

Worker-level governance

Worker agents use the Strands SDK. The GovernedToolHandler subclasses AgentToolHandler.preprocess() to intercept every tool call before execution. This is the fine-grained governance layer: which tools is this specific agent permitted to use?

class GovernedToolHandler(AgentToolHandler):
    """
    Tool handler that evaluates governance policy before every tool execution.
    If preprocess() returns a ToolResult, the tool call is short-circuited.
    If preprocess() returns None, the tool executes normally.
    """

    def __init__(self, tool_registry, agent_id, workflow_id, denied_tools=None):
        super().__init__(tool_registry)
        self.agent_id = agent_id
        self.workflow_id = workflow_id
        self.denied_tools = denied_tools or set()

        # Load from env var if not passed directly
        if not self.denied_tools:
            denied_env = os.environ.get('DENIED_TOOLS', '')
            self.denied_tools = set(t.strip() for t in denied_env.split(',') if t.strip())

    def preprocess(self, tool, tool_config, **kwargs):
        """
        Pre-execution governance check. Deterministic — no LLM.
        Returns a denial ToolResult if the tool is not permitted.
        Returns None to allow the tool call to proceed.
        """
        tool_name = tool.get('name', '')
        tool_use_id = tool.get('toolUseId', '')

        if tool_name in self.denied_tools:
            finding_id = str(uuid.uuid4())
            self._write_finding(finding_id, tool_name, 'deny',
                f'tool_denied:explicit_deny_list:{tool_name}')
            return ToolResult(
                toolUseId=tool_use_id,
                status='error',
                content=[{'text': f"Tool '{tool_name}' is not authorised. Finding: {finding_id}"}],
            )

        # Permit + governance record
        self._write_finding(str(uuid.uuid4()), tool_name, 'permit',
            'tool_permitted:no_constraints_violated')
        return None  # Proceed with execution

Both governance layers write to the same DynamoDB table. A single query by workflowId produces the complete governance trace: Arbiter-level dispatch decisions and Worker-level tool decisions in chronological order.

Every evaluation produces a record

The governance record is produced at evaluation time as a structural byproduct. Not reconstructed after the fact. If the ledger write fails, the action is denied. An action that produces no record is a governance failure.

def write_finding(finding: GovernanceFinding) -> None:
    """Write a governance finding to the ledger table. Raises on failure."""
    table_name = os.environ.get('GOVERNANCE_LEDGER_TABLE')
    if not table_name:
        raise RuntimeError("GOVERNANCE_LEDGER_TABLE not configured")

    table = _dynamodb.Table(table_name)
    table.put_item(Item={
        'findingId': finding.finding_id,        # UUID — write-once key
        'workflowId': finding.workflow_id,       # queryable — all findings for a workflow
        'timestamp': str(finding.timestamp),
        'decision': finding.decision.value,      # permit | deny | escalate | halt
        'requestingAgent': finding.requesting_agent,
        'targetAgent': finding.target_agent,
        'scopeEvaluated': finding.scope_evaluated or 'none',
        'contractEvaluated': finding.contract_evaluated or 'none',
        'reason': finding.reason,                # machine-readable cause string
        'escalationTarget': finding.escalation_target or 'none',
        'residualAuthorityDenial': finding.residual_authority_denial,
        'ttl': int(finding.timestamp) + (90 * 24 * 3600),  # 90-day retention
    })

In governed_process_agent_call(), the ledger write failure explicitly halts dispatch:

    # Write governance record (mandatory — before dispatch or denial)
    try:
        write_finding(finding)
    except Exception as e:
        print(f"GOVERNANCE LEDGER WRITE FAILED: {e} — halting dispatch")
        return None  # No dispatch, no denial, no escalation — just halt

Deploy and observe

# Clone the repository
git clone https://github.com/aws-samples/sample-agentic-fabric
cd sample-agentic-fabric/Arbiter/app

# Install and build
npm install
npm run build

# Deploy (starts in bypass mode)
ENVIRONMENT=dev npx cdk bootstrap --profile your-profile
ENVIRONMENT=dev npx cdk deploy --all --require-approval never --profile your-profile

# 1. Invoke an agent (governance bypass active)
#    Check CloudWatch — you should see "GOVERNANCE_BYPASS active"

# 2. Set GOVERNANCE_BYPASS=false in the Lambda environment
#    Redeploy or update the environment variable directly

# 3. Invoke an agent again
#    Check CloudWatch — you should see "Governance PERMIT: scope_match:..."
#    Check the governance ledger DynamoDB table — a finding should be written

# 4. Add a denied tool to a worker agent config
#    Invoke a workflow that uses that tool
#    Check CloudWatch — "Governance DENY tool 'tool-name'"
#    The agent receives the denial as a ToolResult error