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.
Contract:
- The index path: answers the live object of a published record of
the index, whichever version that record names. The lookups by
spell_id and by source id name one version and do not come through
here; all three share `_meld_live_spell_from_records_locked`.
- Requires the caller to hold `self._lock`.
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,
)
)
return self._meld_live_spell_from_records_locked(
matching_spell_records,
subject="Spell index '{0}'".format(spell_index_id),
frame_name=frame_name,
)
def _meld_live_spell_from_records_locked(
self,
spell_records: list[SpellRecord],
*,
subject: str,
frame_name: str,
) -> object:
"""
Return the already-live object of the first of `spell_records` that has
one, while the command lock is already held.
Purpose:
One reuse-only loop for the static room's three lookups: by spell index
id (every record of the index), by spell_id (the records published for
that spell_id) and by source id (the one record it names). Each record
is melded by its own spell_id, so the answer is always a version the
caller named - or, on the index path, one the index's records name.
Contract:
- Skips records whose existence keeps no single stored object (`many`,
`unique_per_spell_space`, `managed_spell_quantity`) and records with
no owner conduit; melds every other record through its owner root
with `meld_existing_spell(spell=<the record's spell_id>)`, which
builds nothing, and skips a record its owner refuses (ValueError: no
live object).
- Requires the caller to hold `self._lock`; the frame and spell ACL
checks are the caller's.
Args:
spell_records:
Published records to try, in order.
subject:
What the caller asked for, as the refusals name it (for example
"Spell index '<id>'" or "Spell id '<id>'").
frame_name:
Resolved hosted frame name.
Returns:
object: Already-live runtime object of the first record that has one.
Raises:
ValueError:
"<subject> uses unsupported static existence '<name>'." when no
record had a live object and one of them uses an unsupported
existence; otherwise "<subject> is not live in frame '<frame>'.".
"""
for spell_record in spell_records:
# No single stored object to hand out: `many` builds anew, a spellspace object is request-local, and a
# managed spell's objects are leased from a pool (2026-10-15).
if spell_record.existence in {
Existence.many,
Existence.unique_per_spell_space,
Existence.managed_spell_quantity,
}:
continue
owner_conduit_id = spell_record.owner_conduit_id
if not owner_conduit_id:
continue
# Spell owners are roots.
owner_conduit = self._aether.get_root_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 spell_records
if spell_record.existence in {
Existence.many,
Existence.unique_per_spell_space,
Existence.managed_spell_quantity,
}
),
None,
)
if unsupported_spell_record is not None:
raise ValueError(
"{0} uses unsupported static existence '{1}'.".format(
subject,
unsupported_spell_record.existence.name,
)
)
raise ValueError(
"{0} is not live in frame '{1}'.".format(
subject,
frame_name,
)
)
[docs]
def get_spell_by_source_id(
self,
spell_source_id: str,
*,
frame_name: Optional[str] = None,
) -> object:
"""
Return the already-live object of the spell a published source id names.
Contract:
- Melds only the spell the source id names, by its spell_id, through
its owner root (`meld_existing_spell`, reuse-only); never the live
object of another version of its index. Until 0.2.8253 this
resolved the record's index and returned the live object of any
published record of it.
- Enforces the frame command gate and the record's spell-index ACL
before touching runtime state.
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 source id is empty, malformed, not published or
ambiguous in the selected frame, if command ACL denies frame or
spell access, if its existence keeps no single stored object, or
if it 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,
)
self._assert_frame_command_enabled(resolved_frame_name)
self._assert_spell_command_enabled(
spell_record.spell_index_id,
frame_name=resolved_frame_name,
)
return self._meld_live_spell_from_records_locked(
[spell_record],
subject="Spell '{0}'".format(spell_source_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_spell_id(
self,
spell_id: str,
*,
frame_name: Optional[str] = None,
) -> object:
"""
Return the already-live object of the spell that owns a spell_id.
Contract:
- Melds only the records published for that spell_id (one per
Spellbook that publishes it), each by the spell_id, through its
owner root (`meld_existing_spell`, reuse-only); never the live object
of another version of its index. Renamed from `get_spell_by_id` in
0.2.8253, which resolved the spell_id's index and returned the live
object of any published record of it.
- Enforces the frame command gate and the spell-index ACL before
touching runtime state.
Args:
spell_id:
The 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_id is empty, not published or published under more
than one index in the selected frame, if command ACL denies frame
or spell access, if its existence keeps no single stored object, or
if it is not currently live.
"""
with self._entered_command_action(
action_name="get_spell_by_spell_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,
)
self._assert_frame_command_enabled(resolved_frame_name)
self._assert_spell_command_enabled(
spell_index_id,
frame_name=resolved_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
]
return self._meld_live_spell_from_records_locked(
matching_spell_records,
subject="Spell id '{0}'".format(spell_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,
Existence.managed_spell_quantity,
}
is_live = False
owner_conduit_id = spell_record.owner_conduit_id
if owner_conduit_id:
# Spell owners are roots.
owner_conduit = self._aether.get_root_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
)