pulse

package module
v0.29.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 25, 2026 License: MIT Imports: 33 Imported by: 0

README

Pulse

High-performance, self-describing tabular data processing engine written in Go. Ships as a Go library (github.com/frankbardon/pulse) and a CLI binary (cmd/pulse/). Library is primary; CLI is a thin adapter.

Pulse reads and writes .pulse files — a compact binary format with an inline schema, categorical dictionaries, and per-field descriptions. Import from CSV, TSV, NDJSON, JSON array, Parquet, Apache Arrow IPC, or Excel; run aggregations, filters, groupers, windows, features, statistical tests, and regressions; export back to any supported format.

LLM-native by design: every command supports --json, every operator declares its accepted types and streamability in the manifest, an embedded skill pack teaches agents how to operate Pulse, an embedded request-example library gives them runnable templates, and pulse mcp serves the full surface over Model Context Protocol.

Installation

From source
git clone https://github.com/frankbardon/pulse.git
cd pulse
make build
# Binary at ./bin/pulse
Go install
go install github.com/frankbardon/pulse/cmd/pulse@latest

Requires Go 1.22+.

Quickstart

Import a CSV into a .pulse file
pulse import csv --input data.csv --output data.pulse

Schema is inferred from a sample of rows (default 500). To supply an explicit schema:

# Generate a schema template from your data
pulse import schema-template data.csv > schema.json

# Edit schema.json — adjust types, add descriptions
pulse import csv --input data.csv --schema schema.json --output data.pulse
Inspect the file
pulse cohort inspect data.pulse --json

Returns field names, types, byte offsets, descriptions, and categorical dictionaries (truncated by default; pass --full-dict for the full mapping).

Run an aggregation

Create a request file (request.json):

{
  "cohort": {"filename": "data.pulse"},
  "aggregations": [
    {"type": "AGG_COUNT", "field": "id", "label": "total"},
    {"type": "AGG_AVERAGE", "field": "score", "label": "avg_score"}
  ]
}

pulse api process --request request.json --json
Filter, group, and aggregate
{
  "cohort": {"filename": "data.pulse"},
  "filterers": [
    {"type": "FILTER_RANGE", "field": "score", "values": ["80", "100"]}
  ],
  "groups": [
    {"type": "GROUP_CATEGORY", "field": "brand"}
  ],
  "aggregations": [
    {"type": "AGG_COUNT", "field": "id", "label": "count"},
    {"type": "AGG_AVERAGE", "field": "score", "label": "avg_score"},
    {"type": "AGG_STDDEV", "field": "score", "label": "stddev"}
  ]
}
Smart defaults

If you name a field but omit type, Pulse fills in a sensible operator from the schema type — AGG_SUM for numerics, AGG_FREQUENCY for categoricals, GROUP_RANGE (interval 10) for numerics, GROUP_CATEGORY for categoricals, GROUP_DATE ("day") for dates. Disable with --no-defaults or pulse.Options{DisableDefaults: true}. The full rule table lives in descriptor/defaults.go.

Validate before executing
pulse api predict --request request.json --json

Returns the proposed schema, applied defaults, streamability, warnings (e.g., numeric aggregation on a categorical field), and structural errors without touching record data. Pass --strict to treat warnings as errors.

Streaming

For requests Pulse can stream (single-pass aggregations on non-decimal fields, no buffered ops), use --stream to get NDJSON one row per line:

pulse api process --request request.json --stream
pulse api compose --request batch.json --stream

Library equivalent: pulse.ProcessStream(ctx, req) returns a RowIter.

Parallel compose
pulse api compose --request batch.json --parallel 4 --json

Library equivalent: pulse.ComposeParallel(ctx, req, pulse.ComposeOptions{MaxWorkers: 4}). Order-preserving by slot; --no-fail-fast aggregates errors instead of cancelling on first failure.

Export and convert
pulse export csv --input data.pulse --output results.csv
pulse export parquet --input data.pulse --output results.parquet

pulse convert data.csv data.parquet
pulse convert data.xlsx output.tsv --schema schema.json

Format auto-detected from extensions. convert does not write an intermediate .pulse unless --keep-pulse=path.

Sample rows / facet a field
pulse api sample --input data.pulse --count 10
pulse api facet --input data.pulse --field brand
Synthetic data

Generate a deterministic synthetic cohort from a schema spec or from a previously-captured profile:

pulse synth from-schema --spec spec.json --output synth.pulse --seed 42
pulse profile create --input real.pulse --output profile.json --include-correlations
pulse synth from-profile --profile profile.json --output synth.pulse --rows 100000 --seed 42

12 distributions (normal, lognormal, poisson, exponential, pareto, bernoulli, weighted_categorical, regex, uniform, uniform_date, monotonic_from, constant), pairwise correlations, value constraints. See the synthetic-data skill.

CLI Reference

pulse
├── --json [--slim]                       Root manifest (self-description)
├── import
│   ├── csv|tsv|ndjson|jsonarray|         --input FILE --output FILE [--schema FILE]
│   │   parquet|arrow|excel
│   ├── auto SOURCE                       Managed import (auto-detect format)
│   ├── list                              List managed import handles
│   ├── drop HANDLE                       Drop a managed handle
│   ├── predict                           --input FILE [--schema FILE] --json
│   └── schema-template INPUT             Generate editable schema from source
├── export
│   ├── csv|tsv|ndjson|jsonarray|         --input FILE --output FILE
│   │   parquet|arrow|excel
│   └── predict                           --input FILE --format FORMAT --json
├── convert INPUT OUTPUT [--from F] [--to F] [--schema FILE] [--keep-pulse PATH]
│   └── predict INPUT OUTPUT --json
├── cohort
│   ├── inspect PATH [--json] [--full-dict]
│   └── filter --input FILE --output FILE --filter EXPR
├── api
│   ├── process --request FILE [--json] [--stream] [--no-defaults]
│   ├── compose --request FILE [--json] [--stream] [--parallel N] [--no-fail-fast]
│   ├── sample --input FILE --count N
│   ├── facet --input FILE --field NAME
│   └── predict --request FILE --json [--strict]
├── synth
│   ├── from-schema --spec FILE --output FILE [--rows N] [--seed N]
│   └── from-profile --profile FILE --output FILE --rows N [--seed N]
├── profile
│   └── create --input FILE --output FILE [--top-k N] [--include-correlations]
├── skills
│   ├── list [--json]
│   └── show NAME
├── examples
│   ├── search [--query Q] [--tag T ...] [--category C] [--json]
│   └── show NAME [--json]
├── errors
│   ├── lookup CODE [--json]
│   └── list [--domain D] [--query Q] [--json]
└── mcp [--data-dir DIR] [--bind-on-open]   Serve MCP over stdio

Every leaf supports --json for output wrapped in a descriptor.Envelope with format_version: "1.1", data, errors, and warnings. pulse api compose --json wraps a ComposedResponse {responses[], overlays[]} on data — not the legacy []*Response array.

MCP Usage

Pulse ships an MCP (Model Context Protocol) server that exposes the full library surface to AI clients (Claude Desktop, Claude Code, Cursor, any MCP-aware host). One binary, one command:

pulse mcp --data-dir /path/to/data

PULSE_DATA_DIR is honored if --data-dir is omitted. The server speaks JSON-RPC over stdio (logs to stderr).

Wiring into a host

For Claude Desktop, add to ~/Library/Application Support/Claude/claude_desktop_config.json:

{
  "mcpServers": {
    "pulse": {
      "command": "pulse",
      "args": ["mcp"],
      "env": {
        "PULSE_DATA_DIR": "/Users/me/data"
      }
    }
  }
}

For Claude Code:

claude mcp add pulse --env PULSE_DATA_DIR=/Users/me/data -- pulse mcp

Restart the host. Pulse tools appear in the tool list.

Tool surface
Tool Purpose
pulse_manifest Call first. Self-description: commands, operators, accepted types, tests, regressions, distributions, error codes, MCP tools, cohort field types. Cache once per session.
pulse_inspect Read header + schema (no record bytes). Also binds session-scoped field-name enums on action tools.
pulse_predict Validate a request against the schema without executing.
pulse_process Execute one pre-built request.
pulse_compose Execute a batch of requests in one round trip.
pulse_sample Return up to N rows for preview.
pulse_facet Distinct values for a single field.
pulse_import Import a tabular source into a managed .pulse handle (TTL-tracked, default 7d).
pulse_drop Drop a managed handle.
pulse_imports_list Enumerate managed handles with sidecar metadata.
pulse_examples_search Search the embedded request-example library by query, tags, category.
pulse_examples_get Fetch one runnable example by name.
pulse_errors_lookup Per-code Message + Fixup detail (kept out of the manifest for context economy).
pulse_skills_list / pulse_skills_get Embedded skill pack.
Resources and prompts
URI What
pulse://<path> One per .pulse file under the data directory. Read returns descriptor.InspectResult JSON.
pulse-skill://<name> One per embedded skill. Read returns the markdown body.

Two prompts (pulse-bootstrap, pulse-author-request) are registered for hosts that surface them as slash commands.

  1. pulse_manifest once at session start. Cache the result — it is deterministic for a binary version and carries every fact needed to author a valid request.
  2. pulse_import when the user hands the LLM a raw tabular file; skip when the cohort already exists as a managed handle or .pulse under PULSE_DATA_DIR.
  3. pulse_inspect on the handle (or path). Reads header + schema only and binds session-scoped field-name enums on the action tools.
  4. pulse_predict with the authored request. Validates against the schema; each error code carries Fixup metadata so the LLM can repair the request without another round trip.
  5. pulse_process to execute. Use pulse_compose for batching, pulse_sample / pulse_facet for cheap probes.
Schema-bound enums

After a successful pulse_inspect, the server registers session-scoped variants of the action tools whose JSON Schemas embed enums on field-name parameters — picked from a typed list rather than free-texted. Works on SSE / Streamable HTTP transports; on stdio the session does not support tool overrides, so enums are advisory and pulse_predict remains the validation gate. Disable with --bind-on-open=false.

The mcp-integration skill (pulse skills show mcp-integration) is the authoritative reference.

Embedding Pulse in a Go Application

Pulse is library-first. The CLI is a thin adapter over the public API.

package main

import (
    "context"
    "fmt"
    "log"

    "github.com/frankbardon/pulse"
    pio "github.com/frankbardon/pulse/io"
    "github.com/frankbardon/pulse/io/csv"
    "github.com/frankbardon/pulse/types"
)

func main() {
    ctx := context.Background()

    p, err := pulse.New(pulse.Options{DataDir: "/var/data"})
    if err != nil {
        log.Fatal(err)
    }

    // Import a CSV.
    importJob := &pio.ImportJob{
        Source: csv.NewReader(nil, "input.csv"),
        Target: "dataset.pulse",
    }
    report, err := p.Import(ctx, importJob)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("Imported %d rows\n", report.RowsImported)

    // Run an aggregation.
    resp, err := p.Process(ctx, &pulse.Request{
        Cohort: &types.Cohort{Filename: "dataset.pulse"},
        Aggregations: []*types.Aggregation{
            {Type: types.AGG_AVERAGE, Field: "score", Label: "avg"},
        },
    })
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println(resp.Data)

    // Inspect a file.
    result, err := p.Inspect(ctx, "dataset.pulse")
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("Fields: %d\n", result.FieldCount)

    // Validate before execution.
    prediction, err := p.Predict(ctx, &pulse.Request{
        Cohort: &types.Cohort{Filename: "dataset.pulse"},
        Aggregations: []*types.Aggregation{
            {Type: types.AGG_SUM, Field: "revenue"},
        },
    })
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("Warnings: %v\n", prediction.Warnings)
}
Public facade

pulse.Pulse exposes: New, Open, Process, ProcessStream, Compose, ComposeParallel, Import, ImportFile, Drop, Imports, SweepImports, ResolveImport, Export, Convert, Inspect, Predict, Sample, Facet, Synth, Profile, Manifest, plus example/error lookup helpers.

Custom filesystem

Pulse accepts any afero.Fs for testing or non-local backends:

import "github.com/spf13/afero"

// In-memory filesystem for testing
p, _ := pulse.New(pulse.Options{
    FS: afero.NewMemMapFs(),
})

fs.NewMemMap() returns a complete in-memory Config for hermetic tests — no disk I/O.

Operator catalogue

Counts as currently registered (the manifest is the source of truth — pulse --json --slim):

  • 16 aggregators (AGG_*): COUNT, SUM, AVERAGE, MIN, MAX, MEDIAN, STDDEV, RANGE, FREQUENCY, MODE, PERCENTILE, ZSCORE, KURTOSIS, …
  • 9 attributes (ATTR_*): ZSCORE, TSCORE, NORMALIZED, FORMULA, PERCENTILE, DATE_PART, …
  • 5 filterers (FILTER_*): INCLUDE, EXCLUDE, RANGE, EXPRESSION, NULL
  • 5 groupers (GROUP_*): CATEGORY, RANGE, ROUNDED, DATE, QUANTILE
  • 10 windows (WIN_*): LAG, LEAD, ROW_NUMBER, RANK, DENSE_RANK, RUNNING_SUM, RUNNING_AVG, MOVING_AVG, EWMA, PCT_CHANGE
  • 9 features (FEAT_*): LOG, SQRT, BUCKETIZE, ONE_HOT, FREQUENCY_ENCODE, TARGET_ENCODE, DATE_FEATURES, TRAIN_TEST_SPLIT, POLY
  • 20 statistical tests (TEST_*): tier-1 row tests (T, WELCH, CHISQ, ANOVA_F, ANOVA_WELCH, ANOVA_RM, KS, PAIRED_T, PROP_Z, PEARSON_R, SPEARMAN_R, KENDALL_TAU, MANN_WHITNEY_U, WILCOXON_SR, KRUSKAL_WALLIS, BROWN_FORSYTHE, FISHER_EXACT, SHAPIRO_WILK) and tier-2 post-tests (TUKEY_HSD, TREND, variants)
  • 3 regressions (REG_*): OLS, GLM, BAYES_LINEAR — with Resample / Selection modifiers and FEAT_POLY composition cover 13 textbook regression names
  • 12 synth distributions

LLM Skill Pack

Pulse bundles 22 skill documents that teach LLM agents how to operate it. Skills are embedded via //go:embed — no external files.

Discovering skills
pulse skills list
pulse skills list --json
pulse skills show aggregation-guide
Bundled skills
Skill Purpose
getting-started Pulse vocabulary, MCP tool surface, file format, operator catalog
cohort-schema-design Field types, nullability, bit-packing, descriptions
aggregation-guide Aggregator selection (AGG_) and filterer selection (FILTER_)
attribute-composition ATTR_* derived columns: z-score, formula, percentile, date_part
grouper-design CATEGORY, RANGE, ROUNDED, DATE, QUANTILE
window-operations LAG/LEAD/RANK/MOVING_AVG/EWMA partitioning and frame semantics
feature-engineering Pre-filter FEAT_* operators for ML pipelines + leakage trap
statistical-testing Tier-1 row tests and tier-2 post-tests
regression-modeling OLS, GLM, Bayesian linear; modifiers; 13 textbook names mapped
synthetic-data Distributions, correlations, constraints
compose-requests Multi-request batching against one cohort
debugging-with-predict Iterating with pulse_predict / pulse api predict
error-code-reference Reading envelopes; calling pulse_errors_lookup
import-best-practices Schema inference, fail-closed semantics, PULSE_IMPORT_*
export-format-selection CSV / TSV / NDJSON / JSON array / Parquet / Arrow / Excel
financial-cohorts decimal128 semantics for money
mcp-integration MCP tool surface, schema-bound enums, session bootstrap
contributor-workflow Recipes for extending Pulse
From Go
import "github.com/frankbardon/pulse/skills"

for _, s := range skills.List() {
    fmt.Printf("%s: %s\n", s.Name, s.Description)
}

content, ok := skills.Get("aggregation-guide")
if ok {
    agent.AddContext(content)
}

The root manifest (pulse --json) includes a skills[] array so agents can discover available skills in one call.

.pulse File Format

Binary, self-describing, fully transportable:

  • 9-byte header: magic bytes (PULSE\x00\x00\x00) + format version (0x01)
  • Schema block: field count, per-field descriptors (type, name, byte offset, bit position, source column index, optional description capped at 1000 bytes)
  • Dictionary blocks: one per categorical field (string-to-integer mapping stored inline)
  • Record data: fixed-width binary records, one per row

17 field types:

Type Bytes Notes
u8, u16, u32, u64 1, 2, 4, 8 Unsigned integers
f32, f64 4, 8 IEEE 754 floats
date 4 Days since Unix epoch
packed_bool 0 Bit-packed; shares bytes with adjacent packed fields
nullable_bool 0 Tri-state; bit-packed
nullable_u4 0 4-bit unsigned, nullable; bit-packed
nullable_u8, nullable_u16 1, 2 Nullable unsigned integers
categorical_u8, categorical_u16, categorical_u32 1, 2, 4 Dictionary-encoded strings
decimal128 16 Fixed precision/scale; banker's rounding
nullable_decimal128 16 Nullable decimal128

Categorical width auto-selected from sample cardinality during import. Bit-packed types report ByteSize() == 0 — they share bytes with adjacent packed fields. Schema reader rejects unknown type bytes at parse time with ENCODING_INVALID.

Configuration

Three environment variables, all optional:

export PULSE_DATA_DIR=/path/to/data        # Base directory for .pulse cohort files
export PULSE_IMPORTS_DIR=imports           # Subdir for managed-import handles (default "imports")
export PULSE_IMPORT_TTL=7d                 # Default TTL for managed handles ("24h", "30m", "7d", "pin")

Embedders override per-instance via pulse.Options{DataDir, ImportsDir, ImportTTL, FS}. No config files.

Output Format Contract

All --json output is wrapped in a descriptor.Envelope:

{
  "format_version": "1.1",
  "data": { ... },
  "errors": [],
  "warnings": []
}

format_version is "1.1" (bumped from "1.0" for the Compose facade lift — pulse api compose --json now wraps a ComposedResponse object {responses[], overlays[]} on data instead of a raw []*Response array). Errors/warnings use {"code", "message", "details"} entries (empty array when absent, never null). Additive-only: new data fields don't bump the version; renames or removals do.

Development

Build
make build    # Binary at ./bin/pulse
make test     # go test ./...
make fmt      # gofmt
make vet      # go vet
make lint     # staticcheck (auto-installed via go run)
make cover    # Coverage report
make docs     # Build mdBook
make clean    # Remove artifacts

A .env at repo root is auto-loaded by the Makefile.

Run tests
go test ./...
go test ./processing/...
go test ./service/... -v -run TestProcess
go test ./encoding/... -fuzz FuzzPulseFileHeader -fuzztime 30s
Project structure
pulse/
├── pulse.go                Public facade
├── cmd/pulse/              CLI binary (thin adapter)
├── service/                Orchestration: wires processing to encoding
├── processing/             Aggregators, attributes, filterers, groupers
│   ├── window/             WIN_* operators
│   └── feature/            FEAT_* pre-filter feature engineers
├── encoding/               .pulse binary codec
├── io/                     Tabular ↔ .pulse adapters
│   └── csv|tsv|ndjson|jsonarray|arrow|parquet|excel/
├── fs/                     afero-based filesystem abstraction
├── errors/                 Typed error codes (CodedError system)
├── types/                  Request/response structs + streamability table
├── descriptor/             manifest, predict, inspect, envelope (no-execute)
├── skills/                 //go:embed markdown skill pack
├── examples/               //go:embed runnable request examples
├── synth/                  Synthetic data generator
├── docs/                   mdBook source (GitHub Pages)
└── internal/
    ├── cli/                CLI internals
    └── mcp/                MCP server (wraps pulse.Pulse)

Documentation: https://frankbardon.github.io/pulse/.

License

MIT. See LICENSE.

Documentation

Overview

Package pulse is a high-performance, self-describing tabular data processing engine.

Pulse ships as a CLI binary and as an embeddable Go library. The library is the primary deliverable; the CLI is a thin adapter over it.

Index

Constants

View Source 
const (
	LabelModeReplace = types.LabelModeReplace
	LabelModeAugment = types.LabelModeAugment
)

LabelMode constants re-exported for caller ergonomics.

View Source 
const (
	// LookupMultiplicityAssertUnique errors PULSE_LOOKUP_AMBIGUOUS when
	// the key matches more than one row. It is the zero-value default.
	LookupMultiplicityAssertUnique = types.LookupMultiplicityAssertUnique
	// LookupMultiplicityFirst returns the lowest-row-id match.
	LookupMultiplicityFirst = types.LookupMultiplicityFirst
	// LookupMultiplicityAll returns every match, ascending row-id order.
	LookupMultiplicityAll = types.LookupMultiplicityAll
)

LookupMultiplicity mode constants, re-exported so callers can set LookupRequest.Multiplicity without importing the types package.

Variables

This section is empty.

Functions

This section is empty.

Types

type AggregatorComponentsFunc added in v0.20.0 

type AggregatorComponentsFunc func(instance processing.Aggregator) (map[string]any, error)

AggregatorComponentsFunc is the optional emitter an embedder may supply alongside an AggregatorRegistration to make the operator participate in the per-operator components contract. The orchestrator invokes the func ONCE after Aggregate / Finalize and routes the result onto Response.Components.Aggregations[i].Operator; the universal floor ({"n", "n_null"}) is filled unconditionally by the orchestrator. Returning (nil, nil) is the canonical signal for "no operator-specific keys; orchestrator's universal floor is the entire payload" (floor-only operators).

The instance passed in is the value the registration's Factory returned (after Aggregate / Finalize); the func should not call Aggregate / UpdateRow / Finalize on it. The returned map's keys must be a SUBSET of the registration's ComponentSchema.Keys (universal- floor keys allowed but not required) — probe-validation enforces this contract at pulse.New time and runtime mismatches surface as PULSE_EXTENSION_COMPONENT_SCHEMA_MISMATCH.

type AggregatorRegistration added in v0.7.0 

type AggregatorRegistration struct {
	Name        types.AggregationType
	Description string
	Factory     processing.AggregatorFactory
	Streamable  bool
	Accepts     []encoding.FieldType
	Params      []ParamMeta
	// FieldInputs is the optional buffered-projection introspection
	// hook. See FieldInputsFunc. Omit to keep the operator opaque to
	// projection (runtime widens the field set when this operator
	// appears in a request).
	FieldInputs FieldInputsFunc
	// ComponentSchema declares the per-operator components contract
	// surfaced through Response.Components.Aggregations[i].Operator.
	// Leaving the schema empty (zero-value Keys plus
	// empty Mergeability) makes the operator floor-only — the
	// orchestrator emits the universal floor ({"n", "n_null"}) and no
	// extra keys. When ComponentSchema.Keys is non-empty, the
	// ComponentsFunc emitter MUST be supplied and its returned map
	// keys MUST be a subset of the declared key set; probe-validation
	// enforces both at pulse.New time.
	ComponentSchema descriptor.ComponentSchema
	// ComponentsFunc is the optional sibling-interface emitter. When
	// set, the orchestrator routes the returned map onto
	// Response.Components.Aggregations[i].Operator after the
	// aggregator's Aggregate / Finalize call terminates. Nil is the
	// floor-only path (universal floor fills the entire payload).
	// Mirrors processing.MetaAggregator.Components() in shape.
	ComponentsFunc AggregatorComponentsFunc
}

AggregatorRegistration installs a custom AGG_* operator. The factory must obey processing.AggregatorFactory: it builds a fresh Aggregator per Process call against the supplied Aggregation spec + schema.

When Streamable=true the factory MUST return a value that also implements processing.OnlineAggregator. Probe-validation at registration time enforces the contract via PULSE_EXTENSION_STREAMABLE_MISMATCH.

type AttributeEmitType added in v0.7.0 

type AttributeEmitType string

AttributeEmitType is a manifest hint declaring the dtype the attribute emits per record. Defaults to "float64" when empty. 

const (
	AttributeEmitFloat64 AttributeEmitType = "float64"
	AttributeEmitBool    AttributeEmitType = "bool"
	AttributeEmitString  AttributeEmitType = "string"
)

type AttributeMode added in v0.7.0 

type AttributeMode string

AttributeMode declares which streaming tier an attribute factory promises to satisfy. 

const (
	// AttributeModeRowLocal — factory returns a processing.RowLocalAttribute.
	// Equivalent to built-in ATTR_FORMULA / ATTR_DATE_PART.
	AttributeModeRowLocal AttributeMode = "row_local"
	// AttributeModeTwoPass — factory returns a processing.TwoPassAttribute.
	// Equivalent to ATTR_ZSCORE / ATTR_TSCORE / ATTR_NORMALIZED.
	AttributeModeTwoPass AttributeMode = "two_pass"
	// AttributeModeBuffered — factory returns a plain
	// processing.AttributeComputer; no streaming. Equivalent to
	// ATTR_PERCENTILE.
	AttributeModeBuffered AttributeMode = "buffered"
)

type AttributeRegistration added in v0.7.0 

type AttributeRegistration struct {
	Name        types.AttributeType
	Description string
	Factory     processing.AttributeFactory
	Mode        AttributeMode
	Accepts     []encoding.FieldType
	Emits       AttributeEmitType
	Params      []ParamMeta
	// FieldInputs is the optional buffered-projection introspection
	// hook. See FieldInputsFunc.
	FieldInputs FieldInputsFunc
}

AttributeRegistration installs a custom ATTR_* operator.

type BuildIndexResult added in v0.27.0 

type BuildIndexResult = service.BuildIndexResult

BuildIndexResult re-exports service.BuildIndexResult — the outcome of a successful point-lookup sidecar index build: the derived sidecar path (see encoding.SidecarIndexPath) plus the in-memory encoding.Index that was serialized there.

type ChainRequest added in v0.10.0 

type ChainRequest = types.ChainRequest

ChainRequest re-exports types.ChainRequest so callers can use pulse.ChainRequest as the input to ProcessChain.

type ChainResponse added in v0.10.0 

type ChainResponse = types.ChainResponse

ChainResponse re-exports types.ChainResponse.

type ChainStage added in v0.10.0 

type ChainStage = types.ChainStage

ChainStage re-exports types.ChainStage.

type ChangeEvent added in v0.11.0 

type ChangeEvent struct {
	Path      string
	Kind      ChangeKind
	Hash      string
	Timestamp time.Time
}

ChangeEvent is one observed mutation against a watched .pulse path (or any file, when WatchDir is rooted at a non-.pulse directory).

type ChangeKind added in v0.11.0 

type ChangeKind int

ChangeKind identifies the kind of change a Watch / WatchDir emits. 

const (
	// ChangeCreated reports the appearance of a new file in the
	// watched scope. Hash is populated when the file is readable.
	ChangeCreated ChangeKind = iota
	// ChangeModified reports an in-place content change on an
	// already-tracked path. Hash carries the post-change content
	// hash.
	ChangeModified
	// ChangeRemoved reports that a previously-tracked path is gone.
	// Hash is empty.
	ChangeRemoved
	// ChangeRenamed reports an atomic write that promoted a temp
	// file to the watched path. Coalesces a Removed-or-Created
	// pair into a single Renamed event aimed at the destination.
	ChangeRenamed
)

func (ChangeKind) String added in v0.11.0 

func (k ChangeKind) String() string

String returns a stable lowercase identifier for the kind.

type Cohort 

type Cohort struct {
	// contains filtered or unexported fields
}

Cohort represents an opened .pulse file with its parsed schema. It wraps the service-layer Cohort to provide a clean public API.

func (*Cohort) Categorical 

func (c *Cohort) Categorical(name string) (*encoding.Dictionary, bool)

Categorical returns the dictionary for a named categorical field. Returns nil, false if the field is not found or is not categorical.

func (*Cohort) Field 

func (c *Cohort) Field(name string) *encoding.Field

Field returns a pointer to the named field, or nil if not found.

func (*Cohort) RecordCount added in v0.8.3 

func (c *Cohort) RecordCount() (int64, error)

RecordCount returns the number of records in the cohort. For single-file cohorts this is derived from the byte length of the record region divided by the per-record size implied by the schema. For archive-backed cohorts the caller should sum per-shard RecordCount values from Shards() — the underlying service Cohort errors on RecordCount for archives because the byte-region path doesn't apply across shards.

func (*Cohort) Schema 

func (c *Cohort) Schema() *encoding.Schema

Schema returns the cohort's schema.

func (*Cohort) Shards added in v0.8.0 

func (c *Cohort) Shards() []ShardEntry

Shards returns the shard manifest for an archive-backed cohort. Empty for single-file cohorts. The returned slice is a defensive copy.

type ComposeOptions added in v0.2.0 

type ComposeOptions = service.ComposeOptions

ComposeOptions controls parallel execution. See service.ComposeOptions.

type ComposedRequest 

type ComposedRequest = types.ComposedRequest

Type aliases re-exported from the types package so embedders can use pulse.Request instead of types.Request.

type ComposedResponse added in v0.22.0 

type ComposedResponse = types.ComposedResponse

Type aliases re-exported from the types package so embedders can use pulse.Request instead of types.Request.

type DateRangeSpec added in v0.27.0 

type DateRangeSpec = processing.DateRangeSpec

DateRangeSpec is the public alias for processing.DateRangeSpec — the wire-level shape of a single labeled date range ({label, start, end}). It is the shared model authored inline on a GROUP_DATE_RANGES grouper or FILTER_DATE_RANGES filter and registered in an Extensions.RangeTables entry. Start / End are ISO date literals; nil / empty is an open bound.

type DistributionRegistration added in v0.7.0 

type DistributionRegistration struct {
	Name        string
	Description string
	// Factory is the sampler-construction callback. Its concrete
	// signature is finalised when the synth distribution overlay
	// lands; until then this field is held opaque to keep the API
	// shape stable.
	Factory any
	Params  []ParamMeta
}

DistributionRegistration installs a custom synthetic-data distribution kind. The factory shape is finalised alongside the synth wiring in a later phase; for now this registration is validated for naming + duplicates only and the Factory field is held opaque.

type ErrorFixup added in v0.5.0 

type ErrorFixup = errors.Fixup

ErrorFixup is one repair template attached to an error code.

type ErrorMetadata added in v0.5.0 

type ErrorMetadata = errors.LookupResult

ErrorMetadata is the depth-on-demand projection returned by ErrorLookup, ErrorsByDomain, and ErrorsSearch. Carries the code, domain, canonical Message, and materialised Fixup templates.

type Example added in v0.5.0 

type Example = examples.Example

Example is the full record returned by ExampleGet — runnable request JSON plus the indexed metadata.

type ExampleSummary added in v0.5.0 

type ExampleSummary = examples.ExampleSummary

ExampleSummary is the lightweight projection returned by ExamplesSearch.

type ExprFunction added in v0.7.0 

type ExprFunction struct {
	Name        string
	Description string
	// Signature is a human-readable signature surfaced in the
	// manifest and in predict error messages when typecheck fails
	// (e.g. `rank_familiarity(value float64, total_pop bool) float64`).
	Signature string
	// Fn is the Go function. expr-lang accepts any func value;
	// the runtime resolves argument types against the call site.
	Fn any
	// Pure declares the function is side-effect-free and depends
	// only on its arguments. Reserved for a future memoization
	// optimisation; declaring it has no runtime effect today.
	Pure bool
}

ExprFunction describes a function callable from runtime expressions (ATTR_FORMULA / FILTER_EXPRESSION). At pulse.New time the Fn value is registered with the expr-lang engine under Name.

type Extensions added in v0.7.0 

type Extensions struct {
	Aggregators        []AggregatorRegistration
	Attributes         []AttributeRegistration
	Filterers          []FiltererRegistration
	Groupers           []GrouperRegistration
	Windows            []WindowRegistration
	Features           []FeatureRegistration
	Tests              []TestRegistration
	SynthDistributions []DistributionRegistration

	// ExprFunctions are merged into the runtime expression environment
	// used by ATTR_FORMULA and FILTER_EXPRESSION (plus any future
	// expression hook). Each entry is callable from request expressions
	// under its declared Name.
	ExprFunctions []ExprFunction

	// LookupTables expose static keyed maps to the runtime expression
	// environment. Callers reference them as
	//   lookup("table_name", key1, key2, ...)
	// The built-in lookup() function is added to the expression env
	// whenever LookupTables is non-empty. Tables are read-only after
	// pulse.New.
	LookupTables map[string]LookupTable

	// LabelTables expose static ID→label maps used by the output-time
	// categorical-label overlay (per-request Labels slot on Request /
	// SampleRequest / FacetRequest / ExportJob). Each table accepts a
	// single string key (the categorical's resolved dictionary value)
	// and returns a display string. Tables are read-only after
	// pulse.New and do not affect filter / formula / sort semantics —
	// only end-user output rendering.
	LabelTables map[string]LabelTable

	// RangeTables expose named sets of labeled date ranges so a
	// GROUP_DATE_RANGES grouper or FILTER_DATE_RANGES filter can
	// reference an ordered {label, start, end} bucketing by name
	// instead of authoring it inline per request. Each table is
	// validated at pulse.New time via the shared range-compilation
	// pass (the same overlap / duplicate-label / empty / invalid-
	// boundary rules the inline path enforces); a table that nothing
	// references is valid and simply visible in the manifest. Tables
	// are read-only after pulse.New. Operator→table resolution is
	// wired separately; this slot only registers the tables.
	RangeTables map[string]RangeTable
}

Extensions bundles every per-category registration slot plus the expression-environment additions an embedder wants exposed to runtime ATTR_FORMULA and FILTER_EXPRESSION evaluation.

Zero value is the zero-extension case — pulse.New(Options{}) with no Extensions field behaves identically to a Pulse instance shipped without this surface. The registration types live in a separate namespace from the built-in registries (no collision risk) and the runtime treats them identically to built-ins.

type FacetDiscrete added in v0.7.0 

type FacetDiscrete = types.FacetDiscrete

FacetDiscrete is the per-value count list for discrete fields.

type FacetField added in v0.7.0 

type FacetField = types.FacetField

FacetField wraps either a discrete or numeric per-field summary.

type FacetHistogram added in v0.7.0 

type FacetHistogram = types.FacetHistogram

FacetHistogram is the fixed-width binning of a numeric field.

type FacetNumeric added in v0.7.0 

type FacetNumeric = types.FacetNumeric

FacetNumeric is the streaming-stats summary for numeric fields.

type FacetRequest added in v0.7.0 

type FacetRequest = types.FacetRequest

FacetRequest is the input to FacetSchema — multi-field, with optional filters, percentiles, histograms, and additive contribution counts.

type FacetResult added in v0.7.0 

type FacetResult = types.FacetResult

FacetResult is the response from FacetSchema.

type FacetValueCount added in v0.7.0 

type FacetValueCount = types.FacetValueCount

FacetValueCount is one (value, count) tuple inside FacetDiscrete.

type FeatureRegistration added in v0.7.0 

type FeatureRegistration struct {
	Name        types.FeatureType
	Description string
	Factory     feature.Factory
	Streamable  bool
	Accepts     []encoding.FieldType
	Params      []ParamMeta
	// FieldInputs is the optional buffered-projection introspection
	// hook. See FieldInputsFunc.
	FieldInputs FieldInputsFunc
}

FeatureRegistration installs a custom FEAT_* operator. Set Streamable=true when the factory returns a feature.Computer that also implements feature.StreamingComputer.

type FieldInputsFunc added in v0.8.0 

type FieldInputsFunc func(raw json.RawMessage) []string

FieldInputsFunc is the optional introspection callback an extension registration may supply so the buffered-projection extractor can determine which extra schema fields the operator reads beyond the spec's explicit Field/Field2/PartitionBy/etc. references. raw is the operator's Params block. Return value lists schema field names to include; nil/empty means "no extra fields."

Embedders that omit this hook leave their operator opaque to the projection extractor — the runtime then widens the buffered-decode field set to "every field" so the operator stays correct.

type FilterToFileRequest added in v0.11.0 

type FilterToFileRequest struct {
	// SourcePath identifies the cohort to filter. Single-file path,
	// shard archive, or `archive.pulse#shard.pulse` anchor — same
	// dispatch rules as Pulse.FilterToFile.
	SourcePath string `json:"source_path"`

	// Expression is the FILTER_EXPRESSION-style predicate evaluated
	// per row. Empty when Filterers is supplied instead.
	Expression string `json:"expression,omitempty"`

	// Filterers is the structured-predicate alternative. When set,
	// the request is translated into the equivalent expression and
	// passed to the existing FilterToFile engine. Mutually exclusive
	// with Expression.
	Filterers []*types.Filterer `json:"filterers,omitempty"`

	// OutputDir is the directory the result file lands in. Required.
	OutputDir string `json:"output_dir"`

	// OutputName is the optional override for the result basename.
	// When empty, defaults to `{source-hash}_{predicate-hash}.pulse`
	// so dedup is automatic. When set, the engine still validates the
	// caller-supplied path against the deterministic hash and refuses
	// to overwrite an existing file with mismatching content.
	OutputName string `json:"output_name,omitempty"`
}

FilterToFileRequest is the structured request shape for the deterministic filter-to-file operation. It deduplicates by `{source-hash}_{predicate-hash}.pulse` so two callers issuing the same filter against the same source land on the same output file and the engine skips redundant work.

Exactly one of Expression / Filterers must be set. Both empty is an error; both set is rejected so the predicate hash is unambiguous.

type FilterToFileResult added in v0.11.0 

type FilterToFileResult struct {
	OutputPath string `json:"output_path"`
	OutputHash string `json:"output_hash"`
	RowCount   int64  `json:"row_count"`
	ElapsedMs  int64  `json:"elapsed_ms"`
	// Reused is true when the engine returned an existing output file
	// without re-running the filter (content-hash match).
	Reused bool `json:"reused"`
}

FilterToFileResult is the structured response from FilterToFileWithRequest.

type FiltererComponentsFunc added in v0.20.0 

type FiltererComponentsFunc func(instance processing.FiltererBuilder) (map[string]any, error)

FiltererComponentsFunc is the filterer-shape sibling. In v1 every built-in filterer leaves its operator-specific payload empty (uniform floor of {n_in, n_out, n_null_input}); extensions may opt in by supplying a ComponentSchema + this emitter. The instance passed in is the FiltererBuilder the registration's Factory returned.

type FiltererRegistration added in v0.7.0 

type FiltererRegistration struct {
	Name        types.FiltererType
	Description string
	Factory     processing.FiltererFactory
	Accepts     []encoding.FieldType
	Params      []ParamMeta
	// FieldInputs is the optional buffered-projection introspection
	// hook. See FieldInputsFunc. Filterers don't carry a Params
	// block today; the callback receives nil raw bytes and should
	// return the static set of extra fields the filterer reads
	// beyond Filterer.Field.
	FieldInputs FieldInputsFunc
	// ComponentSchema declares the per-operator components contract
	// surfaced through Response.Components.Filterers[i].Operator.
	// v1 ships every built-in filterer as floor-only
	// (uniform {n_in, n_out, n_null_input}); extensions may opt in by
	// declaring a non-empty ComponentSchema alongside a ComponentsFunc.
	// Mismatch between declared schema and emitted keys is rejected at
	// pulse.New time.
	ComponentSchema descriptor.ComponentSchema
	// ComponentsFunc is the optional sibling-interface emitter. When
	// set, the orchestrator routes the returned map onto
	// Response.Components.Filterers[i].Operator after the filter pass
	// terminates. Nil is the floor-only path (the orchestrator's
	// universal floor is the entire payload). Mirrors
	// processing.MetaFilterer.Components() in shape.
	ComponentsFunc FiltererComponentsFunc
}

FiltererRegistration installs a custom FILTER_* operator. Filterers are always row-local streamable today — every built-in filterer evaluates per-row. The runtime records custom filterers as streamable; a buffered-filterer registration shape lands the day Pulse grows a buffered-filterer path.

type GrouperComponentsFunc added in v0.20.0 

type GrouperComponentsFunc func(instance processing.Grouper) (map[string]any, error)

GrouperComponentsFunc is the grouper-shape sibling of AggregatorComponentsFunc. The orchestrator invokes the func ONCE after the grouper's terminal partition pass and routes the result onto Response.Components.Groupers[i].Operator. The universal floor ({total_n, n_null}) is filled unconditionally by the orchestrator. Returning (nil, nil) is the canonical signal for "no operator- specific keys".

type GrouperRegistration added in v0.7.0 

type GrouperRegistration struct {
	Name        types.GroupType
	Description string
	Factory     processing.GrouperFactory
	Streamable  bool
	Accepts     []encoding.FieldType
	Params      []ParamMeta
	// FieldInputs is the optional buffered-projection introspection
	// hook. See FieldInputsFunc.
	FieldInputs FieldInputsFunc
	// ComponentSchema declares the per-operator components contract
	// surfaced through Response.Components.Groupers[i].Operator.
	// Leaving the schema empty makes the operator
	// floor-only — the orchestrator emits the universal floor
	// ({total_n, n_null}) without operator-specific extras. When
	// ComponentSchema.Keys is non-empty, the ComponentsFunc emitter
	// MUST be supplied and its keys MUST match the declared schema.
	ComponentSchema descriptor.ComponentSchema
	// ComponentsFunc is the optional sibling-interface emitter. When
	// set, the orchestrator routes the returned map onto
	// Response.Components.Groupers[i].Operator after the grouper's
	// terminal partitioning pass. Nil is the floor-only path. Mirrors
	// processing.MetaGrouper.Components() in shape.
	ComponentsFunc GrouperComponentsFunc
}

GrouperRegistration installs a custom GROUP_* operator. Set Streamable=true when the factory returns a processing.Grouper that also implements processing.StreamingGrouper (KeyForRow).

type ImportEntry added in v0.5.0 

type ImportEntry = imports.Entry

Type aliases re-exported from the imports package so embedders can use pulse.ImportSpec instead of imports.Spec.

type ImportResult added in v0.5.0 

type ImportResult = imports.Result

Type aliases re-exported from the imports package so embedders can use pulse.ImportSpec instead of imports.Spec.

type ImportSpec added in v0.5.0 

type ImportSpec = imports.Spec

Type aliases re-exported from the imports package so embedders can use pulse.ImportSpec instead of imports.Spec.

type IndexFreshnessReason added in v0.27.0 

type IndexFreshnessReason = service.IndexFreshnessReason

IndexFreshnessReason re-exports service.IndexFreshnessReason — why VerifyIndex reached its Fresh/stale verdict ("stat_mismatch"/"fingerprint_match"/"fingerprint_mismatch").

type IndexInfo added in v0.27.0 

type IndexInfo = service.IndexInfo

IndexInfo re-exports service.IndexInfo — one entry in a Service.ListIndexes result: a sidecar's derived path, its ordered key column names, and its distinct-key / indexed-record summary.

type LabelBinding added in v0.10.1 

type LabelBinding = types.LabelBinding

LabelBinding pairs a categorical field with a label table for output-time translation. See types.LabelBinding for semantics.

type LabelMatch added in v0.11.2 

type LabelMatch struct {
	Key   string `json:"key"`
	Value string `json:"value"`
	// Score is the match confidence in [0,1]: 1.0 is an exact (or
	// exact-key) hit, ~0.9+ a prefix/near-typo, and lower values are
	// fuzzy matches. Callers should treat a high top score that is
	// clearly ahead of the runner-up as confident, and present options
	// to the user when the best score is low or several are close.
	Score float64 `json:"score"`
}

LabelMatch is one (key, value) pair from a label table, where key is the on-disk categorical dictionary value and value is the display label. ResolveLabel returns these ranked by Score.

type LabelMode added in v0.10.1 

type LabelMode = types.LabelMode

LabelMode selects replace vs augment rendering for a binding.

type LabelTable added in v0.10.1 

type LabelTable struct {
	Description string
	// Rows is the canonical static map. Key is the categorical's
	// resolved dictionary value (the on-disk string the field
	// presents, e.g. "US"); value is the display label (e.g.
	// "United States").
	Rows map[string]string
	// Lookup is the function-driven accessor. Returns the label and
	// true on hit; the second return is the miss signal. An error
	// from Lookup surfaces as PULSE_LABEL_LOOKUP_MISS with the
	// embedder's message attached.
	Lookup func(key string) (label string, ok bool, err error)
}

LabelTable exposes an ID→label map for display-time categorical translation. Exactly one of Rows / Lookup must be non-nil.

Rows is the simple path — a string→string map keyed by the resolved categorical dictionary value. Lookup is the escape hatch for tables that wrap an external store (e.g. a code-system service) or that compose the label from multiple sources.

LabelTable is intentionally distinct from LookupTable: LookupTable returns float64 for the numeric expression environment, LabelTable returns string for output rendering. Conflating them would force every embedder to choose one shape over the other; the two surfaces are independent.

type LabelTableInfo added in v0.11.2 

type LabelTableInfo struct {
	Name     string `json:"name"`
	RowCount int    `json:"row_count"`
	// Enumerable reports whether the table exposes a static Rows map
	// (true) or only a function-driven Lookup closure (false). Reverse
	// search via ResolveLabel works only on enumerable tables.
	Enumerable bool `json:"enumerable"`
}

LabelTableInfo is one registered label table as surfaced by LabelTables. RowCount and Enumerable describe whether the table can be reverse-searched (ResolveLabel requires a static Rows map).

type LoadMemberSetResult added in v0.8.3 

type LoadMemberSetResult = processing.LoadMemberSetResult

LoadMemberSetResult mirrors processing.LoadMemberSetResult so callers can inspect drop counts after loading an include-set file.

func LoadMemberSetFromReader added in v0.8.3 

func LoadMemberSetFromReader(r io.Reader, schema *encoding.Schema, fieldName string) (LoadMemberSetResult, error)

LoadMemberSetFromReader is the public alias for the underlying processing-package loader. It reads newline-delimited values from r and returns the best MemberSet impl for the named field on schema (bitset for categorical, uint64 map for integer / date, string map for decimal / fallback). Float fields are rejected.

type LookupKey added in v0.27.0 

type LookupKey = types.LookupKey

LookupKey re-exports types.LookupKey — one ordered key column/value pair in a composite LookupRequest.Keys tuple, so callers can build a composite key without importing the types package.

type LookupMultiplicity added in v0.27.0 

type LookupMultiplicity = types.LookupMultiplicity

LookupMultiplicity re-exports types.LookupMultiplicity — the duplicate-key handling mode on LookupRequest (assert-unique by default; opt into "first" or "all" for a key that may match more than one row).

type LookupRequest added in v0.27.0 

type LookupRequest = types.LookupRequest

LookupRequest re-exports types.LookupRequest so callers can use pulse.LookupRequest as the input to Lookup.

type LookupResult added in v0.27.0 

type LookupResult = types.LookupResult

LookupResult re-exports types.LookupResult — the response from Lookup.

type LookupTable added in v0.7.0 

type LookupTable struct {
	Description string
	// Rows is the canonical static map. The map key is the joined
	// composite key (caller joins arguments with the table's
	// configured separator, conventionally "|").
	Rows map[string]float64
	// Lookup is the function-driven accessor. Returns the value
	// and true on hit; the second return is the miss signal. An
	// error from Lookup surfaces as PULSE_LOOKUP_MISS with the
	// embedder's message attached.
	Lookup func(keys ...string) (value float64, ok bool, err error)
}

LookupTable exposes a static keyed map to the runtime expression environment. Exactly one of Rows / Lookup must be non-nil.

Rows is the simple path — a composite-key map where the caller of lookup() supplies the joined key. Lookup is the escape hatch for embedders that compose keys, perform partial-match fallback, or pull from an external store.

type MemberSet added in v0.8.3 

type MemberSet = processing.MemberSet

MemberSet is the public alias for processing.MemberSet — the read-only set type consumed by FilterToFileBySetAndExpr. Build one via LoadMemberSetFromReader (recommended for newline-delimited files) or by constructing a concrete impl directly.

type Options 

type Options struct {
	// DataDir is the base directory for cohort files.
	// Defaults to PULSE_DATA_DIR if empty and FS is not set.
	DataDir string

	// FS is an optional custom filesystem.
	// When set, DataDir is ignored for filesystem construction.
	FS afero.Fs

	// DisableDefaults turns off the smart-defaults pass that infers
	// operator Type from the named field's schema type when the caller
	// omits it. Defaults to false (defaults enabled). Predict still
	// computes and reports DefaultsApplied independently — this flag
	// governs only what the runtime mutates on the live request.
	DisableDefaults bool

	// DisableComponents suppresses the Response.Components block emitted
	// by every Process / Compose / ProcessChain / Facet response —
	// per-aggregator, per-grouper, per-filterer, crosstab, and run-wide
	// constituent metadata. Defaults to false (components emitted).
	//
	// When true, Response.Components stays nil and the wire form is
	// byte-identical to the pre-Components baseline; format_version is
	// not bumped. The gate sits at the processor's attach helpers and
	// the per-shard merge path, so the MetaAggregator.Components and
	// MetaGrouper.Components construction work is skipped entirely —
	// not built-then-discarded.
	//
	// Per-request override via Request.DisableComponents (*bool): nil
	// inherits this engine setting; explicit true / false wins.
	DisableComponents bool

	// DisableProjection turns off buffered-decode field projection.
	//
	// Projection is ON by default because it is output-transparent: the
	// runtime walks each request to compute the set of schema fields the
	// operators actually read (processing.NeededFields) and skips map
	// writes for fields outside that set. The result payload is identical
	// — only faster and lighter on per-record memory. Extension operators
	// without a registered FieldInputs hook widen the projection to the
	// full schema, so the default is always safe (worst case degenerates
	// to the full-decode behaviour).
	//
	// Set DisableProjection: true to force full-record decode (every
	// schema field populated in the record map regardless of what the
	// request references). Defaults to false (projection enabled).
	DisableProjection bool

	// ImportsDir overrides the managed-imports directory. Defaults to
	// imports.DefaultImportsDir (resolved relative to the Pulse fs
	// root). Honoured before the PULSE_IMPORTS_DIR env var.
	ImportsDir string

	// ImportTTL overrides the default TTL applied to managed imports
	// when the caller does not pass one. Zero falls back to the
	// PULSE_IMPORT_TTL env var, then to imports.DefaultTTL. Negative
	// values pin imports (never expire) by default.
	ImportTTL time.Duration

	// ImportSourceFS is the afero.Fs used to read source files when
	// ImportFile / pulse_import receives an absolute path. Defaults to
	// afero.NewOsFs() on real OS installs, or to FS when FS is an
	// in-memory MemMapFs (so tests stay hermetic). Setting this
	// disables the default jail: the explicit fs IS the boundary.
	ImportSourceFS afero.Fs

	// ImportSourceJailRoot confines absolute source paths to a
	// directory tree. Empty string defaults to os.Getwd() at New
	// time — the natural sandbox for an MCP server or CLI invocation.
	// Ignored when ImportSourceFS is set explicitly.
	ImportSourceJailRoot string

	// Extensions registers domain-specific operators + expression
	// extensions that the runtime treats identically to built-ins.
	// Zero value disables the extension path entirely. See
	// extensions.go for the full surface.
	Extensions Extensions

	// ShardWorkers caps the per-shard parallel worker pool used when
	// Process operates on a shard archive. Zero means runtime.NumCPU();
	// 1 forces strictly serial execution (same byte-for-byte semantics
	// as the serial path). Negative values are rejected at New() time.
	//
	// The parallel reducer engages only when every operator in the
	// request is mergeable per processing.CanMergeRequest. Non-
	// mergeable requests (percentile aggregators, window operators,
	// tier-2 tests, two-pass attributes combined with groupers, ...)
	// fall through to the serial shardIter path with no worker
	// spawning. Worker count is also capped at the shard count — no
	// point spawning more workers than shards.
	//
	// Order semantics: partials merge in shard insertion order (zip
	// central-directory order). Associative+commutative aggregators
	// (count, sum, min, max, null_count, frequency, distinct_count,
	// mode) produce byte-equal results vs the serial path; Welford
	// mean / variance / stddev drift within ULP on well-conditioned
	// inputs (parallel formula, see processing.MergeOnline docstrings).
	ShardWorkers int

	// DecodeWorkers caps the per-cohort parallel decode worker pool the
	// buffered Process path spawns when a single-file cohort is large
	// enough to benefit from segmenting record-decode across workers.
	// Zero means runtime.NumCPU() at dispatch time when the cohort
	// exceeds service.parallelDecodeRecordThreshold (currently 100_000
	// records); 1 forces strictly serial execution regardless of cohort
	// size. Negative values are rejected at New() time.
	//
	// Below the threshold the buffered Process path stays serial
	// regardless of this knob — worker spawn + merge overhead
	// dominates savings on small inputs. Sharded cohorts continue to
	// route through ShardWorkers; the two knobs are orthogonal (one
	// fans out across shards, the other fans out across record
	// segments within a single file or shard payload).
	DecodeWorkers int

	// Strict promotes request-validation warnings into hard errors at
	// runtime. Today this covers the numeric-aggregation-on-categorical
	// check (PULSE_AGG_NOT_MEANINGFUL_FOR_CATEGORICAL); future runtime
	// validations follow the same flag. Defaults to false — Process
	// runs the request and emits warnings through the descriptor
	// Envelope (visible via --json at the CLI boundary). Predict's
	// PredictOptions.Strict remains independently controllable.
	Strict bool

	// ProjectBufferedFields enables buffered-decode field projection.
	//
	// Deprecated: projection is on by default; retained for compat.
	// Setting true is a harmless no-op; use DisableProjection to opt out.
	ProjectBufferedFields bool

	// LabelTablesDir points at a directory of JSON files that the
	// engine auto-registers as LabelTables at pulse.New time. Empty
	// disables the loader (the only source of LabelTables is then
	// Options.Extensions.LabelTables, set programmatically).
	//
	// File format: each *.json file is a mapping of source value to
	// label string, or a wrapped object {"description": "...",
	// "rows": {"k": "v"}}. The filename without the .json extension
	// becomes the registered table name.
	//
	// Honoured after PULSE_LABEL_TABLES_DIR — programmatic value
	// wins. Tables loaded from disk merge into Options.Extensions.
	// LabelTables; collisions are rejected with
	// PULSE_EXTENSION_DUPLICATE.
	LabelTablesDir string

	// RangeTablesDir points at a directory of JSON files that the
	// engine auto-registers as RangeTables at pulse.New time. Empty
	// disables the loader (the only source of RangeTables is then
	// Options.Extensions.RangeTables, set programmatically).
	//
	// File format: each *.json file is a bare array of range objects
	// [{"label":"Q1","start":"2024-01-01","end":"2024-03-31"}, ...] or
	// a wrapped object {"description": "...", "ranges": [ ... ]}. The
	// filename without the .json extension becomes the registered
	// table name.
	//
	// Honoured after PULSE_RANGE_TABLES_DIR — the programmatic value
	// wins. A table name declared both programmatically and on disk is
	// a hard error at pulse.New; the registered ranges are then
	// validated via the shared range-compilation pass, surfacing the
	// matching PULSE_RANGE_* code on any structural failure.
	RangeTablesDir string

	// TemplateDirs is the ordered list of directory roots the engine
	// scans for request templates at pulse.New time. Empty disables the
	// loader entirely — no store is built, and template lookups answer
	// PULSE_TEMPLATE_NOT_FOUND.
	//
	// File format: every *.json file beneath a root is a template
	// document. A template's name is its path relative to its OWN root,
	// minus the .json extension, forward-slash separated — a file at
	// <root>/finance/revenue.json is named "finance/revenue" — so
	// subdirectories namespace for free and the root's own location never
	// leaks into the name.
	//
	// Roots are a precedence list and the FIRST root wins: the same name
	// under a later root is shadowed, not rejected, which is what lets a
	// site override directory sit ahead of a shipped default. A root that
	// does not exist is skipped (an absent optional layer is not a
	// fault); a root that exists but is not a directory is an error.
	//
	// Honoured before PULSE_TEMPLATES_DIR — the programmatic value wins,
	// and the env var (roots separated by os.PathListSeparator) is
	// consulted only when this slice is empty. The store is built
	// eagerly, so a malformed template fails pulse.New with the offending
	// file named.
	//
	// The roots stay live after startup: a lookup re-walks them when its
	// cached snapshot has aged past the store's rescan interval, so files
	// added, changed, and removed are picked up without restarting the
	// process. ReloadTemplates forces that walk immediately for callers
	// who need determinism rather than eventual visibility.
	//
	// Breakage after startup degrades per file rather than globally: a
	// template whose file becomes malformed keeps serving its last-good
	// parse, and the broken state surfaces on ListTemplates rather than
	// failing the catalog. That split is deliberate — at startup a broken
	// document is a deploy error, and afterwards it is almost always a
	// half-written editor save.
	TemplateDirs []string

	// EchoRequest causes execution paths and descriptor operations to
	// populate descriptor.Envelope.Request with the *normalized* request
	// that was executed — smart defaults resolved, per-stage forms
	// captured for ProcessChain. Off by default to keep wire size
	// unchanged for hot paths and existing callers. The streaming
	// process / compose paths skip the echo unconditionally (NDJSON
	// emit per row, no envelope construction). Predict / inspect /
	// facet / chain / join descriptor endpoints honor the flag via
	// PredictOptions and the equivalent option structs.
	//
	// Intended for automation callers that want to log or replay the
	// exact request the engine ran without re-deriving defaults
	// themselves.
	EchoRequest bool

	// AutoLabels are default LabelBindings the engine injects into every
	// read request (Process / Compose / Facet / Sample) before
	// validation, so registered label tables render display strings
	// without the caller specifying bindings per request. Each binding
	// is applied only when its Field is present and categorical in the
	// target cohort's schema and the caller has not already bound that
	// field — a default that does not fit a given cohort is silently
	// skipped, never an error. Bindings whose Table is not registered are
	// rejected at New time. Empty (the default) disables auto-binding
	// entirely; existing behaviour is unchanged.
	AutoLabels []LabelBinding

	// SetInferenceMinPct configures the delimited-cell heuristic used
	// when an importer must classify a column as set_* vs categorical.
	// Threshold is the minimum percentage of non-null sampled cells
	// that must contain the inferred delimiter for the column to be
	// classified as set_*; combined with two other gates (post-split
	// unique token count ≤ 64 + average post-split cardinality > 1)
	// the threshold trades off false-positive misclassifications
	// against missed multi-select columns. Zero falls back to the
	// package default of 30. Values > 100 clamp to 100. Per-import
	// overrides via the managed-import Spec still take precedence over
	// this default.
	SetInferenceMinPct int
}

Options configures a Pulse instance.

type ParamMeta added in v0.7.0 

type ParamMeta struct {
	Name        string `json:"name"`
	Description string `json:"description,omitempty"`
	JSONType    string `json:"json_type"`
	Required    bool   `json:"required,omitempty"`
	Default     any    `json:"default,omitempty"`
}

ParamMeta describes one operator-specific parameter for manifest emission. It is descriptive only — runtime parameter resolution happens inside each factory.

type Profile added in v0.2.0 

type Profile = synth.Profile

Profile is the cohort statistical summary used by from-profile.

type ProfileOptions added in v0.2.0 

type ProfileOptions = synth.ProfileOptions

ProfileOptions modulate which statistics the profiler captures.

type Pulse 

type Pulse struct {
	// contains filtered or unexported fields
}

Pulse is the top-level library facade. It wraps the service layer and provides a clean API for embedding Pulse into Go programs.

func New 

func New(opts Options) (*Pulse, error)

New creates a new Pulse instance with the given options.

func (*Pulse) AddShard added in v0.8.0 

func (p *Pulse) AddShard(ctx context.Context, archivePath, shardPath string) error

AddShard validates the incoming single-file shard against the archive's canonical schema and appends it. Dict growth that the incoming shard introduces is reflected in the rewritten `_schema.pulse` payload before the new shard payload is appended. v1 reads the whole archive into memory and writes it back via temp+rename — semantically equivalent to true in-place append and crash-safe at the canonical-path level.

func (*Pulse) BuildIndex added in v0.27.0 

func (p *Pulse) BuildIndex(ctx context.Context, path string, keyFields []string) (*BuildIndexResult, error)

BuildIndex builds a point-lookup sidecar index for the cohort at path over the ordered key columns named in keyFields (a single element is the degenerate single-key case; more than one produces a composite key, in column order). Delegates to service.Service.BuildIndex for the full algorithm and error surface — see that method's doc comment for the scan/bucket/write contract, the PULSE_INDEX_UNSUPPORTED_SHARDED shard-archive rejection, and the PROCESSING_CONFIG disallowed-key-type rejection.

func (*Pulse) CompactShardArchive added in v0.8.0 

func (p *Pulse) CompactShardArchive(ctx context.Context, archivePath string) error

CompactShardArchive rewrites the archive to eliminate orphaned bytes from prior in-place mutations and refreshes the canonical metadata (aggregate_record_count + shard_count). v1 AddShard / RemoveShard already use temp+rename (no orphan bytes in v1 archives), so Compact primarily serves to refresh canonical metadata that may have drifted if the archive was edited outside Pulse. The whole-archive rewrite pattern is the explicit reclaim path per the design contract §7.1.

func (*Pulse) Compose 

func (p *Pulse) Compose(ctx context.Context, req *ComposedRequest) (*ComposedResponse, error)

Compose executes multiple requests against a cohort and returns a structured ComposedResponse carrying both per-request results and any composition-level overlay output.

Returned shape:

  • Responses []*Response — one entry per request in req.Requests, in input order. Per-request errors surface as a Response with non-empty Errors; the call returns a top-level error only on orchestration failures (validation, cohort open, etc.).
  • Overlays []OverlayLayer — one layer per OverlaySpec in req.Overlays, in declaration order. The Compose-host overlay fold (service.applyComposeOverlays) writes layers post-execution; when req.Overlays is empty the slot is nil/omitempty.
  • Overlays[i].Warnings []OverlayWarning — per-overlay diagnostics emitted by the handler (cohesion failures, missing host coordinates, threshold breaches). Empty when the layer produced no diagnostics.

func (*Pulse) ComposeParallel added in v0.2.0 

func (p *Pulse) ComposeParallel(ctx context.Context, req *ComposedRequest, opts ComposeOptions) (*ComposedResponse, error)

ComposeParallel runs every request in req concurrently across a bounded worker pool and returns a structured ComposedResponse. Workers share the engine's read-only registries; each Process call constructs fresh stateful operators per request, so concurrent execution is safe.

Defaults: MaxWorkers = runtime.GOMAXPROCS(0), no per-request timeout, FailFast = true (set FailFast=false to collect every request's outcome instead of cancelling siblings on first error).

Returned shape mirrors Compose:

  • Responses []*Response — one entry per request in req.Requests, in input order (not completion order). Workers compose the slice deterministically before returning.
  • Overlays []OverlayLayer — one layer per OverlaySpec in req.Overlays, in declaration order. The Compose-host overlay fold runs after every per-request Process has settled, so overlay handlers always observe the full Responses slice.
  • Overlays[i].Warnings []OverlayWarning — per-overlay diagnostics surfaced by distributeComposeWarnings (cohesion failures, missing host coordinates, panel-target overflow). Empty when the layer produced no diagnostics.

func (*Pulse) Convert 

func (p *Pulse) Convert(ctx context.Context, job *pio.ConvertJob) (*pio.ConvertReport, error)

Convert chains import and export with no intermediate file on disk. The job's FS field is set to the Pulse instance's filesystem if not already set.

Labels apply to the export half only — see Export.

func (*Pulse) CountRecords added in v0.10.0 

func (p *Pulse) CountRecords(ctx context.Context, path string) (uint64, error)

CountRecords returns the number of records in the cohort at path without decoding payload bytes. For single-file cohorts this is O(1) — bytes read is bounded by header + schema size, independent of the record count. For shard archives the cost is O(N shards) (zip central-directory walk + reserved `_schema.pulse` SHRD trailer). Anchor paths (`archive#shard.pulse`) report the named shard's count only.

Use this when a planner needs to make a cardinality-dependent decision (sample injection, smaller-side hash join, batch sizing) without paying the per-row decode cost of a full Process.

func (*Pulse) CreateShardArchive added in v0.8.0 

func (p *Pulse) CreateShardArchive(ctx context.Context, archivePath string, shardPaths []string) error

CreateShardArchive writes a fresh Pulse shard archive at archivePath containing the supplied single-file shardPaths. The first shard seeds the canonical schema; remaining shards are validated via structural cohesion + the append-only dictionary prefix rule. The archive is written atomically (temp file + rename) so partial writes never appear at archivePath. See service.CreateShardArchive for the full error surface.

func (*Pulse) Drop added in v0.5.0 

func (p *Pulse) Drop(ctx context.Context, handle string) error

Drop removes a managed import handle (and its sidecar) from the pool. Returns PULSE_IMPORT_SOURCE_MISSING when the handle is unknown.

func (*Pulse) DropIndex added in v0.27.0 

func (p *Pulse) DropIndex(ctx context.Context, path string, keyFields []string) error

DropIndex removes the sidecar point-lookup index built for keyFields against the cohort at path. Delegates to service.Service.DropIndex — see that method's doc comment for the non-interactive (no-confirmation-prompt) contract. Returns PULSE_INDEX_MISSING when no sidecar exists at the derived path and PULSE_INDEX_UNSUPPORTED_SHARDED for shard archive cohorts.

func (*Pulse) ErrorLookup added in v0.5.0 

func (p *Pulse) ErrorLookup(code string) (ErrorMetadata, bool)

ErrorLookup returns the metadata projection for a single error code. Case-sensitive exact match. Returns (ErrorMetadata{}, false) when the code is unknown.

The manifest carries only the alphabetized code-name list; per-code Message + Fixup detail lives behind this facade so per-session bootstrap stays lean. Use ErrorsByDomain / ErrorsSearch to enumerate in bulk.

func (*Pulse) ErrorsByDomain added in v0.5.0 

func (p *Pulse) ErrorsByDomain(domain string) []ErrorMetadata

ErrorsByDomain returns every code's metadata in the named domain (CLI, DATA, ENCODING, PROCESSING, PULSE, SERVICE). Match is case-insensitive. Returns a non-nil empty slice when nothing matches; results are sorted alphabetically by code.

func (*Pulse) ErrorsSearch added in v0.5.0 

func (p *Pulse) ErrorsSearch(query string) []ErrorMetadata

ErrorsSearch returns codes whose Message or Fixup hints contain the query (case-insensitive substring). Results are ranked by match source: description hits before fixup hits before code-name hits; ties resolve alphabetically. Returns a non-nil empty slice when nothing matches.

func (*Pulse) ExampleGet added in v0.5.0 

func (p *Pulse) ExampleGet(name string) (*Example, bool)

ExampleGet returns the example whose _meta.name matches name. The returned Body is the request JSON with the _meta block stripped so it can be handed directly to Process / Predict.

func (*Pulse) ExamplesSearch added in v0.5.0 

func (p *Pulse) ExamplesSearch(query string, tags []string, category string) []ExampleSummary

ExamplesSearch returns summaries from the embedded request-example library matching the given filters. An empty filter is treated as "no constraint" for that dimension. Query is case-insensitive substring search across name, description, and operators; tags is ANDed; category is an exact match. Always returns a non-nil slice (possibly empty) for safe JSON marshaling.

func (*Pulse) Export 

func (p *Pulse) Export(ctx context.Context, job *pio.ExportJob) (*pio.ExportReport, error)

Export converts a .pulse file into tabular output. The job's FS field is set to the Pulse instance's filesystem if not already set.

When job.Labels is non-empty, the facade builds the runtime label resolver from the Service's registered LabelTables and attaches it to job.LabelResolver before Run. The resolver applies replace / augment translation to categorical column values during export.

func (*Pulse) ExtractShard added in v0.8.0 

func (p *Pulse) ExtractShard(ctx context.Context, archivePath, shardBasename string) (io.ReadCloser, error)

ExtractShard returns an io.ReadCloser over the named shard's standalone single-file `.pulse` bytes. Suitable for piping to `pulse inspect -` or writing back to disk.

func (*Pulse) Facet 

func (p *Pulse) Facet(ctx context.Context, path string, field string) ([]string, error)

Facet returns distinct values for the named field in the cohort. Categorical fields short-circuit through the dictionary; numeric fields stream the file collecting distinct float values. For richer summaries (counts, null tallies, statistics, histograms, additive contributions) call FacetSchema instead.

func (*Pulse) FacetSchema added in v0.7.0 

func (p *Pulse) FacetSchema(ctx context.Context, req *FacetRequest) (*FacetResult, error)

FacetSchema runs a multi-field rich facet against the cohort named in req.Cohort. Returns per-field summaries (discrete value counts or numeric statistics), with optional percentiles, fixed-width histograms, and "additive" contribution counts that report what each distinct value of an additive field would yield if it were added to the base filter.

Streamability: requests with no NumericPercentiles run in a single pass; requests with percentiles buffer the requested numeric fields' non-null values and sort once before percentile interpolation.

func (*Pulse) FilterToFile added in v0.8.1 

func (p *Pulse) FilterToFile(ctx context.Context, src, dst, filterExpr string) (int64, error)

FilterToFile reads the .pulse cohort at src, evaluates filterExpr (FILTER_EXPRESSION semantics — same operators, identifiers, expr functions, and lookup tables available to pulse.Process with a FILTER_EXPRESSION filterer) against every record, and writes a new .pulse cohort at dst containing only matching records.

Dispatch matches the rest of the facade. Single-file inputs produce single-file outputs whose header + schema bytes are copied byte-for- byte from src. Shard archives produce shard archives that preserve the per-shard layout: one input shard maps to one output shard at the same basename, in central-directory (insertion) order, with per-shard header + schema + categorical-dictionary bytes copied verbatim. Empty shards survive so shard_count metadata stays stable; the canonical `_schema.pulse` trailer's aggregate_record_count is refreshed to the surviving total. The anchor form `archive.pulse#shard.pulse` resolves a single shard and writes a single-file output.

Returns the number of records written to dst (sum across shards for archive inputs).

func (*Pulse) FilterToFileBySetAndExpr added in v0.8.3 

func (p *Pulse) FilterToFileBySetAndExpr(ctx context.Context, src, dst, includeField string, set MemberSet, filterExpr string) (int64, error)

FilterToFileBySetAndExpr applies an optional MemberSet membership test (record's value for includeField must be in set) combined with an optional FILTER_EXPRESSION (filterExpr) to every record in src, writing the survivors to dst. At least one of (set, filterExpr) must be supplied; if both are present they are AND-combined and the set is tested first so per-row work short-circuits on misses without paying the expr eval cost.

Input-shape dispatch is identical to FilterToFile (single-file, shard archive, anchor). The set must be built against the same canonical schema as src.

Returns the number of records written to dst (sum across shards for archive inputs).

func (*Pulse) FilterToFileWithRequest added in v0.11.0 

func (p *Pulse) FilterToFileWithRequest(ctx context.Context, req *FilterToFileRequest) (*FilterToFileResult, error)

FilterToFileWithRequest is the deterministic, dedup-aware variant of FilterToFile. Same (source content, predicate) inputs produce the same output file hash and output path; if the expected output already exists at the target path, the engine returns it without re-doing the work.

Determinism comes from three guarantees:

  • The .pulse byte format embeds no timestamps and no compression-level variation, so the underlying FilterToFile engine produces byte-identical output for identical inputs.
  • The output is written via atomic rename: a temp file is staged in OutputDir and renamed to the final path on success. A partially-written .pulse file never appears at the target path.
  • The output name defaults to `{source-hash}_{predicate-hash}.pulse` so independent consumers reach the same expected path without coordination.

func (*Pulse) Fs added in v0.2.0 

func (p *Pulse) Fs() afero.Fs

Fs returns the underlying afero.Fs. Embedders (e.g. the MCP server) need this to enumerate .pulse files; processing methods route through service and never expose the filesystem directly.

func (*Pulse) GetTemplate added in v0.28.0 

func (p *Pulse) GetTemplate(name string) (*template.Template, error)

GetTemplate returns the parsed declaration registered under name — target, description, and the full Variable list — so a caller can build a form (or a prompt) from the declaration alone before rendering.

Lookup is exact and case-sensitive, and the name is the derived one: a template's path relative to its own directory root, minus the .json extension, forward-slash separated. A file at <root>/finance/revenue.json is named "finance/revenue" — not "finance/revenue.json" and not the absolute path.

An unregistered name — including every name on an engine with no template directories configured — is PULSE_TEMPLATE_NOT_FOUND carrying the requested name in its details. A name whose file has broken since it was loaded still resolves, to the last-good parse (see ReloadTemplates); only a name that has NEVER parsed is PULSE_TEMPLATE_INVALID here, naming the path so the operator knows which file to open.

The returned template is the engine's own copy and must be treated as read-only; rendering never mutates it.

func (*Pulse) Import 

func (p *Pulse) Import(ctx context.Context, job *pio.ImportJob) (*pio.ImportReport, error)

Import converts tabular source data into a .pulse file. The job's FS field is set to the Pulse instance's filesystem if not already set.

func (*Pulse) ImportFile added in v0.5.0 

func (p *Pulse) ImportFile(ctx context.Context, spec ImportSpec) (*ImportResult, error)

ImportFile auto-detects the source format, converts the source into a managed .pulse file under the imports pool, and returns the resulting handle. Pulse-native sources pass through unchanged (no copy, no sidecar). Sliding-window TTL applies to managed handles — every subsequent Inspect / Predict / Process / Sample / Facet against the handle bumps the expiry forward.

func (*Pulse) Imports added in v0.5.0 

func (p *Pulse) Imports(ctx context.Context) ([]ImportEntry, error)

Imports returns a snapshot of the managed-imports pool. Sweep is not invoked; expired entries are flagged via Entry.Expired so callers can render them. Results are sorted by handle name.

func (*Pulse) Inspect 

func (p *Pulse) Inspect(ctx context.Context, path string) (*descriptor.InspectResult, error)

Inspect reads a .pulse file header and schema, returning structured field information. It never reads record data.

func (*Pulse) LabelTables added in v0.11.2 

func (p *Pulse) LabelTables() []LabelTableInfo

LabelTables returns the registered label tables in name order. It reports each table's row count and whether it is reverse-searchable. Returns nil when no label tables are registered.

func (*Pulse) ListIndexes added in v0.27.0 

func (p *Pulse) ListIndexes(ctx context.Context, path string) ([]IndexInfo, error)

ListIndexes enumerates every sidecar point-lookup index built against the cohort at path. Delegates to service.Service.ListIndexes — see that method's doc comment for the directory-glob + sidecar-read discovery algorithm. Returns an empty (non-nil) slice, not an error, when no sidecar indexes have been built yet. Returns PULSE_INDEX_UNSUPPORTED_SHARDED for shard archive cohorts.

func (*Pulse) ListShards added in v0.8.0 

func (p *Pulse) ListShards(ctx context.Context, archivePath string) ([]ShardEntry, error)

ListShards returns the archive's shard manifest in central- directory order (which equals shard insertion order). Single-file cohorts return an empty slice.

func (*Pulse) ListTemplates added in v0.28.0 

func (p *Pulse) ListTemplates() []template.Summary

ListTemplates returns one summary per registered request template, sorted by name so the order is deterministic across runs and platforms.

Each Summary projects everything a caller needs to CHOOSE a template and build a form for it — name, description, target, declared variable names, and the source file it was loaded from — without carrying the body. The full declarations (types, defaults, enum values) come from GetTemplate.

Shadows names the source paths of same-named templates the listed entry takes precedence over. Template directories are an ordered precedence list and the first root wins; the losing entries are reported here rather than discarded, which is what makes "why is my override not taking effect?" answerable from the listing alone. A shadowed entry deliberately gets no summary of its own — it is not renderable, and a listing whose entries cannot all be fetched would be a trap.

Broken is how a post-startup breakage becomes visible. A template file that has stopped parsing since it was loaded keeps ANSWERING GetTemplate with its last-good copy, so nothing about fetching it would reveal the fault; the flag, with the fault text in Error, is what lets an operator find the bad file without rendering all fifty templates one at a time. A broken entry with an empty Target never parsed at all — the file was already malformed the first time the engine saw it — so it is listed to be SEEN rather than fetched, and asking for it by name returns PULSE_TEMPLATE_INVALID.

Always returns a non-nil slice (possibly empty) for safe JSON marshaling. An engine with no template directories configured lists nothing; that is an ordinary deployment, not a fault.

func (*Pulse) Lookup added in v0.27.0 

func (p *Pulse) Lookup(ctx context.Context, req *LookupRequest) (*LookupResult, error)

Lookup resolves a point lookup (single-key or composite, via req.Keys) against the cohort named in req.Cohort, using the prebuilt sidecar index (see BuildIndex). Returns PULSE_INDEX_MISSING when no sidecar index exists for the requested key fields, PULSE_LOOKUP_NOT_FOUND when the index exists but no record matches, or PULSE_LOOKUP_AMBIGUOUS when more than one record matches and req.Multiplicity is (or defaults to) LookupMultiplicityAssertUnique. See service.Service.Lookup for the full algorithm.

func (*Pulse) Manifest added in v0.5.0 

func (p *Pulse) Manifest(_ context.Context) *descriptor.Manifest

Manifest returns the root Pulse self-description. The manifest is deterministic and process-wide: it does not depend on cohort data or the filesystem. Callers cache the result for a session.

func (*Pulse) Open 

func (p *Pulse) Open(ctx context.Context, path string) (*Cohort, error)

Open reads a .pulse file and returns a Cohort with the parsed schema.

Anchor syntax: a path of the form "archive.pulse#shard.pulse" opens the archive, locates the named shard inside it, and returns a single- shard cohort whose schema comes from the shard's own header (not the canonical schema in `_schema.pulse`). The returned Cohort has an empty Shards slice — anchor-resolved shards stand alone for the purposes of facade methods. Anchors require an archive backing the path; using `#` against a single-file `.pulse` raises PULSE_ARCHIVE_MAGIC_INVALID. A literal `#` in a filename is not supported in v1.

Anchor parsing happens inside service.Service.Open as well, so the other facade methods (Process, Sample, Facet, ...) that receive an anchored Cohort path resolve consistently.

func (*Pulse) Predict 

func (p *Pulse) Predict(ctx context.Context, req *Request) (*descriptor.PredictResult, error)

Predict validates a request against a .pulse file without executing it. It reads only the header and schema, never record data.

func (*Pulse) Process 

func (p *Pulse) Process(ctx context.Context, req *Request) (*Response, error)

Process executes a single processing request against a cohort.

func (*Pulse) ProcessChain added in v0.10.0 

func (p *Pulse) ProcessChain(ctx context.Context, req *ChainRequest) (*ChainResponse, error)

ProcessChain executes a source-rooted linear chain of Process requests. The first stage runs against the cohort identified by req.Cohort; each subsequent stage receives the previous stage's rows as its input. All stages must be mergeable per the v1 chain gate (processing.CanChainRequest); a non-mergeable stage surfaces PULSE_CHAIN_NOT_MERGEABLE with the offending stage index so the caller can fall back to per-stage Process calls.

Why chain? Building a linear plan as N independent Process calls pays N file-open + schema-rebind + per-stage validate costs. ProcessChain collapses those into one open + one synth-schema per intermediate stage, keeping the streaming iterator stack alive across stages.

func (*Pulse) ProcessStream added in v0.2.0 

func (p *Pulse) ProcessStream(ctx context.Context, req *Request) (RowIter, error)

ProcessStream executes a request and returns a pull-based row iterator over the result. Equivalent to Process for any request shape — same gates, same errors — but streaming consumers (HTTP responders, NDJSON writers, downstream pipelines) can drain rows one at a time without buffering the full result in their own memory.

Predict's Streamable flag reports whether the underlying execution avoids buffering inside the engine; ProcessStream wraps the result regardless, so the API is stable for non-streamable requests too.

func (*Pulse) ProcessStreamResult added in v0.11.0 

func (p *Pulse) ProcessStreamResult(ctx context.Context, req *Request) (StreamResult[Row], error)

ProcessStreamResult executes req and returns a StreamResult that yields each result row as a StreamChunk[Row]. The header is built before any rows are read; the chunks channel closes when the operation completes, and the Done channel delivers exactly one StreamTerminator describing the terminal state.

Cancellation: closing ctx delivers a StreamTerminator with Status: StreamCancelled and a non-nil Error matching ctx.Err().

func (*Pulse) Profile added in v0.2.0 

func (p *Pulse) Profile(_ context.Context, path string, opts ProfileOptions) (*Profile, error)

Profile reads a .pulse file at path and returns a statistical summary suitable for from-profile synthesis. The profile retains no individual rows from the source data.

func (*Pulse) RangeTables added in v0.27.0 

func (p *Pulse) RangeTables() []RangeTableInfo

RangeTables returns the registered range tables in name order. Each entry carries the table's range count and its ordered {label, start, end} ranges — the INPUT-direction discovery surface a caller consults before authoring a GROUP_DATE_RANGES / FILTER_DATE_RANGES request that references the table by name. Returns nil when none are registered.

func (*Pulse) ReloadTemplates added in v0.28.0 

func (p *Pulse) ReloadTemplates() error

ReloadTemplates rescans every configured template directory immediately and swaps in the result.

It is an escape hatch, not the mechanism. Templates hot-reload on their own: a lookup whose cached snapshot has aged past the store's rescan interval re-walks the directories first, so a file dropped into a scanned directory becomes renderable, a changed file starts serving its new content, and a deleted one stops resolving — all without restarting the process. The interval is a package constant rather than an Option, because a dial nobody can set better than the store can is a permanent public surface bought for nothing.

What the interval cannot give is determinism. A newly written file can be invisible for up to that interval, which is fine for an operator editing a directory and wrong for a deployment step that writes a template and must render it on the next line, or for a test that would otherwise have to sleep. ReloadTemplates covers exactly that case.

A rescan is a directory walk plus one stat per candidate file; a file whose size and modification time both match the copy already parsed is carried over rather than re-read, so calling this on an unchanged directory costs syscalls and no JSON parsing.

A file going bad after startup degrades PER FILE and does not come back from this call. A template that parsed once and whose file later becomes malformed keeps serving its last-good parse, every other template is untouched, and this returns nil: an error here would tell a caller its whole catalog failed over one half-written editor save. The broken state is observable through ListTemplates instead — Summary.Broken with the fault in Summary.Error — and through GetTemplate for a name that never parsed at all. Repairing the file clears it on the next rescan. An unreadable file degrades the same way; it is also one file.

What this DOES return is a whole-walk fault: a configured root that exists but is not a directory, or a directory that cannot be walked. Those are misconfigurations rather than transient edits, and a failed walk leaves the previously loaded templates entirely in place.

Startup keeps the opposite rule: pulse.New still fails outright on a malformed template, because at startup a broken document is a deploy error the operator should see immediately rather than a keystroke.

An engine with no template directories configured has nothing to rescan and returns nil.

func (*Pulse) RemoveShard added in v0.8.0 

func (p *Pulse) RemoveShard(ctx context.Context, archivePath, shardBasename string) error

RemoveShard rewrites the archive omitting the named shard. The canonical schema is preserved (dictionary entries are never shrunk). Returns PULSE_SHARD_MISSING when the named shard is not in the archive.

func (*Pulse) RenderTemplate added in v0.28.0 

func (p *Pulse) RenderTemplate(name string, vars map[string]any) (*template.Rendered, error)

RenderTemplate resolves the named template and renders it against the supplied variable map, returning the substituted JSON plus the typed request it decoded into. It is the general form, covering all five targets; RenderTemplateRequest is the shorthand for the common one.

Exactly one of Rendered's typed pointers is populated, selected by Rendered.Target — read that pointer (or Rendered.Typed()) and hand it to the matching execution method: Process, Compose, ProcessChain, FacetSchema, SampleWithRequest. There are deliberately no per-execution- mode convenience wrappers: N execution modes would mean N wrappers to keep in sync forever, for no capability gain.

Rendered.JSON is the rendered body before decode, retained because re-marshaling the typed value would not reproduce it — every request struct is dense with omitempty, so a slot that rendered to an explicit zero would silently vanish from a round trip.

Errors are the PULSE_TEMPLATE_* family, surfaced with their codes and details intact: PULSE_TEMPLATE_NOT_FOUND for an unknown name, then whatever the render raises — PULSE_TEMPLATE_VAR_MISSING / _VAR_UNKNOWN / _VAR_TYPE / _VAR_ENUM for the variable map, PULSE_TEMPLATE_UNRESOLVED for a marker with nothing to substitute, and PULSE_TEMPLATE_RENDER_INVALID when the substituted JSON does not fit the target request type.

Rendering never opens a cohort: a template that renders is well-formed against the request SHAPE. Whether it is executable against a particular cohort stays Predict's question.

func (*Pulse) RenderTemplateRequest added in v0.28.0 

func (p *Pulse) RenderTemplateRequest(name string, vars map[string]any) (*Request, error)

RenderTemplateRequest renders the named template and returns the typed *Request directly. It is the 95% path — the caller hands the result straight to Process, Predict, or ProcessStream.

It is RenderTemplate restricted to the "request" target. A template declaring any other target is PULSE_TEMPLATE_TARGET_UNKNOWN: the target is a valid one, just not one this method can return, so the message names both the target the template actually declares and RenderTemplate as the method that handles it. Every other fault is RenderTemplate's, unchanged.

func (*Pulse) ResolveCanonicalSchema added in v0.8.3 

func (p *Pulse) ResolveCanonicalSchema(ctx context.Context, src string) (*encoding.Schema, error)

ResolveCanonicalSchema returns the canonical encoding.Schema for the cohort at src without streaming records. Resolves single-file, shard-archive, and `archive#shard` anchor inputs identically to the rest of the facade.

Useful for callers that need to build an include-set before invoking FilterToFileBySetAndExpr: the set loader requires the schema to pick the best MemberSet impl (bitset / uint64 / string) for the field's type.

func (*Pulse) ResolveImport added in v0.5.0 

func (p *Pulse) ResolveImport(ctx context.Context, handle string) (string, error)

ResolveImport returns the managed-pool path for a handle, or PULSE_IMPORT_SOURCE_MISSING when no such handle exists. Embedders who want to address a managed handle by name (instead of by path) run path through this resolver before passing it to Inspect/Process.

func (*Pulse) ResolveLabel added in v0.11.2 

func (p *Pulse) ResolveLabel(table, query string, limit int) ([]LabelMatch, error)

ResolveLabel reverse-searches a registered label table: given a human-readable query — including a minor misspelling — it returns up to limit matching (key, value) pairs ranked by a confidence Score in [0,1]. Matching is typo-tolerant: the query and each label are normalized (lowercased, punctuation folded to spaces, whitespace collapsed), and the score is the best of an exact/prefix/substring match, a Damerau edit-distance ratio (catches typos and adjacent transpositions), and a character-trigram Dice coefficient (catches word reordering and partial overlap). An exact key match also scores 1.0.

Candidates scoring at or above an internal floor are returned best first. When nothing clears the floor the closest few are returned anyway (with their low scores) so the caller can present "did you mean" options rather than a dead end. An empty query returns the first limit rows in label order (a cheap "browse" mode, Score 0). limit <= 0 defaults to 10.

Returns PULSE_LABEL_TABLE_UNKNOWN when the table is not registered and PULSE_LABEL_TABLE_NOT_ENUMERABLE when the table exposes only a Lookup closure (no static Rows to scan).

func (*Pulse) Sample 

func (p *Pulse) Sample(ctx context.Context, path string, n int) ([]Record, error)

Sample returns up to n rows from the cohort as maps of field name to value.

func (*Pulse) SampleWithRequest added in v0.10.1 

func (p *Pulse) SampleWithRequest(ctx context.Context, req *SampleRequest) (*SampleResult, error)

SampleWithRequest is the labelled variant of Sample. The req struct carries the cohort path (via Cohort.Filename), the row cap (N), and LabelBinding entries that translate categorical values to display labels in the returned rows.

When Labels is empty the call degenerates to the same shape as Sample, returning a SampleResult with no Warnings and no transformation applied to the rows.

func (*Pulse) Service added in v0.7.0 

func (p *Pulse) Service() *service.Service

Service returns the underlying service handle. Exposed so tests (and advanced embedders) can inspect the installed extension registry, FS configuration, and orchestration state.

func (*Pulse) SweepImports added in v0.5.0 

func (p *Pulse) SweepImports(ctx context.Context) ([]string, error)

SweepImports removes every expired managed handle and returns the list of swept handle names. Invoked opportunistically by ImportFile and exposed here for callers that want explicit control (CLI maintenance, periodic ticker, etc.).

func (*Pulse) Synth added in v0.2.0 

func (p *Pulse) Synth(_ context.Context, spec *SynthSpec, output string, opts SynthOptions) (*SynthResult, error)

Synth materializes a synthetic .pulse file at output from spec. The generator is deterministic for a given (spec, opts.Seed) pair: same seed produces a byte-identical file.

func (*Pulse) SynthStream added in v0.11.0 

func (p *Pulse) SynthStream(ctx context.Context, spec *SynthSpec, opts SynthOptions) (StreamResult[Row], error)

SynthStream generates synth respondents as a StreamResult. Chunks carry one map[string]any per generated row, the same shape returned by the buffered Synth path. EstimatedTotal mirrors spec.RowCount.

The current implementation drives the buffered synth pipeline behind the scenes and yields rows from the materialized result; it does not expose true row-at-a-time generation yet but is API-stable so callers can adopt the streaming shape today.

func (*Pulse) VerifyIndex added in v0.27.0 

func (p *Pulse) VerifyIndex(ctx context.Context, path string, keyFields []string) (*VerifyIndexResult, error)

VerifyIndex reports whether the sidecar point-lookup index built for keyFields against the cohort at path is still fresh, using the size+mtime fast-path before paying for a full content-hash recompute. Delegates to service.Service.VerifyIndex — see that method's doc comment for the full fast-path decision tree. Returns PULSE_INDEX_MISSING when no sidecar exists for keyFields and PULSE_INDEX_UNSUPPORTED_SHARDED for shard archive cohorts.

func (*Pulse) VerifyShardArchive added in v0.8.0 

func (p *Pulse) VerifyShardArchive(ctx context.Context, archivePath string) (*VerifyResult, error)

VerifyShardArchive opens the archive and re-validates every shard's header (magic + format_version), structural cohesion against the canonical schema, dictionary prefix rule, and cross-checks each shard's record count against the canonical aggregate. Returns a VerifyResult carrying any errors (PULSE_SHARD_HEADER_INVALID, PULSE_SHARD_SCHEMA_MISMATCH, PULSE_SHARD_DICT_DIVERGENCE) and any non-fatal warnings (PULSE_SHARD_DESCRIPTION_DIVERGENCE, aggregate drift). Returns a non-nil error only when the archive itself cannot be opened (archive corrupt, file missing, etc.); per-shard issues are reported through the result struct so the caller can render the full diagnosis.

func (*Pulse) Watch added in v0.11.0 

func (p *Pulse) Watch(ctx context.Context, target string) <-chan ChangeEvent

Watch observes the cohort path on the Pulse instance's filesystem and returns a channel of ChangeEvent records. Closing the context drains and closes the channel.

func (*Pulse) WatchDir added in v0.11.0 

func (p *Pulse) WatchDir(ctx context.Context, dir string, recursive bool) <-chan ChangeEvent

WatchDir observes every file under dir (and subdirectories when recursive=true) for change events, filtered to WatchOptions.Suffix (defaults to ".pulse"). Closing the context drains and closes the channel.

func (*Pulse) WatchDirWithOptions added in v0.11.0 

func (p *Pulse) WatchDirWithOptions(ctx context.Context, dir string, opts WatchOptions) <-chan ChangeEvent

WatchDirWithOptions is the configurable variant of WatchDir.

func (*Pulse) WatchWithOptions added in v0.11.0 

func (p *Pulse) WatchWithOptions(ctx context.Context, target string, opts WatchOptions) <-chan ChangeEvent

WatchWithOptions is the configurable variant of Watch. PollInterval and CoalesceWindow control responsiveness vs CPU; HashPrefixBytes caps the per-event hash size.

type RangeTable added in v0.27.0 

type RangeTable struct {
	// Description is an optional human-readable summary surfaced in the
	// manifest projection.
	Description string
	// Ranges is the ordered list of labeled date ranges. Must be
	// non-empty and pass the shared validation pass.
	Ranges []DateRangeSpec
}

RangeTable is a named, ordered set of labeled date ranges. It is the reusable, register-once form of the inline range list authored on a GROUP_DATE_RANGES grouper or FILTER_DATE_RANGES filter: a caller registers the table under a name and references it by that name from any number of requests.

Ranges are the ordered {label, start, end} entries. Start / End are ISO date literals (nil / empty means an open bound; both bounds are inclusive). The set is validated at pulse.New time via the shared range-compilation pass — overlap, duplicate label, empty set, and unparseable / inverted boundaries each surface the matching PULSE_RANGE_* code. Order is preserved as authored; the compiler sorts internally for deterministic matching.

type RangeTableInfo added in v0.27.0 

type RangeTableInfo struct {
	Name       string          `json:"name"`
	RangeCount int             `json:"range_count"`
	Ranges     []DateRangeSpec `json:"ranges"`
}

RangeTableInfo is one registered range table as surfaced by RangeTables — the discovery companion for the GROUP_DATE_RANGES / FILTER_DATE_RANGES operators. It carries the table name, its labeled- range count, and the ordered ranges themselves so a caller can turn a table name into the {label, start, end} tuples an operator resolves.

type Record 

type Record = map[string]any

Record is a row of field→value data returned by Sample.

type Request 

type Request = types.Request

Type aliases re-exported from the types package so embedders can use pulse.Request instead of types.Request.

type Response 

type Response = types.Response

Type aliases re-exported from the types package so embedders can use pulse.Request instead of types.Request.

type Row added in v0.2.0 

type Row = service.Row

Row is a single result row in a processing stream.

type RowIter added in v0.2.0 

type RowIter = service.RowIter

RowIter is a pull-based iterator over a processing result. Each call to Next returns the next row or (nil, false, nil) on exhaustion. Close releases underlying resources. Metadata returns the run metadata once available (always present after the iterator is drained).

type SampleRequest added in v0.10.1 

type SampleRequest = types.SampleRequest

SampleRequest is the labelled variant of the Sample entry point.

type SampleResult added in v0.10.1 

type SampleResult struct {
	Rows     []Record
	Warnings []SampleWarning
}

SampleResult bundles the rows and the resolver-side warnings from a labelled Sample call. Warnings carry PULSE_LABEL_COLLISION and PULSE_LABEL_LOOKUP_MISS records; an empty slice means the labels resolved cleanly (or no Labels were requested).

type SampleWarning added in v0.10.1 

type SampleWarning struct {
	Code    string
	Message string
	Details map[string]any
}

SampleWarning is the envelope-ready projection of a single resolver warning. The shape mirrors descriptor.EnvelopeWarning so callers can fold it into a descriptor.Envelope at the CLI / MCP boundary.

type ShardEntry added in v0.8.0 

type ShardEntry = service.ShardEntry

ShardEntry is one shard inside a Pulse shard archive. Re-exported from service so embedders can address pulse.ShardEntry directly.

type ShardInfo added in v0.8.0 

type ShardInfo = descriptor.ShardInfo

ShardInfo is one shard entry as surfaced by Inspect / Predict. Re- exported from descriptor so embedders consuming the no-execute surface can address pulse.ShardInfo directly. Mirrors ShardEntry's shape (filename + record count); the two types are parallel because descriptor/ cannot import service/.

type StreamChunk added in v0.11.0 

type StreamChunk[T any] struct {
	Sequence   int
	Data       T
	Progress   float64
	Components *types.ResponseComponents `json:"components,omitempty"`
}

StreamChunk wraps a single incremental payload. Sequence is a monotonic 0-based counter so consumers can detect gaps or reorder out-of-band telemetry; Progress reports a best-effort fraction in [0.0, 1.0] (or -1.0 when unknown).

Components, when populated, carries the per-operator constituent-parts metadata at chunk boundary — same shape as Response.Components on a buffered Process run. Mergeable / partial aggregators surface their running state on every chunk so consumers can render mid-stream; non-mergeable operators (AGG_MEDIAN, AGG_PERCENTILE) appear only on the terminal chunk (last row before close). nil when the underlying operation produced no Components payload (e.g. SynthStream).

type StreamHeader added in v0.11.0 

type StreamHeader struct {
	// RequestHash is the canonical hash of the originating request,
	// as returned by Request.Hash / SynthSpec.Hash. Identifies the
	// stream's input for caching and deduplication.
	RequestHash string

	// EstimatedTotal is the best-effort row-count estimate for the
	// stream. -1 when unknown (predict not available, source is a
	// live cohort whose size is dynamic, etc.).
	EstimatedTotal int64

	// StartedAt is the wall-clock time the stream began producing.
	StartedAt time.Time
}

StreamHeader is the single-shot prelude describing what the stream is about to produce. Sent before any chunks.

type StreamResult added in v0.11.0 

type StreamResult[T any] struct {
	Header StreamHeader
	Chunks <-chan StreamChunk[T]
	Done   <-chan StreamTerminator
}

StreamResult is the canonical streaming-output shape. Every operation that produces incremental data exposes a *Stream variant returning StreamResult[T]. The receiver pattern: 

res, err := p.ProcessStreamResult(ctx, req)
if err != nil { ... }
for chunk := range res.Chunks {
    ... handle chunk.Data
}
term := <-res.Done
if term.Status != pulse.StreamCompleted { ... }

Backpressure: the Chunks channel carries a small buffer (4). Slow consumers slow the producer; the producer never drops chunks.

type StreamStatus added in v0.11.0 

type StreamStatus int

StreamStatus reports the terminal state of a StreamResult. 

const (
	// StreamCompleted means the operation produced all chunks and
	// finished without error.
	StreamCompleted StreamStatus = iota
	// StreamCancelled means the caller's context was cancelled or
	// timed out before the operation finished.
	StreamCancelled
	// StreamErrored means the operation aborted mid-stream. The
	// terminator's Error field carries the underlying cause.
	StreamErrored
)

func (StreamStatus) String added in v0.11.0 

func (s StreamStatus) String() string

String returns a stable lowercase identifier for the status, suitable for logging and JSON marshalling.

type StreamTerminator added in v0.11.0 

type StreamTerminator struct {
	CompletedAt time.Time
	TotalRows   int64
	Status      StreamStatus
	Error       error
}

StreamTerminator is the single-shot epilogue describing how the stream ended. Sent after the chunks channel closes; the Done channel itself is closed immediately after delivering this value so a receiver that selects on Done sees one value and then channel-closed.

type SynthOptions added in v0.2.0 

type SynthOptions = synth.Options

SynthOptions modulate the deterministic seed and other knobs.

type SynthResult added in v0.2.0 

type SynthResult = synth.Result

SynthResult is the result of a successful Synth call.

type SynthSpec added in v0.2.0 

type SynthSpec = synth.Spec

SynthSpec is the parsed synthesis request shape.

type TestRegistration added in v0.7.0 

type TestRegistration struct {
	Name        types.TestType
	Description string
	Tier        TestTier
	RowFactory  processing.RowTestFactory
	PostFactory processing.PostTestFactory
	Streamable  bool
	Accepts     []encoding.FieldType
	Params      []ParamMeta
	// FieldInputs is the optional buffered-projection introspection
	// hook for tier-1 row tests. Tier-2 post-tests run on materialized
	// result rows rather than source records, so projection doesn't
	// apply — leave nil for tier-2 registrations.
	FieldInputs FieldInputsFunc
}

TestRegistration installs a custom TEST_* operator. Exactly one of RowFactory / PostFactory must be non-nil, matching Tier.

Streamable applies to tier-1 only and indicates whether the test can co-stream with online aggregators (no extra pass over the data).

type TestTier added in v0.7.0 

type TestTier string

TestTier selects which factory shape a TestRegistration provides. Tier-1 row tests fold per-row during the streaming aggregation pass; tier-2 post-tests consume the materialized result row set after windows. 

const (
	TestTierRow  TestTier = "tier1"
	TestTierPost TestTier = "tier2"
)

type VerifyIndexResult added in v0.27.0 

type VerifyIndexResult = service.VerifyIndexResult

VerifyIndexResult re-exports service.VerifyIndexResult — the outcome of a Service.VerifyIndex freshness check.

type VerifyResult added in v0.8.0 

type VerifyResult = service.VerifyResult

VerifyResult carries the structured outcome of VerifyShardArchive. Errors aggregate every fatal cohesion failure discovered while walking the archive's shards; Warnings carry non-fatal divergences (per-field description drift, aggregate-record-count mismatch). An empty Errors slice means the archive is structurally sound.

type WatchOptions added in v0.11.0 

type WatchOptions struct {
	// PollInterval is the stat-poll cadence. Defaults to 250ms.
	// Network filesystems should pass a larger interval (recommend
	// 30s).
	PollInterval time.Duration

	// CoalesceWindow caps how long the watcher waits for a quick
	// follow-up modify before emitting a single event. Defaults to
	// 100ms; pass 0 to disable coalescing entirely.
	CoalesceWindow time.Duration

	// HashPrefixBytes caps how many leading bytes the hash digest
	// reads from each file. Defaults to 64 KiB — enough to cover
	// the .pulse header + schema for any realistic cohort. Pass a
	// negative value to hash the entire file.
	HashPrefixBytes int

	// Recursive controls whether WatchDir descends into
	// subdirectories. Ignored by Watch.
	Recursive bool

	// Suffix narrows the watched fileset by suffix (case-sensitive).
	// Empty matches every file. WatchDir defaults this to ".pulse"
	// to keep traffic focused on cohort files; callers can override
	// with an empty string to watch every file.
	Suffix string
}

WatchOptions tunes the watcher cadence. Zero values fall back to the defaults documented on each field.

type WindowRegistration added in v0.7.0 

type WindowRegistration struct {
	Name        types.WindowType
	Description string
	Factory     window.WindowFactory
	Accepts     []encoding.FieldType
	Params      []ParamMeta
	// FieldInputs is the optional buffered-projection introspection
	// hook. See FieldInputsFunc.
	FieldInputs FieldInputsFunc
}

WindowRegistration installs a custom WIN_* operator. Window operators run buffered today; the runtime records custom windows as non-streamable. A streaming-window shape lands when the runtime gains a streaming-window pipeline.

Source Files

View all Source files

Directories

Path Synopsis
cmd
pulse command
Package main is the entry point for the pulse CLI binary.
 Package main is the entry point for the pulse CLI binary.
Package descriptor provides self-description, manifest, and predict functionality for pulse.
 Package descriptor provides self-description, manifest, and predict functionality for pulse.
Package encoding handles the binary .pulse file format: reading, writing, and schema management.
 Package encoding handles the binary .pulse file format: reading, writing, and schema management.
Package errors provides structured error codes and error handling for pulse.
 Package errors provides structured error codes and error handling for pulse.
Package examples embeds the catalogue of runnable Pulse request JSON files and exposes a search/get API over them.
 Package examples embeds the catalogue of runnable Pulse request JSON files and exposes a search/get API over them.
Package fs provides the filesystem abstraction layer for pulse storage backends.
 Package fs provides the filesystem abstraction layer for pulse storage backends.
Package imports manages tabular-source imports with a TTL-tracked on-disk pool.
 Package imports manages tabular-source imports with a TTL-tracked on-disk pool.
internal
cli
Package cli provides internal CLI wiring and command construction for the pulse binary.
 Package cli provides internal CLI wiring and command construction for the pulse binary.
io
Package io defines the I/O pipeline framework for Pulse: Reader/Writer interfaces, schema inference, and job types (ImportJob, ExportJob, ConvertJob).
 Package io defines the I/O pipeline framework for Pulse: Reader/Writer interfaces, schema inference, and job types (ImportJob, ExportJob, ConvertJob).
arrow
Package arrow provides Arrow IPC (Feather V2) import and export for the pulse I/O pipeline, plus shared Arrow<->Pulse type-mapping helpers used by both this package and io/parquet.
 Package arrow provides Arrow IPC (Feather V2) import and export for the pulse I/O pipeline, plus shared Arrow<->Pulse type-mapping helpers used by both this package and io/parquet.
csv
Package csv provides CSV format adapters for the Pulse I/O pipeline.
 Package csv provides CSV format adapters for the Pulse I/O pipeline.
excel
Package excel provides Excel import and export for the pulse I/O pipeline.
 Package excel provides Excel import and export for the pulse I/O pipeline.
exportoverlay
Package exportoverlay holds the higher-level integration tests for the ExportJob and ConvertJob overlay-embedding wiring.
 Package exportoverlay holds the higher-level integration tests for the ExportJob and ConvertJob overlay-embedding wiring.
format
Package format dispatches tabular Reader construction by format identifier, sitting between the io/ interface definitions and the per-format leaf packages (io/csv, io/tsv, io/ndjson, io/jsonarray, io/parquet, io/arrow, io/excel).
 Package format dispatches tabular Reader construction by format identifier, sitting between the io/ interface definitions and the per-format leaf packages (io/csv, io/tsv, io/ndjson, io/jsonarray, io/parquet, io/arrow, io/excel).
jsonarray
Package jsonarray provides JSON-array import and export for the pulse I/O pipeline.
 Package jsonarray provides JSON-array import and export for the pulse I/O pipeline.
jsonshared
Package jsonshared holds value coercion helpers shared by the ndjson and jsonarray packages.
 Package jsonshared holds value coercion helpers shared by the ndjson and jsonarray packages.
ndjson
Package ndjson provides NDJSON (newline-delimited JSON) import and export for the pulse I/O pipeline.
 Package ndjson provides NDJSON (newline-delimited JSON) import and export for the pulse I/O pipeline.
parquet
Package parquet provides Parquet import and export for the pulse I/O pipeline.
 Package parquet provides Parquet import and export for the pulse I/O pipeline.
tsv
Package tsv provides TSV import and export for the pulse I/O pipeline.
 Package tsv provides TSV import and export for the pulse I/O pipeline.
mcp
Package mcp is the SDK-free core of the Pulse MCP surface.
 Package mcp is the SDK-free core of the Pulse MCP surface.
gosdk
Package gosdk is the thin, reusable adapter that mounts the SDK-free Pulse MCP catalog (github.com/frankbardon/pulse/mcp) onto a caller-supplied github.com/modelcontextprotocol/go-sdk server.
 Package gosdk is the thin, reusable adapter that mounts the SDK-free Pulse MCP catalog (github.com/frankbardon/pulse/mcp) onto a caller-supplied github.com/modelcontextprotocol/go-sdk server.
toolmeta
Package toolmeta holds the metadata table for the MCP tools registered by the Pulse MCP layer.
 Package toolmeta holds the metadata table for the MCP tools registered by the Pulse MCP layer.
Package mcpserve exposes Pulse's Model Context Protocol server as a public entry point.
 Package mcpserve exposes Pulse's Model Context Protocol server as a public entry point.
Package processing provides the single dynamic processing engine for Pulse.
 Package processing provides the single dynamic processing engine for Pulse.
arena
Package arena provides a bump-allocator backed by a single contiguous []byte.
 Package arena provides a bump-allocator backed by a single contiguous []byte.
feature
Package feature implements the FEAT_* operators that run pre-filter to add derived columns to a record stream.
 Package feature implements the FEAT_* operators that run pre-filter to add derived columns to a record stream.
regression
Package regression hosts the REG_* operators that fit a regression model against the filtered record set.
 Package regression hosts the REG_* operators that fit a regression model against the filtered record set.
window
Package window implements the WIN_* window operators for Pulse.
 Package window implements the WIN_* window operators for Pulse.
Package service provides the orchestration layer for pulse operations.
 Package service provides the orchestration layer for pulse operations.
Package skills provides the embedded skill pack for LLM-driven agents.
 Package skills provides the embedded skill pack for LLM-driven agents.
Package synth produces synthetic .pulse cohorts from either a schema declaration ("from-schema") or a statistical profile of a real cohort ("from-profile").
 Package synth produces synthetic .pulse cohorts from either a schema declaration ("from-schema") or a statistical profile of a real cohort ("from-profile").
Package template turns a stored, parameterised JSON document into a validated Pulse request.
 Package template turns a stored, parameterised JSON document into a validated Pulse request.
Package types provides shared type definitions for pulse.
 Package types provides shared type definitions for pulse.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.