Fan-out — map & reduce dans un Web-of-Trust

Les stratégies de reduce — comment ramener un fan-out

Un WoT fan out : un événement se disperse en N enfants (le map). Le reduce les ramène. On ne code pas six logiques : on décompose sur deux axes orthogonaux et on compose une discipline de collecte × un kernel de fusion. Chaque enfant d'un fan-out porte un fan_id stable — c'est lui qui corrèle la dispersion à sa collecte.

src.fan(into=[worker], fan_id="claims", expect="auto")   # MAP : fan_id porté sur chaque enfant

worker >> Reduce("brief", kernel="append")             # collecte ADJACENTE (le join actuel)
Gather("collect", fan_id="claims", n=8, mode="perfect")  # COLLECTEUR distant, ferme à 8, transite
Gather("pool",    fan_id="claims",       mode="latent")  # COLLECTEUR ouvert, tiré à l'étape n

Latent vs Perfect — la discipline de fermeture

Perfect : un nombre attendu n ; quand n events du même fan_id sont arrivés, le collecteur ferme et transite en aval (barrière push). Latent : aucune fermeture ; un StatefulRecord agglomère en continu et une étape n le tire (pull).

mode="perfect", n=8   # ferme à 8, puis transite          → généralise MergeNode.wait_n
mode="latent"         # Record gather:claims agglomère ; "report" fait bag_read("gather:claims")

Les cinq kernels — comment on les écrit

Un WoT à la main — trois collecteurs sur un même fan-out

Traitement de sinistres : on éclate par dossier, puis on ramasse de trois façons différentes en parallèle — un KPI qui se résume au fil de l'eau, la liste complète, et un gros dataset passé par référence.

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 dossier → N

   # (1) collecteur LATENT + fenêtre glissante — KPI résumé au fil de l'eau
   >> Gather("kpi_pool", fan_id="claims", mode="latent",
             kernel=Window(size=3, summarize=SUMMARY_PROMPT), consume_at="report")

   # (2) collecteur PERFECT + append — liste complète, ferme à N, transite
   >> Gather("all_claims", fan_id="claims", mode="perfect", kernel="append")

   # (3) collecteur PERFECT + OKF — gros dataset → ref + schéma, pas d'inline
   >> Gather("dataset", fan_id="claims", mode="perfect",
             kernel=Okf(store="clickhouse", partial=True))

   >> report.to_loop())     # report lit ce dont il a besoin, sans overflow

OKF — la ref est dans le bag, avec son schéma et son interface d'accès partiel

Quand le kernel est Okf, le reduce n'inline pas la donnée : il écrit l'objet lourd dans un store (réutilise NodeRecord/resolve_node) et pose dans le bag un item = la ref + le schéma évolutif (nb de lignes, une description par champ) + une interface d'interaction. On récupère alors seulement certaines parties — pas d'overflow.

# item OKF dans le bag (compact — ce que l'agent VOIT, jamais la donnée) :
{ "ref": "okf:claims#v3",
  "schema": { "rows": 1240,
              "columns": { "branch": {"type":"str",   "desc":"agence"},
                           "risk":   {"type":"float", "desc":"score 0-1"} } } }

# rendu en contexte via un frame schéma-only :
<tckm domain="okf" ref="okf:claims#v3"/>     → "okf:claims#v3 — 1240 lignes; branch, risk"

# l'interface d'interaction — on récupère QUE une tranche, jamais tout :
okf_read("okf:claims#v3", fields=["branch","risk"], rows="0:50")   # tranche
okf_read("okf:claims#v3", where="risk>0.8", limit=20)             # filtre
okf_stats("okf:claims#v3", field="amount")                        # agrégat, pas les lignes

Déjà là : le join adjacent (_node_join, MergeNode), le digest progressif (dry_context), le .transform par-event, l'agglomération durable (ContextRLMRecord), le by-ref (NodeRecord/resolve_node) et le schéma compact (BAG_INFO + render_bag). Neuf : un fan_id stable multi-hop, le gather clé par fan_id, l'enum latent/perfect, plusieurs collecteurs par fan-out, le record OKF unifié + okf_read partiel, la fenêtre à N messages, le gate consume-at-step-n. Le reste est du câblage déclaratif ; le framework exécute la barrière.

Le seam fan_id + gather — spec d'implémentation

Le pré-requis de tout l'axe « gather ». Deux moitiés : fan_id (un id de fan-out stable qui survit à tous les hops) et gather (le collecteur clé par ce fan_id, en latent ou perfect). Aucun des deux n'existe aujourd'hui — voici le plan, ancré sur l'existant.

# spec : une PILE (gère les fans imbriqués ET le multi-collecteur)
@dataclass
class FanFrame:
    fan_id: str        # id stable du fan-out
    seq: int           # index de cet enfant (0..n-1)
    expected: int      # n attendu

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

fid = f"{node}#{parent_event_id}"                 # déterministe → resume-safe
for i, child in enumerate(items):
    child.fan_ctx = [*parent.fan_ctx, FanFrame(fid, i, len(items))]   # hérite + pousse

def carry_fan(src, dst):
    if src.fan_ctx and not dst.fan_ctx:
        dst.fan_ctx = list(src.fan_ctx)
    return dst
# appelé : advance()/child() (wot_record.py:46), .transform (core.py:2071),
#          le voyageur _bag sur 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 peut redélivrer)
    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                               # pas notre fan-out → passe-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)        # INCRÉMENTAL
    if mode == "perfect" and len(g.seen) >= g.expected:
        g.result = finalize_kernel(g); g.closed = True
        return g.result                         # ferme + TRANSITE en aval
    return None                                 # latent, ou perfect pas plein → rien ne sort

KERNELS = {                                       # apply = incrémental ; finalize = clore
  "append":    (push,          lambda g: g.kstate["xs"]),        # _node_join / _merge_dicts
  "window":    (window_fold,   lambda g: g.kstate["digest"]),    # étend 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"]),       # la REF, pas la donnée
}

Ordre : ① fan_ctx + stamping + carry_fan (test : fan_id intact à 3 hops) → ② gather perfect × append (ferme à N, sort la liste) → ③ brancher les kernels → ④ latent + consume_at → ⑤ ambient + multi-collecteurs → ⑥ timeout/dedup.