Skip to content

symfonic.capabilities.human.registration

registration

Named interaction points: ask_user is a registration, not a code path.

The shipped engine had two of everything — a dedicated ask_user mint path guarded by one flag, and a generic interrupt minter guarded by another, each with its own resume route and its own scope check. The difference between them is three facts, not two subsystems: which metadata key holds the payload, which identifier correlates the event, and whether the deployment may open cross-scope redemption. All three live on the registration.

InteractionRegistration dataclass 

InteractionRegistration(name: str, payload_schema: Any, response_schema: Any, validate_response: Callable[[Any, Any], None] | None = None, cross_scope_allowed: bool = False, built_in: bool = False, metadata: Mapping[str, Any] = dict())

One registered pause point, and everything both ends of it need.

ask_user classmethod 

ask_user(*, payload_schema: Any, response_schema: Any, validate_response: Callable[[Any, Any], None] | None = None, cross_scope_allowed: bool = False, metadata: Mapping[str, Any] | None = None) -> InteractionRegistration

The built-in, in the one shape it is allowed to have.

Source code in src/symfonic/capabilities/human/registration.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
@classmethod
def ask_user(
    cls,
    *,
    payload_schema: Any,
    response_schema: Any,
    validate_response: Callable[[Any, Any], None] | None = None,
    cross_scope_allowed: bool = False,
    metadata: Mapping[str, Any] | None = None,
) -> InteractionRegistration:
    """The built-in, in the one shape it is allowed to have."""
    return cls(
        name=ASK_USER,
        payload_schema=payload_schema,
        response_schema=response_schema,
        validate_response=validate_response,
        cross_scope_allowed=cross_scope_allowed,
        built_in=True,
        metadata=metadata or {},
    )

InteractionRegistry 

InteractionRegistry()

The one place a name is resolved, on both ends of a pause.

Source code in src/symfonic/capabilities/human/registration.py 
111
112
def __init__(self) -> None:
    self._registrations: dict[str, InteractionRegistration] = {}

names 

names() -> tuple[str, ...]

Sorted, so a refusal message reads the same on every worker.

Source code in src/symfonic/capabilities/human/registration.py 
139
140
141
def names(self) -> tuple[str, ...]:
    """Sorted, so a refusal message reads the same on every worker."""
    return tuple(sorted(self._registrations))