embed_train

How it differs

Not a partition_loop map. Shared mutable W is trained over multi-pass pair streams:

load vocab counts → filter min_token_count
  → stream positive pairs (bounded pending)
  → SGD update W (inprocess | processes SHM | Ray actor)
  → snapshot vectors → StorageSession (parquet or qdrant)
  → observation / geometry metrics

SGNS math lives in framework/algorithms/embed/.
Orchestration lives in processing/embed_train/ + embed_vocab/.

Shared train loop (P2)

embed_vocab/train_loop.py:

• material_groups
• skip_numpy_geometry_for_output (qdrant)
• run_train_over_groups(backend, groups, train_one_group=…)

Backends (inprocess, processes, ray_actor) implement only train_one_group + transport.
Do not duplicate the outer loop / plan-done logging.

Streaming contract

• streaming/pending.py — bounded in-flight;
• streaming/async_sink.py — observation channel (drop-oldest for telemetry);
• streaming/backpressure.py — put_with_backpressure;
• capacities from runtime.streaming.* in DRTML.

Live observations use a separate StorageSession / PG connection.