From 56914abff0e82b7519603e047a787c64c51b4143 Mon Sep 17 00:00:00 2001 From: Ovsep Avakian Date: Thu, 30 Jul 2026 20:35:20 +0300 Subject: [PATCH 1/4] Add typed payload accessors to the signal package A payload is an any, because a mesh carries mixed types down one pipe. Every component that reads one therefore has to get it back out, and a bare type assertion is either unchecked -- and takes down the whole run when something upstream changes its payload type -- or checked, and three lines every time. As[T] reports what went wrong, AsOrDefault carries on with a default, and neither panics on a nil signal or a nil payload. The per-type shorthands wrap only the fallible form: AsOrDefault already infers T from the default, so an AsIntOrDefault would be longer than the generic it wraps. AsFloat64OrDefault is the one exception, because an untyped 0 infers int and AsOrDefault(s, 0) would silently return the default for a perfectly good float64 payload. AsNumber is the loose one, widening float32/int/int64/uint64 and encoding bool as 1 and 0. It answers "is this a measurement at all" for code that does not know which components produce which payloads -- a structured signal often uses its payload as a type tag and keeps the values in scalars. The README, the runnable Example and the quick-start now read payloads this way instead of asserting on PayloadOrNil. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 6 +- docs/wiki/101.-Quick-start.md | 8 ++- docs/wiki/201.-Signals.md | 52 +++++++++++++++ example_test.go | 6 +- signal/doc.go | 3 + signal/typed.go | 102 +++++++++++++++++++++++++++++ signal/typed_test.go | 119 ++++++++++++++++++++++++++++++++++ 7 files changed, 288 insertions(+), 8 deletions(-) create mode 100644 signal/typed.go create mode 100644 signal/typed_test.go diff --git a/README.md b/README.md index bf13bca..02d4345 100644 --- a/README.md +++ b/README.md @@ -59,8 +59,8 @@ func run() error { component.WithInputs("i1", "i2"), component.WithOutputs("res"), component.WithActivationFunc(func(this *component.Component) error { - word1 := this.InputByName("i1").Signals().FirstPayloadOrDefault("").(string) - word2 := this.InputByName("i2").Signals().FirstPayloadOrDefault("").(string) + word1 := signal.AsOrDefault(this.InputByName("i1").Signals().First(), "") + word2 := signal.AsOrDefault(this.InputByName("i2").Signals().First(), "") return this.OutputByName("res").PutSignals(signal.New(word1 + word2)) })) if err != nil { @@ -71,7 +71,7 @@ func run() error { component.WithInputs("i1"), component.WithOutputs("res"), component.WithActivationFunc(func(this *component.Component) error { - str := this.InputByName("i1").Signals().FirstPayloadOrDefault("").(string) + str := signal.AsOrDefault(this.InputByName("i1").Signals().First(), "") return this.OutputByName("res").PutSignals(signal.New(strings.ToUpper(str))) })) if err != nil { diff --git a/docs/wiki/101.-Quick-start.md b/docs/wiki/101.-Quick-start.md index e52ffec..42f90fa 100644 --- a/docs/wiki/101.-Quick-start.md +++ b/docs/wiki/101.-Quick-start.md @@ -38,7 +38,11 @@ worker := func(name string, reduce func([]int) int) *component.Component { // Read all numbers buffered on the input port var nums []int for _, s := range this.InputByName("in").Signals().All() { - nums = append(nums, s.PayloadOrNil().(int)) + n, err := signal.AsInt(s) // a non-int payload is an error, not a panic + if err != nil { + return err + } + nums = append(nums, n) } // Reduce them and put the result on the output port @@ -179,7 +183,7 @@ The mesh behaves like a computational graph: you build it from components, initi Read the summary from the reporter's output port: ```go -result := fm.ComponentByName("report").OutputByName("out").Signals().FirstPayloadOrNil().(string) +result := signal.AsOrDefault(fm.ComponentByName("report").OutputByName("out").Signals().First(), "") fmt.Println("Result:", result) ``` diff --git a/docs/wiki/201.-Signals.md b/docs/wiki/201.-Signals.md index 2da66dd..d8fe76d 100644 --- a/docs/wiki/201.-Signals.md +++ b/docs/wiki/201.-Signals.md @@ -40,6 +40,58 @@ Example of creating a simple signal: mySignal := signal.New("example payload") // A signal with a string payload ``` +## Reading typed payloads + +A payload is `any`, because a mesh carries mixed types down one pipe. Every component that reads one therefore has to get it back out, and a bare type assertion is either unchecked — panicking the whole run when something upstream changes its payload type — or checked, and three lines every time. The `signal` package provides two accessors instead: + +```go +// Reports what went wrong; never panics +temperature, err := signal.As[float64](sig) +if err != nil { + return err +} + +// Carries on with a default when the payload is missing or of another type +retries := signal.AsOrDefault(sig, 0) +``` + +`As` fails rather than panicking on purpose: a component reading a payload it did not produce cannot know that something upstream changed, and finding out by taking down the mesh is not a useful way to be told. A nil signal and a nil payload are both handled — the fallible form returns an error, the defaulting form returns the default. + +Shorthands save the type parameter on the fallible form: + +| Shorthand | Equivalent to | +|---|---| +| `signal.AsInt(sig)` | `signal.As[int](sig)` | +| `signal.AsFloat64(sig)` | `signal.As[float64](sig)` | +| `signal.AsString(sig)` | `signal.As[string](sig)` | +| `signal.AsBool(sig)` | `signal.As[bool](sig)` | +| `signal.AsGroup(sig)` | `signal.As[*signal.Group](sig)` | + +`AsGroup` is for a signal whose payload is another `signal.Group` — see [Signal Groups](#signal-groups) below. + +There are no matching `AsIntOrDefault`-style shorthands, because `AsOrDefault` infers the type from the default you pass and is shorter than any shorthand would be: + +```go +signal.AsOrDefault(sig, 0) // int +signal.AsOrDefault(sig, "") // string +signal.AsOrDefault(sig, false) // bool +``` + +> [!IMPORTANT] +> One catch: an untyped `0` infers `int`, so `signal.AsOrDefault(sig, 0)` on a **float64** payload quietly returns `0` rather than the payload. Pass `0.0`, or use `signal.AsFloat64OrDefault(sig, 0)` — the one defaulting shorthand that exists, for exactly this reason. + +### AsNumber + +`signal.AsNumber(sig)` returns `(float64, bool)` and is the loose one. It accepts `float64`, `float32`, `int`, `int64` and `uint64`, and encodes `bool` as `1` and `0`: + +```go +if value, ok := signal.AsNumber(sig); ok { + // the signal carries a measurement +} +``` + +It exists for the code that has to decide whether a signal is a measurement **at all**. A structured signal commonly uses its payload as a type tag — `"air"`, `"venous_blood"` — and keeps the real values in [scalars](#scalars), so anything drawing or logging a mesh needs to tell a number from a name without knowing which components produce which. + ## Signal Groups While individual signals are useful for most cases, there are scenarios where working with a **group of signals** simplifies the design. A signal group aggregates multiple signals, each potentially carrying a different type of payload. diff --git a/example_test.go b/example_test.go index d5a0fd0..1a89bc1 100644 --- a/example_test.go +++ b/example_test.go @@ -22,8 +22,8 @@ func Example() { component.WithInputs("i1", "i2"), component.WithOutputs("res"), component.WithActivationFunc(func(this *component.Component) error { - word1 := this.InputByName("i1").Signals().FirstPayloadOrDefault("").(string) - word2 := this.InputByName("i2").Signals().FirstPayloadOrDefault("").(string) + word1 := signal.AsOrDefault(this.InputByName("i1").Signals().First(), "") + word2 := signal.AsOrDefault(this.InputByName("i2").Signals().First(), "") return this.OutputByName("res").PutSignals(signal.New(word1 + word2)) })) must(err) @@ -32,7 +32,7 @@ func Example() { component.WithInputs("i1"), component.WithOutputs("res"), component.WithActivationFunc(func(this *component.Component) error { - str := this.InputByName("i1").Signals().FirstPayloadOrDefault("").(string) + str := signal.AsOrDefault(this.InputByName("i1").Signals().First(), "") return this.OutputByName("res").PutSignals(signal.New(strings.ToUpper(str))) })) must(err) diff --git a/signal/doc.go b/signal/doc.go index 8bbbf52..2a82393 100644 --- a/signal/doc.go +++ b/signal/doc.go @@ -6,4 +6,7 @@ // [Group.ForEach] does not update the receiver on success; label changes on // grouped signals should be done with [Group.Map] / [Group.MapPayloads] so // replaced signals are stored in the new group (github.com/hovsep/fmesh#203). +// +// A payload is an any. [As] and [AsOrDefault] read one back out as a concrete +// type without the panic a bare assertion risks. package signal diff --git a/signal/typed.go b/signal/typed.go new file mode 100644 index 0000000..b568f58 --- /dev/null +++ b/signal/typed.go @@ -0,0 +1,102 @@ +package signal + +import ( + "errors" + "fmt" +) + +// Typed reads of a payload, which is an any because a mesh carries mixed types +// down one pipe. Two shapes: one reports what went wrong, one carries on with a +// default. Neither panics. + +// As returns the payload as T, failing rather than panicking when the payload is +// another type — a component cannot know that something upstream changed its +// payload type, and taking down the mesh is not a useful way to be told. +func As[T any](s *Signal) (T, error) { + var zero T + if s == nil { + return zero, errors.New("signal is nil") + } + + payload, err := s.Payload() + if err != nil { + return zero, fmt.Errorf("signal payload: %w", err) + } + + typed, ok := payload.(T) + if !ok { + return zero, fmt.Errorf("signal payload is %T, not %T", payload, zero) + } + return typed, nil +} + +// AsOrDefault returns the payload as T, or the default if it is missing or of +// another type. +// +// T is inferred from defaultValue, so an untyped 0 means int: pass 0.0, or use +// AsFloat64OrDefault, when the payload is a float64. +func AsOrDefault[T any](s *Signal, defaultValue T) T { + if s == nil { + return defaultValue + } + + value, ok := s.PayloadOrDefault(defaultValue).(T) + if !ok { + return defaultValue + } + return value +} + +// AsFloat64 returns the payload as a float64. +func AsFloat64(s *Signal) (float64, error) { return As[float64](s) } + +// AsFloat64OrDefault returns the payload as a float64, or the default. The other +// types need no such shorthand: AsOrDefault(s, 0) already infers int, while +// AsOrDefault(s, 0) for a float64 payload infers int and returns the default. +func AsFloat64OrDefault(s *Signal, defaultValue float64) float64 { + return AsOrDefault(s, defaultValue) +} + +// AsInt returns the payload as an int. +func AsInt(s *Signal) (int, error) { return As[int](s) } + +// AsString returns the payload as a string. +func AsString(s *Signal) (string, error) { return As[string](s) } + +// AsBool returns the payload as a bool. +func AsBool(s *Signal) (bool, error) { return As[bool](s) } + +// AsGroup returns the payload as a group, for a signal carrying other signals. +func AsGroup(s *Signal) (*Group, error) { return As[*Group](s) } + +// AsNumber reports the payload as a float64 when it carries float64, float32, +// int, int64 or uint64, or a bool as 1 and 0. Narrower integer types are not +// covered: widen at the source rather than adding cases here. +// +// Loose on purpose: it answers "is this a measurement at all" for code that does +// not know which components produce which payloads. +func AsNumber(s *Signal) (float64, bool) { + if s == nil { + return 0, false + } + + switch v := s.PayloadOrNil().(type) { + case float64: + return v, true + case float32: + return float64(v), true + case int: + return float64(v), true + case int64: + return float64(v), true + case uint64: + return float64(v), true + case bool: + if v { + return 1, true + } + return 0, true + default: + return 0, false + } +} diff --git a/signal/typed_test.go b/signal/typed_test.go new file mode 100644 index 0000000..3c34b55 --- /dev/null +++ b/signal/typed_test.go @@ -0,0 +1,119 @@ +package signal + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestAs(t *testing.T) { + t.Run("returns the payload as the asked-for type", func(t *testing.T) { + got, err := As[float64](New(3.5)) + require.NoError(t, err) + assert.InDelta(t, 3.5, got, 1e-9) + }) + + t.Run("a payload of another type is an error, not a panic", func(t *testing.T) { + // An unchecked assertion here takes down the mesh run when something + // upstream changes what it publishes. + got, err := As[float64](New("not a number")) + + require.Error(t, err) + require.ErrorContains(t, err, "is string, not float64") + assert.Zero(t, got) + }) + + t.Run("a nil signal is an error", func(t *testing.T) { + _, err := As[float64](nil) + require.ErrorContains(t, err, "signal is nil") + }) + + t.Run("a nil payload is reported as such", func(t *testing.T) { + _, err := As[float64](New(nil)) + require.Error(t, err) + }) +} + +func TestAsOrDefault(t *testing.T) { + assert.InDelta(t, 3.5, AsOrDefault(New(3.5), 0.0), 1e-9) + assert.InDelta(t, 7.0, AsOrDefault(New("wrong type"), 7.0), 1e-9, + "the wrong type degrades to the default rather than failing the run") + assert.InDelta(t, 7.0, AsOrDefault(nil, 7.0), 1e-9) + + assert.Equal(t, 7, AsOrDefault(New(7), 0), "T is inferred from the default") + assert.Equal(t, "x", AsOrDefault(New("x"), "")) + assert.True(t, AsOrDefault(New(true), false)) +} + +func TestAsOrDefaultInfersFromTheDefault(t *testing.T) { + // An untyped 0 makes T int, so a float64 payload silently yields the + // default. AsFloat64OrDefault exists for exactly this. + assert.Equal(t, 0, AsOrDefault(New(2.5), 0)) + assert.InDelta(t, 2.5, AsFloat64OrDefault(New(2.5), 0), 1e-9) +} + +func TestTypedShorthands(t *testing.T) { + f, err := AsFloat64(New(1.5)) + require.NoError(t, err) + assert.InDelta(t, 1.5, f, 1e-9) + + i, err := AsInt(New(4)) + require.NoError(t, err) + assert.Equal(t, 4, i) + + s, err := AsString(New("hi")) + require.NoError(t, err) + assert.Equal(t, "hi", s) + + b, err := AsBool(New(true)) + require.NoError(t, err) + assert.True(t, b) + + assert.InDelta(t, 9.0, AsFloat64OrDefault(New("x"), 9), 1e-9) +} + +func TestAsGroup(t *testing.T) { + inner := NewGroup(1, 2) + + got, err := AsGroup(New(inner)) + require.NoError(t, err) + assert.Equal(t, 2, got.Len()) + + _, err = AsGroup(New(1)) + require.Error(t, err) +} + +func TestAsNumber(t *testing.T) { + tests := []struct { + name string + payload any + want float64 + ok bool + }{ + {name: "float64", payload: 2.5, want: 2.5, ok: true}, + {name: "float32", payload: float32(2.5), want: 2.5, ok: true}, + {name: "int", payload: 3, want: 3, ok: true}, + {name: "int64", payload: int64(4), want: 4, ok: true}, + {name: "uint64", payload: uint64(5), want: 5, ok: true}, + {name: "true is one", payload: true, want: 1, ok: true}, + {name: "false is zero", payload: false, want: 0, ok: true}, + // A structured signal uses its payload as a type tag and keeps the real + // values in scalars. Telling the two apart is what this is for. + {name: "a type tag is not a measurement", payload: "venous_blood", ok: false}, + {name: "nil payload", payload: nil, ok: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, ok := AsNumber(New(tt.payload)) + assert.Equal(t, tt.ok, ok) + assert.InDelta(t, tt.want, got, 1e-9) + }) + } + + t.Run("nil signal", func(t *testing.T) { + _, ok := AsNumber(nil) + assert.False(t, ok) + }) +} From 1249af0af801820b4a882c97de9b2e1b525098ca Mon Sep 17 00:00:00 2001 From: Ovsep Avakian Date: Thu, 30 Jul 2026 20:35:40 +0300 Subject: [PATCH 2/4] Add multi-edge wiring helpers to the port package Wiring code reads as a list of edges, and a bare error from PipeTo gives no clue which of a dozen lines produced it. MultiPipe takes the list and names the edge that failed; MultiForward does the same for the forwarding case, where signals are copied immediately rather than a connection registered. A nil port on either end is reported rather than dereferenced. That is the common case, not a defensive nicety: OutputByName and InputByName return nil for a name no port has, so without the check a typo in a wiring list surfaces as a nil dereference somewhere well past the typo, with nothing naming it. Co-Authored-By: Claude Opus 5 (1M context) --- docs/wiki/302.-Ports.md | 13 +++++++++ docs/wiki/303.-Pipes.md | 19 +++++++++++++ port/doc.go | 3 ++ port/wiring.go | 61 +++++++++++++++++++++++++++++++++++++++++ port/wiring_test.go | 60 ++++++++++++++++++++++++++++++++++++++++ 5 files changed, 156 insertions(+) create mode 100644 port/wiring.go create mode 100644 port/wiring_test.go diff --git a/docs/wiki/302.-Ports.md b/docs/wiki/302.-Ports.md index 7355fb6..d9c416f 100644 --- a/docs/wiki/302.-Ports.md +++ b/docs/wiki/302.-Ports.md @@ -143,6 +143,19 @@ err := port.ForwardWithMap(source, dest, func(s *signal.Signal) *signal.Signal { _ = port.PutSignalGroups(group1, group2) ``` +### Forwarding several pairs at once + +`port.MultiForward` takes a list of source/destination pairs and says which one failed: + +```go +err := port.MultiForward( + port.Pair{From: outer.InputByName("time"), To: inner.InputByName("time")}, + port.Pair{From: outer.InputByName("data"), To: inner.InputByName("data")}, +) +``` + +Failures name both ends (`forwarding time -> time: ...`). A nil port is reported the same way (`cannot forward: -> time`) rather than panicking, which matters because `InputByName` returns `nil` for a name that does not exist — a typo in the list surfaces as an error naming the typo. See also [MultiPipe](https://github.com/hovsep/fmesh/wiki/303.-Pipes#wiring-many-pipes-at-once) for the same shape applied to pipes. + ## Port metadata and description Port descriptions are set at construction via the `WithDescription` option. Labels and scalars use the mutating API (`AddLabel`, `SetLabels`, `AddScalar`, etc.) — see [Labels page](https://github.com/hovsep/fmesh/wiki/202.-Metadata). diff --git a/docs/wiki/303.-Pipes.md b/docs/wiki/303.-Pipes.md index dfdc798..39cbd05 100644 --- a/docs/wiki/303.-Pipes.md +++ b/docs/wiki/303.-Pipes.md @@ -64,6 +64,25 @@ _ = c3.OutputByName("o1").PipeTo(sink.InputByName("i1")) You can find runnable fan-in/fan-out examples in the [examples repository](https://github.com/hovsep/fmesh-examples). +## Wiring many pipes at once + +Assembly code reads as a list of edges, and a bare error from `PipeTo` gives no clue which of a dozen lines produced it. `port.MultiPipe` takes the list and reports the edge that failed: + +```go +if err := port.MultiPipe( + port.Pipe{From: c1.OutputByName("o1"), To: c2.InputByName("i1")}, + port.Pipe{From: c2.OutputByName("o1"), To: c3.InputByName("i1")}, + port.Pipe{From: c3.OutputByName("o1"), To: sink.InputByName("i1")}, +); err != nil { + // e.g. piping o1 -> i1: pipe must go from output to input + return err +} +``` + +A nil port on either end is reported the same way (`cannot pipe: o1 -> `) instead of panicking. That matters because `OutputByName`/`InputByName` return `nil` for a name no port has, so a typo in a wiring list becomes an error that names the offending edge. + +`port.MultiForward` is the equivalent for [signal forwarding](https://github.com/hovsep/fmesh/wiki/302.-Ports#forwarding-several-pairs-at-once), which copies signals immediately rather than registering a connection. + ## Cyclic Connections ![](https://github.com/user-attachments/assets/7bec7fc4-12ec-4583-99ec-611d4a7ad88d) diff --git a/port/doc.go b/port/doc.go index fd91d4b..9434b07 100644 --- a/port/doc.go +++ b/port/doc.go @@ -5,4 +5,7 @@ // port's signal buffer out through its pipes and clears the source. Unlike // the copy-on-write signal package, port types mutate in place: Set*, Add* // and Remove* methods modify the receiver. +// +// [MultiPipe] and [MultiForward] take a list of edges instead of one call per +// edge, and name the edge that failed. package port diff --git a/port/wiring.go b/port/wiring.go new file mode 100644 index 0000000..8112063 --- /dev/null +++ b/port/wiring.go @@ -0,0 +1,61 @@ +package port + +import "fmt" + +// Pipe is a single wiring edge, from an output port to an input port. +type Pipe struct { + From *Port + To *Port +} + +// MultiPipe wires several 1:1 connections and says which one failed. A nil port +// is reported rather than panicked on: the usual cause is a port name that does +// not exist, and the name is what the caller needs told. +func MultiPipe(pipes ...Pipe) error { + for _, p := range pipes { + if p.From == nil || p.To == nil { + return fmt.Errorf("cannot pipe: %s", p) + } + if err := p.From.PipeTo(p.To); err != nil { + return fmt.Errorf("piping %s: %w", p, err) + } + } + return nil +} + +// String describes the edge, naming whichever end is missing. +func (p Pipe) String() string { + return fmt.Sprintf("%s -> %s", portName(p.From), portName(p.To)) +} + +// Pair is a source and destination port, for forwarding rather than piping. +type Pair struct { + From *Port + To *Port +} + +// MultiForward forwards signals across several 1:1 pairs, reporting nil ports +// and naming the pair that failed, as MultiPipe does. +func MultiForward(pairs ...Pair) error { + for _, pair := range pairs { + if pair.From == nil || pair.To == nil { + return fmt.Errorf("cannot forward: %s", pair) + } + if err := ForwardSignals(pair.From, pair.To); err != nil { + return fmt.Errorf("forwarding %s: %w", pair, err) + } + } + return nil +} + +// String describes the pair, naming whichever end is missing. +func (p Pair) String() string { + return fmt.Sprintf("%s -> %s", portName(p.From), portName(p.To)) +} + +func portName(p *Port) string { + if p == nil { + return "" + } + return p.Name() +} diff --git a/port/wiring_test.go b/port/wiring_test.go new file mode 100644 index 0000000..4d2a3a9 --- /dev/null +++ b/port/wiring_test.go @@ -0,0 +1,60 @@ +package port + +import ( + "testing" + + "github.com/hovsep/fmesh/signal" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestMultiPipe(t *testing.T) { + t.Run("wires every edge", func(t *testing.T) { + out1, out2 := mustOutput("o1"), mustOutput("o2") + in1, in2 := mustInput("i1"), mustInput("i2") + + require.NoError(t, MultiPipe( + Pipe{From: out1, To: in1}, + Pipe{From: out2, To: in2}, + )) + + require.NoError(t, out1.PutSignals(signal.New(1))) + require.NoError(t, out1.Flush()) + assert.True(t, in1.HasSignals()) + }) + + t.Run("names the edge that failed", func(t *testing.T) { + // A bare error from PipeTo gives no clue which of a dozen lines + // produced it. + err := MultiPipe(Pipe{From: mustInput("wrong_way"), To: mustInput("i1")}) + + require.Error(t, err) + assert.Contains(t, err.Error(), "wrong_way -> i1") + }) + + t.Run("a missing port is reported, not panicked on", func(t *testing.T) { + err := MultiPipe(Pipe{From: nil, To: mustInput("i1")}) + + require.ErrorContains(t, err, " -> i1") + }) +} + +func TestMultiForward(t *testing.T) { + t.Run("forwards every pair", func(t *testing.T) { + src1, src2 := mustInput("s1"), mustInput("s2") + dst1, dst2 := mustInput("d1"), mustInput("d2") + require.NoError(t, src1.PutSignals(signal.New(1))) + require.NoError(t, src2.PutSignals(signal.New(2))) + + require.NoError(t, MultiForward(Pair{From: src1, To: dst1}, Pair{From: src2, To: dst2})) + + assert.True(t, dst1.HasSignals()) + assert.True(t, dst2.HasSignals()) + }) + + t.Run("a missing port is reported, not panicked on", func(t *testing.T) { + err := MultiForward(Pair{From: nil, To: mustInput("d1")}) + + require.ErrorContains(t, err, " -> d1") + }) +} From 7294fc8609179d4a9145447e62d211fa46f1f22d Mon Sep 17 00:00:00 2001 From: Ovsep Avakian Date: Thu, 30 Jul 2026 20:35:57 +0300 Subject: [PATCH 3/4] Add activation function combinators A component has one activation function, but the thing it does is often several things in a fixed order -- read the controls, advance the state, publish the result -- and writing them as one closure is how a component grows to two hundred lines with no seam anywhere. Sequential, When, RequireInputs and Pipeline build the single function a component is constructed with. They are values passed to WithActivationFunc, not hooks: OnActivation stays the tool for behavior added from outside the component. When and RequireInputs guard on inputs differently, and the distinction is easy to get backwards. When skips the activation, so a partial set of inputs is cleared at drain time and never comes back; RequireInputs suspends and keeps what already arrived. Port names are resolved before any signal is looked at. Collection.ByNames silently skips names it cannot find, and AllHaveSignals on the resulting empty collection is vacuously true, so ByNames("typo").AllHaveSignals() reports ready and the guard quietly becomes a no-op. RequireInputs fails on such a name rather than suspending the component for the rest of the run -- nothing ever arrives on a port that does not exist -- and every name is checked before any signal, or a port that is merely empty hides the typo behind a wait. Pipeline likewise errors instead of dereferencing a nil port. Pipeline reads its inputs in the order named rather than through the collection's map iteration, so a stage that folds or pairs up its group gets the caller's order rather than an arbitrary one, and a stage returning a nil group is an error rather than a nil dereference at the output port. Co-Authored-By: Claude Opus 5 (1M context) --- component/compose.go | 111 +++++++++++++++ component/compose_test.go | 181 +++++++++++++++++++++++++ component/doc.go | 3 + docs/wiki/301.-Component.md | 101 ++++++++++++-- docs/wiki/602.-Patterns-and-recipes.md | 6 + 5 files changed, 394 insertions(+), 8 deletions(-) create mode 100644 component/compose.go create mode 100644 component/compose_test.go diff --git a/component/compose.go b/component/compose.go new file mode 100644 index 0000000..a2c5c76 --- /dev/null +++ b/component/compose.go @@ -0,0 +1,111 @@ +package component + +import ( + "fmt" + + "github.com/hovsep/fmesh/signal" +) + +// Combinators over ActivationFunc: they build the single function a component is +// constructed with. Use OnActivation hooks instead when the behavior is added +// from outside the component rather than composed by it. + +// Sequential runs activation functions in order, stopping at the first error. +// +// Errors pass through as the component's own, so a stage returning +// ErrWaitingForInputs suspends the component. +func Sequential(funcs ...ActivationFunc) ActivationFunc { + return func(this *Component) error { + for _, f := range funcs { + if err := f(this); err != nil { + return err + } + } + return nil + } +} + +// When runs fn only if the predicate holds. +func When(predicate func(*Component) bool, fn ActivationFunc) ActivationFunc { + return func(this *Component) error { + if !predicate(this) { + return nil + } + return fn(this) + } +} + +// HasSignalsOn reports whether every named input port exists and carries +// something. A name no port has is never ready; RequireInputs reports it. +func HasSignalsOn(portNames ...string) func(*Component) bool { + return func(this *Component) bool { + for _, name := range portNames { + p := this.InputByName(name) + if p == nil || !p.HasSignals() { + return false + } + } + return true + } +} + +// RequireInputs suspends the component until every named port has signals, +// keeping whatever has already arrived. Contrast When, which skips the +// activation and so lets the partial inputs be cleared. +// +// A name no port has fails the activation instead of suspending it forever, and +// every name is checked before any signal, or an empty port hides the typo +// behind a wait. +func RequireInputs(portNames ...string) ActivationFunc { + return func(this *Component) error { + for _, name := range portNames { + if this.InputByName(name) == nil { + return fmt.Errorf("required input port %q does not exist", name) + } + } + + if !this.Inputs().ByNames(portNames...).AllHaveSignals() { + return ErrWaitingForInputsKeep + } + return nil + } +} + +// PipelineStage transforms a group of signals on its way through a pipeline. +type PipelineStage func(signals *signal.Group) (*signal.Group, error) + +// Pipeline reads the named input ports in the order given, passes their signals +// through each stage in turn, and writes the result to one output port. +// +// A port name no port has, and a nil group from a stage, are errors rather than +// nil dereferences. A stage that drops everything returns an empty group. +func Pipeline(inputPortNames []string, outputPortName string, stages ...PipelineStage) ActivationFunc { + return func(this *Component) error { + out := this.OutputByName(outputPortName) + if out == nil { + return fmt.Errorf("pipeline output port %q does not exist", outputPortName) + } + + signals := signal.NewGroup() + for _, name := range inputPortNames { + in := this.InputByName(name) + if in == nil { + return fmt.Errorf("pipeline input port %q does not exist", name) + } + signals = signals.With(in.Signals().All()...) + } + + for i, stage := range stages { + next, err := stage(signals) + if err != nil { + return fmt.Errorf("pipeline stage %d failed: %w", i, err) + } + if next == nil { + return fmt.Errorf("pipeline stage %d returned a nil group", i) + } + signals = next + } + + return out.PutSignalGroups(signals) + } +} diff --git a/component/compose_test.go b/component/compose_test.go new file mode 100644 index 0000000..8275f04 --- /dev/null +++ b/component/compose_test.go @@ -0,0 +1,181 @@ +package component + +import ( + "errors" + "testing" + + "github.com/hovsep/fmesh/signal" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestSequential(t *testing.T) { + t.Run("runs in order", func(t *testing.T) { + var order []string + note := func(s string) ActivationFunc { + return func(*Component) error { order = append(order, s); return nil } + } + + require.NoError(t, Sequential(note("a"), note("b"), note("c"))(nil)) + assert.Equal(t, []string{"a", "b", "c"}, order) + }) + + t.Run("stops at the first error", func(t *testing.T) { + reached := false + boom := func(*Component) error { return errors.New("boom") } + after := func(*Component) error { reached = true; return nil } + + err := Sequential(boom, after)(nil) + + require.ErrorContains(t, err, "boom") + assert.False(t, reached, "nothing after a failed stage runs") + }) + + t.Run("a waiting stage suspends the component", func(t *testing.T) { + // The error passes through as the component's own, so a stage can + // suspend exactly as it would if the activation were one function. + wait := func(*Component) error { return ErrWaitingForInputsKeep } + + err := Sequential(wait, func(*Component) error { return nil })(nil) + + require.ErrorIs(t, err, ErrWaitingForInputsKeep) + }) +} + +func TestWhenAndRequireInputs(t *testing.T) { + newC := func(t *testing.T) *Component { + t.Helper() + c, err := New("c", WithInputs("a", "b"), + WithActivationFunc(func(*Component) error { return nil })) + require.NoError(t, err) + return c + } + + t.Run("When skips the work when the predicate fails", func(t *testing.T) { + c := newC(t) + ran := false + + err := When(HasSignalsOn("a"), func(*Component) error { ran = true; return nil })(c) + + require.NoError(t, err) + assert.False(t, ran, "an idle component simply does nothing") + }) + + t.Run("When runs the work when it holds", func(t *testing.T) { + c := newC(t) + require.NoError(t, c.InputByName("a").PutSignals(signal.New(1))) + ran := false + + require.NoError(t, When(HasSignalsOn("a"), func(*Component) error { ran = true; return nil })(c)) + assert.True(t, ran) + }) + + t.Run("RequireInputs waits instead of skipping", func(t *testing.T) { + // A partial set of inputs must be kept, not silently dropped by + // returning nil. + c := newC(t) + require.NoError(t, c.InputByName("a").PutSignals(signal.New(1))) + + err := RequireInputs("a", "b")(c) + + require.ErrorIs(t, err, ErrWaitingForInputsKeep) + }) + + t.Run("RequireInputs passes once everything has arrived", func(t *testing.T) { + c := newC(t) + require.NoError(t, c.InputByName("a").PutSignals(signal.New(1))) + require.NoError(t, c.InputByName("b").PutSignals(signal.New(2))) + + require.NoError(t, RequireInputs("a", "b")(c)) + }) + + t.Run("a port that does not exist is never carrying anything", func(t *testing.T) { + // A misspelled name must not read as ready: an empty selection of ports + // trivially satisfies "all of them have signals". + c := newC(t) + require.NoError(t, c.InputByName("a").PutSignals(signal.New(1))) + + assert.False(t, HasSignalsOn("typo")(c)) + assert.False(t, HasSignalsOn("a", "typo")(c)) + }) + + t.Run("RequireInputs fails on a port that does not exist", func(t *testing.T) { + c := newC(t) + + err := RequireInputs("typo")(c) + + require.ErrorContains(t, err, `required input port "typo" does not exist`) + require.NotErrorIs(t, err, ErrWaitingForInputs, + "nothing arrives on a port that does not exist, so waiting would suspend the component forever") + }) + + t.Run("an empty port does not hide a misspelled one behind a wait", func(t *testing.T) { + // Every name is checked before any signal is, or the typo is only ever + // reported once the real port happens to be full. + c := newC(t) + + err := RequireInputs("a", "typo")(c) + + require.ErrorContains(t, err, `"typo" does not exist`) + }) +} + +func TestPipeline(t *testing.T) { + c, err := New("c", WithInputs("in"), WithOutputs("out"), + WithActivationFunc(func(*Component) error { return nil })) + require.NoError(t, err) + require.NoError(t, c.InputByName("in").PutSignals(signal.New(2), signal.New(3))) + + double := func(g *signal.Group) (*signal.Group, error) { + return g.Map(func(s *signal.Signal) *signal.Signal { + return signal.New(s.PayloadOrDefault(0).(int) * 2) + }), nil + } + + require.NoError(t, Pipeline([]string{"in"}, "out", double, double)(c)) + + got := c.OutputByName("out").Signals() + require.Equal(t, 2, got.Len()) + assert.Equal(t, 8, got.First().PayloadOrDefault(0), "two doublings of 2") + + t.Run("a failing stage says which one", func(t *testing.T) { + bad := func(*signal.Group) (*signal.Group, error) { return nil, errors.New("nope") } + + err := Pipeline([]string{"in"}, "out", double, bad)(c) + + require.ErrorContains(t, err, "pipeline stage 1 failed") + }) + + t.Run("a stage returning a nil group is an error, not a panic", func(t *testing.T) { + var noGroup *signal.Group + nilGroup := func(*signal.Group) (*signal.Group, error) { return noGroup, nil } + + err := Pipeline([]string{"in"}, "out", nilGroup)(c) + + require.ErrorContains(t, err, "pipeline stage 0 returned a nil group") + }) + + t.Run("ports that do not exist are errors, not panics", func(t *testing.T) { + require.ErrorContains(t, Pipeline([]string{"in"}, "typo", double)(c), + `pipeline output port "typo" does not exist`) + require.ErrorContains(t, Pipeline([]string{"typo"}, "out", double)(c), + `pipeline input port "typo" does not exist`) + }) +} + +func TestPipelineReadsInputsInOrder(t *testing.T) { + // The order is the one the caller wrote, not whatever a map iteration + // produced: a stage that folds or pairs up its group depends on it. + c, err := New("c", WithInputs("a", "b", "d"), WithOutputs("out"), + WithActivationFunc(func(*Component) error { return nil })) + require.NoError(t, err) + require.NoError(t, c.InputByName("a").PutSignals(signal.New("a"))) + require.NoError(t, c.InputByName("b").PutSignals(signal.New("b"))) + require.NoError(t, c.InputByName("d").PutSignals(signal.New("d"))) + + require.NoError(t, Pipeline([]string{"d", "a", "b"}, "out")(c)) + + payloads, err := c.OutputByName("out").Signals().AllPayloads() + require.NoError(t, err) + assert.Equal(t, []any{"d", "a", "b"}, payloads) +} diff --git a/component/doc.go b/component/doc.go index 386e45c..b0db7d6 100644 --- a/component/doc.go +++ b/component/doc.go @@ -5,4 +5,7 @@ // [New] and functional options (WithInputs, WithActivationFunc, ...); after a // run, each cycle exposes an [ActivationResult] per component describing // whether and how it activated. Component types mutate in place. +// +// One activation function can be composed from several with [Sequential], +// [When], [RequireInputs] and [Pipeline]. package component diff --git a/docs/wiki/301.-Component.md b/docs/wiki/301.-Component.md index b333fb2..193ec71 100644 --- a/docs/wiki/301.-Component.md +++ b/docs/wiki/301.-Component.md @@ -81,12 +81,16 @@ func(this *component.Component) error { // Approach 1: Using All() to get a slice for _, sig := range this.InputByName("i1").Signals().All() { - sum += sig.PayloadOrNil().(int) + n, err := signal.AsInt(sig) + if err != nil { + return err + } + sum += n } - // Approach 2: Using ForEach() (cleaner for iteration) - // _, err := this.InputByName("i1").Signals().ForEach(func(sig *signal.Signal) error { - // sum += sig.PayloadOrNil().(int) + // Approach 2: Using ForEach() (cleaner for iteration), tolerating odd payloads + // err := this.InputByName("i1").Signals().ForEach(func(sig *signal.Signal) error { + // sum += signal.AsOrDefault(sig, 0) // return nil // }) @@ -100,6 +104,74 @@ Explanation: * The function calculates the sum of all integer payloads received on input port ***i1***. * A new signal with the computed sum is sent to output port ***o1***. * The loop iterates over all signals in the input port's buffer, allowing flexible handling of multiple signals. +* Payloads are read with `signal.AsInt` rather than a bare type assertion, so a payload of an unexpected type is an error instead of a panic — see [Reading typed payloads](https://github.com/hovsep/fmesh/wiki/201.-Signals#reading-typed-payloads). + +## Composing activation functions + +A component has one activation function, which is the right number: it is the thing the component does. But "the thing it does" is often several things in a fixed order — read the controls, advance the state, publish the result — and writing them as one closure is how a component grows to two hundred lines with no seam anywhere. + +The `component` package provides combinators that build one `ActivationFunc` out of several: + +```go +component.WithActivationFunc(component.Sequential( + component.RequireInputs("config", "data"), + readControls, + advanceState, + publishResult, +)) +``` + +`Sequential` runs the functions in order and stops at the first error. The error is the component's own, not a hook failure, so a stage that returns `component.ErrWaitingForInputsKeep` suspends the component exactly as it would have if the whole activation were written as one function. + +### When vs RequireInputs + +Both guard on inputs, and the difference matters: + +| Combinator | Behavior | Use when | +|---|---|---| +| `component.When(pred, fn)` | Skips `fn` and returns `nil` — the component simply does nothing this cycle | The component is **idle** until something arrives | +| `component.RequireInputs(names...)` | Returns `ErrWaitingForInputsKeep` — the component is suspended and **keeps** what already arrived | The component **needs** every named input before it can compute anything | + +```go +// Idle: nothing on "trigger" means nothing to do +component.When(component.HasSignalsOn("trigger"), doWork) + +// Needs both: hold the one that arrived and wait for the other +component.Sequential(component.RequireInputs("left", "right"), sum) +``` + +Getting this backwards is easy and silently drops signals: `When` where `RequireInputs` was meant lets the cycle end normally, so the partial inputs are cleared and never come back. + +`HasSignalsOn` is the predicate `When` is usually given, and reports `false` for a name the component has no port for. `RequireInputs` **fails** the activation on such a name (`required input port "..." does not exist`) instead of suspending — nothing ever arrives on a port that does not exist, so waiting on one would suspend the component for the rest of the run. + +### Pipeline + +`Pipeline` covers the common shape of reading inputs, transforming the group, and writing one output: + +```go +component.WithActivationFunc(component.Pipeline( + []string{"in"}, // input ports, read in the order named + "out", // one output port + dropEmpty, // stages, applied in order + normalize, +)) +``` + +A stage is a `component.PipelineStage` — a transformation of the group on its way through: + +```go +func(signals *signal.Group) (*signal.Group, error) +``` + +A failing stage names itself in the error (`pipeline stage 1 failed: ...`). Three things worth knowing: + +* Inputs are read in the order the names are given, so a stage that folds or pairs up its group gets the caller's order rather than an arbitrary one. +* A stage that drops everything returns an **empty** group. Returning `nil` is an error, reported as such. +* A port name the component has no port for fails the activation, naming the port. + +### Combinators or hooks? + +These combinators are values you hand to `WithActivationFunc`: nothing about them is initialization, and nothing needs a name or a registry. `OnActivation` hooks chain functions too, and are the better tool when something **outside** the component is adding behavior to it — see [Hooks](https://github.com/hovsep/fmesh/wiki/501.-Hooks). Combinators are for a component composing its own behavior. ## Stateful component @@ -187,7 +259,7 @@ factorizer, _ := component.New("factorizer", component.WithActivationFunc(func(this *component.Component) error { inner := buildFactorizationMesh() // your inner *fmesh.FMesh - _, err := this.InputByName("in").Signals().ForEach(func(sig *signal.Signal) error { + err := this.InputByName("in").Signals().ForEach(func(sig *signal.Signal) error { // Bridge: outer signal → inner mesh input _ = inner.ComponentByName("starter").InputByName("in").PutSignals(sig) @@ -311,10 +383,10 @@ s, _ := component.New("sum", return component.ErrWaitingForInputsKeep } - inputNum1 := this.InputByName("i1").Signals().FirstPayloadOrDefault(0) - inputNum2 := this.InputByName("i2").Signals().FirstPayloadOrDefault(0) + inputNum1 := signal.AsOrDefault(this.InputByName("i1").Signals().First(), 0) + inputNum2 := signal.AsOrDefault(this.InputByName("i2").Signals().First(), 0) - return this.OutputByName("o1").PutSignals(signal.New(inputNum1.(int) + inputNum2.(int))) + return this.OutputByName("o1").PutSignals(signal.New(inputNum1 + inputNum2)) }), ) ``` @@ -328,6 +400,19 @@ Two sentinel errors control waiting behavior: Use `ErrWaitingForInputsKeep` when every signal is important and you want to collect multiple signals on each port. Use `ErrWaitingForInputs` when the presence of signals on specific ports matters more than their content. +The same guard is available as a combinator, which also checks that the named ports exist instead of quietly ignoring names it cannot find: + +```go +component.WithActivationFunc(component.Sequential( + component.RequireInputs("i1", "i2"), + func(this *component.Component) error { + // reached only when both ports have signals + }, +)) +``` + +See [Composing activation functions](#composing-activation-functions). + By using this mechanism, you can control when a component should activate and ensure proper synchronization in your mesh. ## Component Hooks diff --git a/docs/wiki/602.-Patterns-and-recipes.md b/docs/wiki/602.-Patterns-and-recipes.md index 44f177c..89e9853 100644 --- a/docs/wiki/602.-Patterns-and-recipes.md +++ b/docs/wiki/602.-Patterns-and-recipes.md @@ -280,10 +280,14 @@ A whole mesh can hide behind one component whose activation function runs it — Build the inner mesh **once** (in the wrapper's constructor, closed over by the activation function), not on every activation — component state inside the inner mesh then persists across ticks. +Those four phases are exactly what `component.Sequential` expresses, with `component.RequireInputs` as the validate phase and `port.MultiForward` for the sense and feedback edges — see [Composing activation functions](https://github.com/hovsep/fmesh/wiki/301.-Component#composing-activation-functions). + ### Convention-based wiring With many components, explicit pipe lists get tedious. Adopt a port-naming convention and wire by lookup — e.g. at assembly time, pipe the clock's output to every component that has an input named `time`, or match inputs named `_` to the corresponding subsystem output. Keep port names as shared constants in one package — pipes should never depend on magic strings. +Collect the matched edges into a `[]port.Pipe` and hand them to `port.MultiPipe(edges...)`: a convention that fails to match resolves to a nil port, and `MultiPipe` reports it as a named edge rather than letting it become a nil dereference at flush time. + ### Table-driven activation functions When many components share behavior that differs only in data, generate the activation function from a declarative table instead of writing near-identical closures: @@ -303,6 +307,8 @@ func activationFromTable(handlers map[string]func(*component.Component, *signal. Each component instance is then just a table; the uniform dispatch lives in one place. +Table-driven dispatch splits an activation by **data**. To split one by **phase** — guard, then compute, then publish — compose it from smaller activation functions instead: [Composing activation functions](https://github.com/hovsep/fmesh/wiki/301.-Component#composing-activation-functions). + --- ## Metadata techniques From 50a2d35f6a239334927ed387a571a905081742a9 Mon Sep 17 00:00:00 2001 From: Ovsep Avakian Date: Thu, 30 Jul 2026 20:36:37 +0300 Subject: [PATCH 4/4] Record the rules the new helpers established design.md gains the invariant behind the guard bug the combinators had to fix: lookups by name are silently forgiving -- ByName returns nil, ByNames skips, and Every on the empty result is vacuously true -- so anything taking port names as strings must resolve them all before asking about signals. It also records why AsFloat64OrDefault survives when its siblings were dropped, so nobody restores the symmetry, and adds signal.As[T] to the approved-generics list, which the accessors had quietly broken. Comment hygiene gains a length rule. The three new files were first written with multi-paragraph headers and per-function essays; that material is documentation, and of the two copies the one in source is the one that rots. One line is the default, a short second paragraph the ceiling, and anything longer belongs in docs/wiki. The files were compacted to match: 46 lines of comment removed, every constraint kept. hooks.md separates OnActivation hooks from the combinators, naming.md covers combinator naming and the component-level predicate, and testing.md requires covering the name that resolves to no port -- the case whose absence let the vacuous-truth bug through. Also fixes signal.Group.ForEach in four wiki snippets that showed it returning (*Group, error). It returns error only, so none of them compiled. Co-Authored-By: Claude Opus 5 (1M context) --- .agent/docs/design.md | 26 ++++++++++++++++++++---- .agent/docs/hooks.md | 4 ++++ .agent/docs/naming.md | 10 +++++++++ .agent/docs/testing.md | 3 +++ docs/wiki/202.-Metadata.md | 2 +- docs/wiki/203.-Collections-and-Groups.md | 2 +- docs/wiki/601.-Tips-&-tricks.md | 13 ++++++++++++ 7 files changed, 54 insertions(+), 6 deletions(-) diff --git a/.agent/docs/design.md b/.agent/docs/design.md index 1e61c9d..5ad06fc 100644 --- a/.agent/docs/design.md +++ b/.agent/docs/design.md @@ -15,16 +15,17 @@ Architecture overview (the concept → type → package table and the execution **Fan-out shares pointers.** Output→input fan-out forwards the same `*Signal` pointers to all destinations. Do not add deep-copy to `ForwardSignals` or `Flush`. -**No generics in data flow.** `signal`/`meta` use `any`/`float64`; FBP requires mixed-type signal flows in one group. Approved generics elsewhere: `hook.Group[T]` (typed hook registry) and `component.MustGetTyped[T]` (state accessor). Do not add more without approval. +**No generics in data flow.** `signal`/`meta` use `any`/`float64`; FBP requires mixed-type signal flows in one group. Approved generics elsewhere: `hook.Group[T]` (typed hook registry), `component.MustGetTyped[T]` (state accessor), and `signal.As[T]`/`signal.AsOrDefault[T]` (payload accessors — they read a payload out, they do not make the flow typed). Do not add more without approval. **Minimise `reflect`.** Only when no alternative exists. Current approved use: `reflect.TypeOf(payload).Comparable()` in `ContainsPayload` — always nil-guard before calling `.Comparable()`. ## Package notes -- **`signal`** — `payload` is `[]any{value}` (single-element slice so `nil` is valid). Predicate combinators and label constructors live in `predicates.go`. `ForEach` returns `(*Group, error)`. +- **`signal`** — `payload` is `[]any{value}` (single-element slice so `nil` is valid). Predicate combinators and label constructors live in `predicates.go`. `ForEach`/`ForEachIf` return `error` only (as on every collection type — see [naming.md](naming.md)). Typed payload accessors live in `typed.go`: `As[T]` (error on nil signal / missing payload / wrong type), `AsOrDefault[T]`, fallible per-type shorthands over `As`, and `AsNumber` (loose `(float64, bool)` widening — `float64`/`float32`/`int`/`int64`/`uint64`, `bool` as 1/0). None of them panic; that is the point of having them. There are deliberately **no** `AsIntOrDefault`-style shorthands: `AsOrDefault` infers `T` from the default and is shorter. The sole exception is `AsFloat64OrDefault`, which exists because an untyped `0` infers `int`, so `AsOrDefault(s, 0)` silently returns the default for a float64 payload — do not "restore symmetry" by adding the others back. - **`meta`** — `Labels` (string k/v) and `Scalars` (string→float64). `Keys()`/`Values()` return sorted slices for determinism. `Merge(other)` is the one non-mutating method on both types. `Every(pred)` on empty = `true` (vacuous truth). `ForEach` returns `error`. Constructors: `NewLabels()`, `NewScalars()`. -- **`port`** — `Flush()` fans out then clears source. `PipeTo` is output→input only. Both return `error`. `PipeTo` validates direction at call time. -- **`component`** — `State` is `map[string]any`, persistent across cycles and across `Run`s (see [runtime.md](runtime.md)). Constructors use functional options: `component.New(name, opts...) (*Component, error)`. Ports come in two creation styles: name-based (`WithInputs`/`AddInputs`, `WithIndexedInputs("i", 1, 3)` → `i1..i3`) and attach-based (`AttachInputPorts` for pre-built `port.NewInput` ports with options). `LoopbackPipe(out, in)` wires a component to itself (such a mesh never stops naturally). `ErrWaitingForInputs`/`ErrWaitingForInputsKeep` are scheduler control-flow sentinels, not failures. +- **`port`** — `Flush()` fans out then clears source. `PipeTo` is output→input only. Both return `error`. `PipeTo` validates direction at call time. `wiring.go` holds the declarative multi-edge helpers: `Pipe`/`MultiPipe` (registers connections) and `Pair`/`MultiForward` (copies signals now); both name the failing edge and report nil ports instead of dereferencing them. +- **Name lookups are silently forgiving — helpers taking port names must not be.** `Collection.ByName` returns `nil` for a name no port has, `Collection.ByNames` skips such names entirely, and `AllHaveSignals()`/`Every()` on the resulting empty collection is vacuously `true`. So `ByNames("typo").AllHaveSignals()` reports *ready*. Any helper that accepts port names as strings must resolve every name before asking anything about signals, and report the name it could not resolve. +- **`component`** — `State` is `map[string]any`, persistent across cycles and across `Run`s (see [runtime.md](runtime.md)). Constructors use functional options: `component.New(name, opts...) (*Component, error)`. Ports come in two creation styles: name-based (`WithInputs`/`AddInputs`, `WithIndexedInputs("i", 1, 3)` → `i1..i3`) and attach-based (`AttachInputPorts` for pre-built `port.NewInput` ports with options). `LoopbackPipe(out, in)` wires a component to itself (such a mesh never stops naturally). `ErrWaitingForInputs`/`ErrWaitingForInputsKeep` are scheduler control-flow sentinels, not failures. `compose.go` holds the `ActivationFunc` combinators — `Sequential`, `When`+`HasSignalsOn`, `RequireInputs`, `Pipeline`+`PipelineStage` — which compose a component's *own* activation (contrast `OnActivation` hooks, which are for behavior added from outside; see [hooks.md](hooks.md)). `When` skips, `RequireInputs` suspends and keeps: never substitute one for the other, as skipping where waiting was meant drops the partial inputs at drain time. - **`hook`** — lives at `internal/hook` (not public API). Generic `hook.Group[T]`, ordered, fail-fast `Trigger`. Three hook levels (mesh/component/port); see [hooks.md](hooks.md). - **`cycle`** — has its own `Any`/`Every`/`Count` on its collection type, independent of `signal.Group`. @@ -59,6 +60,23 @@ Comments must add information beyond the signature. Omit a comment entirely rath - No usage guidance ("Use X to do Y") or examples in type definition comments; those belong in method godocs or external docs - Method comments: one line where possible +**Length — short and on point:** + +One line is the default. A rationale that genuinely needs more gets a second short paragraph of +two or three sentences, and that is the ceiling. + +- **State the constraint, not the story.** "A name no port has fails the activation instead of + suspending forever" beats a paragraph reconstructing how a reader might get it wrong. +- **No file-header essays.** A file-level comment names what the file holds in a sentence. If a + file needs several paragraphs to introduce itself, the material is documentation, not a comment. +- **A paragraph belongs in `docs/wiki/`.** The wiki teaches — narrative, examples, when-to-use-which + tables. Godoc reminds. Name the concept in the comment and let the page carry it; duplicating it + in source means two things to keep in sync, and the source copy is the one that rots. +- **Cut the prose voice.** Comments that argue with the reader ("which is the right number", "and + that is the point", "the whole reason this exists") are essay, not documentation. +- Explain the **non-obvious**: an invariant, a footgun, an ordering requirement, a reason the + obvious implementation is wrong. Never narrate what the next line plainly does. + ## Dead code policy Do not keep unused exported symbols "for future use". Remove them immediately: diff --git a/.agent/docs/hooks.md b/.agent/docs/hooks.md index c7c39cf..867bada 100644 --- a/.agent/docs/hooks.md +++ b/.agent/docs/hooks.md @@ -26,6 +26,10 @@ closures are the only registration path. - **`OnActivation` is special**: its hooks are `ActivationFunc`s appended after the main activation function and run **sequentially in the same activation** — they share the error path (first error aborts the chain and becomes the activation error). +- **`OnActivation` hooks vs activation combinators**: `component/compose.go` also chains + `ActivationFunc`s, but those are combinators — plain values passed to `WithActivationFunc`, + with no registry, no name, and no initialization step. Use a hook when something *outside* + the component adds behavior to it; use a combinator when the component composes its own. - **`AfterActivation` always runs** — success, error, panic, or waiting; a `finally` block. - Outcome hooks (`OnSuccess`/`OnError`/`OnPanic`/`OnWaitingForInputs`) fire before `AfterActivation`. Distinguish waiting modes via `ctx.Result.Code()` diff --git a/.agent/docs/naming.md b/.agent/docs/naming.md index f6a4fbd..d2e9113 100644 --- a/.agent/docs/naming.md +++ b/.agent/docs/naming.md @@ -94,6 +94,16 @@ need to wrap `error` where nothing can go wrong. Prefer combinators over inline closures: `Not`, `And`, `Or`, `HasLabel`, `LabelEquals`, `LabelContains`, `HasAllLabels`, `HasAnyLabel`. +Component-level: `component.HasSignalsOn(names...)` returns `func(*Component) bool`, the predicate +`component.When` takes. + +## Combinators + +Functions that take and return an `ActivationFunc` are named after what they do to the activation, +with no prefix: `Sequential`, `When`, `RequireInputs`, `Pipeline`. `With*` stays reserved for +constructor options, so `WithActivationFunc(Sequential(...))` reads as option-wrapping-value rather +than two options. + ## Stuttering Do not repeat the package name in a type or function name. Within the `meta` package, use diff --git a/.agent/docs/testing.md b/.agent/docs/testing.md index f564219..7046aaf 100644 --- a/.agent/docs/testing.md +++ b/.agent/docs/testing.md @@ -10,6 +10,7 @@ Shared implementations live in `internal/testutil` (usable from `integration_tests/` and other external test packages); in-package tests of `fmesh` and `port` keep local copies (import cycle) - Use `assert.InDelta` for float64 comparisons (tolerance `1e-9` for exact values, larger for computed averages) +- Comments explain why a case exists (the bug it pins down), never what the assertion does — one or two lines, same brevity rule as source (see [design.md](design.md)) ## What to cover @@ -19,3 +20,5 @@ - Cross-entity aggregation on `signal.Group`: `AvgScalar`/`MinScalar`/`MaxScalar` return `signal.ErrScalarNotFoundInGroup` when no element has the named scalar; `SumScalar` returns 0 - Group metadata separation: group's own Labels/Scalars must not bleed into element Labels/Scalars and vice versa - `signal.Group` batch methods (`WithLabelOnEach`, `WithScalarOnEach`, etc.) must preserve the group's own metadata on the returned group +- Anything taking a port name as a string: cover the name that resolves to no port. An unresolved name reaches the assertion as an empty collection (vacuously ready) or a nil port (a panic at the first dereference), so the passing test proves nothing unless it names a port that does not exist +- Typed payload accessors: a wrong payload type, a nil payload, and a nil signal must all return an error or the default — never panic diff --git a/docs/wiki/202.-Metadata.md b/docs/wiki/202.-Metadata.md index 55e2383..cf049ff 100644 --- a/docs/wiki/202.-Metadata.md +++ b/docs/wiki/202.-Metadata.md @@ -344,7 +344,7 @@ _, _ = component.New("router", component.WithInputs("in"), component.WithOutputs("high-priority", "low-priority", "normal"), component.WithActivationFunc(func(c *component.Component) error { - _, err := c.InputByName("in").Signals().ForEach(func(sig *signal.Signal) error { + err := c.InputByName("in").Signals().ForEach(func(sig *signal.Signal) error { priority := sig.Labels().ValueOrDefault("priority", "normal") switch priority { case "high": diff --git a/docs/wiki/203.-Collections-and-Groups.md b/docs/wiki/203.-Collections-and-Groups.md index de4dcdb..44bc394 100644 --- a/docs/wiki/203.-Collections-and-Groups.md +++ b/docs/wiki/203.-Collections-and-Groups.md @@ -186,7 +186,7 @@ There are two main approaches for iterating over signals: ```go // Approach 1: Using ForEach() — recommended -_, err := port.Signals().ForEach(func(s *signal.Signal) error { +err := port.Signals().ForEach(func(s *signal.Signal) error { fmt.Printf("Signal payload: %v\n", s.PayloadOrNil()) return nil }) diff --git a/docs/wiki/601.-Tips-&-tricks.md b/docs/wiki/601.-Tips-&-tricks.md index 99c095d..05820e0 100644 --- a/docs/wiki/601.-Tips-&-tricks.md +++ b/docs/wiki/601.-Tips-&-tricks.md @@ -133,6 +133,19 @@ if port := c.InputByName("data"); port != nil { } ``` +Helpers that take port names by string do this check for you and report the offending name: `component.RequireInputs`, `component.Pipeline`, `port.MultiPipe`, and `port.MultiForward`. Prefer them over hand-rolled lookups when a typo would otherwise go unnoticed. + +## Typed payload reads + +Reach for `signal.As[T]` / `signal.AsOrDefault` instead of asserting on `PayloadOrNil()`. An unchecked assertion panics the whole run when an upstream component changes what it publishes: + +```go +n, err := signal.AsInt(sig) // error on a wrong or missing payload +n = signal.AsOrDefault(sig, 0) // or carry on with a default (T comes from the default) +``` + +See [Reading typed payloads](https://github.com/hovsep/fmesh/wiki/201.-Signals#reading-typed-payloads). + ## PutPayloads shortcut Instead of wrapping payloads in `signal.New`, use `PutPayloads`: