///|
/// A side-channel collector for values pushed by memo compute closures.
///
/// Typed buffers (`per_memo`, `prev_push_sets`) live on the handle; runtime
/// holds only type-erased closures in SlotMeta. See the design doc
/// `docs/superpowers/specs/2026-04-19-accumulator-api-design.md`.
pub struct Accumulator[T] {
priv rt : Runtime
priv slot_id : @incr_types.AccumulatorId
priv label : String?
// Typed per-memo buffer: the values this memo pushed in its latest
// successful compute. Read by Memo::accumulated(_peek).
priv per_memo : @hashmap.HashMap[CellId, Array[T]]
// Snapshot of per_memo taken at BEFORE_CLOSURE; restored on ON_ABORT.
priv prev_push_sets : @hashmap.HashMap[CellId, Array[T]]
// Per-memo revision bumped in finalize_memo when push-set differs
// across recomputes. Reader records this in its accumulator_reads.
priv push_revised_at : @hashmap.HashMap[CellId, Revision]
}
///|
/// Creates a new accumulator bound to the given runtime.
///
/// Constructs the typed buffer state and the SlotMeta closures that the
/// runtime uses to drive snapshot/finalize/dispose phases.
pub fn[T : Eq] Accumulator::Accumulator(
rt : Runtime,
label? : String,
) -> Accumulator[T] {
let per_memo : @hashmap.HashMap[CellId, Array[T]] = @hashmap.HashMap([])
let prev_push_sets : @hashmap.HashMap[CellId, Array[T]] = @hashmap.HashMap([])
let push_revised_at : @hashmap.HashMap[CellId, Revision] = @hashmap.HashMap([])
// snapshot_and_clear: move current buffer into prev, replace per_memo[M]
// with a fresh array. Called BEFORE_CLOSURE.
let snapshot_and_clear = (m_id : CellId) => {
match per_memo.get(m_id) {
Some(current) => {
prev_push_sets.set(m_id, current)
per_memo.set(m_id, [])
}
None =>
// Memo had no prior pushes; record empty prev so diff catches
// first-ever pushes correctly.
prev_push_sets.set(m_id, [])
}
}
// restore_buffer: undo snapshot. Called ON_ABORT.
let restore_buffer = (m_id : CellId) => {
match prev_push_sets.get(m_id) {
Some(prev) => {
per_memo.set(m_id, prev)
prev_push_sets.remove(m_id)
}
None =>
// Shouldn't happen — BEFORE_CLOSURE always writes prev_push_sets.
()
}
}
// finalize_memo: atomic commit-phase work. Compare prev vs current;
// bump push_revised_at if different; drop snapshot; gc empty buffer.
let finalize_memo = (m_id : CellId, rev : Revision) => {
let prev = match prev_push_sets.get(m_id) {
Some(p) => p
None => []
}
let current = match per_memo.get(m_id) {
Some(c) => c
None => []
}
if !prev.equal(current) {
push_revised_at.set(m_id, rev)
}
prev_push_sets.remove(m_id)
if current.length() == 0 {
per_memo.remove(m_id)
}
}
// dispose_memo: remove memo's entries when the memo itself is disposed.
let dispose_memo = (m_id : CellId) => {
per_memo.remove(m_id)
prev_push_sets.remove(m_id)
push_revised_at.remove(m_id)
}
// clear_new_run_buffer: ON_ABORT cleanup for a slot touched in this run
// that was NOT in prev_contributions. Drops the in-flight pushes but
// keeps push_revised_at and prev_push_sets intact so a prior run's
// revision marker survives the abort. prev_push_sets has no entry for
// m_id in this case (snapshot_and_clear wasn't called this run), so
// not touching it is correct.
let clear_new_run_buffer = (m_id : CellId) => per_memo.remove(m_id)
let push_revised_at_for = (m_id : CellId) => {
match push_revised_at.get(m_id) {
Some(r) => r
None => Revision::initial()
}
}
let has_buffer_for = (m_id : CellId) => {
match per_memo.get(m_id) {
Some(arr) => arr.length() > 0
None => false
}
}
let meta : SlotMeta = {
label,
disposed: false,
snapshot_and_clear,
restore_buffer,
finalize_memo,
dispose_memo,
clear_new_run_buffer,
push_revised_at_for,
has_buffer_for,
}
let slot_id = rt.register_accumulator_slot(meta)
{ rt, slot_id, label, per_memo, prev_push_sets, push_revised_at }
}
///|
/// Runtime-side metadata for an Accumulator slot. Holds type-erased closures
/// captured at Accumulator construction with `T : Eq` in scope; the typed
/// buffers live on the Accumulator[T] handle itself.
///
/// This struct is intentionally package-private — runtime code dispatches
/// through the closures; no outside caller needs to construct or read it.
priv struct SlotMeta {
label : String?
mut disposed : Bool
// Phase closures (typed Accumulator handle captured at construction):
snapshot_and_clear : (@incr_types.CellId) -> Unit
restore_buffer : (@incr_types.CellId) -> Unit
finalize_memo : (@incr_types.CellId, @incr_types.Revision) -> Unit
dispose_memo : (@incr_types.CellId) -> Unit
// ON_ABORT cleanup for slots this aborted run touched but that were
// NOT in prev_contributions at BEFORE_CLOSURE. Clears per_memo[M]
// (the in-flight pushes) while preserving push_revised_at[M], so a
// prior run's revision marker is not rolled back. Without this,
// consumers with a pre-existing stored read would miss invalidation.
clear_new_run_buffer : (@incr_types.CellId) -> Unit
// Verify-side typed data access:
push_revised_at_for : (@incr_types.CellId) -> @incr_types.Revision
// True iff the slot still has a non-empty per_memo[M] buffer. Used in
// AFTER_CLOSURE to rebuild accumulator_contributions reverse index.
has_buffer_for : (@incr_types.CellId) -> Bool
}
///|
/// Internal: register a new accumulator slot and return its ID.
/// Called by Accumulator::Accumulator. Monotonic — IDs never reused.
///
/// Pushes into `accumulator_slots` and the parallel
/// `accumulator_snapshots` cache so the two arrays stay index-aligned.
fn Runtime::register_accumulator_slot(
self : Runtime,
meta : SlotMeta,
) -> @incr_types.AccumulatorId {
let id = self.next_accumulator_id
self.next_accumulator_id = id + 1
self.accumulator_slots.push(meta)
self.accumulator_snapshots.push(meta)
self.check_accumulator_cache_invariant()
{ runtime_id: self.core.runtime_id, id }
}
///|
/// Verify-side view of a SlotMeta's disposal flag.
impl @shared.SlotSnapshot for SlotMeta with fn disposed(self) -> Bool {
self.disposed
}
///|
/// Verify-side accessor for the slot's per-memo push_revised_at revision.
///
/// The local binding is load-bearing: `self.push_revised_at_for` names
/// both the closure field on `SlotMeta` and this trait method, so
/// `self.push_revised_at_for(cell_id)` would be parsed as a recursive
/// trait-method call and infinite-loop. Binding the closure to a local
/// name disambiguates lexically — a parens-wrap `(self.field)(arg)`
/// works too but is a parser subtlety that's easy to break in a later
/// refactor.
impl @shared.SlotSnapshot for SlotMeta with fn push_revised_at_for(
self,
cell_id,
) -> Revision {
let f = self.push_revised_at_for
f(cell_id)
}
///|
/// Internal: guarantee a memo has been computed and is up-to-date in the
/// current revision, WITHOUT recording an ordinary dep on the caller's
/// tracking frame. Used by Memo::accumulated.
///
/// # Behavior
///
/// - If the cell is disposed, returns Ok(()) silently (caller handles
/// disposal separately; this is defense-in-depth for other call sites).
/// - If the memo has never been computed (verified_at == initial revision),
/// triggers the first compute via the memo's stored compute closure.
/// - Otherwise runs `pull_verify`, which may recompute if stale.
/// - Neither path calls `record_dependency`. The synthetic accumulator
/// dep is recorded separately in the accumulator read flow.
///
/// Both `data.compute()` and `pull_verify` push a fresh tracking frame
/// for the target memo while recomputing, so any `get()` calls inside the
/// target's closure record deps on the target's frame — never on the
/// caller's frame.
fn Runtime::ensure_computed_untracked(
self : Runtime,
cell_id : CellId,
) -> Result[Unit, CycleError] raise Failure {
if self.is_cell_disposed(cell_id) {
return Ok(())
}
let data = self.get_memo_data(cell_id)
// "Never been computed" is tracked explicitly via has_been_computed
// rather than via verified_at == Revision::initial() because a fresh
// runtime starts at initial, so a successful first compute would leave
// verified_at == initial, indistinguishable from "never computed".
if !data.has_been_computed {
match (data.compute)() {
Ok(_) => Ok(())
Err(e) => Err(e)
}
} else {
match self.pull_verify(cell_id) {
Ok(_) => Ok(())
Err(e) => Err(e)
}
}
}
///|
/// Returns the slot's unique ID.
pub fn[T] Accumulator::id(self : Accumulator[T]) -> @incr_types.AccumulatorId {
self.slot_id
}
///|
/// Returns the optional label provided at construction.
pub fn[T] Accumulator::label(self : Accumulator[T]) -> String? {
self.label
}
///|
/// Returns true iff this accumulator has been disposed.
pub fn[T] Accumulator::is_disposed(self : Accumulator[T]) -> Bool {
self.rt.accumulator_slots[self.slot_id.id].disposed
}
///|
/// Disposes the accumulator. Idempotent. Clears typed buffers and removes
/// reverse-index entries from rt.accumulator_contributions. The slot_id
/// stays allocated — monotonic, never reused.
pub fn[T] Accumulator::dispose(self : Accumulator[T]) -> Unit {
let slot = self.rt.accumulator_slots[self.slot_id.id]
if slot.disposed {
return
}
// Remove slot_id from every contributing memo's reverse index.
for m_id in self.per_memo.keys() {
match self.rt.accumulator_contributions.get(m_id) {
Some(set) => {
set.remove(self.slot_id)
if set.is_empty() {
self.rt.accumulator_contributions.remove(m_id)
}
}
None => ()
}
}
self.per_memo.clear()
self.prev_push_sets.clear()
self.push_revised_at.clear()
slot.disposed = true
}
///|
/// Appends a value to this accumulator, keyed by the currently-computing
/// memo's CellId.
///
/// # Errors
///
/// - `fail` if called outside a tracked compute context
/// - `fail` if the top tracking frame is not a Memo (MVP restriction)
/// - `fail` if this accumulator has been disposed
/// - `abort` (via `check_cross_runtime`) if slot belongs to a different runtime
pub fn[T] Accumulator::push(
self : Accumulator[T],
value : T,
) -> Unit raise Failure {
// Cross-runtime check (abort — pre-existing pattern).
self.rt.check_cross_runtime(self.slot_id.runtime_id, "Accumulator")
if self.rt.static_recompute_depth > 0 {
fail("Accumulator::push is unsupported inside static Derived recompute")
}
let slot = self.rt.accumulator_slots[self.slot_id.id]
if slot.disposed {
fail("Accumulator::push to disposed accumulator")
}
// Top-frame check.
let frame = match self.rt.top_active_query() {
Some(q) => q
None => fail("Accumulator::push called outside a tracked compute")
}
let m_id = frame.cell_id
// Top-frame must be a PullMemo / HybridMemo cell.
match self.rt.core.cell_index[m_id.id] {
PullMemo(_) | HybridMemo(_) => ()
_ =>
fail(
"Accumulator::push only valid inside Derived / ReachableDerived compute",
)
}
// Append to typed buffer.
let buf = match self.per_memo.get(m_id) {
Some(existing) => existing
None => {
let fresh : Array[T] = []
self.per_memo.set(m_id, fresh)
fresh
}
}
buf.push(value)
// Stage touched slot on AccumulatorCommitHook (committed in AFTER_CLOSURE).
// ensure_for_cell lazy-creates the entry when before_recompute took its
// fast-path (no slots existed at recompute-start time but this accumulator
// was registered mid-recompute). The line-449 cell_index check above
// guarantees we're inside a Memo/HybridMemo frame.
self.rt.accumulator_commit_hook
.ensure_for_cell(m_id)
.ensure_touched()
.add(self.slot_id)
}
///|
///|
/// Debug representation.
pub fn[T] Accumulator::debug(self : Accumulator[T]) -> String {
let lbl = match self.label {
Some(s) => "\"" + s + "\""
None => "None"
}
"Accumulator(id=" +
self.slot_id.id.to_string() +
", label=" +
lbl +
", disposed=" +
self.is_disposed().to_string() +
")"
}