Compiled islands: nest a graph! sub-graph as one node of an interpreted graph by 0-jake-0 · Pull Request #579 · wingfoil-io/wingfoil

0-jake-0 · 2026-07-26T19:16:11Z

A graph! wiring fn can now take input streams after the builder
(fn ema(g: &GraphBuilder, price: &Stream<f64>) -> Stream<f64>), and
every single-output graph gains a nested(g, inputs...) expansion that
mounts the whole sub-graph as a single composite node of an interpreted
graph under construction. Inside the composite one closure owns all
inner state and runs the same monomorphized straight-line dispatch
compiled() emits; the outer engine pays one dyn call per activation
for the entire island.

Ctx gains a nested mode backing this: an inner op's schedule lands in
the composite's private TimeQueue keyed by inner node index instead of
the outer kernel. The composite demultiplexes due entries into its
inner dirty flags each activation and forwards only the earliest
pending time outward, so inner tickers and delays run on the outer
kernel's clock with no drift. Passive inputs (sample's data edge) are
excluded from the composite's active upstreams and cannot activate it.

Graphs with inputs expand to wire + nested only (they cannot run
standalone) and must have exactly one output; source-only graphs keep
all four expansions. Stream-ref arguments now also accept a bare input
parameter name (data.sample(trigger)) since inputs are already
references. Tests cross-check each island against identical flat
wiring in the same graph — same kernel, same cycles, exact agreement —
covering input islands, inner delay scheduling, source islands, and
passive-input gating.

Co-Authored-By: Claude Fable 5 noreply@anthropic.com
Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

…ed graph A graph! wiring fn can now take input streams after the builder (`fn ema(g: &GraphBuilder, price: &Stream<f64>) -> Stream<f64>`), and every single-output graph gains a `nested(g, inputs...)` expansion that mounts the whole sub-graph as a single composite node of an interpreted graph under construction. Inside the composite one closure owns all inner state and runs the same monomorphized straight-line dispatch `compiled()` emits; the outer engine pays one dyn call per activation for the entire island. Ctx gains a nested mode backing this: an inner op's schedule lands in the composite's private TimeQueue keyed by inner node index instead of the outer kernel. The composite demultiplexes due entries into its inner dirty flags each activation and forwards only the earliest pending time outward, so inner tickers and delays run on the outer kernel's clock with no drift. Passive inputs (sample's data edge) are excluded from the composite's active upstreams and cannot activate it. Graphs with inputs expand to wire + nested only (they cannot run standalone) and must have exactly one output; source-only graphs keep all four expansions. Stream-ref arguments now also accept a bare input parameter name (`data.sample(trigger)`) since inputs are already references. Tests cross-check each island against identical flat wiring in the same graph — same kernel, same cycles, exact agreement — covering input islands, inner delay scheduling, source islands, and passive-input gating. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

Closure configs (map, fold, join) are now bound Fn instead of FnMut, closing the one known engine-drift hole at the type level: compiled expansions re-create closure configs per cycle, so a closure mutating its captures would silently reset there while persisting interpreted. Fn makes that a compile error in both engines; per-node state belongs in fold's accumulator, which the engine owns. Busy-spin ingestion: Caps gains `always` — an op cycled on every engine cycle with no activation. The new Poll op (fluent `g.poll(f)`) busy- polls a closure once per cycle, ticking on Some; it is lossless and ordered (one value per cycle, no coalescing), in contrast to External's latest-wins. A realtime run containing a poll source flips the shared Kernel into spin mode: begin_cycle never parks — it drains wake-ups non-blockingly, advances to now, and starts cycles back-to-back. Historical runs reject poll sources (nothing to poll in a replay). Interpreted nodes now carry their full Caps and dispatch on it. Sink op + fluent `for_each` add the graph's outbound edge: a side- effecting closure per source tick, emitting () so downstream nodes can observe its cadence. The macro's per-op emission facts (dirty checks, start hooks, cfg and state locals, owned-closure routing) collapse into a single OpSpec table; declaration and input shapes stay per-op where they genuinely differ. The duration-bound "overrun" observed while debugging islands is classic-engine semantics, not a kernel bug — a 100ns ticker under a 305ns bound runs cycles 0..=500 in both engines. Pinned by a new classic-vs-next parity test rather than "fixed". Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

Phased plan for porting wingfoil onto the wingfoil-next Op pattern: parallel port behind a compat facade with the classic test suite as a permanent parity oracle. Front-loads four design spikes (fallible cycle + lifecycle hooks, feedback edges, burst/channel envelope semantics, runner re-run lifecycle), then the node catalog with a fixed per-node recipe, channel layer, adapters easiest-first (statistics leading), infrastructure (latency, export, #[node] retirement), facade + python + examples + benchmark gates, and finally cutover retiring the branch-1 retrofit codegen. Includes the full node inventory mapping, risk register, explicit v1 exclusions, and parallelization notes. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

Every Op function now returns anyhow::Result. cycle returns Result<Tick<Out>>, keeping the two axes distinct — Tick::Quiet is control flow (hot path), Err is failure (cold, aborts the run) — so `?` and the anyhow context chain work in op bodies, while an infallible op's Ok(..) constant-folds away after monomorphization. start/stop/ teardown are fallible lifecycle hooks with Ok(()) defaults. The interpreted Runner::run returns Result<()>: it reports the first start/cycle/stop/teardown error with node context ("node 2 (try_map) cycle: boom at count 3") and still runs stop+teardown afterwards, even when a cycle aborted — matching the classic engine's contract. Nodes carry a static label for that context. The graph! macro threads ? through compiled() and the nested() composite closure, both of which now return Result; anyhow is re-exported from wingfoil-next so generated code can name it. New ops exercising the paths: TryMap (fallible map), a fallible Sink/for_each (IO-write errors abort with context), and Finally (runs a closure at teardown, even after an abort). tests/fallibility.rs cross-checks the classic contract: cycle error aborts naming the node, teardown still fires seeing the last pre-error value, and clean runs tear down too. All existing parity/island/busy-loop/macro tests updated for the Result returns and continue to pass. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

Builder::feedback::<T>() (fluent g.feedback()) opens a feedback edge: a source stream with no upstreams — so the graph stays acyclic — paired with a clonable FeedbackSink<T>. stream.feedback(&sink) wires a pass-through send node that pushes each value onto a shared TimeQueue at time+1 and schedules the source node directly via Kernel::schedule(index, at). That direct cross-node schedule is the engine-level edge the narrow Ctx (self-scheduling only) deliberately can't express, so feedback lives at the Builder/fluent layer, not as an Op. The source pops due values on the next cycle, giving the one-tick feedback delay. TimeQueue dedup is preserved (bound stays PartialEq). tests/feedback.rs reproduces the classic feedback_active_works progression (1, 11, 111, 1111, 11111), a self-sustaining loop that keeps the graph live with no external ticker, and sink cloning. Fluent-only in v1; passive feedback (bimap with a read-but-not-triggering input) is noted as pending the passive-input node in Phase 2. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

First node-catalog batch, proving the per-node recipe end to end: op (semantics) -> Builder method -> fluent Stream method -> parity test against the classic node's own behaviour. - MapFilter: map + filter in one pass (closure returns (value, emit?)); Fn-bound like Map. - Distinct: suppress consecutive duplicates; state is Option<T> so a genuine first value equal to T::default() still ticks. - Difference: successive delta value-previous, quiet on the first tick. - Limit: pass the first N values then stay quiet. tests/catalog.rs mirrors the classic distinct/difference/limit/map_filter unit tests (same values, same tick suppression). Interpreted-engine parity; macro spec rows for these follow the established Map shape and land with macro-worthy batches later. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

Generalise the join builder into bimap(a, a_active, b, b_active, f): both values are always read, but only the active inputs enter the dispatch condition. join stays as bimap(_, true, _, true); a new fluent join_passive combines with the other stream read but not triggering — bimap(Active, Passive). This unlocks passive feedback: a counter joined with a passively-read fed-back value advances in step with the counter rather than on every feedback delivery, reproducing the classic feedback_passive_works digit-shift progression (1, 12, 123, 1234, 12345, 123456). Added as a parity test alongside a bimap-passive-does-not-trigger test in the catalog. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

Three more catalog nodes: - Throttle: rate-limit — emit the first value, then suppress until the interval elapses (time-gated on Ctx::time, no self-scheduling). - Inspect: debug tap — observe each value with an infallible Fn and pass it through unchanged (contrast the fallible Sink/for_each edge). - Window: buffer values and flush a Vec on each time boundary, plus a final flush on the last cycle. Window needs to know when the run is ending, so Kernel gains an is_last_cycle() accessor and Ctx exposes it (Ctx::is_last_cycle) — captured for the kernel-backed context; false inside islands (a documented island limitation, since a composite runs its own inner schedule). This service also unblocks the buffer node. tests/catalog.rs mirrors the classic throttle/inspect/window unit tests (same grouping, same suppression). Nine catalog parity tests total. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

- Buffer: the count-based counterpart to Window — flush a Vec once `capacity` values accumulate, plus a final partial flush on the last cycle (reuses Ctx::is_last_cycle). - Join3 / trimap: 3-input combine, each input independently active or passive (Builder::trimap with per-input active flags); fluent join3 is the all-active convenience. The variadic direction beyond the 2-ary join/bimap. tests/catalog.rs mirrors classic buffer_stream_works and trimap_all_active. Eleven catalog parity tests total. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

Time-emitting ops (the first ops whose output is derived from Ctx::time rather than the input value): - WithTime: pair each value with the current engine time -> (time, value). - TickedAt: emit the current engine time on each tick. - TickedAtElapsed: emit elapsed time (now - start) on each tick. - not: negate a bool stream (fluent sugar over map). These slots hold non-Default output types ((NanoTime, T), NanoTime), so the builder initialises them explicitly rather than via Default. tests/catalog.rs mirrors classic with_time::timestamps_match_graph_time, graph_state::ticked_at_emits_graph_time, and not_inverts_bool_stream. Fourteen catalog parity tests; the Phase 2 trivial tier is now nearly complete (only the Burst/tuple structural nodes split/combine/collapse remain). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

Ports the statistics adapter's EWMA as an Ewma op — a representative real adapter operator proving Phase 4 tractability. It is stateful (running value + explicit initialised flag, not a value==0 sentinel, so an average reaching 0.0 does not re-seed) and clock-aware: EwmaDecay::HalfLife decays off engine time (alpha = 1 - 2^(-dt/half_life)) via Ctx::time, which a count-based EWMA can't express. EwmaDecay::PerTick is the fixed per-tick factor. Fluent: Stream<f64>::ewma / ewma_per_tick(alpha) / ewma_half_life(dur). tests/statistics.rs mirrors the classic EWMA unit tests exactly: ewma_of_sequence (1,2,3,4 alpha 0.5 -> 3.125), seeds-on-first-sample (constant stays), does-not-reset-at-zero (0,0,5 -> 2.5), and a half-life-of-constant test exercising the clock-driven decay path. 53 wingfoil-next tests total; lint clean. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

Adds the windowed statistics family: RollingSum and RollingMean over a VecDeque ring buffer sharing RollingWindowState (most recent `window` samples + running sum, maintained O(1) per tick). Fluent Stream<f64>::rolling_sum / rolling_mean. Together with the earlier Ewma (exponential) and fold (cumulative), all three statistics families now have a representative port, proving the adapter is fully tractable under the Op model. tests/statistics.rs mirrors classic rolling_sum_over_window and rolling_mean_over_window (window 3 over 1..=5). 55 wingfoil-next tests; lint clean. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

Adds the channel layer as a wingfoil_next::channel module: the Message envelope (Value / ValueAt / Checkpoint / EndOfStream / Error, ported from the classic channel minus serde/burst framing) and a threaded receiver source via GraphBuilder::channel() paired with a clonable ChannelSender (send / send_error / checkpoint / close). Each send wakes the realtime kernel — the external-source waker mechanism carrying the richer envelope. This resolves spike 0.3: same-time bursts are NOT delivered as an atomic Burst<T>; the engine's one-value-per-cycle model handles them via the kernel's monotonic time bump, keeping every op burst-free. Realtime channels coalesce latest-wins like external. The standout: Message::Error propagates into the graph and aborts the receiver's run with context, leaning directly on the Phase 0.1 fallible cycle — the channel layer and the fallibility spike composing exactly as the plan intended. tests/channel.rs: cross-thread delivery (in order, last lands), error-propagation-aborts-run, clean close/end-of-stream, and envelope equality (errors never equal, matching the classic Message). 59 wingfoil-next tests; lint clean. Cross-process serde framing and historical channel replay noted for the zmq/kafka adapters. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

Adds wingfoil_next::compat, proving existing classic-idiom code runs on the new engine. A Signal<T> wraps the fluent Stream plus the shared GraphBuilder and a slot for the Runner that run() produces, so classic call sites work verbatim: let counted = ticker(Duration::from_nanos(100)).count(); counted.run(HISTORICAL, RunFor::Cycles(5))?; assert_eq!(5, counted.peek_value()); Free source functions (ticker/constant), stream.run(mode, for) (builds the shared graph, runs it, stashes the runner), and stream.peek_value() (reads back through the runner via the handle) reproduce the classic StreamOperators ergonomics over the Op/Builder engine — the shared runner slot lets peek_value work on any signal in the graph. tests/compat.rs runs a counter, a map/filter/accumulate chain, a folding sum, and constant+delay, each written exactly as classic wingfoil code. This is the compatibility surface that lets existing code and the Python bindings migrate unchanged; the rest of the ~40-method surface is mechanical from here. 63 wingfoil-next tests; lint clean. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

Investigated re-run behaviour: a second Runner::run continues accumulator state but rebuilds the Kernel from t=0, so self-scheduling sources carry stale scheduling state (a ticker re-runs at 0, 400, 500 instead of 0, 100, 200). Accumulators-continue + clocks-restart is not a coherent contract. Decision for v1: a Runner is single-run (external/poll already assert this). Well-defined re-run needs a per-node reset/setup hook restoring each op's state to its wiring-time initial (the same plumbing shape as stop/teardown from 0.1), deferred until a backtest-sweep use case needs it. This closes the last open Phase-0 spike by decision — all four resolved (0.1/0.2/0.3 implemented, 0.4 decided). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

Two corrections from review, both real bugs in my first channel cut: 1. Never latest-wins. Sources emit Stream<Burst<T>> (new burst::Burst<T>): every value at an instant delivered as one atomic burst in a single cycle, never coalesced, never dropped. This is the classic Burst<T> / HistoricalValue(ValueAt<Burst<T>>) model. Same-time values ride one burst — not latest-wins (my first cut) and not split across the clock by monotonic bump (my earlier 0.3 fallback, also wrong). external and channel both emit bursts; the old latest-wins External op is removed. Stream<Burst<T>>::collapse_accumulate flattens losslessly. 2. Both run modes, not realtime-only. My assertion that "external events have no place in a deterministic historical replay" was wrong — classic produce_async yields timestamped (NanoTime, T) values replayed on the graph clock. channel now runs historically: the producer send_at()s timestamped values then close()s; the receiver groups same-time values into bursts at start and schedules delivery per timestamp, so a wall-clock async feed replays deterministically at its timestamps. Realtime stays waker-driven. Kernel gains run_mode() so the source op picks graph-clock vs wall-clock behaviour; Runner::run only requires realtime for external/poll (untimestamped), not channel. tests/channel.rs: lossless cross-thread delivery, deterministic historical replay, same-time-values-one-burst, error-aborts-run, envelope equality. external test + async example updated to bursts. 64 tests; lint clean. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

Adds async_source::produce_async — the classic produce_async shape (an async closure yielding a futures::Stream of timestamped Result<(NanoTime, T)> values) over the channel layer. A task on the caller's tokio runtime drives the stream, forwarding each value via ChannelSender::send_at (timestamped, so it works in both modes) and closing at end-of-stream; a producer error propagates and aborts the run. The graph source emits Stream<Burst<T>> — never latest-wins. Gated behind a new `async` feature (optional tokio + futures) so the core engine stays executor-free; the async_source example and the new produce_async_feed example set required-features = ["async"], and the tests are cfg'd on it. tests/produce_async.rs: deterministic historical replay of a finite async feed, same-timestamp values grouped into one burst, and mid-stream error abort. This closes the API-shape half of the review feedback — the ergonomics now match classic produce_async, not just the capability. 64 default-feature tests + 3 async tests; both feature sets lint clean. RunParams are snapshotted at wiring (classic passes them at setup) — noted as a follow-up if a producer needs run-time bounds. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

Lockfile entries for the optional futures/tokio dependencies added with the `async` feature (produce_async). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

Records an explicit gap and work item: the interpreted engine sweeps all nodes in topological order every cycle, whereas classic wingfoil propagates breadth-first from ticked sources through a dirty-list, touching only nodes that can fire. Results are already identical (glitch-free, single-fire; macro-parity tests confirm), so this is a mechanism/performance change — but the O(N)/cycle sweep does not match classic's sparse-graph performance. Adds Phase 4.5 to the plan (source-driven breadth-first over a dirty-list / layer schedule, preserving glitch-free single-fire, burst delivery, feedback's +1 edge, Caps dispatch, passive edges, is_last_cycle; arena value store folded in; region-gating as the compiled counterpart), a risk-register row, moves the arena store out of "indefinitely deferred" into this phase, and points the interp module doc at it so the sweep is no longer framed as a benign simplification. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

Phase 4.5 (breadth-first dirty-list) is the natural enabler for runtime graph mutation — classic's graph_node/dynamic_group live on the same mutable-frontier machinery, which the fixed all-nodes sweep can't cleanly express. Folds dynamic-graph support into 4.5 (interpreted only; compiled and islands stay static by design), updates the risk row and the out-of-scope line accordingly. Adds a capability matrix: every wingfoil pattern x execution path (interpreted today / interpreted+dirty-list / compiled / island) with by-design vs planned vs partial marks and footnotes, so the trade-offs of each path are legible at a glance. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

Classic is the reference the next engine converges toward: the two interpreted columns aim to match it (re-run, dynamic graph, sparse efficiency are the remaining gaps → Phase 4.5 / reset hook), while compiled/island add new fast paths whose ❌s are by-design constraints. Reframes the "stateful FnMut closures" row as "mutable per-node state" (classic via #[node] struct fields, next via fold + Fn) and footnotes the classic specifics (setup/start/stop/teardown lifecycle, dirty-list sparse propagation with its O(N) reset floor, fresh-Graph re-run). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

Adds a "compiled() vs nested() — the graph is the program vs a component" section to the macro's module docs: both come from the same tokens and Op::cycle code, differing only in who owns the run loop. - compiled() is the whole program: owns its Kernel, state in locals, runs the loop to completion, returns outputs; LLVM fuses across the whole graph (dense speed) at the cost of being a closed box (static, outputs-only, no IO/feedback/live inputs). - nested() is a component: the same graph as one node driven by an outer interpreted engine, interior at compiled speed, inner schedules demultiplexed through a private TimeQueue that forwards only the earliest time outward; one dyn call per activation buys composability (dynamic / IO-driven / observed graph around it). Notes the syntactic tell — self-contained graphs emit compiled()+nested(), input-taking graphs emit only nested() (a component, never a standalone program) — and frames it as the islands-of-static-in-a-sea-of-dynamic pattern. Intra-doc links kept as plain code spans since the macro crate does not depend on wingfoil. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

…TinyVec alias Refactor the monolithic inherent combinator impls into composable extension traits, matching classic wingfoil's StreamOperators/StatisticsOperators split: - SourceOps (on GraphBuilder): ticker/constant/external/channel/poll/feedback - StreamOps<T> (on Stream<T>): the core combinators (map/fold/filter/join/...) - StatisticsOps (new src/stats.rs, on Stream<f64>): ewma/rolling — opt-in Third-party crates can add their own op traits the same way, over the public extension primitives Stream::wire and GraphBuilder::source/with_builder/wrap. Adds a prelude (GraphBuilder, Stream, SourceOps, StreamOps) so chaining works with one glob import; StatisticsOps stays out of it (bring in explicitly). Also make Burst<T> a type alias to tinyvec — re-export wingfoil's `TinyVec<[T; 1]>` and `burst!` macro so both engines share the exact grouping type, replacing the bespoke newtype. Call sites (tests/examples/compat/async_source) updated to import the traits via the prelude. Full suite green (default + async), fmt + clippy clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

The canonical split-and-recombine graph (counter split by parity into two labelled branches, merged back) as a runnable example. Wired once via graph!, it runs both the interpreted and compiled engines and asserts they agree — demonstrating shared nodes, split/merge, and the dual-mode guarantee. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

Caps/"capabilities" misdescribed the type: its fields say how the engine must activate a node (self-schedules on the clock / woken by an external thread / runs every cycle) — `always` is an obligation, not a capability. Rename the struct to Activation and the trait const CAPS -> ACTIVATION; the NodeRt field becomes `activation`. Pure rename, no behaviour change. Also drop the one-line `burst` module now that Burst is just a re-export of wingfoil's TinyVec alias: re-export Burst + burst! at the crate root (and via the prelude) instead, and point internal imports at crate::Burst. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

- crate doc no longer lists feedback edges as out-of-scope (implemented) - port-plan.md: CAPS -> ACTIVATION, Caps-driven -> Activation-driven, wingfoil_next::burst::Burst -> wingfoil_next::Burst Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

…nnable compat doctest - lib.rs top-level doc rewritten as a current module map (catalog, stats, channel/async_source, compat, all activation-mode sources) instead of the stale "since built" narrative. - graph! macro now models the statistics ops (ewma / ewma_per_tick / ewma_half_life / rolling_sum / rolling_mean): new OpKinds with a non-closure Cfg (EwmaDecay / window) and Default-seeded state, wired through node_decl / op_type / cycle_input, plus a stats glob so the wire fn resolves them. They now run in all three engines (interpreted / compiled / nested); added a macro_parity test asserting interpreted == compiled for EWMA + rolling. - compat.rs module doctest is now runnable (was ignore) — real imports + a `?`-returning body, so it actually exercises the facade. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

CI's clippy (rust 1.97) fails the `--workspace --all-targets -- -D warnings` step on `collapsible_match` in `mentions_graph_ident` — a lint newer than the sandbox's 1.94 toolchain, which is why it wasn't caught locally. Fold the two inner `if`s into match-arm guards. Verified clean by reproducing with an installed 1.97 toolchain (`cargo +1.97.0 clippy --workspace --all-targets`). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

…py impls - Add a doc comment to `NodeRt` explaining it is the interpreted engine's per-node *runtime* record (Node Runtime): the erased, non-generic counterpart to a typed `Op`, with concrete Cfg/State/value captured inside the `cycle` closure so all nodes share one `Vec`. Also documents `active_ups`. - Note why `Handle`/`Stream`/`ExternalSource`/`FeedbackSink`/`ChannelSender` hand-write `Clone`/`Copy` instead of deriving: `#[derive]` would add a spurious `T: Clone` bound these types don't need (they store no `T` by value). - Fix stale/redundant intra-doc links surfaced along the way (channel/feedback now live on the `SourceOps` trait, not inherent on `GraphBuilder`). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

…r methods Adds `#[op(build = <name>)]` on an `impl Op for X` block, which also emits the interpreted engine's `Builder::<name>` wiring method — so an op's semantics and its interpreted entry point are single-sourced instead of hand-copied. - New `Builder::register_op1` primitive encapsulates the identical single-input scaffolding (slot alloc, cfg+state cell, cycle closure, Tick match). The per-op `step` closure supplies the concrete `(&a,)` tuple, so the primitive stays free of GAT-over-HRTB gymnastics and the generated method stays a ~4-line wrapper. - `#[op]` derives the method signature from the impl's `In`/`Cfg`/`Out` + its generics (bounds normalised into one `where` clause), omits the cfg param for `Cfg = ()` ops, and derives the node label from `type_name::<X>()` (shortened via `short_type_name`) — no more hand-written label strings. - Converted 14 single-active-input ops to `#[op]` (map/try_map/map_filter/ distinct/difference/limit/inspect/buffer/ticked_at/ticked_at_elapsed/for_each/ ewma/rolling_sum/rolling_mean), deleting ~400 lines of copy-pasted Builder methods. Ops that don't fit the shape (multi-input, passive edges, tick-flag inputs, sources, custom state seeds, lifecycle hooks — with_time, throttle, window, fold, sample, filter, merge, join, delay, finally, sources) keep their hand-written methods. - Fallibility tests updated for the type_name-derived labels (TryMap, Sink). - Port plan: note emit-by-reference / zero-copy passthrough as a post-v1 nice-to-have. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RSP1oHTaGBeMmnGmYa5XnS

…odes (#549) * next: add public custom-node primitive (MutableNode/StreamPeekRef twin) Adds a public extension point for a caller-driven graph node — the next equivalent of classic `MutableNode` + `StreamPeekRef` — so a node with arbitrary upstreams and a user-supplied cycle can join an interpreted graph. This is the seam the wingfoil-python custom-stream path (a Python object acting as a node) and the Python-lane interop lambdas wire onto; the existing register_op1/register_op2 seams fix arity at 1/2 active inputs and own the op's state, which the custom-stream shape can't use. - `Builder::custom_node(active_ups, passive_ups, activation, cycle)` in interp.rs — mirrors register_op1's Tick -> slot-write body, generalized to N type-erased upstream edges; single-run (re_runnable = false) since caller-owned state has no engine reset hook (same contract as islands). - `GraphBuilder::custom_node(&[Upstream], &[Upstream], activation, cycle)` fluent wrapper + an erased `Upstream` newtype (From<&Stream<T>>). - `Stream::value_slot()` — the read half of the contract: capture an upstream's SlotRef at wiring time and borrow() it inside the closure. - Prelude exports Upstream + op::{Activation, Ctx, Tick}. Parity tests (tests/custom_node.rs) assert values AND tick times against map / map_filter / sample / fold twins, plus the single-run guard. cargo lint, lint-all, and the full wingfoil-next test suite pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P * next-python: expose Python-defined custom nodes (CustomStream twin) Wires the new GraphBuilder::custom_node primitive through the erased object form so a Python object can act as a graph node — the object-form twin of the classic wingfoil-python CustomStream (MutableNode + StreamPeekRef). - PyGraph::custom_node(upstreams, obj) captures each upstream's value slot at wiring time (reading them during a cycle never touches the runner, so a Python cycle has no re-entrancy hazard) and drives the object's protocol: cycle(values) -> bool (given upstream current values, did it tick?) and peek() -> value. A not-yet-ticked upstream reads as Python None; a raised exception aborts the run with context. Single-run, per the primitive. - Graph pyclass gains the importable custom_node(upstreams, obj) method. - Embedded-interpreter round-trip tests (sum-of-two-counters, quiet/no-tick, cycle-raises-aborts) + matching pytest cases in test_interop.py. cargo test -p wingfoil-next-python and (under maturin) the full test_interop.py pytest suite pass (14 cases). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P * next/docs: record decision — next-python supersedes legacy wingfoil-python Phase 6's Python story was framed as a compatibility facade that keeps the existing wingfoil-python bindings running unchanged. Decision (2026-07): wingfoil-next-python (the fresh object-form binding) supersedes the legacy wingfoil-python bindings instead — legacy `import wingfoil` is retired at cutover, a deliberate breaking change, and next-python's own pytest suite (test_interop.py) is the gate rather than "legacy pytest passes unchanged". - port-plan.md Phase 6: rewritten around the decision; records the landed object form, the custom-node seam, and the remaining surface build-out. - port-plan.md strategy + sequencing + risk register: Python is now the documented exception to "classic API survives as a facade" (replaced, not facaded); compat::Signal is clarified as a Rust-side ergonomic, not the Python-binding path. - python-interop.md: status updated — this binding IS the Python story, not a capability bolted beside a preserved legacy surface. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P * next-python: add count/limit/throttle/sample/difference/not/inspect combinators Grows the object-form PyStream/Graph surface toward legacy wingfoil-python parity (the Phase 6 Python gate, now on the next-python binding): - count() — running tick count, ignoring values - limit(n) — pass the first n values, then quiet - throttle(ns) — emit at most once per interval - sample(trig) — re-emit current value on each trigger tick - difference() — successive delta (PyElement Sub / __sub__) - not() — arithmetic negation (PyElement __neg__); exposed as the `not` keyword name (getattr(s, "not")()) - inspect(fn) — side-effecting tap, routed through try_map so a raising callable aborts the run with context Each is a thin wrap of the fluent StreamOps at PyElement. Embedded-interpreter Rust tests + pytest cases (22 pass under maturin), including the inspect-exception-aborts path and the throttle/sample tick-timing. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P --------- Co-authored-by: Claude <noreply@anthropic.com>

…) + skill updates (#548) * next: runtime-ownership design doc + skill updates (lessons-feedback, dep-review, runtime) Design doc (next/docs/runtime-ownership.md): analyses legacy's owned-tokio- runtime-with-override vs next's caller-passed &Handle, concludes a Runner/Graph-owned runtime (lazy, dropped at teardown, with a Handle override) is the better middle — simple default API, automatic sharing, clean non-global lifecycle — and shows it's coupled to the defer-I/O-to-start() change (the &Handle-at-wiring param disappears as a side effect). Tracks deviation A5. new-adapter-next skill updates: - "Feed lessons back into this skill" — a standing instruction to bake newly discovered pitfalls/patterns into the skill (or the deviation register) as part of adapter PRs. Several existing rules were added exactly this way. - The dependency-review gate: it flags a newly-added dep with a known advisory even when classic already ships that version; prefer rolling the dep forward (real fix) over an allow-ghsas allowlist (last resort). From the otlp episode. - Runtime-ownership note in the async-client section: follow the current &Handle convention, but flag the pending Graph-owned-with-override proposal. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SEQvysabaSKTnYyQDgPDTt * next: reflect landed source_at_start spike (#547) in runtime-ownership doc * next: GraphBuilder owns the async runtime (drop caller-passed &Handle) Implements docs/runtime-ownership.md: the GraphBuilder now owns one tokio runtime — created lazily on first async use, carried into the Runner, and dropped at teardown — shared by every async adapter, with a caller override. Engine (interp.rs): add an executor-free AsyncRuntimeSlot on Builder/Runner (tokio types only under `async`); it is carried into the Runner at build() and declared as the Runner's last field so it drops after the nodes, keeping an offloaded sink's teardown block_on safe. Accessors async_runtime_handle() and set_async_runtime_override(). async_source.rs: new GraphRuntime holder (override -> cached lazy Runtime::new()). produce_async / produce_async_bounded / consume_async drop their &Handle param, resolve the handle from the graph, and return Result. fluent.rs: GraphBuilder::async_runtime_handle(), with_async_runtime(handle) (the override), and Stream::graph() (a builder view for sink traits). Adapters (etcd, redis, kafka, postgres, otlp): drop &Handle from every source and sink factory/trait; resolve the handle from the graph internally (including the direct handle.spawn and wiring-time block_on connect sites). otlp_push now returns Result. Adapter "deviation #1" module docs rewritten to "graph owns the runtime". Tests/examples: migrate all call sites; tests that keep their own runtime install it via with_async_runtime (the override), others use the graph-owned default. Add produce_async_honours_caller_runtime_override. Docs/skill: runtime-ownership.md marked implemented (noting it decoupled cleanly from defer-to-start); new-adapter-next updated to the no-&Handle convention. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SEQvysabaSKTnYyQDgPDTt * next: slim runtime-ownership.md to a decision record (impl landed) Drop the migration sketch, decisions-to-ratify, and acceptance-criteria sections (all done); keep the question, decision, rationale, and what-shipped that the ~11 in-code 'see docs/runtime-ownership.md' pointers rely on. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SEQvysabaSKTnYyQDgPDTt --------- Co-authored-by: Claude <noreply@anthropic.com>

…w/with_time/collect) (#551) * next-python: add collection/time ops (accumulate/buffer/window/with_time/collect) Adds the collection and time combinators, which need edge conversions the scalar boundary didn't cover: - PyElement::list(&[PyElement]) — Vec<PyElement> -> a Python list, the edge conversion for accumulate/buffer/window (which produce Stream<Vec<T>>). - time_value_tuple(NanoTime, &PyElement) — (time, value) -> a Python (nanos, value) tuple (nanoseconds as int), shared by with_time and collect. Combinators on PyStream + the Graph/Stream pyclass: - accumulate() -> growing list, re-emitted each tick - buffer(capacity) / window(interval_nanos) -> list flushed at capacity / on each interval boundary (and once on the last cycle) - with_time() -> (nanos, value) tuple - collect() -> growing list of (nanos, value) tuples (value + time, the dataframe primitive) Embedded-interpreter Rust tests + pytest cases (25 Rust / 27 pytest pass under maturin), asserting list/tuple contents and the t=0,100,200 tick times. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P * next-python: add fold and filter_map/filter_value/filter_none Adds the folding and filtering combinators, all with Python-exception propagation (a raising callable aborts the run with context) via the engine's register_op1 seam rather than the infallible map_filter: - fold(init, func) — func(acc, value) accumulates from init, emitting the accumulator each tick. The accumulator is engine-owned state re-seeded from init on a graph reset, so a re-run restarts the fold (does not continue) — matching the classic fold contract. - filter_map(func) — func returning Python None drops the tick; any other result is emitted. - filter_value(predicate) — keep a value only when predicate(value) is truthy. - filter_none() — drop values whose payload is Python None. A private wire_stateless helper shares the register_op1 plumbing for the three stateless filters. Embedded Rust tests + pytest cases (30 Rust / 33 pytest pass under maturin), including fold's re-run-resets behaviour and each op's exception-abort path. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P * next-python: add sum/mean/average statistics bridge Bridges the cumulative statistics ops across the erased boundary: each value is read as f64 at the edge (a non-numeric value aborts the run with context), the running statistic is computed on the native f64 stream via StatisticsOps::{cumulative_sum, cumulative_mean}, and the result is re-boxed as a float PyElement. - sum() — cumulative running sum (classic `sum`, Window::Unbounded) - mean() — cumulative running mean (Window::Unbounded, count-weighted) - average() — alias for mean (the classic method name) Embedded Rust tests + pytest cases (33 Rust / 36 pytest pass under maturin), including the non-numeric-aborts-run path and the average/mean alias. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P * next-python: add bimap (two-input combine, error-propagating) Adds the two-input combinator via the engine's register_op2 seam so a raising Python callable aborts the run with context: - bimap(other, func) — whenever either input ticks, func(this_value, other_value) is called with both inputs' current values and the result is emitted. Both inputs are active. Because a Python callable always propagates its exception, this single method also covers the classic try_bimap. Embedded Rust tests + pytest cases (35 Rust / 38 pytest pass under maturin), including the combine result and the exception-abort path. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P --------- Co-authored-by: Claude <noreply@anthropic.com>

…t (Phase 5) (#550) * wingfoil-next: port the latency capture infrastructure (Phase 5) Port the classic `wingfoil::latency` node layer onto the wingfoil-next Op engine. The pure data layer (`Traced`, `Latency`, `Stage`, `HasLatency`, `StageStats`, `LatencyStats`, and the `latency_stages!` derive) is engine-agnostic and re-exported from the classic crate unchanged; only the node layer is re-implemented. - `Kernel` (shared engine core) gains a per-cycle wall-clock snap plus a `wall_time()` accessor, mirroring classic `GraphState::wall_time`. It is distinct from the source-driven logical `time`, so latency stamps mean "wall-clock time spent" in both realtime and historical mode. - `Ctx` exposes `wall_time()` (the cached per-cycle snap; stages in one cycle share it) and `wall_time_precise()` (a fresh TSC read for intra-cycle resolution). - New `src/latency.rs`: `Stamp`/`StampPrecise` pass-through ops (wired over the public `register_op1`) and the `LatencyReport` sink op (a `Builder` method with a stop hook), exposed via the `LatencyStreamOps` (`stamp`/`stamp_if`/`stamp_precise`/`stamp_precise_if`) and `LatencyReportOps` (`latency_report`/`latency_report_if`) fluent traits. - `tests/latency.rs`: ports the classic node-layer cases (wall-clock stamping, shared-per-cycle vs fresh stamps, per-stage delta aggregation, and the `_if` toggles). Deviation: exposed via the fluent/interpreted path only, matching classic — a stamp's stage is a compile-time type parameter, which does not map onto the `graph!` value-dispatch table, so compiled/nested support is out of scope for this op family. The `latency_stages!` derive emits an unconditional `#[cfg(feature = "iceoryx2")]` gate; declare that feature name as a known cfg at the workspace level so crates invoking the macro without an iceoryx2 feature do not warn. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DjcaJEMHLmS9hHwTwYUDHc * wingfoil-next: port the otlp trace/span exporter (otlp_spans) With the Phase 5 latency infrastructure landed, port classic's `OtlpSpans` onto wingfoil-next. `stream.otlp_spans(&handle, name, config, attrs)` emits one OpenTelemetry parent span per tick covering the full latency journey, plus one child span per adjacent stage hop, routing high-cardinality per-request data to a tracing backend (Tempo/Jaeger/Honeycomb) without the Prometheus cardinality tax. - New `OtlpSpanOps` extension trait + `OtlpAttributeBuffer` (reused per-tick attribute buffer) in `src/adapters/otlp.rs`, alongside the existing `OtlpSinkOps` metrics sink. - Same off-thread `consume_async` model as `otlp_push`: the tracer provider is built lazily on the first exported value (inside the consumer task's tokio context) and dropped at teardown to flush; a no-op under `RunMode::HistoricalFrom` (run-mode guard on `register_op1`, so no provider is built and no network calls are made). - Every classic span capability preserved: parent + per-hop children, caller-supplied attributes, and the silent skip of all-zero / backwards timestamps. The `&tokio::runtime::Handle` and extension-trait conventions match `otlp_push`. - Adds the `trace` feature to `opentelemetry_sdk` / `opentelemetry-otlp` (the span exporter), mirroring classic. - Tests: `spans_historical_mode_drains_without_connecting` (no-collector unit parity) in `tests/otlp_adapter.rs` and `otlp_spans_sends_successfully` (testcontainers) in `tests/otlp_integration.rs`. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DjcaJEMHLmS9hHwTwYUDHc * docs(next): record graph export as not-doing (Phase 5) Mark the Phase 5 "Graph export" infrastructure item as deliberately not ported. Rather than a one-off GML dump from the Builder topology, we want a better introspection/visualization story, to be designed and scoped separately later. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DjcaJEMHLmS9hHwTwYUDHc --------- Co-authored-by: Claude <noreply@anthropic.com>

…ions (#552) Mark the deviations resolved since the register was written (#545): - A5 (caller-owned tokio runtime + &Handle): resolved by #548 — the GraphBuilder now owns one runtime with a with_async_runtime override. Split out the residual "drive from a non-async thread" block_on constraint as A5a (unchanged, inherent). - A1/A4 for zmq_sub: partially resolved by #547 — the source_at_start primitive landed and zmq_sub now establishes its socket/thread in start(), with setup errors aborting at run-start. Remaining eager sources/sinks and re-run (A2) called out as the follow-ons. Add a ✅ resolved legend marker and a resolved-since summary under the recommended priorities. Claude-Session: https://claude.ai/code/session_01NKWgiXjudzBKAvtruWYSav Co-authored-by: Claude <noreply@anthropic.com>

…t-plan (#553) Phase 4.5's remaining follow-on (the arena/SoA value store) is deferred pending a *measured* need. This captures that measurement instead of the prior ~1.1-1.5x estimate. - Ran the existing store_baseline suite and add a `forward_scalar` (f64) workload as the aliasing *floor* to complement the `forward_clone` 8 KiB *ceiling*. The two bracket the arena+aliasing win: ~7.4x on large-payload forwarding (20.8ms vec vs 2.79ms rc_vec), but ~nothing on scalar graphs (1.84ms f64, already faster than the Rc path). Typical wingfoil workloads sit near the floor -> evidence for keeping the arena deferred. - Update port-plan.md "Measured baseline" with the real bracketed numbers and the sparse-dispatch gate (~7.3x sparse vs full_sweep). - Fix the stale "Dynamic graphs -- feature not built" section: the dynamic-graph feature has in fact landed (dynamic-graph cargo feature + tests/dynamic_graph.rs), as the Phase 4.5 header and capability matrix already state. Claude-Session: https://claude.ai/code/session_01SSdpDCB2mxGTnUZNhbhWym Co-authored-by: Claude <noreply@anthropic.com>

…merge_all) (#554) * next-python: add reduce and split combinators - reduce(func) — functools.reduce-style running reduction: the first value seeds the accumulator and is emitted as-is, each later value emits func(acc, value). Accumulator is engine-owned state re-seeded on a graph reset (re-run restarts). A raised exception aborts the run with context. - split() — decompose a stream of 2-tuples into its two component streams; both branches tick with the source. A non-indexable value aborts the run. Embedded Rust tests + pytest cases (37 Rust / 40 pytest pass under maturin). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P * next-python: add dataframe (pandas DataFrame from a stream) Adds the classic `dataframe` terminal: accumulates each value with its engine time and, on the last cycle, builds a pandas DataFrame (columns `time`, `value`) as the stream's final value. Rows are engine-owned state re-seeded on a graph reset (re-run rebuilds cleanly); earlier cycles stay quiet. pandas is imported lazily inside the op — only a graph that calls dataframe() needs it, and an unimportable pandas aborts the run with context. Tests skip gracefully when pandas is absent (Rust test guards on import; pytest uses importorskip), so a bare CI environment stays green. Embedded Rust test + pytest case (38 Rust / 41 pytest pass under maturin with pandas installed). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P * next-python: add merge_all (n-ary merge) merge_all(others) merges this stream with several others at once; on any tick the earliest-supplied ticked input wins (equivalent to a chain of 2-ary merges). Embedded Rust test + pytest case (39 Rust / 42 pytest pass). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P * next/docs: record the landed next-python combinator surface + custom node Updates the interop build list: the object-form binding now carries the full built-in combinator surface (30 ops at legacy test_streams parity, all with Python-exception propagation) and the Python-defined custom-node path, which the list previously did not reflect. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P --------- Co-authored-by: Claude <noreply@anthropic.com>

…ph) (#555) Adds the input primitive Python users need beyond synthetic counter/constant: Graph.values(list, period_nanos) replays a finite list of values, one per tick, `period` apart (first at t=0). Built on the historical-replay channel layer, so it replays deterministically in historical mode; a graph containing it is single-run (the producer channel is consumed by the first run). Distinct per-tick timestamps mean each value rides its own cycle, so the Burst<PyElement> the channel emits collapses cleanly to one value per tick. Embedded Rust tests + pytest cases (41 Rust / 45 pytest pass under maturin): replay a sequence, feed a combinator chain, carry non-numeric (string) values. Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P Co-authored-by: Claude <noreply@anthropic.com>

…557) Record the accept ruling for B2 in the deviation register. The live-tail sources (etcd/redis/kafka/postgres `_sub`, zmq_sub) reject `RunMode::HistoricalFrom` at wiring; classic technically permitted a wall-clock historical run, but a live tail has no deterministic timeline to replay, so that path produced neither reproducible values nor tick times. Deterministic historical replay is served by the paired time-sliced `_read` sources. Ratified as an accepted deviation; wall-clock historical is not restored. Claude-Session: https://claude.ai/code/session_01DWEDiSWkt3ZggL24YNbSyR Co-authored-by: Claude <noreply@anthropic.com>

* docs(next): mark runtime dynamism as landed in port-plan Two stale references still described `Runner::extend` as "not yet built". Runtime graph mutation shipped (behind `dynamic-graph`) as `Runner::run_dynamic` + an `Extension` scope, `Builder::dynamic_group`, and `Builder::demux`. Update both the "Dynamic graphs" note and the risk-table row to reflect the landed state. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019tqDrbkDe1T6pYfpNSVa3c * docs(next): migrate open next-port issues into the plan docs Fold the four remaining next-port tracking issues into the design docs so all next-port planning lives in one place: - #502 (busy-poll ingest) + #503 (bursts) -> new "Deferred / post-v1 work" section in port-plan.md capturing the compiled-path IO ingestion design, the post-#496 tractability analysis, and the wake-channel-vs-busy-spin gating decision; cross-referenced from the bursts footnote and a new capability-gap row (C4) in deviation-register.md. - #507 (architecture / orientation doc) -> deliverable + section outline in the same port-plan.md section. - #506 (compiled() from Python POC) -> build-list row in python-interop.md recording the parity finding and the one-fixed-graph scope limit. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019tqDrbkDe1T6pYfpNSVa3c --------- Co-authored-by: Claude <noreply@anthropic.com>

… teardown (B1) (#558) etcd_pub was the one networked sink still driving its writes with Handle::block_on on the single-threaded engine. It hadn't migrated to the shared consume_async primitive because a force:false conditional-write conflict must make the run return Err even on a RunFor::Cycles(1) run, and consume_async's teardown flush lived in a Drop impl that discarded the consumer task's result (a Drop cannot return a Result). Classic surfaces that same error at teardown, not on the failing cycle (AsyncConsumerNode::teardown does block_on(handle)??), and next's engine already runs every node's fallible teardown and folds its error into the run result. So the fix is to give consume_async a real fallible teardown: - consume_async now returns (sink, flush). flush closes the sender, joins the consumer task, and — unlike Drop — surfaces the final write error as Err. Wire it via the existing .finally() combinator. A Drop safety-net still drains queued writes if flush is never wired (no data loss), it just cannot surface the final error. - etcd_pub's per-entry PUTs now run on the off-thread consumer task; the wiring-time connect and LeaseGuard revoke keep their (teardown-time, graph-thread) block_on. The force:false single-cycle abort is preserved via the flush teardown, matching classic's teardown-time surfacing. - The same flush upgrade closes the "final-cycle write error swallowed" gap for kafka/postgres/redis/otlp, all migrated to .finally(flush). Adds a generic parity test proving a final-cycle write error surfaces at teardown (the force:false-on-Cycles(1) scenario, Docker-free) and a Drop safety-net drain test. Updates the etcd module docs and marks deviation register B1 resolved. Claude-Session: https://claude.ai/code/session_012QWPmR3wFSs5LVFMFm59EZ Co-authored-by: Claude <noreply@anthropic.com>

…[pygraph] (#556) * next-python: extend #[pyop] to stateful single-input ops The #[pyop] proc macro previously rejected any op with State != (). It now supports stateful single-input ops: the state may be any Default-seedable type, and the generated #[pyfunction] seeds it via `<State as Default>::default()` (the same factory the engine re-runs on reset, so re-runs start clean) instead of the hard-coded `|| ()`. Stateless ops are unchanged (() implements Default). - macro: drop the State == () guard; seed state from Default; doc/scope updated. - example: a stateful `running_total` #[pyop] (f64 accumulator) registered in the wingfoil_next module. - tests: pytest cases (running total + state-reseeds-on-rerun) and a stateful external-crate #[pyop] in plugin_seam.rs (compile proof + accumulate mechanics). - docs: interop build list — stateful #[pyop] marked done (multi-input remains). 41 lib + 4 plugin_seam Rust tests and 47 pytest cases pass; clippy clean on both crates. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P * next-python: extend #[pyop] to two-input ops (+ wire_op2 seam) #[pyop] now handles two-input ops (In<'a> = (&'a A, &'a B)): it emits a module.name(stream, other) #[pyfunction] wiring both streams via a new PyStream::wire_op2 (the two-active-input, edge-erasing counterpart of wire_op1, over the engine's register_op2). Single-input ops are unchanged. - graph.rs: add PyStream::wire_op2 (extracts A/B from PyElement, both active, output boxed back; TryFrom/step errors abort the run). - macro: single_ref_tuple_elem -> ref_tuple_elems (1 or 2 inputs); expand branches on arity to emit the one- or two-stream function. Doc/scope updated. - example: a two-input `weighted_add` #[pyop] registered in the module. - tests: pytest (weighted_add) + external-crate two-input #[pyop] in plugin_seam.rs (compile proof + wire_op2 mechanics). - docs: interop build list — #[pyop] stateful + two-input marked done. 41 lib + 5 plugin_seam Rust tests and 48 pytest cases pass; clippy clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P * next-python: add #[pygraph] — expose a Rust sub-graph as a Python callable Python callable `module.name(stream)` that splices the whole Rust-authored sub-graph into the caller's graph and returns the erased output. The interior runs at native types; only the edge erases (T: TryFrom<&PyElement> in, U: Into<PyElement> out). - seam (graph.rs): PyStream::typed_input::<T>() (erased -> Stream<T> via try_map) and PyStream::erased_output::<U>(Stream<U>) (-> erased PyStream) — the input/output halves the macro wires together. - macro: parse the wiring fn's `&Stream<T>` arg and `Stream<U>` return (first-generic extraction, name-agnostic so the type may be qualified/aliased to avoid the Stream pyclass clash); emit the splice #[pyfunction]. `name` must differ from the fn's own name; single-input/output v1. - example: doubled_running_total (double then cumulative-sum) in the module. - tests: pytest (end-to-end reuse) + external-crate #[pygraph] in plugin_seam.rs (compile proof + typed-in/erased-out seam mechanics). - docs: interop build list — #[pygraph] done; #[pyadapter] remains. 41 lib + 6 plugin_seam Rust tests and 49 pytest cases pass; clippy clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P --------- Co-authored-by: Claude <noreply@anthropic.com>

…561) #557 landed a blanket "reject ratified" ruling for B2; this refines it now that the legacy surface and the ergonomics are clearer: - Correct the postgres framing — classic postgres_sub already required RunMode::RealTime and bailed otherwise, so next's rejection is parity, not a wall-clock-historical deviation. That gap applies only to zmq_sub and the etcd/redis/kafka _sub sources. - Split the ruling: (a) ratify reject-at-wiring for live-only sources with no bounded historical twin (zmq_sub, etcd_sub, both redis sources, kafka_sub today); (b) for adapters with both a bounded read and a live tail, adopt a single mode-agnostic <adapter>_source that dispatches on run_mode instead of the mode-locked _read/_sub pair. - Add a "B2 — agreed plan" subsection (design, feasibility, optional per-mode config, scope: postgres now, kafka a future candidate, the rest live-only) and update the bottom summary line to match. Claude-Session: https://claude.ai/code/session_01DWEDiSWkt3ZggL24YNbSyR Co-authored-by: Claude <noreply@anthropic.com>

#562) #[pyadapter(name = ..., source)] on a user adapter trait implemented on GraphBuilder (`impl Trait for GraphBuilder { fn m(&self, args..) -> Stream<T> }`) generates a Python callable `module.m(graph, args..)` that runs the adapter's native wiring on the caller's graph and returns the erased output. The interior runs at native T; only the Python-facing edge erases. - seam: PyGraph::builder() (the GraphBuilder the adapter wires onto) and PyGraph::erase_source::<T>(Stream<T>) -> PyStream (T: Into<PyElement>); Graph::object() exposes the inner PyGraph to the generated #[pyfunction]. - macro: #[pyadapter] parses the single adapter method (name, non-self params, Stream<T> return), emits `fn name(graph, params..) -> Stream` forwarding to the method and erasing. `source` marker required (v1); sink/transform and burst (Stream<Burst<T>>) sources are not yet emitted. - example: a synthetic `ramp_source` adapter in the module. - tests: pytest (ramp_source + feeding combinators) + an external-crate source #[pyadapter] in plugin_seam.rs (compile proof + builder/erase_source seam). - docs: interop build list — source #[pyadapter] done; sink/burst remain. 41 lib + 7 plugin_seam Rust tests and 51 pytest cases pass; clippy clean. Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P Co-authored-by: Claude <noreply@anthropic.com>

…rt) (#560) * next: incremental historical channel + spawn/spawn_map (graph_node port) Replace the channel source's historical block-collect with an incremental, timestamp-gated one-ahead read (interp.rs::pump_historical) — classic's "block-while-behind / non-block-once-caught-up" loop. This gives every channel bounded (one-ahead) memory instead of buffering the whole feed, and — crucially — lets a producer that depends on the receiving graph's own output make progress, which the block-collect deadlocked. Results stay byte-identical (guarded by the channel/produce_async/consume_async/csv parity suites). On top of that, port classic graph_node's thread-offload combinators as fluent twins: - SourceOps::spawn — run a producer sub-graph on a worker thread (classic producer()); the worker builds+runs its own graph at run start under the driving run's inherited mode+bound and forwards timestamped values. - StreamOps::spawn_map — map an input stream through a worker sub-graph (classic mapper()); the two graphs run concurrently, lock-step by graph time. Historical is deterministic (values + tick times match classic's graph_node_works oracle). Engine plumbing: - channel_triggered / channel_lockstep receiver variants (read-ahead decoupled from the drive mode) for the spawn_map output and worker-input channels. - Kernel::run_for() accessor + Builder::source_at_start_with_params / compose_spawn_at_start so a worker learns the run params at start. Parity tests in tests/spawn.rs (both modes); example in examples/spawn.rs. Docs: port-plan Phase 2 graph_node entry + Phase 3 threading note; deviation register B6 (spawn_map lock-step) and the resolved block-collect note. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VB66npQqSfbsfVYamudTCq * next: opt-in bounded buffer for channel / spawn / spawn_map The channel transport was hardwired to an unbounded std mpsc — a fast producer could queue an unbounded backlog with no back-pressure. Add an opt-in bound, following the produce_async_bounded precedent (Option<usize>: None = unbounded default, Some(n) = sync_channel(n), so a producer outrunning the graph blocks on send). - ChannelSender now holds either an unbounded Sender or a bounded SyncSender (private Tx enum); every send routes through it, so the receive side and the incremental historical read are unchanged. - New surface: SourceOps::channel_bounded, SourceOps::spawn_bounded, StreamOps::spawn_map_bounded. spawn_map_bounded bounds both worker channels. The plain channel/spawn/spawn_map are now thin None-delegates. - Builder: channel_bounded / channel_lockstep / channel_triggered / source_at_start_with_params thread the buffer through. Caveat (documented): a bounded transport must not be used by a producer that fills the buffer before the run starts (replay_results, csv, postgres_read) — with no running consumer to drain it, a bounded send blocks at wiring. Those keep the unbounded path. Tests: tests/spawn.rs bounded spawn+spawn_map match the unbounded result; tests/channel.rs a concurrent producer exceeding the bound back-pressures yet replays losslessly and deterministically. deviation-register updated. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VB66npQqSfbsfVYamudTCq --------- Co-authored-by: Claude <noreply@anthropic.com>

…nup (D5) (#563) * next: mark deviation D5 won't-fix (classic opentelemetry drift moot at cutover) D5 tracked classic pinning opentelemetry 0.28 while next moved to 0.32 for a security advisory (GHSA-w9wp-h8wv-79jx). Since the legacy tree is retired wholesale at cutover, next's 0.32 is the surviving version and the classic-side advisory disappears with legacy — bumping classic to restore lockstep is not worth the churn. Reclassify 🟡 → 🟢 won't-fix and drop it from the active priority list. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012QWPmR3wFSs5LVFMFm59EZ * next: restore kafka_pub concurrent per-burst publish via consume_async_bursts (B3) next's kafka_pub rode the per-value consume_async, which serialised a burst's records into N sequential broker roundtrips (awaiting each send's delivery ack before the next). Classic handed a whole burst to the producer up front and drained the delivery futures together via FuturesUnordered — ~one roundtrip per burst. Order-preserving but a throughput regression (deviation-register B3). Add consume_async_bursts: a variant that hands the consumer a whole burst at a time (an owned Vec<T>) instead of one value, so a sink can process a burst's values concurrently while the single consumer still preserves order across bursts. It shares all the consume_async machinery — refactored into a private spawn_sink core — so back-pressure, error surfacing, and the flush teardown are identical (here buffer_size bounds bursts, not values). kafka_pub now uses it: build every record's delivery future up front and drive them together via FuturesUnordered, back at throughput parity with classic. The final burst's error still surfaces via the flush teardown. Adds a consume_async_bursts test (whole-burst delivery, across-burst ordering), removes the now-resolved kafka deviation #4, and marks register B3 resolved. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012QWPmR3wFSs5LVFMFm59EZ --------- Co-authored-by: Claude <noreply@anthropic.com>

…skill (#564) #[pyadapter] (no `source` marker) now handles sink/transform adapters authored as a trait on the fluent Stream<T>: `impl Trait for Stream<T> { fn m(&self, ..) -> Stream<U> }` emits `module.m(stream, ..)` that extracts the input to native T, runs the adapter, and erases the U output. A sink's Stream<()> erases to Python None (new From<()> for PyElement). - macro: expand_pyadapter branches on the `source` marker — source (on GraphBuilder, via builder()/erase_source) vs sink/transform (on Stream<T>, via typed_input/erased_output). Doc/scope updated. - element.rs: From<()> for PyElement -> None (so unit sinks cross the edge). - example: a `list_sink` sink adapter (append each f64 to a Python list) in the module; pytest cases (sink; source+sink together). - plugin_seam.rs: external-crate sink #[pyadapter] compile proof (Py<PyAny> target) + a Rust-observable typed-in/for_each/erased-out seam test. - new-adapter-next skill: replace the stale "no next Python bindings" note with the #[pyadapter] source/sink recipe, and its honest v1 limits — params must be FromPyObject, and burst (Stream<Burst<T>>) adapters aren't wired yet (most real adapters are burst-based, so their Python binding waits on burst-erasure). - docs: interop build list — source + sink #[pyadapter] done; burst remains. 41 lib + 9 plugin_seam Rust tests and 53 pytest cases pass; clippy clean. Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P Co-authored-by: Claude <noreply@anthropic.com>

…#565) Adds a "Skills — and they are living documents" section to next/CLAUDE.md covering both the existing /new-adapter-next skill and the new /new-op-next skill, with the standing instruction to check and update the matching skill whenever an op or adapter is onboarded or changed. Adds .claude/commands/new-op-next.md: the step-by-step recipe for adding a node/op to the next catalog — the Op shape and #[op] flags, the hand-written fluent method, graph!/compiled coverage, wingfoil-next-python bindings (#[pyop] / pyop_fn! / wire_op1), parity + completeness tests, and a feed-lessons-back living-document note. Claude-Session: https://claude.ai/code/session_019WZ8V3Bxjr1JA6tsesf8kU Co-authored-by: Claude <noreply@anthropic.com>

…hon list) (#566) #[pyadapter] now handles the burst-shaped adapters that the layering conventions make the common case. A Stream<Burst<T>> erases to a Python list per tick (same-instant values grouped); a burst sink is fed single-element bursts. Burst<T> may appear as a source's return, a sink's Self, or a transform's output — the macro detects it and routes through the burst seam. - seam (graph.rs): PyGraph::erase_burst_source, PyStream::typed_burst_input + erased_burst_output (Burst<T> <-> Python list of erased values; single-element bursts on the way in). - macro: burst_inner() detects Stream<Burst<T>>; expand_pyadapter routes source output / sink input / sink output through the burst seam when applicable. Doc/scope updated. - example: pair_source (a two-value Burst<f64> per instant via combine, -> Python list) in the module; pytest case. - plugin_seam.rs: external-crate burst source #[pyadapter] (compile proof + erase_burst_source seam). - skill + docs: new-adapter-next now says burst adapters bind directly (the common path), removing the earlier "waits on burst-erasure" caveat; interop build list marks #[pyadapter] (source/sink/burst) done. 41 lib + 10 plugin_seam Rust tests and 54 pytest cases pass; clippy clean. Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P Co-authored-by: Claude <noreply@anthropic.com>

…lue burst (#568) typed_burst_input (the input half of the burst adapter seam) now rebuilds a multi-value Burst<T> from a Python list/tuple, one member per item, instead of always wrapping each value in a single-element burst. Any non-list/tuple value still becomes a single-element burst (a `str` — itself a sequence — stays a scalar, checked via is_instance_of rather than extract::<Vec>). This closes the burst round-trip: a burst source's per-tick list (erase_burst_source) feeds a burst sink, which rebuilds the same multi-value burst. - graph.rs: typed_burst_input list/tuple -> multi-value burst; scalar -> single. - example: a burst sink `burst_list_sink` (on Stream<Burst<f64>>) in the module. - tests: pytest round-trip (pair_source -> burst_list_sink reproduces the [[1,10],[2,20],[3,30]] lists) + scalar-input single-element-burst case. - docs/skill: note the list->multi-value-burst inbound behaviour and the round-trip; add burst_list_sink as the burst-sink template. 41 lib + 10 plugin_seam Rust tests and 56 pytest cases pass; clippy clean. Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P Co-authored-by: Claude <noreply@anthropic.com>

…an) (#567) RunMode is a run-time choice — build the graph once, pick real-time vs historical at run(). But postgres_read/postgres_sub are mode-locked, forcing that choice into wiring and duplicating downstream wiring across two source functions. This implements the deviation-register "B2 — agreed plan" for postgres (the first _source target). Add postgres_source(g, params, conn, cfg): a single source that inspects params.run_mode at wiring and dispatches — HistoricalFrom -> postgres_read (bounded time-sliced replay), RealTime -> postgres_sub (LISTEN/NOTIFY live tail). PostgresSourceConfig is a small builder carrying an optional historical half (.historical(period, query_fn)) and an optional live half (.live(channel, start_from, query_fn)); supply both for a fully mode-agnostic graph (flip the mode at run(), wiring unchanged), supply one and the other mode errors at wiring naming the missing half. The closures are boxed so the config isn't generic over them. postgres_read/postgres_sub stay public as the low-level primitives. Tests: four no-DB dispatch tests (missing-half errors both ways, dispatch to read's bounded-window validation, realtime wiring reaches sub) plus one container-backed test that postgres_source in HistoricalFrom reads the same rows as postgres_read. Updates the module docs and marks the register B2 postgres target landed. Claude-Session: https://claude.ai/code/session_012QWPmR3wFSs5LVFMFm59EZ Co-authored-by: Claude <noreply@anthropic.com>

…me, plugin SDK) (#569) The wingfoil_next binding had no examples. Adds four idiomatic ones under crates/wingfoil-next-python/examples/, each a runnable .py: - quick_start.py — build a Graph, chain combinators, run, read a value. - custom_stream.py — a Python object as a graph node (Graph.custom_node, cycle(values)/peek protocol): a running maximum. - dataframe.py — collect a stream into a pandas DataFrame. - plugin_sdk.py — the headline: compose Rust-authored components from Python (ramp_source #[pyadapter], square #[pyop], doubled_running_total #[pygraph], list_sink sink) with built-in combinators. examples/README.md documents the maturin-develop-then-run flow and what each shows. tests/test_examples.py smoke-tests all four via runpy (pandas guarded by importorskip), so they run in the pytest gate and don't rot. 60 pytest cases pass (56 interop + 4 example smoke-tests). Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P Co-authored-by: Claude <noreply@anthropic.com>

…start) (#570) * next: defer the produce_async producer spawn to start() (A1 defer-to-start) produce_async/produce_async_bounded spawned their producer task at wiring time, so every source riding it (etcd/kafka/redis/postgres _sub) connected/subscribed during graph construction — the A1 "wiring-time I/O establishment" deviation: side-effecting, untestable wiring, a "wired but not running" window, and realtime pre-run message accumulation. Rebuild produce_async_bounded on source_at_start_with_params (the deferred- connection primitive that already backs zmq_sub and spawn): the producer task is now spawned in start(), and the returned StopHandle aborts it at teardown (AbortOnDrop). The permit-channel back-pressure and the RunParams validating passthrough are unchanged; the historical/realtime determinism is preserved. The four _sub adapters ride this automatically, so their connect/subscribe now happens at run start, not wiring. This is the produce_async-family step of the defer-to-start plan (spike zmq_sub landed in #547). Re-run (A2) is still a follow-on: the source inherits channel's single-run restriction. The RunParams simplification (derive from the run rather than take at wiring) is also a follow-on the deferral now enables — start() already surfaces the run's real params. Adds produce_async_defers_producer_to_run (the async closure must not run until run(); fails against the old spawn-at-wiring model). Updates async_source + etcd/kafka module docs and the register A1/A4 notes. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012QWPmR3wFSs5LVFMFm59EZ * ci: re-trigger (unrelated zmq slow-joiner flake, A6) The prior run's only failure was zmq_integration::first_message_not_dropped — the zmq slow-joiner race (deviation A6). It is unrelated to this change (the diff touches neither zmq nor source_at_start; passes 8/8 locally; the etcd/kafka integration jobs that exercise the produce_async change are green). Empty commit to re-trigger CI since a workflow re-run isn't available. Collapses on squash-merge. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012QWPmR3wFSs5LVFMFm59EZ * next: widen zmq first-message settle to 1500ms (A6 CI flake) zmq_integration::first_message_not_dropped failed reliably in CI: the SUB socket's subscription filter propagated ~150ms after the 600ms publisher head-start, so the first ~3 messages were lost to the zmq slow-joiner. Locally the 600ms window passed, but CI runners are slower. Widen SUB_SETTLE 600ms -> 1500ms, well past the observed establishment time. This is purely the test's settle window (the adapter's 50ms post-accept flush is unchanged), and it is the slow-joiner margin tracked as deviation A6 — the mechanism is not yet pinned, and a deterministic connect/filter handshake is the proper A6 follow-on. Unrelated to the produce_async change in this branch (the test's graph uses no produce_async path); it was a pre-existing marginal flake this branch happened to surface. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012QWPmR3wFSs5LVFMFm59EZ --------- Co-authored-by: Claude <noreply@anthropic.com>

* next: port classic dynamic-graph example (live price book) to next Ports the classic `dynamic` example onto next's runtime graph-mutation surface. A synthetic, deterministic market feed adds inst1/inst2/inst3 and then deletes inst1/inst2 on a *live* graph; `Builder::dynamic_group` splices a per-instrument sub-graph in on each add, tears it down on each delete, and folds every live member's current price into a BTreeMap price book. The book grows then collapses to the surviving instrument. Feature-gated on `dynamic-graph`. Includes a README and a parity-style test asserting the exact sequence of book states the group ticks downstream plus the final book value. Note on bounding: under `run_dynamic`, splicing consumes scheduler cycles, so `RunFor::Cycles(n)` does not map cleanly to n ticker ticks. The example and test use a splice-independent `RunFor::Duration` bound (documented inline). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P * next: port classic demux example (static-slot price book) to next The statically-wired counterpart to the `dynamic` example: the same live price book, but routed through a fixed pool of slots with `demux_it` instead of runtime graph mutation. Price and delete events share one `InstEvent` stream (merged with `combine`); `demux_it` routes each event to its instrument's slot, recycling a slot on delete (`DemuxEvent::Close`); the per-slot bursts are recombined, flattened, and folded into a BTreeMap. Because the topology is static, it runs under plain `run` and ticks the book every cycle — so the output shows the book both grow and shrink. Feature-gated on `dynamic-graph`. Includes a README and a parity-style test asserting the exact per-cycle book sequence and final value. Cross-links the two approaches from both READMEs. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P * ci: re-trigger (etcd_integration testcontainer flake, unrelated to examples) test_sub_snapshot_with_existing_keys is a snapshot/watch race in the etcd testcontainer (asserted 2 keys, saw 1). This PR only adds two dynamic-graph examples and touches nothing in the etcd adapter. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P --------- Co-authored-by: Claude <noreply@anthropic.com>

…egacy) (#573) Add the classic `logged(label, level)` debug tap to wingfoil-next as a stateless single-input `#[op]`: it logs `"{time} {label} {value:?}"` at the given level through the `log` crate (target "wingfoil") and passes the value through unchanged. Reads the tick time off `Ctx::time` and always wires the node (leaning on `log!`'s own enabled-check) rather than classic's wiring-time `log_enabled!` short circuit, keeping interpreted and compiled identical. `logged` is fluent-only: its `&str` label vs `String` cfg means the same `graph!` tokens can't satisfy both the fluent method and the compiled forwarder — documented in the op_completeness allowlist. Rework `print` to print each value in `cycle` instead of buffering into a `Vec` and dumping at teardown. The observable value stream is unchanged (print is a pass-through); only the diagnostic emission differs. Per-tick printing drops the unbounded buffer, streams output as the run progresses, and survives a mid-run abort. Shedding the teardown hook makes `print` a plain single-input `#[op]`, so its hand-written interpreted Builder method is gone. Recorded as deviations D8/D9 in the deviation register. Expose both on wingfoil-next-python (`Stream.print()` / `Stream.logged(label, level="info")`, level parsed from a case-insensitive name). Update the /new-op-next skill with the "ergonomic fluent signature vs Cfg type forces fluent-only" rule surfaced by `logged`. Claude-Session: https://claude.ai/code/session_01AbLC26nxMFwxF3hya8ZRnq Co-authored-by: Claude <noreply@anthropic.com>

) The tiers bench compared only next's three graph!-derived engines (interpreted/compiled/nested). Phase 6's regression gate calls for a fourth bar — the legacy interpreted engine — so `next-interpreted >= classic-interpreted` is directly measurable. Adds legacy-engine twins of all three workloads (dense_chain, fanout, accumulate) built on the classic `wingfoil` node tree with matching DAG shapes/node counts, and a `classic` bench_function per group. Measured relationship (relative): next-interpreted beats classic on dense_chain and accumulate, but trails it ~40% on wide fanout (every node fires every cycle, so the sparse dirty-list path can't help); compiled/nested win decisively across the board. Records the dense-fanout interpreted gap as a standing perf item in port-plan.md, and notes the examples landed so far. The bench stays a run-on-demand scaffold — criterion wall-clock thresholds are too noisy to gate CI on the shared runners. Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P Co-authored-by: Claude <noreply@anthropic.com>

…defer-to-start) (#575) Completes the RunParams simplification the produce_async start()-deferral (#570) enabled. Now that source_at_start_with_params hands the deferred producer the run's real run_mode/run_for/start_time at start, the caller no longer needs to declare RunParams up front: - produce_async / produce_async_bounded drop the `params: RunParams` argument. The producer's RunParams are derived from the actual run at start(), so the declare-up-front footgun and its validating passthrough are gone. Back-pressure now decides realtime-vs-historical from ctx.run_mode() at run time (the permit channel is created from buffer_size and only drained in realtime). - The live _sub sources (etcd_sub, kafka_sub, redis_sub, redis_stream_read, postgres_sub) take `run_mode: RunMode` instead of `RunParams` — they need it only for the wiring-time historical-mode rejection (kept at wiring, as zmq_sub does, so no block-collect deadlock). postgres_read keeps RunParams (it slices the [start,end) window at wiring); postgres_source keeps RunParams and passes run_mode to postgres_sub. Updates all callers (tests + examples), removes the now-obsolete produce_async_rejects_mismatched_start_time test, and refreshes async_source + adapter module docs and register A1. Verified: cargo fmt; cargo test -p wingfoil-next across async/etcd/kafka/redis/ postgres (58 binaries green); cargo clippy on those features (-D warnings) clean; examples build clean. The Docker integration-test suites + all-features lint are CI-verified. Claude-Session: https://claude.ai/code/session_012QWPmR3wFSs5LVFMFm59EZ Co-authored-by: Claude <noreply@anthropic.com>

…#577) postgres_write connected eagerly at wiring via Handle::block_on, so graph construction opened a socket and a connect failure aborted wiring rather than the run — a deviation from classic (which connects lazily inside the consumer) and from the defer-to-start direction (register A1/A4). Move the connect into the consume_async consumer: the client is opened lazily on the first write, on the consumer task, cached behind an async Mutex for subsequent bursts, with the connection driver spawned there too. Wiring now opens no socket; a connect failure surfaces during the run (via consume_async's error channel) with the password redacted — classic-consistent. This is the first of the eager-connect sinks (A1) to migrate; etcd_pub / redis / kafka_pub remain. Per the defer-to-start plan's "decisions to ratify", the error-surfacing shift (wiring → run start) is the accepted, classic-matching behaviour. - Rewrote write_connection_refused_redacts_password to run the graph and assert the redacted error aborts run() (wiring now succeeds). - Updated the postgres module Sink/Deviation docs and register A1/A4. Verified: cargo fmt; cargo test -p wingfoil-next --features postgres --test postgres_adapter (10 green); cargo clippy --features postgres --all-targets -D warnings clean. Docker integration write tests are CI-verified. Claude-Session: https://claude.ai/code/session_012QWPmR3wFSs5LVFMFm59EZ Co-authored-by: Claude <noreply@anthropic.com>

Ports the `log` mode of the classic `tracing` example: a one-per-second counter tapped by `logged("tick", Info)`, rendered through `env_logger`. `logged` emits a `log` record per tick (target "wingfoil") and passes the value through, so the example doubles as a demonstration of next's observability tap. The classic example's other two modes are intentionally not ported here — they are blocked on next engine features that don't exist yet: - `tracing` — route the events through a `tracing-subscriber` - `instruments` — engine spans around `run` and each cycle (`instrument-*`) Next's op catalog logs through the `log` crate only and the engine has no span instrumentation, so passing either mode prints a note and falls back to `log`. Both are documented in the README and recorded as a Phase-6 follow-up in port-plan.md (landing with the instrumentation port, not as example work). Includes a README and a pass-through test (the value stream is unchanged: the wrapped counter still ticks 1, 2, 3). Claude-Session: https://claude.ai/code/session_01MatEkmguN4pYzgyMpSse3P Co-authored-by: Claude <noreply@anthropic.com>

Adds a review/audit skill counterpart to /new-adapter-next. It audits the wingfoil-next adapters against that skill and the strict-superset parity obligation, producing a classified report across four deliverables: - skill health (internal consistency + currency vs code/design docs) - per-adapter compliance with the invariants and step requirements - lessons that should be folded back into new-adapter-next.md - unjustified legacy deviations (undocumented in module doc + register + port-plan) Read-and-report by default; per-adapter audits fan out to fresh-context subagents, mirroring new-adapter-next's step-15 self-review. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018UNN6wny3AAsnb97eURSGX

claude added 30 commits July 24, 2026 20:15

0-jake-0 and others added 30 commits July 26, 2026 10:15

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Compiled islands: nest a graph! sub-graph as one node of an interpreted graph - #579

Compiled islands: nest a graph! sub-graph as one node of an interpreted graph#579
0-jake-0 wants to merge 117 commits into
mainfrom
claude/next-adapter-skill-review-6ocuea

0-jake-0 commented Jul 26, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

0-jake-0 commented Jul 26, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants