Registries

sakshi.registries

Sakshi registry classes.

The module registry tracks alternative implementations of each phase with hot-swap support. The phase registry maps cycle phase names to OODA phases and runs the cycle-completion callback chain.

DEFAULT_PHASE_REGISTRY: dict[str, PhaseSlot] = {'PERCEIVE': PhaseSlot(ooda_phase=(OODAPhase.OBSERVE), description='Sense environment state'), 'INTERPRET': PhaseSlot(ooda_phase=(OODAPhase.ORIENT), description='Anomaly detection and knowledge integration'), 'EVAL': PhaseSlot(ooda_phase=(OODAPhase.ORIENT), description='Goal divergence from current state'), 'INTEND': PhaseSlot(ooda_phase=(OODAPhase.DECIDE), description='Candidate selection and goal commitment'), 'PLAN': PhaseSlot(ooda_phase=(OODAPhase.DECIDE), description='Policy selection and tool-chain planning'), 'ACT': PhaseSlot(ooda_phase=(OODAPhase.ACT), description='Execute the committed plan')} module-attribute

CycleCompleteCallback = Callable[[CycleTrace], Awaitable[None]] module-attribute

ModuleRegistry

Map phases to alternative module implementations with hot-swap.

Each phase can have multiple registered modules. One is active at any time (initially the highest priority). A swap call changes the active module without modifying the registry.

register(phase_name: str, module_name: str, priority: int = 1, metadata: dict[str, Any] | None = None) -> None

Register a module as an alternative for a phase.

Parameters:

Name	Type	Description	Default
phase_name	str	Cycle phase name.	required
module_name	str	Unique name of the module / service.	required
priority	int	Lower is higher priority (1 = default / primary).	1
metadata	dict[str, Any] | None	Optional metadata about the module.	None

get_modules_for_phase(phase_name: str) -> list[RegisteredModule]

Return all registered modules for a phase, sorted by priority.

get_active_module(phase_name: str) -> RegisteredModule | None

Return the currently active module for a phase.

Returns the swap override if one exists; otherwise returns the highest-priority registered module. Returns None if the phase has no registered modules.

swap_module(phase_name: str, target_module_name: str) -> bool

Swap the active module for a phase.

Parameters:

Name	Type	Description	Default
phase_name	str	Phase to swap module for.	required
target_module_name	str	Name of the module to activate.	required

Returns:

Type	Description
bool	True if the swap succeeded, False if the module was not found.

restore_default(phase_name: str) -> None

Restore the default (highest-priority) module for a phase.

Removes any swap override, reverting to priority-based selection.

get_swap_history(phase_name: str) -> list[dict[str, Any]]

Return the swap history for a phase.

RegisteredModule dataclass

A module registered as an alternative for a cycle phase.

name: str instance-attribute

priority: int instance-attribute

metadata: dict[str, Any] = field(default_factory=dict) class-attribute instance-attribute

PhaseRegistry

Registry that drives a cognitive cycle through its phases.

Usage:

registry = PhaseRegistry(event_bus=bus)
registry.on_cycle_complete(my_persistence_callback)

await registry.start_cycle(cycle_id="cycle-001")
await registry.record_phase_output("PERCEIVE", perception_result)
await registry.record_phase_output("INTERPRET", interpret_result)
await registry.record_phase_output("EVAL", eval_result)
await registry.record_phase_output(
    "INTEND", intend_result, np_snapshot=np_snapshot
)
await registry.record_phase_output("PLAN", plan_result)
await registry.record_phase_output("ACT", act_result)
trace = await registry.finalize_cycle()

Parameters:

Name	Type	Description	Default
event_bus	EventBus	An object satisfying sakshi.protocols.EventBus. Used to publish a cycle-complete event.	required
phase_config	dict[str, PhaseSlot] | None	Optional override for the default phase mapping.	None
cycle_complete_event_type	str	String event type the registry emits when a cycle finalizes. Hosts can override to align with their bus's event-name conventions.	'sakshi.cycle.complete'

list_phases() -> list[str]

Return ordered list of registered phase names.

get_phase_config(phase_name: str) -> PhaseConfig

Return PhaseConfig for a phase.

Raises:

Type	Description
KeyError	If the phase name is not registered.

on_cycle_complete(callback: CycleCompleteCallback) -> None

Register a callback to fire when finalize_cycle() is called.

start_cycle(cycle_id: str) -> None async

Begin a new cycle, resetting the current trace.

record_phase_output(phase_name: str, output: dict[str, Any], *, np_snapshot: NPStateSnapshot | None = None) -> None async

Record the output of a completed phase.

At INTEND, np_snapshot may carry the dominant thought-seed's NP state and prior type.

finalize_cycle() -> CycleTrace async

Close the current cycle and fire all registered callbacks.

Raises:

Type	Description
PhaseTransitionError	If no cycle is active.

Returns:

Type	Description
CycleTrace	The finalized CycleTrace.

get_current_trace() -> CycleTrace | None

Return the in-progress trace, or None if no cycle is active.

get_last_completed_trace() -> CycleTrace | None

Return the most recently finalized trace, or None.

PhaseSlot dataclass

OODA mapping plus optional service names for one cycle phase.

ooda_phase: OODAPhase instance-attribute

description: str = '' class-attribute instance-attribute

services: list[str] = field(default_factory=list) class-attribute instance-attribute