# emitter side
rt.emit_event(topic: bytes, payload: bytes, schema: bytes | None = None) -> EmitResult
    # payload length is bounded by MAX_EVENT_PAYLOAD_BYTES (default 4096); overflow fails the call
    # payload bytes are billed 1 cell/byte against the emitter (same model as calldata)
    # EmitResult contains the in-block synchronous fire outcomes
    # plus the list of sub_ids overflowed to H+1
    # Ordering: descending by bid; top MAX_SYNC_FIRES_PER_TOPIC fire synchronously,
    # the rest are split into multiple system defer txs

rt.force_unsubscribe_event(sub_id: SubscriptionId) -> ()
    # Callable only by the emitter; used for eviction.
    # gas_remaining and occupied cells are both refunded to the subscriber
    # (the emitter neither pays nor receives anything).
    # bid has already been sunk and is NOT refunded (see §2.6).

# subscriber side
rt.subscribe_event(
    emitter: ActorAddr,
    topic: bytes,
    handler: str,
    gas_prepaid: u64,
    bid: u64 = 0,                        # bid amount that drives rank; 0 means pure first-come-first-served
                                         # (placed after every positive bid)
    schema: bytes | None = None,         # optional 32-byte payload schema identifier
) -> SubscriptionId
    # Subscriber is debited four items at once:
    #   - SUBSCRIPTION_REGISTRATION_FEE  cycles  (write compute, NOT refunded)
    #   - delta cells                            (EventSub record + index entry growth;
    #                                              fully refunded on unsubscribe / reap)
    #   - gas_prepaid                    cycles  (fire prepay pool;
    #                                              residual refunded on unsubscribe)
    #   - bid                            cycles  (already sunk, NOT refunded;
    #                                              recorded in the subscription's bid field)

rt.update_bid(sub_id: SubscriptionId, additional_bid: u64) -> u64
    # Callable only by the subscriber. Adds to bid (cannot decrease). Returns the new cumulative bid.
    # Takes effect immediately by reinserting into the (emitter, topic) ordered index.
    # Note: for an async segment whose ordering was already locked by an emit in the current tx,
    #       update_bid does NOT affect that emit's fire order — the new bid only affects this
    #       subscription's rank in subsequent emits (see §2.3 "async segment ordering is locked").

rt.topup_subscription(sub_id: SubscriptionId, additional_gas: u64) -> u64
    # Any account may top up a subscription's gas_remaining. Returns the new balance.
    # Top-ups are non-refundable, do not change bid or REGISTRATION_FEE; used to keep
    # long-lived subscriptions alive before gas_remaining is drained.
    # The funder pays additional_gas cycles, which enter the subscription prepay pool.

rt.unsubscribe_event(sub_id: SubscriptionId) -> u64
    # Callable only by the subscriber. Returns the refunded gas_remaining.
    # Cells occupied by the subscription are also refunded;
    # SUBSCRIPTION_REGISTRATION_FEE and bid are NOT refunded.

StatePrefix::EventSub      = 0x18    // Single subscription record (located by sub_id)
StatePrefix::EventSubIndex = 0x19    // (emitter, topic) → ordered subscriber index

sub_id = keccak256(emitter ‖ subscriber ‖ topic ‖ u64_be(sub_height)) ‖ 0x00

key   = 0x18 || keccak256(sub_id)[..32] || zero-pad   // 54-byte StateKey
value = ActorEventSubscription {
    emitter_addr:  Address,
    topic:         BoundedBytes<64>,
    subscriber:    Address,
    handler_name:  BoundedString,
    gas_prepaid:   u64,                  // Total prepaid budget at registration
    gas_remaining: u64,                  // Current balance; subscription expires when zero
    bid:           u64,                  // Cumulative bid (monotonically increasing via update_bid);
                                         // already sunk; drives rank
    sub_height:    BlockHeight,
}

key   = 0x19 || keccak256(emitter_addr || topic)[..32] || zero-pad   // 54-byte StateKey
value = {
    declared_schema: Option<[u8; 32]>,
    subscribers: Vec<(schema: Option<[u8; 32]>, bid_inv: u64, sub_height: u64, sub_id: SubscriptionId)>,
}
    // bid_inv = u64::MAX - bid, so high bid sorts first
    // Sort key: lexicographic (bid_inv, sub_height, sub_id)
    //   ⇒ higher bid wins; on tie, earlier subscription wins; tiebreak by sub_id

fn emit_event(&mut self, topic: &[u8], payload: &[u8], schema: Option<&[u8]>) -> HostResult<EmitResult> {
    // 1. payload length + caps: per-tx emit count, per-tx total synchronous fire count, fan-out depth
    require!(payload.len() <= MAX_EVENT_PAYLOAD_BYTES);    // 4096 byte cap (a bound, not a charge)
    self.check_emit_caps(topic)?;

    // 2. Load the subscription index (already sorted by (bid_inv, sub_height, sub_id))
    let sub_ids = self.load_sub_index(self.current_actor(), topic)?;

    // 3. For an explicit schema, apply §2.1's declaration and mismatch rule
    //    BEFORE any cell charge: a mismatch returns EventSchemaMismatch here,
    //    so a failing typed emit consumes no payload cells. A new or changed
    //    declaration separately charges 32 cells to the emitter.
    if let Some(schema) = schema {
        self.check_or_declare_schema(self.current_actor(), topic, schema)?;   // may charge 32 cells
    }

    // 4. Payload cells, charged only once the emit is going to publish
    self.charge_emitter_cells(payload.len() as u64)?;      // 1 cell/byte against emitter

    // 5. Tier split: top K sync, rest queued async (ordering locked at this point)
    let k = MAX_SYNC_FIRES_PER_TOPIC.min(sub_ids.len());
    let (sync_subs, async_subs) = sub_ids.split_at(k);

    let mut sync_results = Vec::with_capacity(k);
    let mut deferred_subs = Vec::new();
    let mut zombies = Vec::new();

    // --- Sync segment: top K fire inside the emitter call stack ---
    for sid in sync_subs {
        let sub = self.load_sub_record(*sid)?;

        // 3a. Lazy GC: gas_remaining can't even cover one minimal fire, skip + mark zombie
        if sub.gas_remaining < MIN_FIRE_COST {
            zombies.push(*sid);
            sync_results.push(EmitOutcome::SkippedExpired(*sid));
            continue;
        }

        // 3b. Snapshot: subscriber state + global checkpoint
        let snapshot = self.pvm.snapshot();

        // 3c. Use sub.gas_remaining as this call's budget cap;
        //     consumed cycles/cells do NOT come out of the emitter's budget
        let r = self.call_actor_with_isolated_gas(
            sub.subscriber,
            &sub.handler_name,
            payload,
            sub.gas_remaining,
        );

        match r {
            Ok(res) => {
                self.deduct_subscription_budget(*sid, res.gas_used)?;
                sync_results.push(EmitOutcome::Ok(res));
            }
            Err(e) => {
                // 3d. On failure: rollback snapshot; emitter state untouched
                self.pvm.rollback(snapshot);
                self.deduct_subscription_budget(*sid, e.gas_consumed)?;
                sync_results.push(EmitOutcome::Err(e));
            }
        }
    }

    // --- Async segment: split overflow sub_ids into multiple system defer txs, all queued to H+1 ---
    //
    // **Ordering locked**: the ordering of async_subs is fixed at this emit (based on the current
    // (bid_inv, sub_height, sub_id) ordering at emit time). Even if a subscriber calls update_bid
    // between block H and block H+1 to raise its bid, the fire order of *this emit's* async
    // segment is not changed — the new bid only affects the rank for the *next* emit. This aligns
    // with the "decided at emit time" semantics of the sync segment, and rules out the
    // "emit-then-bump-bid to cut in line on the current async segment" arbitrage.
    //
    // **Split policy**: each system defer tx carries at most ASYNC_FIRES_PER_DEFER_TX = 64 subs.
    // 448 overflow subs split into ⌈448/64⌉ = 7 defer txs, all targeting H+1;
    // if H+1's total cycle budget cannot absorb them, the existing defer admission gate naturally
    // pushes the excess into H+2 / H+3.
    let emit_id = self.next_emit_id_in_tx();          // index of this emit within the tx; used in receipt causality
    for chunk in async_subs.chunks(ASYNC_FIRES_PER_DEFER_TX) {
        self.enqueue_event_defer_tx(EmitOrigin {
            emitter:      self.current_actor(),
            topic:        topic.to_vec(),
            payload:      payload.to_vec(),
            sub_ids:      chunk.to_vec(),
            origin_block: self.current_block(),
            origin_tx:    self.current_tx_hash(),
            emit_id,
        }, self.current_block() + ASYNC_FIRE_DEFERRAL_BLOCKS)?;
    }
    deferred_subs.extend_from_slice(async_subs);

    // --- Zombie cleanup: remove from index list and delete subscription records.
    //     Freed cells are credited back to each subscriber (NOT to emitter / reaper).
    if !zombies.is_empty() {
        self.trim_sub_index(self.current_actor(), topic, &zombies)?;
        for sid in &zombies {
            let sub = self.load_sub_record(*sid)?;
            self.credit_cells(sub.subscriber, self.compute_freed_cells(&sub))?;
            self.delete_sub_record(*sid)?;
        }
    }

    Ok(EmitResult {
        sync_outcomes: sync_results,
        deferred_subs,                  // caller can observe which sub_ids are queued to H+1
        emit_id,                        // index of this emit within the tx; for receipt correlation
    })
}

fn remove_subscription(&mut self, sid: SubscriptionId, caller: Address) -> HostResult<u64> {
    let sub = self.load_sub_record(sid)?;

    // Authorization: subscriber (unsubscribe) or matching emitter (force_unsubscribe)
    require!(caller == sub.subscriber || caller == sub.emitter_addr);

    // 1. Refund residual gas (cycles) to the subscriber.
    //    REGISTRATION_FEE and bid were burned at subscribe / update_bid time;
    //    the sub.bid field is only a sort signal here, no further handling required.
    let cycles_refund = sub.gas_remaining;
    self.credit_balance(sub.subscriber, cycles_refund)?;

    // 2. Compute the storage delta freed by removing this record + index entry,
    //    and refund those cells to the subscriber regardless of who called us.
    let cells_freed = self.compute_freed_cells(&sub);
    self.credit_cells(sub.subscriber, cells_freed)?;

    self.trim_sub_index(sub.emitter_addr, &sub.topic, &[sid])?;
    self.delete_sub_record(sid)?;
    Ok(cycles_refund)
}

fn apply_bid(&mut self, subscriber: Address, sid: SubscriptionId, bid_delta: u64) -> HostResult<()> {
    require!(bid_delta > 0);                               // zero-delta calls are not allowed (no-op)
    self.debit_balance(subscriber, bid_delta)?;            // debit subscriber immediately
    self.burn_cycles(bid_delta)?;                          // route into burn pool (same channel as EIP-1559 basefee)
    self.bump_sub_bid(sid, bid_delta)?;                    // accumulate into sub.bid and reinsert into the index
    Ok(())
}

@emit("liquidation_imminent")
def maybe_liquidate(self, position_id):
    if self.below_threshold(position_id):
        return ("trigger", position_id)              # decorator auto-emits ("liquidation_imminent", ...)
    return ("noop", None)                            # also emits, payload = ("noop", None)

def settle_batch(self, ctx, orders):
    for order in orders:
        result = self.try_match(order)

        if result.matched:
            # Emit #1: successful match, fired inside a branch
            ctx.emit("order_matched", {"id": order.id, "price": result.price})
        elif result.expired:
            # Emit #2: expired, different topic
            ctx.emit("order_expired", {"id": order.id})
        # Neither matched nor expired → no event; the function continues

    # Emit #3: batch summary, coexisting with the N per-iteration emits above
    ctx.emit("batch_settled", {"count": len(orders)})
    return ("ok", len(orders))

@on_event(
    emitter="0xLENDING_PROTOCOL",
    topic="liquidation_imminent",
    gas=500_000,
    bid=100_000,                                     # bid 100k cycles to claim a sync-window slot
)
def on_liquidation(self, position_id):
    self.bid_for_collateral(position_id)

// node/types/src/constants.rs (EVENT_SUBSCRIPTION_SYSTEM_ACTOR = 0x1D)
EVENT_SUBSCRIPTION_SYSTEM_ACTOR = Address::from_low_u64(0x1D)

struct EmitOrigin {
    emitter:      Address,
    topic:        BoundedBytes<64>,
    payload:      BoundedBytes<MAX_EVENT_PAYLOAD_BYTES>,
    sub_ids:      Vec<SubscriptionId>,         // ≤ ASYNC_FIRES_PER_DEFER_TX = 64
    origin_block: BlockHeight,                  // block of the original emit
    origin_tx:    H256,                         // tx hash of the original emit
    emit_id:      u32,                          // index of this emit within the tx (disambiguates multiple emits)
}

struct AsyncEmitReceipt {
    // ... general receipt fields (gas_used, events, status, ...) ...
    triggered_by_emit: Option<EmitOriginRef>,  // set only for system event defer txs; None for any other defer tx
}

struct EmitOriginRef {
    origin_block:  BlockHeight,
    origin_tx:     H256,
    emit_id:       u32,
}