Fan-out — map & reduce in a Web-of-Trust

The reduce strategies — how to bring back a fan-out

A WoT fans out: an event scatters into N children (the map). The reduce brings them back. We don't code six logics: we decompose along two orthogonal axes and compose one collection discipline × one merge kernel. Each child of a fan-out carries a stable fan_id — that is what correlates the scatter to its collection.

src.fan(into=[worker], fan_id="claims", expect="auto")   # MAP : fan_id carried on each child

worker >> Reduce("brief", kernel="append")             # ADJACENT collection (the current join)
Gather("collect", fan_id="claims", n=8, mode="perfect")  # distant COLLECTOR, closes at 8, transits
Gather("pool",    fan_id="claims",       mode="latent")  # open COLLECTOR, pulled at step n

Latent vs Perfect — the closure discipline

Perfect: an expected count n; when n events of the same fan_id have arrived, the collector closes and transits downstream (push barrier). Latent: no closure; a StatefulRecord agglomerates continuously and a step n pulls it (pull).

mode="perfect", n=8   # closes at 8, then transits         → generalizes MergeNode.wait_n
mode="latent"         # Record gather:claims agglomerates ; "report" runs bag_read("gather:claims")

The five kernels — how they are written

A hand-built WoT — three collectors on a single fan-out

Claims processing: we split by file, then gather in three different ways in parallel — a KPI that summarizes on the fly, the complete list, and a big dataset passed by reference.

hive = Hive(app, name="claims-wot", type="wot", tracer=tracer)
triage, assess, report = hive.stage("triage", triage_fn), assess_agent, report_agent

(hive.from_input()
   >> triage
   .fan(into=[assess], fan_id="claims", expect="auto")          # MAP : 1 file → N

   # (1) LATENT collector + sliding window — KPI summarized on the fly
   >> Gather("kpi_pool", fan_id="claims", mode="latent",
             kernel=Window(size=3, summarize=SUMMARY_PROMPT), consume_at="report")

   # (2) PERFECT collector + append — complete list, closes at N, transits
   >> Gather("all_claims", fan_id="claims", mode="perfect", kernel="append")

   # (3) PERFECT collector + OKF — big dataset → ref + schema, no inline
   >> Gather("dataset", fan_id="claims", mode="perfect",
             kernel=Okf(store="clickhouse", partial=True))

   >> report.to_loop())     # report reads what it needs, without overflow

OKF — the ref is in the bag, with its schema and its partial-access interface

When the kernel is Okf, the reduce does not inline the data: it writes the heavy object into a store (reuses NodeRecord/resolve_node) and places in the bag an item = the ref + the evolving schema (number of rows, one description per field) + an interaction interface. You then retrieve only certain parts — no overflow.

# OKF item in the bag (compact — what the agent SEES, never the data) :
{ "ref": "okf:claims#v3",
  "schema": { "rows": 1240,
              "columns": { "branch": {"type":"str",   "desc":"agency"},
                           "risk":   {"type":"float", "desc":"score 0-1"} } } }

# rendered into context via a schema-only frame :
<tckm domain="okf" ref="okf:claims#v3"/>     → "okf:claims#v3 — 1240 rows; branch, risk"

# the interaction interface — you retrieve ONLY a slice, never everything :
okf_read("okf:claims#v3", fields=["branch","risk"], rows="0:50")   # slice
okf_read("okf:claims#v3", where="risk>0.8", limit=20)             # filter
okf_stats("okf:claims#v3", field="amount")                        # aggregate, not the rows

Already there: the adjacent join (_node_join, MergeNode), the progressive digest (dry_context), the per-event .transform, the durable agglomeration (ContextRLMRecord), the by-ref (NodeRecord/resolve_node) and the compact schema (BAG_INFO + render_bag). New: a stable multi-hop fan_id, the gather keyed by fan_id, the latent/perfect enum, multiple collectors per fan-out, the unified OKF record + partial okf_read, the N-message window, the consume-at-step-n gate. The rest is declarative wiring; the framework executes the barrier.

The fan_id + gather seam — implementation spec

The prerequisite of the entire "gather" axis. Two halves: fan_id (a stable fan-out id that survives every hop) and gather (the collector keyed by that fan_id, in latent or perfect mode). Neither exists today — here is the plan, anchored on what already exists.

# spec : a STACK (handles nested fans AND multi-collector)
@dataclass
class FanFrame:
    fan_id: str        # stable id of the fan-out
    seq: int           # index of this child (0..n-1)
    expected: int      # expected n

class WotRecord(StatefulRecord):
    fan_ctx: list[FanFrame] = field(default_factory=list)   # top = innermost fan

fid = f"{node}#{parent_event_id}"                 # deterministic → resume-safe
for i, child in enumerate(items):
    child.fan_ctx = [*parent.fan_ctx, FanFrame(fid, i, len(items))]   # inherit + push

def carry_fan(src, dst):
    if src.fan_ctx and not dst.fan_ctx:
        dst.fan_ctx = list(src.fan_ctx)
    return dst
# called : advance()/child() (wot_record.py:46), .transform (core.py:2071),
#          the _bag traveler on handoff, header x-fan-ctx (cast.py)

@sr_init(entity="gather", initial="open", states=["open","closed"])
class GatherRecord(StatefulRecord):
    fan_id: str = ""; mode: str = "perfect"; expected: int = 0
    seen: set = field(default_factory=set)      # {seq} — dedup (Ray may redeliver)
    kernel: str = "append"; kstate: dict = field(default_factory=dict)
    result: Any = None; closed: bool = False

async def _node_gather(rec, ev, *, gather_node, fan_id, n, mode, kernel):
    if fan_id not in [f.fan_id for f in ev.fan_ctx]:
        return ev                               # not our fan-out → pass-through
    g = ensure_gather(gather_node, fan_id, mode, n or fan_expected(ev, fan_id), kernel)
    seq = fan_seq(ev, fan_id)
    if seq in g.seen: return None               # dedup / idempotence
    g.seen.add(seq); apply_kernel(g, ev)        # INCREMENTAL
    if mode == "perfect" and len(g.seen) >= g.expected:
        g.result = finalize_kernel(g); g.closed = True
        return g.result                         # closes + TRANSITS downstream
    return None                                 # latent, or perfect not full → nothing emitted

KERNELS = {                                       # apply = incremental ; finalize = close
  "append":    (push,          lambda g: g.kstate["xs"]),        # _node_join / _merge_dicts
  "window":    (window_fold,   lambda g: g.kstate["digest"]),    # extends dry_context
  "transform": (user_fn,       lambda g: user_fn(end=True)),     # .transform
  "rlm":       (rlm.add_message, lambda g: rlm_record),          # ContextRLMRecord
  "okf":       (okf_write,     lambda g: g.kstate["ref"]),       # the REF, not the data
}

Order: ① fan_ctx + stamping + carry_fan (test: fan_id intact at 3 hops) → ② gather perfect × append (closes at N, emits the list) → ③ plug in the kernels → ④ latent + consume_at → ⑤ ambient + multi-collectors → ⑥ timeout/dedup.