Adding a Grouper

Embedding Pulse, not contributing to it? This recipe adds a built-in in internal/processing/, which embedders cannot import. To add your own operator, implement the matching contract in the public extend package instead — see Extension Points.

Audience: Pulse internals contributors adding a new GROUP_* operator — a bucketing function that maps each record to a group key the orchestrator uses to partition the aggregation.

The recipe mirrors the aggregator recipe; the grouper-specific moving parts are the per-bucket key contract, the grouper registry, and the v0.20.0 ComponentSchema / MetaGrouper emission path.

1. Declare the type constant

Add the new constant to types/types.go and the slice returned by types.AllGroupTypes():

const (
    // ... existing constants ...
    GROUP_DECILE GroupType = "GROUP_DECILE"
)

func AllGroupTypes() []GroupType {
    return []GroupType{
        // ... existing entries, alphabetised ...
        GROUP_DECILE,
    }
}

2. Implement and register

Implement the grouper in internal/processing/. Register the factory in grouperRegistry (internal/processing/registry.go). The interface choice depends on whether the grouper can run in the streaming path:

• Grouper — buffered-only. Group(rows) returns a slice of bucket assignments. Used by quantile-class groupers that need the full sorted view.
• StreamingGrouper / StreamableGrouper — streaming. The per-record KeyFor(record) / KeyForRow(record) returns a bucket key as the iterator walks rows; the orchestrator accumulates per key.
• MultiKeyStreamingGrouper — multi-emit streaming. KeysForRow returns N keys per row; used by set-typed groupers (GROUP_SET_PER_ELEMENT).

3. Tests

Add tests in internal/processing/grouper_test.go (or the per-grouper test file) before the implementation. Cover empty input, single-value input, null-bearing input, the Include filter slot if your grouper honours it, and the streaming-vs-buffered parity assertions where applicable.

4. Declare the capability metadata

Add a row to internal/descriptor/capabilities_groupers.go with the grouper’s params, accepted field types, and the ComponentSchema that follows. TestManifestOperatorsComplete enforces a capability row per registered grouper.

5. Declare the ComponentSchema (Response.Components contract)

Every registered grouper MUST declare a ComponentSchema on its capability row in internal/descriptor/capabilities_groupers.go, tagged with one of three mergeability classes:

The groupSchema helper in capabilities_groupers.go prepends the universal floor ({"total_n", "n_null"}) automatically — do not list those keys in your extra slice:

{
    Name:        string(types.GROUP_DECILE),
    Category:    "grouper",
    Description: "Partition the numeric field into ten equally-sized quantile buckets.",
    AcceptsTypes: numericFieldTypesAnalyticsNoDecimal,
    ComponentSchema: groupSchema(None,
        ComponentKey{Name: "edges", Type: "[]float64", Description: "Decile boundary values (length 9)."},
        ComponentKey{Name: "buckets", Type: "map[string]int", Description: "Per-decile record counts keyed by label."},
    ),
},

6. Emit per-operator component values at runtime

Implement MetaGrouper (internal/processing) on the grouper type and add a compile-time assertion in the grouper implementation file:

type decileGrouper struct {
    edges   []float64
    buckets map[string]int
    // ...
}

func (g *decileGrouper) Components() (map[string]any, error) {
    return map[string]any{
        "edges":   g.edges,
        "buckets": g.buckets,
    }, nil
}

var _ MetaGrouper = (*decileGrouper)(nil)

The orchestrator owns the universal floor {total_n, n_null} — the post-filter record walker fills it unconditionally. Your Components() MUST NOT re-emit those keys.

total_n is the number of (record, bucket) assignments: the sum of your buckets[].count when Components() returns a buckets list of []map[string]any with int counts (as GROUP_CATEGORY and GROUP_SET_PER_ELEMENT do), otherwise the orchestrator’s own assignment count. For single-key groupers that is the number of keyed records; for multi-key streaming groupers (GROUP_SET_PER_ELEMENT) a record contributes once per selected label, so total_n can exceed the record count.

Merging under the parallel reducers

The parallel reducers (ShardWorkers over a shard archive, DecodeWorkers over a large single file) build one grouper per partition and fold the per-key aggregator buckets themselves. Your grouper’s own components state (the counters behind Components()) is folded through MergeableGrouper.MergeGrouperState(other) (internal/processing/grouper_merge.go); other is a fresh instance of the same concrete type from the same spec. Implement it and return true from GroupType.Mergeable() (types/streamability.go) only when the fold is associative — a sum, or the same first/last-write rule the serial tracker applies — so the merged Components() equals a serial instance’s. The reducers sum the assignment count across partitions and hand it to processing.FinalizeGroupedStream, so total_n / n_null match the serial path even for a bucket-less grouper. A grouper not Mergeable() runs those requests serially (processing.CanMergeRequestWithExtensions).

An extension grouper takes the same paths through the public contract: declare Streamable: true and Mergeable: true on its pulse.GrouperRegistration and, if it emits components, implement extend.MergeableGrouper.MergeState(other); the adapter hands MergeState your own value. A grouper that emits no components needs no method (the adapter folds nothing). Probe-validation refuses a declaration the value cannot honour (PULSE_EXTENSION_MERGEABLE_MISMATCH). See Extension points — Grouper.

The full Response.Components contract for groupers — streaming behaviour by mergeability class, the orchestrator-owned floor, the extension-side parity contract — lives in the response-components skill. Embedder extensions implement the same MetaGrouper interface via ComponentsFunc or directly; see Extension Points.

7. The Include inclusion-list slot

If your grouper honours an Include whitelist of allowed bucket labels, update internal/processing/grouper.go + internal/processing/grouper_set.go so the include filter is applied at key-emission time, not on the output table after the fact. The contract is enforced by the following gates:

• TestGroupCategory_IncludeFiltersLabels
• TestGroupSetValue_IncludeFiltersCompositeKey
• TestGroupSetPerElement_IncludeFilters

8. Update the grouper-design skill

Add a section in skills/grouper-design.md covering the new grouper’s bucket-naming convention, the params it accepts, the streamability classification, and any cardinality bound.

9. Update CLAUDE.md

There is no registered-grouper count or list in CLAUDE.md to update — CLAUDE.md never hardcodes registered counts; the manifest is the source of truth. The Update Demand operator row (.claude/reference/update-demand.md) requires instead the atomic skill skills/op-group-<kebab>.md, the capability declaration in internal/descriptor/capabilities_groupers.go — and an internal/examples/<dir>/*.json example whose _meta.operators names the operator (TestEveryOperatorHasAnExampleTag). Edit CLAUDE.md only if the operator introduces a contract it states directly, and mind TestClaudeMdSizeBudget — long-form prose belongs in .claude/reference/.

10. Run the gates

go test ./internal/skills/ -run TestSkillsCoverAllComponents
go test ./internal/descriptor/ -run TestManifestOperatorsComplete
go test ./internal/processing/ -run TestGroup

The Update Demand row for groupers (and the Group.Include slot) covers all of these in one PR; see The Update Demand.