cadrumo.application.operator_surface.command_ports module

Inward command-surface ports consumed by outer composition roots.

The command graph is an outer concern, but the values exchanged with another composition root are application contracts. Keeping these records here lets the CLI and the retained harness share one schema, policy, and dispatch vocabulary without either side importing the other. Implementations remain free to resolve their own command graph; this module contains no Typer, Click, entrypoint, or process-launching code.

class CommandNodeKind(*values)

Bases: StrEnum

The position a command node occupies in an outer command graph.

ROOT

GROUP

LEAF

class ParameterKind(*values)

Bases: StrEnum

Whether a command input is positional or option-shaped.

ARGUMENT

OPTION

class JsonType(*values)

Bases: StrEnum

The JSON scalar represented by a command input.

STRING

INTEGER

NUMBER

BOOLEAN

class CommandWriteRoute(*values)

Bases: StrEnum

The storage scope through which a command may write.

NONE

PROFILE_BOUND

BOOTSTRAP_ROOT

class ProfileAuthenticationPosture(*values)

Bases: StrEnum

The authentication posture declared by a command registration.

NOT_APPLICABLE

RESUME_FALLBACK

SELF_AUTHENTICATING

class MachineSecretPresence(*values)

Bases: StrEnum

Presence condition for a machine-secret payload variant.

ABSENT

PRESENT

class CommandCapabilityClass(capabilities, side_effects, performance)

Bases: object

Capability/effect/performance classification supplied by the graph.

Parameters:

• capabilities (frozenset[Capability])

• side_effects (frozenset[SideEffect])

• performance (PerformanceClass)

capabilities: frozenset[TypeAliasType]

side_effects: frozenset[TypeAliasType]

performance: TypeAliasType

property expanded_capabilities: frozenset[Capability]

Return the transitive capability implications used by policy checks.

class CommandParameterMetadata(name, kind, cli_flag, off_flag, json_type, required, is_flag, multiple, choices, default, help)

Bases: object

One language-specific parameter projection from a command declaration.

Parameters:

• name (str)

• kind (ParameterKind)

• cli_flag (str)

• off_flag (str)

• json_type (JsonType)

• required (bool)

• is_flag (bool)

• multiple (bool)

• choices (tuple[str, ...])

• default (CommandParameterDefault)

• help (str)

name: str

kind: ParameterKind

cli_flag: str

off_flag: str

json_type: JsonType

required: bool

is_flag: bool

multiple: bool

choices: tuple[str, ...]

default: TypeAliasType

help: str

class MachineSecretFieldMetadata(name, json_type)

Bases: object

Value-free field shape for one machine-secret payload.

Parameters:

• name (str)

• json_type (Literal['string'])

name: str

json_type: Literal['string']

class MachineSecretVariantConditionMetadata(option_name, presence)

Bases: object

Condition selecting one machine-secret payload variant.

Parameters:

• option_name (str)

• presence (MachineSecretPresenceValue)

option_name: str

presence: TypeAliasType

class MachineSecretPayloadMetadata(key, fields, condition, maximum_bytes, same_scope_exclusive, duplicate_keys_forbidden, extra_fields_forbidden)

Bases: object

Value-free machine-secret payload contract exposed to a consumer.

Parameters:

• key (str)

• fields (tuple[MachineSecretFieldMetadata, ...])

• condition (MachineSecretVariantConditionMetadata | None)

• maximum_bytes (int)

• same_scope_exclusive (bool)

• duplicate_keys_forbidden (bool)

• extra_fields_forbidden (bool)

key: str

fields: tuple[MachineSecretFieldMetadata, ...]

condition: MachineSecretVariantConditionMetadata | None

maximum_bytes: int

same_scope_exclusive: bool

duplicate_keys_forbidden: bool

extra_fields_forbidden: bool

class ProfileAuthenticationContractMetadata(fields, maximum_bytes, same_scope_exclusive, stdin_exclusive_across_scopes, descriptors_must_differ_across_scopes, duplicate_keys_forbidden, extra_fields_forbidden)

Bases: object

Value-free profile-authentication shape and collision rules.

Parameters:

• fields (tuple[MachineSecretFieldMetadata, ...])

• maximum_bytes (int)

• same_scope_exclusive (bool)

• stdin_exclusive_across_scopes (bool)

• descriptors_must_differ_across_scopes (bool)

• duplicate_keys_forbidden (bool)

• extra_fields_forbidden (bool)

fields: tuple[MachineSecretFieldMetadata, ...]

maximum_bytes: int

same_scope_exclusive: bool

stdin_exclusive_across_scopes: bool

descriptors_must_differ_across_scopes: bool

duplicate_keys_forbidden: bool

extra_fields_forbidden: bool

class CommandPolicyMetadata(capabilities, side_effects, performance, write_route, destructive, handoff, live_write)

Bases: object

Graph declaration from which an execution policy is projected.

Parameters:

• capabilities (frozenset[Capability])

• side_effects (frozenset[SideEffect])

• performance (PerformanceClass)

• write_route (CommandWriteRouteValue)

• destructive (bool)

• handoff (bool)

• live_write (bool)

capabilities: frozenset[TypeAliasType]

side_effects: frozenset[TypeAliasType]

performance: TypeAliasType

write_route: TypeAliasType

destructive: bool

handoff: bool

live_write: bool

class CommandRegistrationMetadata(command, schema_name, schema_owner, schema_source_sha256, cli_path, parameters_by_language, help_by_language, hidden, policy, handler_owner, source_sha256, machine_secret_payloads=(), profile_authentication=ProfileAuthenticationPosture.NOT_APPLICABLE)

Bases: object

One command registration projected without an entrypoint object.

Parameters:

• command (str)

• schema_name (str)

• schema_owner (str)

• schema_source_sha256 (str)

• cli_path (tuple[str, ...] | None)

• parameters_by_language (tuple[tuple[str, tuple[CommandParameterMetadata, ...] | None], ...])

• help_by_language (tuple[tuple[str, str], ...])

• hidden (bool | None)

• policy (CommandPolicyMetadata | None)

• handler_owner (str | None)

• source_sha256 (str | None)

• machine_secret_payloads (tuple[MachineSecretPayloadMetadata, ...])

• profile_authentication (ProfileAuthenticationPostureValue)

command: str

schema_name: str

schema_owner: str

schema_source_sha256: str

cli_path: tuple[str, ...] | None

parameters_by_language: tuple[tuple[str, tuple[CommandParameterMetadata, ...] | None], ...]

help_by_language: tuple[tuple[str, str], ...]

hidden: bool | None

policy: CommandPolicyMetadata | None

handler_owner: str | None

source_sha256: str | None

machine_secret_payloads: tuple[MachineSecretPayloadMetadata, ...]

profile_authentication: TypeAliasType

property help: dict[str, str]

Return the language-indexed help projection.

property parameters: dict[str, tuple[CommandParameterMetadata, ...] | None]

Return the language-indexed parameter projection.

class LiveNodeRegistrationMetadata(path, kind, loader_owner, handler_owner, source_sha256, policy)

Bases: object

One live command-tree node projected by an outer graph adapter.

Parameters:

• path (tuple[str, ...])

• kind (CommandNodeKind)

• loader_owner (str | None)

• handler_owner (str)

• source_sha256 (str | None)

• policy (CommandPolicyMetadata | None)

path: tuple[str, ...]

kind: CommandNodeKind

loader_owner: str | None

handler_owner: str

source_sha256: str | None

policy: CommandPolicyMetadata | None

class CommandRegistrationProjection(commands, nodes, profile_authentication_contract)

Bases: object

Complete registration projection shared by outer consumers.

Parameters:

• commands (tuple[CommandRegistrationMetadata, ...])

• nodes (tuple[LiveNodeRegistrationMetadata, ...])

• profile_authentication_contract (ProfileAuthenticationContractMetadata)

commands: tuple[CommandRegistrationMetadata, ...]

nodes: tuple[LiveNodeRegistrationMetadata, ...]

profile_authentication_contract: ProfileAuthenticationContractMetadata

class CommandExecutionPolicy(classification, write_route, destructive=False, handoff=False, live_write=False)

Bases: object

Validated immutable execution policy returned by the command port.

Parameters:

• classification (CommandCapabilityClass)

• write_route (CommandWriteRouteValue)

• destructive (bool)

• handoff (bool)

• live_write (bool)

classification: CommandCapabilityClass

write_route: TypeAliasType

destructive: bool

handoff: bool

live_write: bool

class RecoveryHandoffContract(**data)

Bases: BaseModel

Value-free machine discovery for a two-way secret handoff.

Parameters:

data (Any)

handoff_option: str

handoff_direction: Literal['write']

verification_option: str

verification_direction: Literal['read']

required_together: bool

json_fields: tuple[str, ...]

maximum_bytes: int

strict_utf8_object: bool

duplicate_extra_missing_fields_refused: bool

descriptors_closed: bool

reserved_descriptors: tuple[int, ...]

descriptors_must_differ: bool

collides_with: tuple[str, ...]

windows_handle_bootstrap: str

class VerbLeafKind(*values)

Bases: StrEnum

Kind of resolved outer command leaf.

COMMAND

CALLBACK

class VerbParameter(**data)

Bases: BaseModel

One input parameter in an application-owned verb schema.

Parameters:

data (Any)

name: str

kind: ParameterKind

cli_flag: str

off_flag: str

json_type: JsonType

required: bool

is_flag: bool

multiple: bool

choices: tuple[str, ...]

default: bool | int | float | str | list[Any] | None

help: str

property_schema()

Project this parameter into the consumer-facing JSON schema.

Return type:

dict[str, Any]

class ResolvedVerbLeaf(**data)

Bases: BaseModel

One command key resolved to its canonical outer path.

Parameters:

data (Any)

subject_leaf_key: str

cli_path: tuple[str, ...]

alias_paths: tuple[tuple[str, ...], ...]

kind: VerbLeafKind

class VerbLeafResolutionFailure(**data)

Bases: BaseModel

One failed command-key-to-schema resolution.

Parameters:

data (Any)

subject_leaf_key: str

attempted_cli_path: tuple[str, ...]

resolved_cli_path: tuple[str, ...]

reason: str

class VerbInputSchema(**data)

Bases: BaseModel

Application-owned input schema supplied by an outer command adapter.

Parameters:

data (Any)

command_key: str

cli_path: tuple[str, ...]

parameters: tuple[VerbParameter, ...]

machine_secret_payloads: tuple[MachineSecretPayloadMetadata, ...]

recovery_handoff_contract: RecoveryHandoffContract | None

profile_authentication: ProfileAuthenticationPostureValue

profile_authentication_contract: ProfileAuthenticationContractMetadata

help: str

property resolved_leaf: ResolvedVerbLeaf

Return the canonical command-leaf identity represented by this schema.

property required_inputs: tuple[VerbParameter, ...]

Return parameters that must be supplied by a caller.

json_schema()

Project the input contract into a strict JSON object schema.

Return type:

dict[str, Any]

exception SchemaResolutionError(failures)

Bases: CadrumoError

Raised by an outer adapter when its live schema projection is incomplete.

Parameters:

failures (tuple[VerbLeafResolutionFailure, ...])

Return type:

None

code

cli_argv_for(schema, arguments)

Encode named schema arguments into the canonical command argv tail.

Return type:

list[str]

Parameters:

• schema (VerbInputSchema)

• arguments (Mapping[str, object])

class CommandSchemaPort(*args, **kwargs)

Bases: Protocol

Application boundary for an outer command graph’s schema projection.

command_schema_refs()

Return every registered result-schema identity.

Return type:

tuple[CommandSchemaRef, ...]

command_schema_type(command)

Return the registered result schema for one command identity.

Return type:

TypeAliasType

Parameters:

command (str)

command_schema_types()

Return the immutable command-to-result-schema projection.

Return type:

Mapping[str, TypeAliasType]

command_registration_projection()

Return the complete registration metadata projection.

Return type:

CommandRegistrationProjection

build_verb_input_schemas(command_keys)

Build input schemas for the requested command identities.

Return type:

Mapping[str, VerbInputSchema]

Parameters:

command_keys (tuple[str, ...])

class CommandMetadataPort(*args, **kwargs)

Bases: Protocol

Application boundary for command identity and exposure metadata.

cli_path_for_command_key(command_key)

Resolve one command identity to its canonical path.

Return type:

tuple[str, ...]

Parameters:

command_key (str)

is_exposable_command(command_key)

Return whether an identity is callable on the external surface.

Return type:

bool

Parameters:

command_key (str)

command_search_terms(command_key)

Return graph-authored semantic search terms for one command.

Return type:

tuple[str, ...]

Parameters:

command_key (str)

global_flags()

Return root options accepted alongside every command path.

Return type:

frozenset[str]

class CommandPolicyPort(*args, **kwargs)

Bases: Protocol

Application boundary for policy lookup by canonical command path.

command_execution_policy_for_cli_path(cli_path)

Return type:

CommandExecutionPolicy

Parameters:

cli_path (tuple[str, ...])

class CommandSurfacePort(*args, **kwargs)

Bases: CommandSchemaPort, CommandMetadataPort, CommandPolicyPort, Protocol

Complete read-only command surface consumed by the retained harness.

assert_schema_coverage(resolution_errors)

Raise one typed error when an outer adapter cannot project a leaf.

Return type:

None

Parameters:

resolution_errors (tuple[VerbLeafResolutionFailure, ...])