from typing import TYPE_CHECKING, Optional
from melder.nexus.rift.command_system.command_system import (
CommandSystem,
)
from melder.aether.spellbook.existence.existence import Existence
if TYPE_CHECKING:
from melder.nexus.frame_descriptor.spell_record import SpellRecord
[docs]
class StaticCommandSystem(CommandSystem):
"""
Internal
Static-room command surface.
Purpose:
Keep the shared command infrastructure while owning the static-safe
command surface for `StaticRiftSpace`.
Contract:
- Inherits all shared selected-target, ACL, and workstation behavior
from `CommandSystem`.
- Owns live-only spell runtime retrieval under the shared getter names.
- Owns static-specific spell status and reuse-only command helpers.
- Does not expose topology mutation or direct `meld(...)` because those
methods now live on the capability surface instead of being denied
after inheritance.
- Leaves already-bound workstation objects outside post-bind policing.
Registration:
MELDER KERNEL - guarded. Built by `StaticRiftSpace` through the room's
command-system factory seam.
Subsystem Context:
The static posture of the command family, beside
`CapabilityCommandSystem` and `CodegenCommandSystem`. All three inherit
shared infrastructure from `CommandSystem` and differ only in the
vocabulary they own.
System Context:
The third contract line records a genuine architectural correction and
is worth reading closely: topology mutation and direct `meld(...)` are
not DENIED here after inheritance - they simply LIVE ON the capability
surface instead. That difference matters. Inheriting a dangerous method
and then refusing it means the method exists on the object, appears in
introspection, and must be defended at every call site. Never inheriting
it means the static surface is honestly narrow, and agents enumerating
the room's commands see the truth rather than a list of things that will
reject them.
"Leaves already-bound workstation objects outside post-bind policing" is
the matching honesty: once an object is in the canvas it is the caller's,
and pretending to police it afterwards would promise a containment this
room cannot actually enforce.
AGENT_ACCESS: internal
AGENT_PURPOSE:
access: internal. Static-room command surface. Melder kernel machinery: read it to
understand the runtime, do not drive it directly.
"""
_STATIC_COMMAND_METHOD_NAMES: tuple[str, ...] = (
"meld_existing_spell",
"describe_spell_status_by_source_id",
"describe_spell_status_by_id",
"describe_spell_status_by_index_id",
)
@staticmethod
def _raise_static_runtime_helper_absent(method_name: str) -> None:
"""
Raise the static-room absent-method contract for direct conduit helpers.
Purpose:
Keep static rooms aligned to the published command surface by
making capability-only conduit-object helpers absent at runtime.
Contract:
- Always raises `AttributeError`.
- Uses the requested method name in the failure message.
Args:
method_name:
Public helper name that static rooms must not expose.
Returns:
None.
Raises:
AttributeError:
Always, because static rooms do not expose the helper.
"""
raise AttributeError(
"Static command surface does not expose '{0}'.".format(method_name)
)
[docs]
def get_conduit_cloud(
self,
*,
frame_name: Optional[str] = None,
) -> None:
"""
Static rooms do not expose the live conduit-cloud runtime object.
Args:
frame_name:
Unused static-room frame selector.
Returns:
None:
Raises:
AttributeError:
Always, because static rooms do not expose direct conduit-cloud
runtime access.
"""
self._raise_static_runtime_helper_absent("get_conduit_cloud")
[docs]
def get_conduit_by_id(
self,
conduit_id: str,
*,
frame_name: Optional[str] = None,
) -> None:
"""
Static rooms do not expose live conduit runtime objects by id.
Args:
conduit_id:
Unused conduit id.
frame_name:
Unused static-room frame selector.
Returns:
None:
Raises:
AttributeError:
Always, because static rooms do not expose direct conduit
runtime-object access.
"""
self._raise_static_runtime_helper_absent("get_conduit_by_id")
[docs]
def get_conduit_by_name(
self,
conduit_name: str,
*,
frame_name: Optional[str] = None,
) -> None:
"""
Static rooms do not expose live conduit runtime objects by name.
Args:
conduit_name:
Unused conduit name.
frame_name:
Unused static-room frame selector.
Returns:
None:
Raises:
AttributeError:
Always, because static rooms do not expose direct conduit
runtime-object access.
"""
self._raise_static_runtime_helper_absent("get_conduit_by_name")
def _get_spell_by_index_id_locked(
self,
spell_index_id: str,
*,
frame_name: str,
) -> object:
"""
Resolve one static-room spell runtime object by stable spell-index id while
the command lock is already held.
Args:
spell_index_id:
Stable SpellIndex id to resolve.
frame_name:
Resolved hosted frame name.
Returns:
object: Already-live spell runtime object.
Raises:
ValueError:
If the spell-index id is empty, unpublished, command-disabled, uses
unsupported static existence, or is not currently live.
"""
if not spell_index_id:
raise ValueError("spell_index_id cannot be empty.")
self._assert_frame_command_enabled(frame_name)
self._assert_spell_command_enabled(
spell_index_id,
frame_name=frame_name,
)
descriptor = self._rift._get_required_command_projection(
frame_name
).frame_descriptor
matching_spell_records = [
spell_record
for spell_record in descriptor.spell_records_by_key.values()
if spell_record.spell_index_id == spell_index_id
]
if len(matching_spell_records) == 0:
raise ValueError(
"Spell index id '{0}' was not found in frame '{1}'.".format(
spell_index_id,
frame_name,
)
)
for spell_record in matching_spell_records:
if spell_record.existence in {
Existence.many,
Existence.unique_per_spell_space,
}:
continue
owner_conduit_id = spell_record.owner_conduit_id
if not owner_conduit_id:
continue
owner_conduit = self._aether._get_conduit_by_id(
owner_conduit_id,
frame_name,
)
try:
return owner_conduit.meld_existing_spell(
spell=spell_record.spell_id,
)
except ValueError:
continue
unsupported_spell_record = next(
(
spell_record
for spell_record in matching_spell_records
if spell_record.existence in {
Existence.many,
Existence.unique_per_spell_space,
}
),
None,
)
if unsupported_spell_record is not None:
raise ValueError(
"Spell index '{0}' uses unsupported static existence '{1}'.".format(
spell_index_id,
unsupported_spell_record.existence.name,
)
)
raise ValueError(
"Spell index '{0}' is not live in frame '{1}'.".format(
spell_index_id,
frame_name,
)
)
[docs]
def get_spell_by_source_id(
self,
spell_source_id: str,
*,
frame_name: Optional[str] = None,
) -> object:
"""
Return one already-live spell runtime object using a published spell source id.
Args:
spell_source_id:
Published spell source id in `spellbook_id:spell_id` form.
frame_name:
Optional frame name. When omitted, the room default frame is
used.
Returns:
object: Already-live spell runtime object.
Raises:
ValueError:
If the spell is not published in the selected frame or is not
currently live.
"""
with self._entered_command_action(
action_name="get_spell_by_source_id",
frame_name=frame_name,
), self._lock:
resolved_frame_name = self._resolve_runtime_frame_name(frame_name)
spell_record = self._get_required_published_spell_record_by_source_id(
spell_source_id,
frame_name=resolved_frame_name,
)
return self._get_spell_by_index_id_locked(
spell_record.spell_index_id,
frame_name=resolved_frame_name,
)
[docs]
def describe_spell_status_by_source_id(
self,
spell_source_id: str,
*,
frame_name: Optional[str] = None,
) -> dict:
"""
Return static availability status for one published spell source id.
Args:
spell_source_id:
Published spell source id in `spellbook_id:spell_id` form.
frame_name:
Optional frame name. When omitted, the room default frame is
used.
Returns:
dict: Static spell status payload.
"""
with self._entered_command_action(
action_name="describe_spell_status_by_source_id",
frame_name=frame_name,
), self._lock:
resolved_frame_name = self._resolve_runtime_frame_name(frame_name)
if not spell_source_id:
raise ValueError("spell_source_id cannot be empty.")
spellbook_id, spell_id = spell_source_id.split(":", 1)
descriptor = self._rift._get_required_command_projection(
resolved_frame_name
).frame_descriptor
matching_spell_records = [
spell_record
for spell_record in descriptor.spell_records_by_key.values()
if (
spell_record.origin_spellbook_id == spellbook_id
and spell_record.spell_id == spell_id
)
]
if len(matching_spell_records) == 0:
return {
"frame_name": resolved_frame_name,
"spell_source_id": spell_source_id,
"is_published": False,
"is_command_enabled": False,
"is_static_supported": False,
"is_live": False,
"is_available": False,
"reason": "not_published",
}
if len(matching_spell_records) > 1:
return {
"frame_name": resolved_frame_name,
"spell_source_id": spell_source_id,
"is_published": True,
"is_command_enabled": False,
"is_static_supported": False,
"is_live": False,
"is_available": False,
"reason": "ambiguous_spell_source_id",
}
return self._describe_static_spell_status(
matching_spell_records[0],
frame_name=resolved_frame_name,
)
[docs]
def describe_spell_status_by_id(
self,
spell_id: str,
*,
frame_name: Optional[str] = None,
) -> dict:
"""
Return static availability status for one current spell id.
Args:
spell_id:
Current spell id to resolve.
frame_name:
Optional frame name. When omitted, the room default frame is
used.
Returns:
dict: Static spell status payload.
"""
with self._entered_command_action(
action_name="describe_spell_status_by_id",
frame_name=frame_name,
), self._lock:
resolved_frame_name = self._resolve_runtime_frame_name(frame_name)
descriptor = self._rift._get_required_command_projection(
resolved_frame_name
).frame_descriptor
matching_spell_records = [
spell_record
for spell_record in descriptor.spell_records_by_key.values()
if spell_record.spell_id == spell_id
]
if len(matching_spell_records) == 0:
return {
"frame_name": resolved_frame_name,
"spell_id": spell_id,
"is_published": False,
"is_command_enabled": False,
"is_static_supported": False,
"is_live": False,
"is_available": False,
"reason": "not_published",
}
if len(matching_spell_records) > 1:
return {
"frame_name": resolved_frame_name,
"spell_id": spell_id,
"is_published": True,
"is_command_enabled": False,
"is_static_supported": False,
"is_live": False,
"is_available": False,
"reason": "ambiguous_spell_id",
}
return self._describe_static_spell_status(
matching_spell_records[0],
frame_name=resolved_frame_name,
)
[docs]
def describe_spell_status_by_index_id(
self,
spell_index_id: str,
*,
frame_name: Optional[str] = None,
) -> dict:
"""
Return static availability status for one stable spell-index id.
Args:
spell_index_id:
Stable SpellIndex id to resolve.
frame_name:
Optional frame name. When omitted, the room default frame is
used.
Returns:
dict: Static spell status payload.
"""
with self._entered_command_action(
action_name="describe_spell_status_by_index_id",
frame_name=frame_name,
), self._lock:
resolved_frame_name = self._resolve_runtime_frame_name(frame_name)
descriptor = self._rift._get_required_command_projection(
resolved_frame_name
).frame_descriptor
matching_spell_records = [
spell_record
for spell_record in descriptor.spell_records_by_key.values()
if spell_record.spell_index_id == spell_index_id
]
if len(matching_spell_records) == 0:
return {
"frame_name": resolved_frame_name,
"spell_index_id": spell_index_id,
"is_published": False,
"is_command_enabled": False,
"is_static_supported": False,
"is_live": False,
"is_available": False,
"reason": "not_published",
}
if len(matching_spell_records) > 1:
return {
"frame_name": resolved_frame_name,
"spell_index_id": spell_index_id,
"is_published": True,
"is_command_enabled": False,
"is_static_supported": False,
"is_live": False,
"is_available": False,
"reason": "ambiguous_spell_index_id",
}
return self._describe_static_spell_status(
matching_spell_records[0],
frame_name=resolved_frame_name,
)
[docs]
def get_spell_by_index_id(
self,
spell_index_id: str,
*,
frame_name: Optional[str] = None,
) -> object:
"""
Return one already-live spell runtime object by stable spell-index id.
Args:
spell_index_id:
Stable SpellIndex id to resolve.
frame_name:
Optional frame name. When omitted, the room default frame is
used.
Returns:
object: Already-live spell runtime object.
Raises:
ValueError:
If the spell index is not published in the selected frame or
does not currently have a live creation.
"""
self.check_cleaned()
with self._entered_command_action(
action_name="get_spell_by_index_id",
frame_name=frame_name,
), self._lock:
resolved_frame_name = self._resolve_runtime_frame_name(frame_name)
return self._get_spell_by_index_id_locked(
spell_index_id,
frame_name=resolved_frame_name,
)
[docs]
def get_spell_by_id(
self,
spell_id: str,
*,
frame_name: Optional[str] = None,
) -> object:
"""
Return one already-live spell runtime object by current spell id.
Args:
spell_id:
Current spell id to resolve.
frame_name:
Optional frame name. When omitted, the room default frame is
used.
Returns:
object: Already-live spell runtime object.
Raises:
ValueError:
If the spell is not published in the selected frame or does
not currently have a live creation.
"""
with self._entered_command_action(
action_name="get_spell_by_id",
frame_name=frame_name,
), self._lock:
resolved_frame_name = self._resolve_runtime_frame_name(frame_name)
spell_index_id = self._get_required_published_spell_index_id_by_spell_id(
spell_id,
frame_name=resolved_frame_name,
)
return self._get_spell_by_index_id_locked(
spell_index_id,
frame_name=resolved_frame_name,
)
[docs]
def meld_existing_spell(
self,
conduit_id: str,
spell_name: Optional[str] = None,
*,
spell: Optional[object] = None,
spellframe: Optional[object] = None,
binding_name: Optional[str] = None,
frame_name: Optional[str] = None,
) -> object:
"""
Return one already-live spell runtime object through a selected conduit.
Purpose:
Keep the reuse-only spell-activation helper on the static command
surface without inheriting the broader capability activation set.
Args:
conduit_id:
Conduit id that should perform the reuse-only resolution.
spell_name:
Optional logical spell name key.
spell:
Optional spell id string or spell object.
spellframe:
Optional spellframe / protocol / frame key.
binding_name:
Optional binding name for resolution.
frame_name:
Optional hosted frame name. When omitted, the room default
frame is used.
Returns:
object: Already-live runtime object returned by the conduit.
"""
self.check_cleaned()
with self._entered_command_action(
action_name="meld_existing_spell",
frame_name=frame_name,
), self._lock:
resolved_frame_name = self._resolve_runtime_frame_name(frame_name)
conduit = self._get_conduit_by_id_locked(
conduit_id,
frame_name=resolved_frame_name,
)
return conduit.meld_existing_spell(
spell_name=spell_name,
spell=spell,
spellframe=spellframe,
binding_name=binding_name,
)
def _describe_static_spell_status(
self,
spell_record: SpellRecord,
*,
frame_name: str,
) -> dict:
"""
Build one static spell status payload from a published spell record.
Args:
spell_record:
Published spell record to inspect.
frame_name:
Hosted frame name.
Returns:
dict: Static spell status payload.
"""
is_command_enabled = False
try:
self._assert_frame_command_enabled(frame_name)
self._assert_spell_command_enabled(
spell_record.spell_index_id,
frame_name=frame_name,
)
is_command_enabled = True
except ValueError:
is_command_enabled = False
is_static_supported = spell_record.existence not in {
Existence.many,
Existence.unique_per_spell_space,
}
is_live = False
owner_conduit_id = spell_record.owner_conduit_id
if owner_conduit_id:
owner_conduit = self._aether._get_conduit_by_id(
owner_conduit_id,
frame_name,
)
is_live = owner_conduit.has_live_creation(spell=spell_record.spell_id)
if not is_command_enabled:
reason = "command_disabled"
elif not is_static_supported:
reason = "unsupported_static_existence"
elif not is_live:
reason = "not_live"
else:
reason = "available"
return {
"frame_name": frame_name,
"spell_source_id": "{0}:{1}".format(
spell_record.origin_spellbook_id,
spell_record.spell_id,
),
"spell_id": spell_record.spell_id,
"spell_index_id": spell_record.spell_index_id,
"spell_name": spell_record.spell_name,
"binding_name": spell_record.binding_name,
"existence": spell_record.existence.name,
"is_published": True,
"is_command_enabled": is_command_enabled,
"is_static_supported": is_static_supported,
"is_live": is_live,
"is_available": (
is_command_enabled and is_static_supported and is_live
),
"reason": reason,
}
[docs]
def list_supported_command_methods(self) -> tuple[str, ...]:
"""
Return the public command methods supported by static rooms.
Returns:
tuple[str, ...]: Shared command names plus static-owned helper
names in stable presentation order.
"""
self.check_cleaned()
with self._entered_command_action(
action_name="list_supported_command_methods",
frame_name=None,
):
return self._list_supported_command_methods_tuple() + (
self._STATIC_COMMAND_METHOD_NAMES
)