import threading
from contextlib import contextmanager
from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Tuple
if TYPE_CHECKING:
from melder.aether.conduit.conduit import Conduit
from melder.aether.spellbook.spell import Spell
from melder.nexus.acl.frame_acl_compiled_access_surface import (
CompiledFrameACLAccessSurface,
)
from melder.nexus.frame_descriptor.spell_record import SpellRecord
from melder.nexus.rift.rift import Rift
from melder.nexus.rift.rift_space.rift_space import RiftSpace
from melder.nexus.rift.rift_space.workstation import Workstation
from melder.aether.aether import Aether
from melder.utilities.general_base.cleanable import Cleanable
from melder.utilities.helpers.id_builder import IDBuilder
[docs]
class CommandSystem(Cleanable):
"""
Internal
Room-local shared command infrastructure plus common read/target helpers.
Purpose:
Provide shared command infrastructure above the viewer/workstation
split without owning every room-specific command vocabulary.
Contract:
- Uses the owning `Rift` command projections as the shared command
substrate.
- Enforces compiled command ACL state on direct fetch paths before
exposing frame/conduit/spell runtime objects.
- Uses the owned workstation for active-target attribute/method getters
and method execution.
- Does not store results itself. Callers that want persistence must
bind returned values into the workstation explicitly.
- Leaves room-owned topology mutation and spell activation commands to
room-specific subclasses instead of pretending every room owns the
same broad public surface.
Lifecycle:
Owned by one `RiftSpace`. Cleanup drops references to the owning room
and workstation but does not clean those children itself.
Threading:
Room-confined. It holds no cross-room state and takes no lock of its
own; the ACL projections it enforces against are refreshed by the owning
Rift.
Registration:
MELDER KERNEL. The three postures are melder-internal and constructed by their
owning room (for example `CapabilityRiftSpace` builds
`CapabilityCommandSystem`), with no user injection seam.
Subsystem Context:
The MEDIATED command layer above the viewer/workstation split.
`FrameViewer` answers read questions, `Workstation` holds bindings and
the active target, and this class is the controlled surface through
which getters and executes actually run. Room-specific subclasses add
the vocabulary that does not belong to every room.
System Context:
Two rules define this class and both are about refusing convenience.
First, ACL IS ENFORCED ON THE DIRECT FETCH PATH - compiled command ACL
state is checked BEFORE any frame, conduit, or spell runtime object is
exposed. Checking after would mean the object had already escaped.
Second, IT DOES NOT STORE RESULTS. A caller that wants persistence must
bind the returned value into the workstation explicitly. Auto-storing
would make every read silently extend object lifetime and quietly
populate a room's canvas with things nobody chose to keep.
The subclass split exists because pretending every room owns the same
broad public surface is exactly the lie the room modes exist to prevent:
`CapabilityCommandSystem` owns topology mutation and direct activation,
`StaticCommandSystem` owns live-only retrieval and reuse-only
activation, and `CodegenCommandSystem` owns the validate/execute seams
plus the full research command family.
AGENT_ACCESS: internal
AGENT_PURPOSE:
access: internal. Room-local shared command infrastructure plus common read/target
helpers. Melder kernel machinery: read it to understand the runtime, do not drive it
directly.
"""
__slots__ = Cleanable.__slots__ + [
"_id",
"_owner_space_id",
"_lock",
"_rift",
"_space",
"_workstation",
]
_DENIED_RAW_RUNTIME_OBJECT_ACCESS_METHOD_NAMES: frozenset[str] = frozenset()
_RAW_RUNTIME_OBJECT_ACCESS_DENIED_MESSAGE_TEMPLATE: str = (
"Command surface does not allow raw runtime-object access method '{0}'."
)
_aether = Aether()
def __init__(
self,
*,
rift: Rift,
space: RiftSpace,
workstation: Workstation,
) -> None:
"""
Internal
Initialize one room-local command system.
Args:
rift:
Owning `Rift` that manages applied projection state.
space:
Owning `RiftSpace`.
workstation:
Room-local workstation owned by the same space.
Returns:
None.
Raises:
TypeError:
If `space` or `workstation` is None.
"""
super().__init__()
if rift is None:
raise TypeError("rift cannot be None.")
if space is None:
raise TypeError("space cannot be None.")
if workstation is None:
raise TypeError("workstation cannot be None.")
type(self)._aether = Aether()
self._id: str = IDBuilder.create_id()
self._owner_space_id: str = space.space_id
self._lock: threading.RLock = threading.RLock()
self._rift: Rift = rift
self._space: RiftSpace = space
self._workstation: Workstation = workstation
[docs]
def cleanup(self) -> None:
"""
Internal
Idempotently clear command-system-owned references.
Contract:
- Safe to call more than once.
- Clears only command-system-owned references.
- Does not cleanup the owning `RiftSpace` or its workstation,
because those remain owned by the room itself.
Returns:
None.
"""
if self._cleaned:
return
with self._lock:
if self._cleaned:
return
self._cleaned = True
del self._owner_space_id
del self._rift
del self._space
del self._workstation
del self._id
del self._lock
@property
def command_system_id(self) -> str:
"""
Return the stable command-system identifier.
Returns:
str: Stable command-system id.
"""
self.check_cleaned()
with self._lock:
return self._id
@property
def owner_space_id(self) -> str:
"""
Return the owning room identifier.
Returns:
str: Owning `RiftSpace` id.
"""
self.check_cleaned()
with self._lock:
return self._owner_space_id
def _get_conduit_by_id_locked(
self,
conduit_id: str,
*,
frame_name: str,
) -> Conduit:
"""
Resolve one conduit object while the command lock is already held.
Contract:
- Enforces raw-runtime access, frame command enablement, and
conduit-level ACL checks before touching Aether runtime state.
- Resolves through `Aether.get_conduit_by_id`, which covers the
frame's live roots and their attached lesser lineage, named or
anonymous, at any depth.
- Keeps this surface's errors: a missing frame raises the frame
error from `_get_required_runtime_frame`, and an id that is not
live in the frame raises "Conduit id '<id>' was not found in
frame '<frame>'." chained from Aether's error.
- Requires the caller to hold `self._lock`.
Args:
conduit_id:
Conduit id to resolve.
frame_name:
Resolved hosted frame name.
Returns:
object: Live conduit object or matching lesser conduit object.
Raises:
ValueError:
If runtime-object access is denied, the frame/conduit ACL gate
fails, or the conduit cannot be found in the frame.
"""
self._assert_raw_runtime_object_access_allowed("get_conduit_by_id")
self._assert_frame_command_enabled(frame_name)
self._assert_conduit_command_enabled(
conduit_id,
frame_name=frame_name,
)
try:
return self._aether.get_conduit_by_id(
conduit_id,
frame_name,
)
except ValueError as error:
# Re-raise in this surface's wording: the frame error first, else the id error.
self._get_required_runtime_frame(frame_name)
raise ValueError(
"Conduit id '{0}' was not found in frame '{1}'.".format(
conduit_id,
frame_name,
)
) from error
def _get_conduit_by_name_locked(
self,
conduit_name: str,
*,
frame_name: str,
) -> Conduit:
"""
Resolve the exact authorized identity published under a conduit name.
Contract:
The caller holds the command lock. Raw/frame access precedes published
selection; the existing ID resolver enforces the selected ID's ACL and
searches root and lesser runtime scopes. Never resolve a second time
by name: a released name may already identify an unauthorized new ID.
Reject same-shell reuse under a different name. The returned object is
borrowed; this does not grant a lease or prevent subsequent cleanup.
Args:
conduit_name: Exact published name of an active root or lesser scope.
frame_name: Resolved frame whose command projection supplies authority.
Returns:
Conduit: Live scope matching the selected ID and requested name.
Raises:
ValueError: Access denied, name missing/ambiguous, ID retired or the
live name changed during lookup. Refresh/retry against current state.
RuntimeError: The selected conduit was cleaned during the borrowed read.
"""
self._assert_raw_runtime_object_access_allowed("get_conduit_by_name")
self._assert_frame_command_enabled(frame_name)
conduit_id = self._get_required_published_conduit_id_by_name(
conduit_name, frame_name=frame_name,
)
conduit = self._get_conduit_by_id_locked(conduit_id, frame_name=frame_name)
if conduit.name != conduit_name:
raise ValueError(
f"Conduit name '{conduit_name}' changed during lookup in frame '{frame_name}'. "
"Retry discovery against the current named scope."
)
return conduit
def _get_spell_by_index_id_locked(
self,
spell_index_id: str,
*,
frame_name: str,
) -> object:
"""
Resolve one spell runtime object by stable spell-index id while the command
lock is already held.
Contract:
- Enforces raw-runtime access plus frame/spell command ACL checks
before touching runtime conduit state.
- Resolves through the command projection descriptor truth rather
than viewer state.
- Requires the caller to hold `self._lock`.
Args:
spell_index_id:
Stable SpellIndex spell-index id to resolve.
frame_name:
Resolved hosted frame name.
Returns:
object: Live spell runtime object.
Raises:
ValueError:
If the spell-index id is empty, unpublished, command-disabled, or
not found in the owner spellbooks for the frame.
"""
if not spell_index_id:
raise ValueError("spell_index_id cannot be empty.")
self._assert_raw_runtime_object_access_allowed(
"get_spell_by_index_id"
)
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:
owner_conduit_id = spell_record.owner_conduit_id
if not owner_conduit_id:
continue
owner_conduit = self._get_conduit_by_id_locked(
owner_conduit_id,
frame_name=frame_name,
)
spell = owner_conduit.get_spell_by_index_id(spell_index_id)
if spell is not None:
return spell
raise ValueError(
"Spell index id '{0}' was not found in the owner spellbooks for frame '{1}'.".format(
spell_index_id,
frame_name,
)
)
def _get_spell_by_record_locked(
self,
spell_record: SpellRecord,
*,
frame_name: str,
) -> Spell:
"""
Resolve the Spell one published record names while the command lock is
already held.
Purpose:
The exact counterpart of `_get_spell_by_index_id_locked`: a record
names one spell by its spell_id, and this returns that spell - active
or parked - from the Spellbook of the record's owner conduit, never
the spell the record's SpellIndex has active.
Contract:
- Enforces the frame command gate and the spell-index ACL of the
record's index before touching runtime state, then resolves the
owner conduit through `_get_conduit_by_id_locked`, which applies the
conduit ACL.
- Reads the owner's Spellbook with `find_spell_by_spell_id`, a
lock-free read of its two id maps; the Spell returned is borrowed.
- Requires the caller to hold `self._lock`.
Args:
spell_record:
Published spell record whose spell should be returned.
frame_name:
Resolved hosted frame name.
Returns:
Spell: The spell the record's spell_id names.
Raises:
ValueError:
If command ACL denies frame, spell or conduit access, if the record
has no owner conduit, if the owner conduit has no Spellbook, or if
that Spellbook no longer holds the spell_id.
"""
source_id = "{0}:{1}".format(
spell_record.origin_spellbook_id,
spell_record.spell_id,
)
self._assert_frame_command_enabled(frame_name)
self._assert_spell_command_enabled(
spell_record.spell_index_id,
frame_name=frame_name,
)
owner_conduit_id = spell_record.owner_conduit_id
if not owner_conduit_id:
raise ValueError(
"Spell '{0}' has no owner conduit in frame '{1}'.".format(
source_id,
frame_name,
)
)
owner_conduit = self._get_conduit_by_id_locked(
owner_conduit_id,
frame_name=frame_name,
)
spellbook = owner_conduit._spellbook
if spellbook is None:
raise ValueError(
"Owner conduit for spell '{0}' has no spellbook.".format(
spell_record.spell_id
)
)
spell = spellbook.find_spell_by_spell_id(spell_record.spell_id)
if spell is not None:
return spell
raise ValueError(
"Spell '{0}' was not found in the owner spellbook for frame '{1}'.".format(
source_id,
frame_name,
)
)
def _get_target_method_locked(
self,
method_name: str,
) -> Callable[..., object]:
"""
Resolve one callable from the current workstation target while the
command lock is already held.
Contract:
- Requires the caller to hold `self._lock`.
- Enforces the same validation contract as
`get_target_method(...)`.
Args:
method_name:
Method name to resolve from the active workstation target.
Returns:
Callable[..., object]: Bound callable from the current target.
Raises:
ValueError:
If `method_name` is empty.
AttributeError:
If the current target does not expose the requested attribute.
RuntimeError:
If the resolved attribute is not callable.
"""
if not method_name:
raise ValueError("method_name cannot be empty.")
target = self._workstation.get_target()
method = getattr(target, method_name)
if not callable(method):
raise RuntimeError(
"Target attribute '{0}' is not callable.".format(method_name)
)
return method
def _list_supported_command_methods_tuple(self) -> Tuple[str, ...]:
"""
Return the shared stable command-method vocabulary without any gate or
room-policy filtering.
Contract:
- Acts as the single source of truth for the base command surface.
- Returns stable presentation order for discovery/reporting.
- Does not perform gating, memory emission, or room-specific
filtering.
Returns:
Tuple[str, ...]: Shared base command-method names in stable order.
"""
return (
"link_frame",
"get_nexus_frame",
"describe_spells_in_conduit",
"get_resolution_state",
"get_active_spellspace",
"find_spell_id",
"find_spell_key",
"get_spell_permissions",
"snapshot_state",
"get_spell_by_source_id",
"get_spell_by_index_id",
"get_spell_by_spell_id",
"get_target_attribute",
"get_target_method",
"execute_target_method",
)
[docs]
def describe_spells_in_conduit(
self,
conduit_id: str,
*,
frame_name: Optional[str] = None,
) -> List[dict[str, Any]]:
"""
Return the spell description payloads exposed by one conduit.
Purpose:
Provide a structured spell-description surface for one runtime
conduit through the shared command API.
Contract:
- Resolves the conduit through command ACL and descriptor truth
before touching runtime conduit state.
- Returns the lower conduit runtime's current spell description
payloads as a list without additional filtering.
- Uses one top-level command action boundary for gate admission and
memory emission.
Args:
conduit_id:
Conduit id whose spell descriptions should be returned.
frame_name:
Optional frame name. When omitted, the room default frame is
used.
Returns:
List[dict[str, Any]]: Runtime spell description payloads for the
conduit.
"""
self.check_cleaned()
with self._entered_command_action(
action_name="describe_spells_in_conduit",
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.describe_spells_in_conduit()
[docs]
def get_resolution_state(
self,
conduit_id: str,
*,
frame_name: Optional[str] = None,
) -> object:
"""
Return the conduit-scoped resolution state for one conduit.
Purpose:
Expose the lower conduit runtime's current resolution-state object
through the shared command surface.
Contract:
- Resolves the conduit through command ACL and descriptor truth
before touching runtime conduit state.
- Returns the live lower-runtime resolution-state object directly.
- Uses one top-level command action boundary for gate admission and
memory emission.
Args:
conduit_id:
Conduit id whose resolution state should be returned.
frame_name:
Optional frame name. When omitted, the room default frame is
used.
Returns:
object: Live conduit-scoped resolution-state object.
"""
self.check_cleaned()
with self._entered_command_action(
action_name="get_resolution_state",
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.get_resolution_state()
[docs]
def get_active_spellspace(
self,
conduit_id: str,
*,
frame_name: Optional[str] = None,
) -> object:
"""
Return the active spellspace for one conduit, if any.
Purpose:
Expose the lower conduit runtime's active spellspace surface
through the shared command API.
Contract:
- Resolves the conduit through command ACL and descriptor truth
before touching runtime conduit state.
- Returns the currently active spellspace object or whatever the
lower runtime exposes for the no-active-spellspace case.
- Uses one top-level command action boundary for gate admission and
memory emission.
Args:
conduit_id:
Conduit id whose active spellspace should be returned.
frame_name:
Optional frame name. When omitted, the room default frame is
used.
Returns:
object: Active spellspace object or lower-runtime sentinel value.
"""
self.check_cleaned()
with self._entered_command_action(
action_name="get_active_spellspace",
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.get_active_spellspace()
[docs]
def find_spell_id(
self,
conduit_id: str,
spellframe: str,
spell_name: str,
binding_name: str,
*,
frame_name: Optional[str] = None,
) -> object:
"""
Return the current spell id resolved from logical spell identifiers.
Purpose:
Mirror the lower conduit spell-id lookup on the shared command
surface using logical spell identity fields.
Contract:
- Resolves the conduit through command ACL and descriptor truth
before touching runtime conduit state.
- Defers lookup semantics to the lower conduit runtime instead of
re-implementing spell identity matching in the command layer.
- Uses one top-level command action boundary for gate admission and
memory emission.
Args:
conduit_id:
Conduit id whose spell inventory should be queried.
spellframe:
Logical spellframe key.
spell_name:
Logical spell name.
binding_name:
Logical binding name.
frame_name:
Optional frame name. When omitted, the room default frame is
used.
Returns:
object: Current spell id resolved by the lower conduit runtime.
"""
self.check_cleaned()
with self._entered_command_action(
action_name="find_spell_id",
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.find_spell_id(spellframe, spell_name, binding_name)
[docs]
def find_spell_key(
self,
conduit_id: str,
spellframe: str,
spell_name: str,
binding_name: str,
*,
frame_name: Optional[str] = None,
) -> object:
"""
Return the spellbook key resolved from logical spell identifiers.
Purpose:
Mirror the lower conduit spellbook-key lookup on the shared command
surface using logical spell identity fields.
Contract:
- Resolves the conduit through command ACL and descriptor truth
before touching runtime conduit state.
- Defers lookup semantics to the lower conduit runtime instead of
re-implementing spell identity matching in the command layer.
- Uses one top-level command action boundary for gate admission and
memory emission.
Args:
conduit_id:
Conduit id whose spell inventory should be queried.
spellframe:
Logical spellframe key.
spell_name:
Logical spell name.
binding_name:
Logical binding name.
frame_name:
Optional frame name. When omitted, the room default frame is
used.
Returns:
object: Spellbook key resolved by the lower conduit runtime.
"""
self.check_cleaned()
with self._entered_command_action(
action_name="find_spell_key",
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.find_spell_key(spellframe, spell_name, binding_name)
[docs]
def get_spell_permissions(
self,
conduit_id: str,
spell_id: str,
*,
frame_name: Optional[str] = None,
) -> object:
"""
Return the permissions string for one spell inside one conduit.
Purpose:
Expose one lower-runtime spell-permissions lookup through the
shared command surface.
Contract:
- Resolves the conduit through command ACL and descriptor truth
before touching runtime conduit state.
- Returns the exact permissions value exposed by the lower conduit
runtime.
- Uses one top-level command action boundary for gate admission and
memory emission.
Args:
conduit_id:
Conduit id whose spell permissions should be queried.
spell_id:
Current spell id to inspect.
frame_name:
Optional frame name. When omitted, the room default frame is
used.
Returns:
object: Lower-runtime permissions value for the spell.
"""
self.check_cleaned()
with self._entered_command_action(
action_name="get_spell_permissions",
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.get_spell_permissions(spell_id)
[docs]
def snapshot_state(
self,
conduit_id: str,
*,
frame_name: Optional[str] = None,
) -> Dict[str, Any]:
"""
Return a detached snapshot of one conduit state payload.
Purpose:
Expose one lower-runtime conduit state snapshot through the shared
command surface.
Contract:
- Resolves the conduit through command ACL and descriptor truth
before touching runtime conduit state.
- Returns the lower conduit runtime's detached snapshot payload.
- Uses one top-level command action boundary for gate admission and
memory emission.
Args:
conduit_id:
Conduit id whose state snapshot should be returned.
frame_name:
Optional frame name. When omitted, the room default frame is
used.
Returns:
Dict[str, Any]: Detached conduit state snapshot payload.
"""
self.check_cleaned()
with self._entered_command_action(
action_name="snapshot_state",
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.snapshot_state()
[docs]
def get_spell_by_source_id(
self,
spell_source_id: str,
*,
frame_name: Optional[str] = None,
) -> object:
"""
Return the Spell a published spell source id names.
Purpose:
Hand out the runtime Spell behind one published record, addressed the
way the frame viewer addresses it: `spellbook_id:spell_id`.
Contract:
- Answers exactly the spell the source id names, from its owner's
Spellbook (`find_spell_by_spell_id`), never another version of its
SpellIndex; the version an index has active is
`get_spell_by_index_id`'s answer. Nexus publishes the version an
index has active, so a published record names that version. Until
0.2.8253 this method resolved the record's index and returned the
spell the index had active.
- Enforces raw-runtime access, the frame command gate, the record's
spell-index ACL and the owner conduit's ACL before touching runtime
state (`_get_spell_by_record_locked`).
- Uses one top-level command action boundary for gate admission and
memory emission.
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: The live `Spell` the source id names.
Raises:
ValueError:
If the spell source id is empty, malformed, not published or
ambiguous in the selected frame, if command ACL denies frame,
spell or conduit access, if the record has no owner conduit, or
if the owner's Spellbook no longer holds the spell_id.
"""
self.check_cleaned()
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_raw_runtime_object_access_allowed(
"get_spell_by_source_id"
)
return self._get_spell_by_record_locked(
spell_record,
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 live spell object by stable spell index id.
Args:
spell_index_id:
Stable SpellIndex spell-index id to resolve.
frame_name:
Optional frame name. When omitted, the room default frame is
used.
Returns:
object: Live spell object.
Raises:
ValueError:
If the spell index is not published in the selected frame or
command ACL denies spell access.
"""
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 Spell that owns a spell_id.
Purpose:
Hand out the runtime Spell behind one spell_id - the SHA256 `bind`
and `bind_inactive` return - through the shared command surface.
Contract:
- Answers exactly the spell that owns the spell_id, never another
version of its SpellIndex; the version an index has active is
`get_spell_by_index_id`'s answer. The spell_id must be published in
the room's projection of the frame, and Nexus publishes the version
an index has active, so a parked version's spell_id is refused
(ValueError), not answered with the active one. Renamed from
`get_spell_by_id` in 0.2.8253, whose owner-Book lookup answered a
parked spell_id with the spell its index had active.
- Enforces raw-runtime access, the frame command gate and the
spell-index ACL of the published record before touching runtime
state, then reads the owner Spellbook's id maps
(`Spellbook.find_spell_by_spell_id`) without walking its live index
map.
- Uses one top-level command action boundary for gate admission and
memory emission.
Args:
spell_id:
The spell_id to resolve.
frame_name:
Optional frame name. When omitted, the room default frame is
used.
Returns:
object: The live `Spell` that owns the spell_id.
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 no conduit of the frame claims the spell_id,
if the owner conduit has no Spellbook, or if that Spellbook no
longer holds the spell_id.
"""
self.check_cleaned()
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)
self._assert_raw_runtime_object_access_allowed("get_spell_by_spell_id")
self._assert_frame_command_enabled(resolved_frame_name)
spell_index_id = self._get_required_published_spell_index_id_by_spell_id(
spell_id,
frame_name=resolved_frame_name,
)
self._assert_spell_command_enabled(
spell_index_id,
frame_name=resolved_frame_name,
)
owner_conduit = self._aether._get_conduit_by_spell_id(
spell_id,
resolved_frame_name,
)
spellbook = owner_conduit._spellbook
if spellbook is None:
raise ValueError(
"Owner conduit for spell '{0}' has no spellbook.".format(spell_id)
)
# The owner Book's id maps answer the spell_id lookup in O(1) - the spell that owns the
# spell_id, active or parked - without walking its live index map, which a concurrent
# bind or removal could resize mid-walk.
spell = spellbook.find_spell_by_spell_id(spell_id)
if spell is not None:
return spell
raise ValueError(
"Spell id '{0}' was not found in the owner spellbook.".format(spell_id)
)
[docs]
def get_target_attribute(self, attribute_name: str) -> object:
"""
Return one attribute value from the current workstation target.
Args:
attribute_name:
Attribute name to retrieve from the active target.
Returns:
object: Retrieved attribute value.
Raises:
ValueError:
If `attribute_name` is empty.
AttributeError:
If the target does not expose the requested attribute.
"""
self.check_cleaned()
with self._entered_command_action(
action_name="get_target_attribute",
frame_name=None,
), self._lock:
if not attribute_name:
raise ValueError("attribute_name cannot be empty.")
target = self._workstation.get_target()
return getattr(target, attribute_name)
[docs]
def get_target_method(
self,
method_name: str,
) -> Callable[..., object]:
"""
Return one method/callable from the current workstation target.
Args:
method_name:
Method name to retrieve from the active target.
Returns:
Callable[..., object]: Bound callable from the target.
Raises:
ValueError:
If `method_name` is empty.
AttributeError:
If the target does not expose the requested method.
RuntimeError:
If the resolved attribute is not callable.
"""
self.check_cleaned()
with self._entered_command_action(
action_name="get_target_method",
frame_name=None,
), self._lock:
return self._get_target_method_locked(method_name)
[docs]
def execute_target_method(
self,
method_name: str,
*args: Any,
bind_as_name: Optional[str] = None,
bind_as_store: str = "objects",
bind_result_weak_ref: Optional[bool] = None,
**kwargs: Any
) -> object:
"""
Execute one method on the current workstation target.
Args:
method_name:
Method name to execute on the active target.
*args:
Positional arguments passed to the method.
bind_as_name:
Optional workstation binding name for the return value.
bind_as_store:
Store to use when binding the return value.
bind_result_weak_ref:
Optional workstation reference-mode override for the bound
result. `True` forces weak storage, `False` forces strong
storage, and `None` uses the room/workstation default.
**kwargs:
Keyword arguments passed to the method.
Returns:
object: Method return value.
"""
self.check_cleaned()
with self._entered_command_action(
action_name="execute_target_method",
frame_name=None,
), self._lock:
method = self._get_target_method_locked(method_name)
result = method(*args, **kwargs)
if bind_as_name is not None:
self._bind_result(
bind_as_name=bind_as_name,
bind_as_store=bind_as_store,
value=result,
bind_result_weak_ref=bind_result_weak_ref,
)
return result
[docs]
def link_frame(self, frame_name: str) -> None:
"""
Engage one target frame on this Rift through the shared command surface.
Args:
frame_name:
Target frame name to link.
Returns:
None.
"""
self.check_cleaned()
with self._entered_command_action(
action_name="link_frame",
frame_name=frame_name,
), self._lock:
self._rift.create_frame_link(frame_name)
[docs]
def get_nexus_frame(self, frame_name: Optional[str] = None) -> object:
"""
Return one rooted Nexus-managed conduit through the shared command surface.
Args:
frame_name:
Optional explicit Nexus frame name.
Returns:
object: Root conduit for the resolved Nexus-managed frame.
"""
self.check_cleaned()
with self._entered_command_action(
action_name="get_nexus_frame",
frame_name=frame_name,
), self._lock:
return self._rift.get_nexus_frame(frame_name=frame_name)
[docs]
def list_supported_command_methods(self) -> Tuple[str, ...]:
"""
Return the public command methods supported by this room surface.
Purpose:
Give callers a cheap, explicit way to discover the current
command-surface vocabulary instead of guessing from room type or
trial-and-error errors.
Returns:
Tuple[str, ...]: Supported public command method 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()
@contextmanager
def _entered_command_action(
self,
*,
action_name: str,
frame_name: Optional[str],
category: str = "command",
) -> Any:
"""
Enter one explicit top-level public command action.
Args:
action_name:
Stable public command method name that completed successfully.
frame_name:
Optional caller-supplied frame name associated with the action.
Returns:
Any: Context manager that guarantees symmetric RiftGate release and
successful top-level memory emission.
"""
with self._entered_action_hook_scope_if_available(
category=category,
action_name=action_name,
):
rift_gate = self._begin_command_action()
command_succeeded = False
try:
yield
command_succeeded = True
finally:
self._finish_command_action(
rift_gate=rift_gate,
action_name=action_name,
frame_name=frame_name,
emit_memory=command_succeeded,
)
def _begin_command_action(self) -> Optional[Any]:
"""
Enter one top-level command action under RiftGate control.
Contract:
- Verifies that the command system is still live.
- Resolves the owning room Rift gate when one is available.
- Calls `admit()` before registering one active ticket.
- Returns the gate object so the caller can guarantee symmetric
release in a `finally` block.
Returns:
Optional[Any]: Owning room Rift gate, or None when no gate is bound
to the room.
"""
self.check_cleaned()
rift_gate = self._get_rift_gate_if_available()
if rift_gate is None:
return None
# Ticket-first admission (drain-race fix 2026-07-12): one verb
# acquires a VISIBLE ticket before validating state, so a
# projection refresh's drain can never observe zero tickets while
# this command sits between admit and register.
rift_gate.admit_ticket()
return rift_gate
def _finish_command_action(
self,
*,
rift_gate: Optional[Any],
action_name: str,
frame_name: Optional[str],
emit_memory: bool,
) -> None:
"""
Exit one top-level command action and optionally emit command memory.
Contract:
- Releases the RiftGate ticket before emitting memory so the gate
never stays held across memory callbacks.
- Emits command memory only when the caller marks the command as a
successful top-level action.
- Never suppresses gate-release failures.
Args:
rift_gate:
Rift gate returned by `_begin_command_action(...)`.
action_name:
Stable public command method name.
frame_name:
Optional caller-supplied frame name.
emit_memory:
True when the command completed successfully and should emit one
memory record.
Returns:
None.
"""
if rift_gate is not None:
rift_gate.unregister_ticket()
if emit_memory:
self._emit_command_memory_if_enabled(
action_name=action_name,
frame_name=frame_name,
)
def _emit_command_memory_if_enabled(
self,
*,
action_name: str,
frame_name: Optional[str],
) -> None:
"""
Emit one command memory record when the owning room enables memory output.
Args:
action_name:
Stable public command method name.
frame_name:
Optional caller-supplied frame name.
Returns:
None.
"""
memory_system = self._get_memory_system_if_available()
if memory_system is None or not memory_system.memory_enabled:
return
resolved_frame_name = self._resolve_memory_frame_name(frame_name)
if resolved_frame_name is None:
return
memory_system.create_and_emit_memory(
frame_name=resolved_frame_name,
action_name=action_name,
metadata={
"surface": "command",
"command_system_id": self._id,
"owner_space_id": self._owner_space_id,
},
)
def _get_memory_system_if_available(self) -> Optional[Any]:
"""
Return the owning room's memory system when one is available.
Returns:
Optional[Any]: Room-local memory system, or None when unavailable.
"""
try:
return self._space.memory_system
except AttributeError:
return None
def _get_rift_gate_if_available(self) -> Optional[Any]:
"""
Return the owning room's Rift gate when one is available.
Returns:
Optional[Any]: Room-local Rift gate, or None when unavailable.
"""
try:
return self._space.rift_gate
except AttributeError:
return None
@contextmanager
def _entered_action_hook_scope_if_available(
self,
*,
category: str,
action_name: str,
) -> Any:
"""
Enter the owning room's action-hook scope when one is available.
Args:
category:
Action category.
action_name:
Stable public action name.
Returns:
Any: Hook scope context manager.
"""
try:
action_scope = self._space._entered_action_hook_scope(
category=category,
action_name=action_name,
)
except AttributeError:
yield
return
with action_scope:
yield
def _resolve_memory_frame_name(
self,
frame_name: Optional[str],
) -> Optional[str]:
"""
Resolve the frame name that should be recorded in emitted command memories.
Args:
frame_name:
Optional caller-supplied frame name.
Returns:
Optional[str]: Concrete frame name for memory emission, or None when
the command has no resolvable frame context.
"""
try:
return self._resolve_runtime_frame_name(frame_name)
except Exception:
return None
def _assert_raw_runtime_object_access_allowed(
self,
method_name: str,
) -> None:
"""
Enforce raw runtime-object access policy for this command surface.
Contract:
- Base `CommandSystem` owns the shared raw-runtime getter
vocabulary and defaults to allowing it.
- Denials are represented explicitly through the class-level
`_DENIED_RAW_RUNTIME_OBJECT_ACCESS_METHOD_NAMES` set instead of
hidden no-op hook overrides.
- Subclasses that need room-specific wording may override the class
message template without replacing the assertion method itself.
Args:
method_name:
Public command-system method attempting raw runtime-object
exposure.
Returns:
None.
Raises:
ValueError:
If the requesting method is denied raw runtime-object access in
the current command-surface mode.
"""
if method_name not in type(self)._DENIED_RAW_RUNTIME_OBJECT_ACCESS_METHOD_NAMES:
return
raise ValueError(
type(self)._RAW_RUNTIME_OBJECT_ACCESS_DENIED_MESSAGE_TEMPLATE.format(
method_name
)
)
def _assert_frame_command_enabled(self, frame_name: str) -> None:
"""
Enforce frame-level command access for one hosted frame.
Args:
frame_name:
Hosted frame whose command gate should be checked.
Returns:
None.
Raises:
ValueError:
If the frame does not enable command access.
"""
compiled_access_surface = self._get_required_compiled_access_surface(frame_name)
if compiled_access_surface.command_frame_enabled:
return
raise ValueError(
"Command access is disabled for frame '{0}'.".format(frame_name)
)
def _assert_conduit_command_enabled(
self,
conduit_id: str,
*,
frame_name: str,
) -> None:
"""
Enforce conduit-level command access for one published conduit id.
Args:
conduit_id:
Published conduit id being resolved.
frame_name:
Hosted frame the conduit belongs to.
Returns:
None.
Raises:
ValueError:
If the conduit is not command-enabled in the target frame.
"""
compiled_access_surface = self._get_required_compiled_access_surface(frame_name)
if conduit_id in compiled_access_surface.enabled_conduit_ids:
return
raise ValueError(
"Command access to conduit '{0}' is disabled in frame '{1}'.".format(
conduit_id,
frame_name,
)
)
def _assert_spell_command_enabled(
self,
spell_index_id: str,
*,
frame_name: str,
) -> None:
"""
Enforce spell-level command access for one stable spell index id.
Args:
spell_index_id:
Stable spell index id being resolved.
frame_name:
Hosted frame the spell belongs to.
Returns:
None.
Raises:
ValueError:
If the spell index is not command-enabled in the target
frame.
"""
compiled_access_surface = self._get_required_compiled_access_surface(frame_name)
if spell_index_id in compiled_access_surface.enabled_spell_index_ids:
return
raise ValueError(
"Command access to spell index '{0}' is disabled in frame '{1}'.".format(
spell_index_id,
frame_name,
)
)
def _get_enabled_published_conduit_records(
self,
frame_name: str,
) -> Tuple[Any, ...]:
"""
Return published conduit records that are command-enabled in one frame.
Args:
frame_name:
Hosted frame to query.
Returns:
Tuple[Any, ...]: Published conduit records enabled for command
access in the target frame.
Raises:
ValueError:
If command access is disabled for the frame.
"""
self._assert_frame_command_enabled(frame_name)
compiled_access_surface = self._get_required_compiled_access_surface(frame_name)
enabled_conduit_ids = set(compiled_access_surface.enabled_conduit_ids)
descriptor = self._rift._get_required_command_projection(
frame_name
).frame_descriptor
return tuple(
conduit_record
for conduit_record in descriptor.conduit_records_by_id.values()
if conduit_record.conduit_id in enabled_conduit_ids
)
def _get_required_published_conduit_id_by_name(
self,
conduit_name: str,
*,
frame_name: str,
) -> str:
"""
Return the published conduit id matching one conduit name.
Contract:
- Resolves against published descriptor truth, not runtime conduit
registries.
- Raises on missing or ambiguous published matches.
Args:
conduit_name:
Published conduit name to resolve.
frame_name:
Hosted frame to query.
Returns:
str: Published conduit id for the named conduit.
Raises:
ValueError:
If the conduit name is empty, missing, or ambiguous.
"""
if not conduit_name:
raise ValueError("conduit_name cannot be empty.")
descriptor = self._rift._get_required_command_projection(
frame_name
).frame_descriptor
matching_conduit_ids = [
conduit_record.conduit_id
for conduit_record in descriptor.conduit_records_by_id.values()
if conduit_record.payload.conduit_name == conduit_name
]
if len(matching_conduit_ids) == 0:
raise ValueError(
"Conduit name '{0}' was not found in frame '{1}'.".format(
conduit_name,
frame_name,
)
)
if len(matching_conduit_ids) > 1:
raise ValueError(
"Conduit name '{0}' is ambiguous in frame '{1}'.".format(
conduit_name,
frame_name,
)
)
return matching_conduit_ids[0]
def _get_required_published_spell_index_id_by_spell_id(
self,
spell_id: str,
*,
frame_name: str,
) -> str:
"""
Return the stable spell index id matching one published spell id.
Contract:
- Resolves through published descriptor truth so ACL checks stay on
stable spell-index identity.
- Accepts multiple matching records only when they collapse to one
spell-index id.
Args:
spell_id:
Current published spell id to resolve.
frame_name:
Hosted frame to query.
Returns:
str: Stable published spell index id.
Raises:
ValueError:
If the spell id is empty, missing, or ambiguous across
published spell indexes.
"""
if not spell_id:
raise ValueError("spell_id cannot be empty.")
descriptor = self._rift._get_required_command_projection(
frame_name
).frame_descriptor
matching_spell_index_ids = {
spell_record.spell_index_id
for spell_record in descriptor.spell_records_by_key.values()
if spell_record.spell_id == spell_id
}
if len(matching_spell_index_ids) == 0:
raise ValueError(
"Spell id '{0}' was not found in frame '{1}'.".format(
spell_id,
frame_name,
)
)
if len(matching_spell_index_ids) > 1:
raise ValueError(
"Spell id '{0}' is ambiguous in frame '{1}'.".format(
spell_id,
frame_name,
)
)
return next(iter(matching_spell_index_ids))
def _get_required_published_spell_record_by_source_id(
self,
spell_source_id: str,
*,
frame_name: str,
) -> Any:
"""
Return the published spell record matching one spell source id.
Args:
spell_source_id:
Published spell source id in `spellbook_id:spell_id` form.
frame_name:
Hosted frame to query.
Returns:
Any: Matching published spell record.
"""
if not spell_source_id:
raise ValueError("spell_source_id cannot be empty.")
if ":" not in spell_source_id:
raise ValueError(
"spell_source_id must be in 'spellbook_id:spell_id' form."
)
origin_spellbook_id, spell_id = spell_source_id.split(":", 1)
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.origin_spellbook_id == origin_spellbook_id
and spell_record.spell_id == spell_id
)
]
if len(matching_spell_records) == 0:
raise ValueError(
"Spell '{0}' was not found in frame '{1}'.".format(
spell_source_id,
frame_name,
)
)
if len(matching_spell_records) > 1:
raise ValueError(
"Spell '{0}' is ambiguous in frame '{1}'.".format(
spell_source_id,
frame_name,
)
)
return matching_spell_records[0]
def _get_required_compiled_access_surface(
self,
frame_name: str,
) -> CompiledFrameACLAccessSurface:
"""
Return the compiled ACL access surface for one hosted frame.
Args:
frame_name:
Hosted frame whose compiled ACL surface is required.
Returns:
CompiledFrameACLAccessSurface:
Compiled frame ACL access surface for the target frame.
Raises:
ValueError:
If the room has no command projection for the requested frame.
"""
return self._rift._get_required_command_projection(
frame_name
).compiled_access_surface
def _bind_result(
self,
*,
bind_as_name: str,
bind_as_store: str,
value: object,
bind_result_weak_ref: Optional[bool] = None,
) -> None:
"""
Bind a returned value back into the workstation.
Contract:
- Routes binding into exactly one workstation store.
- Raises for unsupported stores instead of silently discarding the
value.
Args:
bind_as_name:
Binding name for the returned value.
bind_as_store:
Target workstation store.
value:
Value to bind.
bind_result_weak_ref:
Optional workstation reference-mode override for the bound
result.
Returns:
None.
"""
if bind_as_store == "objects":
self._workstation.bind_object(
bind_as_name,
value,
weak_ref=bind_result_weak_ref,
)
return
if bind_as_store == "attributes":
self._workstation.bind_attribute(
bind_as_name,
value,
weak_ref=bind_result_weak_ref,
)
return
if bind_as_store == "methods":
self._workstation.bind_method(
bind_as_name,
value,
weak_ref=bind_result_weak_ref,
)
return
raise ValueError(
"Unsupported workstation store '{0}'.".format(bind_as_store)
)
def _resolve_runtime_frame_name(
self,
frame_name: Optional[str],
) -> str:
"""
Resolve the runtime frame name for direct Aether-backed getters.
Args:
frame_name:
Optional explicit frame name.
Returns:
str: Resolved runtime frame name.
"""
resolved_frame_name = (
frame_name
if frame_name is not None
else self._rift._get_default_runtime_frame_name()
)
if resolved_frame_name is None:
raise ValueError("Rift has no default runtime frame.")
return resolved_frame_name
def _get_required_runtime_frame(self, frame_name: str) -> object:
"""
Return one live Aether frame by name or raise.
Args:
frame_name:
Runtime frame name to resolve.
Returns:
object: Live `AethericFrame`.
"""
try:
return self._aether._aetheric_frames[frame_name]
except KeyError as exc:
raise ValueError(
"Aetheric frame '{0}' does not exist.".format(frame_name)
) from exc