Skip to content

symfonic.platform.governance_declaration

governance_declaration

Which rungs a governance pipeline occupies, and who answers them.

Split from :mod:symfonic.platform.governance_rungs when that module passed its budget, and the seam is the honest one: this is declaration -- what a composed pipeline contributes to a plan -- and run_phase next door is execution. A reader asking "where does governance attach?" and one asking "what happens when it runs?" were reading one file for two questions.

kernel_stages 

kernel_stages(pipeline: GovernancePipeline, decisions: Any = None) -> tuple[tuple[StageDescriptor, ...], dict[str, Any]]

The descriptors and handlers for whichever rungs have stages.

A rung with no stages contributes nothing at all -- the same rule compose applies one layer down. A declared stage that examines nothing is indistinguishable in a trace from one that looked and found nothing.

Source code in src/symfonic/platform/governance_declaration.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
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
def kernel_stages(
    pipeline: GovernancePipeline, decisions: Any = None
) -> tuple[tuple[StageDescriptor, ...], dict[str, Any]]:
    """The descriptors and handlers for whichever rungs have stages.

    A rung with no stages contributes nothing at all -- the same rule
    ``compose`` applies one layer down. A declared stage that examines nothing
    is indistinguishable in a trace from one that looked and found nothing.
    """
    stranded = sorted(
        phase
        for phase in UNDISPATCHED_PHASES
        if pipeline.for_phase(phase) is not None
    )
    if stranded:
        names = ", ".join(
            stage
            for phase in stranded
            for stage in (pipeline.for_phase(phase) or pipeline).stage_names
        )
        raise ConfigurationError(
            f"this pipeline composes {names}, which run at governance phase(s) "
            f"{stranded} -- and the kernel dispatches no rung for them today "
            "(pre-tool and post-tool are accepted by the fold and never "
            "invoked). Compose without those ports rather than shipping a gate "
            "that reports success and never examines a call."
        )

    descriptors: list[StageDescriptor] = []
    handlers: dict[str, Any] = {}

    for gov_phase, rung, stage_id in PHASE_RUNGS:
        # Budget reads the model's draft, even on a turn with no tools.
        # Its EFFECT classification is not its kernel dispatch placement.
        names = tuple(
            name for name in pipeline.stage_names
            if name in CARRIED_STAGES
            or ("egress" if name == "budget" else str(pipeline.rulebook.rule_for(name).phase))
            == gov_phase
        )
        selected = pipeline.select(names)
        if selected is None:
            continue
        descriptors.append(
            StageDescriptor(
                stage_id=stage_id,
                phase=rung,
                capability=GOVERNANCE_CAPABILITY,
                priority=-800,
                # Ingress publishes intent to the immutable resolved snapshot;
                # effect hands back amended tool calls. Egress only observes.
                kind=(
                    StageKind.RESOLUTION
                    if gov_phase in ("ingress", "effect")
                    else StageKind.COMPILATION
                ),
                effects=(
                    frozenset({"model_call"})
                    if {"intent_filter", "metacognition"}.intersection(names)
                    else frozenset()
                ),
            )
        )
        handlers[stage_id] = _handler(selected, gov_phase, decisions)

    return tuple(descriptors), handlers