Cycle¶
sakshi.cycle
¶
Cognitive cycle runtime classes.
The blackboard provides inter-phase state sharing within a single
cycle. The history buffer keeps a rolling window of past traces for
metacognitive pattern analysis. Trace pruners reduce a finished
CycleTrace to the slice the meta-cycle should reason over.
CognitiveBlackboard
¶
Async-safe mutable blackboard for inter-phase state sharing.
Each BlackboardKey gets its own asyncio.Lock so phases that
write to different keys do not block each other.
Lifecycle
clear()at start of cycle.set()/get()during phase execution.snapshot()at finalize.
set(key: BlackboardKey, value: Any) -> None
async
¶
Set a value on the blackboard (async-safe per key).
get(key: BlackboardKey) -> Any | None
async
¶
Return a value from the blackboard, or None if unset.
get_or_default(key: BlackboardKey, default: Any) -> Any
async
¶
Return a value from the blackboard, or default if unset.
clear() -> None
async
¶
Clear all data from the blackboard.
Holds every per-key lock simultaneously while clearing so a
concurrent set() cannot interleave a write between the
last lock release and self._data.clear(). Locks are taken
in a deterministic key-value order; no other method on this
class acquires more than one lock at a time, so the
multi-lock acquisition cannot deadlock against single-lock
callers.
keys() -> list[BlackboardKey]
async
¶
Return a list of currently set keys.
snapshot(cycle_id: str = '') -> BlackboardSnapshot
async
¶
Create an immutable snapshot of current blackboard state.
Returns a deep copy so mutations after the snapshot do not affect it. If a value is not deep-copyable, the snapshot stores the live reference and emits a warning — callers should treat such snapshots as best-effort, not isolation-safe.
CycleHistory
¶
Rolling buffer of CycleTrace snapshots.
- Fixed
max_size(default 100). - When full, prunes the oldest 50% before adding.
- Backward lookup via
get_n_prev_cycle(n).
Thread / coroutine safety is not provided: cycles are sequential (one cycle finalizes before the next starts).
max_size = max_size
instance-attribute
¶
add(trace: CycleTrace) -> None
¶
Add a finalized trace to the history.
If the buffer is at capacity, prunes the oldest half first.
get_n_prev_cycle(n: int) -> CycleTrace | None
¶
Return the Nth previous cycle trace.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
n
|
int
|
0 = most recent, 1 = previous, etc. |
required |
Returns:
| Type | Description |
|---|---|
CycleTrace | None
|
The trace, or None if |
get_phase_output_across_cycles(phase_name: str, n_cycles: int) -> list[dict[str, Any]]
¶
Return phase outputs from the most recent N cycles.
Skips cycles that did not record the requested phase. Results are ordered most-recent-first.
search_cycles(predicate: Callable[[CycleTrace], bool]) -> list[CycleTrace]
¶
Filter cycle history by a predicate function.
Returns the matching traces in oldest-first order.
get_latest() -> CycleTrace | None
¶
Return the most recent trace, or None if empty.
SinceAnomalyPruner
¶
Keep every phase from the last phase that recorded an anomaly forward.
A phase is considered to have recorded an anomaly if its output
dict carries any of the keys in anomaly_keys with a truthy
value. The default looks for anomaly, anomalies, and
anomaly_event to match the conventions used elsewhere in the
package.
TracePruner
¶
Bases: Protocol
Reduce a CycleTrace to the slice the meta-cycle should reason over.
prune(trace: CycleTrace) -> CycleTrace
¶
WhereExpectationFiredPruner
¶
Keep phases whose output flagged at least one expectation.
A phase is considered to have fired an expectation if its
output dict contains a non-empty expectations list, an
expectation_violations list, or a truthy expectation_fired
flag. The host can extend the keys checked by passing
expectation_keys.