Skip to content

symfonic.platform.governance

governance

The governance door: one call a composition root makes.

The rung mapping and the reading of a turn live in :mod:symfonic.platform.governance_rungs; this file is the entry point and nothing else, so the thing an adopter imports is the thing they read first.

GovernanceStages 

GovernanceStages(capability: Any, decisions: Any = None)

A composable wrapper around a built governance pipeline.

The object a composition root passes to Agent(capabilities=[...]). It holds the capability rather than subclassing it: governance answers "what does this turn get, and why", and this answers "which rungs does that run on" -- two questions, and merging them is what would put kernel imports back inside the contained package.

Source code in src/symfonic/platform/governance.py 
30
31
32
def __init__(self, capability: Any, decisions: Any = None) -> None:
    self._capability = capability
    self._decisions = decisions

capability property 

capability: Any

The governed pipeline, for a caller that wants the trace directly.

contribute 

contribute(request: Any) -> Any

Declare the rungs this pipeline runs on, and the handlers for them.

Classification and reflection may call a model and declare that effect when composed. Refusal and deterministic checks need no additional effect grant.

Source code in src/symfonic/platform/governance.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def contribute(self, request: Any) -> Any:
    """Declare the rungs this pipeline runs on, and the handlers for them.

    Classification and reflection may call a model and declare that
    effect when composed. Refusal and deterministic checks need no
    additional effect grant.
    """
    from symfonic.kernel.contracts.contributions import CapabilityContribution

    descriptors, handlers = kernel_stages(
        self._capability.pipeline, self._decisions
    )
    return CapabilityContribution(
        capability=GOVERNANCE_CAPABILITY,
        stages=descriptors,
        handlers=handlers,
        effect_grants=frozenset(effect for stage in descriptors for effect in stage.effects),
    )

governance 

governance(**ports: Any) -> GovernanceStages

Build the safety layer and bind it to the rungs that run it.

One call, in the shape memory_capabilities(store, scope) established: sensible defaults, every port injectable, nothing required.

Agent(provider, capabilities=[governance(reflector=my_reflector)])

Every keyword is forwarded to GovernanceCapability.compose, which leaves a stage out rather than in-but-inert when its port is missing.

Credential-shaped tool results and errors stop the turn before the next model call. POST_TOOL cannot deliver a sanitized replacement: this is a refusal boundary, not redaction of the transcript or already emitted events. Default patterns may also match benign text; deployments can narrow them explicitly through patterns.

decisions is the exception, and it is held here rather than forwarded: what governance decided about a rung is a fact about the kernel mapping, not about the safety pipeline, which does not know a rung exists. Passing a sink -- any object with record(decision), such as DecisionLog -- turns those decisions into a readable record::

log = DecisionLog()
Agent(provider, capabilities=[governance(guards=[...], decisions=log)])

Omitted, nothing is recorded: a deployment that never asked for a decision log does not accumulate one for the life of the process.

Source code in src/symfonic/platform/governance.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def governance(**ports: Any) -> GovernanceStages:
    """Build the safety layer and bind it to the rungs that run it.

    One call, in the shape ``memory_capabilities(store, scope)`` established:
    sensible defaults, every port injectable, nothing required.

        Agent(provider, capabilities=[governance(reflector=my_reflector)])

    Every keyword is forwarded to ``GovernanceCapability.compose``, which
    leaves a stage out rather than in-but-inert when its port is missing.

    Credential-shaped tool results and errors stop the turn before the next
    model call. POST_TOOL cannot deliver a sanitized replacement: this is a
    refusal boundary, not redaction of the transcript or already emitted events.
    Default patterns may also match benign text; deployments can narrow them
    explicitly through ``patterns``.

    ``decisions`` is the exception, and it is held here rather than forwarded:
    what governance *decided about a rung* is a fact about the kernel mapping,
    not about the safety pipeline, which does not know a rung exists. Passing
    a sink -- any object with ``record(decision)``, such as ``DecisionLog`` --
    turns those decisions into a readable record::

        log = DecisionLog()
        Agent(provider, capabilities=[governance(guards=[...], decisions=log)])

    Omitted, nothing is recorded: a deployment that never asked for a decision
    log does not accumulate one for the life of the process.
    """
    from symfonic.capabilities.governance import GovernanceCapability

    decisions = ports.pop("decisions", None)
    return GovernanceStages(GovernanceCapability.compose(**ports), decisions)