symfonic.capabilities.prompting.capability

capability

The prompting capability, as something Agent can be handed — W2 phase 3.

Everything this module needs already existed. :mod:.assembly has been the bridge to the kernel's vocabulary since T3.2.1 — "the capability declares a stage and produces a value; it never calls the kernel" — and the compiler that turns contributions into a prompt has been written and tested the whole time. What was missing was the twenty lines that hand both to the kernel at once, and a kernel that would run them.

So this is deliberately thin. If it needed to be clever, the seam would be wrong.

Prompting is the first capability migrated for a structural reason rather than an arbitrary one: capabilities/memory/contribution.py converts a memory declaration into a PromptContribution, and hydration describes itself as the last memory step before the prompt compiler. Memory has nowhere to deliver until this exists.

PromptingCapability dataclass

PromptingCapability(sources: Sequence[Any] = (), priority: int = -900, options: dict[str, Any] = dict())

Compile the layered prompt and contribute it to the turn's assembly.

Usage is the whole point of the phase::

Agent(provider, capabilities=[PromptingCapability(sources=[...])])

sources are the prompt contributions this capability compiles — the same values the capability's own compiler has always taken. They are held on the config rather than read from a global, so two agents in one process do not share a prompt.

contribute

contribute(request: CapabilityRequest) -> CapabilityContribution

Declare the stage and supply the handler that answers it.

Both in one value, which is what lets fold_contributions refuse a declaration nothing runs. The compile happens here, at contribution time, so the stage descriptor can carry the compiled digest — that is what makes "this plan compiled this prompt" checkable from the plan alone, without the plan carrying the prompt text.

Source code in src/symfonic/capabilities/prompting/capability.py

def contribute(self, request: CapabilityRequest) -> CapabilityContribution:
    """Declare the stage and supply the handler that answers it.

    Both in one value, which is what lets ``fold_contributions`` refuse a
    declaration nothing runs. The compile happens *here*, at contribution
    time, so the stage descriptor can carry the compiled digest — that is
    what makes "this plan compiled this prompt" checkable from the plan
    alone, without the plan carrying the prompt text.
    """
    static, awaited = split_by_door(as_contributions(self.sources))
    digest_options, deferred_gate = _digest_options(self.options)
    compiled = compile_prompt(
        PromptCompileRequest(contributions=static, **digest_options)
    )
    descriptor = prompt_stage_descriptor(
        compiled,
        priority=self.priority,
        awaited_sources=len(awaited),
        deferred_render_gate=deferred_gate,
    )

    async def handle(context: Any) -> StageResult[Any]:
        """Project the compiled prompt onto the turn being assembled.

        Reports ``NO_CHANGE`` with a reason when the compile produced
        nothing, rather than contributing an empty assembly. An empty
        contribution and a capability that decided not to act are different
        facts, and only one of them is a reason to look at the sources.
        """
        # Resolution ran first (STG-7), so the snapshot is complete and
        # frozen. Recompiling with it is what makes this capability compile
        # the *turn's* prompt rather than only its static configuration.
        resolved = _contributions_from(getattr(context, "resolved", None))
        turn_request = getattr(context, "request", None)
        if turn_request is None:  # pragma: no cover - defensive
            return no_change("no turn request on the stage context")
        # The turn's scope reaches every source through the compile request:
        # ``resolve_source`` passes ``scope_path`` down, which is how a
        # source bound at construction still renders for the tenant being
        # served. Without it the system prompt named whichever scope the
        # bundle was folded with, on every turn.
        scope_path = _scope_path_of(turn_request)
        session_id = str(getattr(turn_request, "session_id", "") or "")
        options = _turn_options(self.options, scope_path, session_id)
        # The asynchronous door (S01). The stage handler is already a
        # coroutine the kernel awaits, so awaiting the compile here costs no
        # new concurrency surface -- and it is what lets a contribution whose
        # content lives behind an ``await`` reach the prompt at all. A
        # source that only offers the synchronous ``read`` is still called,
        # not wrapped: ``resolve_source_async`` calls it directly.
        turn_compiled = (
            await compile_prompt_async(
                PromptCompileRequest(
                    contributions=as_contributions(self.sources) + resolved,
                    **options)
            )
            if resolved or self.sources
            else compiled
        )
        run_id = getattr(context, "run_id", "")
        receipt = None
        if run_id:
            receipt = PromptAdmissionReceipt(
                run_id=run_id,
                scope_path=scope_path,
                session_id=session_id,
                compiler_scope_path=options.get("scope_path", ""),
                compiler_session_id=options.get("session_id", ""),
                admitted=turn_compiled.budget.admitted,
                dropped=turn_compiled.budget.dropped,
            )
        if not turn_compiled.text:
            return no_change(
                "the prompt compiled to nothing: no contribution survived "
                "its gates, or none was supplied",
                counts=prompt_counts(turn_compiled),
                prompt_admission=receipt,
            )
        assembly = to_prompt_assembly(
            turn_compiled,
            prompt=turn_request.prompt,
            attachments=tuple(turn_request.attachments),
            history=tuple(turn_request.history),
            append_user_message=getattr(turn_request, "append_user_message", True),
        )
        return applied(
            _compose(context, assembly),
            counts=prompt_counts(turn_compiled),
            prompt_admission=receipt,
        )

    return CapabilityContribution(
        capability="prompting",
        stages=(descriptor,),
        handlers={PROMPTING_STAGE: handle},
        # No effects: prompt assembly is a pure function of the plan and the
        # request (STG-7). Claiming one would widen this capability's own
        # admission for nothing.
        effect_grants=frozenset(),
    )