Transform

The mapper family: three ways to say "put this there"

A descriptor is protobuf's runtime form of a schema: the field list, types, and nesting of a message, available without any generated code. The mapper family operates purely on descriptors and DynamicMessages (messages built and read through the descriptor at runtime), so the same rules work on every message type in the system, including google.protobuf.Struct, protobuf's generic JSON-shaped bag.

The base mapper, ProtoFieldMapper, speaks a three-form text rule language. Paths are dotted field paths; a source can be a path or a literal:

• target = source copies a value into a field.
• target += source appends a value to a repeated field.
• -field clears a field entirely.

dev-tools/protomolt transform mapper core, docs/transform/mapping.md

mapper.mapInPlace(builder, List.of(
    "title = body",
    "tags += \"proto\"",
    "-scratch"
));

One design detail carries real weight in production. When a mapping fails, the mapper throws a MappingException whose Category distinguishes ABSENT_INTERMEDIATE (an intermediate message on the path was simply unset, so "skip" is a sane response) from GENERAL (something is actually broken). Callers branch on the enum instead of parsing error strings.

CEL rules: filter, selector, target, fallback

The text rules are deliberately small. For anything conditional or computed, the CEL mapper wraps the same base mapper. CEL is Google's Common Expression Language, a tiny sandboxed expression language made for exactly this kind of "look at a message, compute a value" work. A CelMappingRule has four parts:

• filter (optional): a boolean expression deciding whether the rule fires at all.
• selector (optional): the expression producing the value to write.
• target: the field path receiving the value.
• fallback (optional): a text rule that runs when the selector cannot produce a value.

dev-tools/protomolt transform mapper cel, docs/transform/mapping.md

new CelProtoMapper(mapper, cel).map(builder, List.of(
    new CelMappingRule(
        "input.lang == 'en'",   // filter
        "input.title",          // selector
        "search_title",         // target path
        List.of())              // text-rule fallback
));

Rules compose progressively: the mapper works on a builder, so a later rule sees everything an earlier rule wrote. There are per-call extra bindings, and candidate matching helpers (tryMap, mapFirstCandidate) for the "try these shapes in order" case. A runnable sample ships in the repo as CelMappingSample.

One CEL engine for the whole tree

transform/mapper/cel is the only CEL integration in the repository. Validation, quality scoring, projection, metric filters, shapes, mapping rules, and the action catalog all compile their expressions through the same CelEnvironmentFactory and CelEvaluator (backed by the dev.cel implementation). Only the root variable name changes by consumer: validation and quality bind this, projection binds source, the mapper binds input (the pipeline executor overrides it to target). Compiled programs and compile failures are both cached, so a malformed expression fails fast on every use instead of being re-diagnosed per message.

The third mapper module, protomolt-mapper-metadata, reads named metadata out of message contents at runtime: a map from metadata name to CEL selector, evaluated against the message. This is distinct from meta.v1 schema annotations, which declare facts about fields. Before any selector evaluates, it is type-checked in two environments: a per-descriptor validation environment for input, plus precompilation in the caller's own evaluator environment. The caches are bounded (MAX_ENVIRONMENTS = 64, MAX_VALIDATED_EXPRESSIONS = 1024), so a long-running process cannot grow memory by seeing many message types.

Try it: a mapping playground

This is the shape every mapping surface in ProtoMolt shares: text rules and CEL rules applied in order, each later rule seeing earlier writes. Toggle the CEL rules, switch the canned document, and watch the result.

{
  "id": "doc-1042",
  "lang": "en",
  "title": "Streaming joins without a database",
  "body": "Two live gRPC streams can be joined in process, with bounded buffers instead of a database.",
  "tags": ["grpc", "streams"],
  "canonical_url": "https://example.dev/joins",
  "scratch": "editor note"
}

heading = title
tags += "ingested"
-scratch

{
  "id": "doc-1042",
  "lang": "en",
  "title": "Streaming joins without a database",
  "body": "Two live gRPC streams can be joined in process, with bounded buffers instead of a database.",
  "tags": ["grpc", "streams", "ingested"],
  "canonical_url": "https://example.dev/joins",
  "heading": "Streaming joins without a database",
  "search_title": "Streaming joins without a database",
  "route": "long-form",
  "origin": "https://example.dev/joins"
}

Validation: measured against the standard, not asserted

Validation rules are protobuf custom options on the schema, read off the descriptor at runtime, so the same checks apply to generated classes and dynamic messages alike. One resilience detail makes this work everywhere: when a descriptor was linked without the extension registry (it arrived over reflection, say), the options are re-parsed from the descriptor's unknown fields. Rules survive the journey instead of being silently dropped.

ProtoValidator is dialect neutral. It evaluates an internal constraint model (FieldConstraints, MessageConstraints), and rule sources plug in through a small Java SPI (Service Provider Interface, the JDK's classpath-discovery mechanism):

dev-tools/protomolt protobuf validation, docs/transform/validation.md

public interface ValidationRuleSource {
    Optional<FieldConstraints>   fieldConstraints(FieldDescriptor field);
    Optional<MessageConstraints> messageConstraints(Descriptor message);
}

Every configured source is consulted for every field, and all violations are merged: no source silently wins. The default chain is the built-in validate.v1 reader plus anything else found by ServiceLoader. Dropping the protomolt-protobuf-validation-protovalidate module on the classpath is all it takes to enforce (buf.validate.field) and (buf.validate.message) annotations too, including predefined rules, custom CEL, Any and FieldMask rules, and the well-known-type string and bytes formats. The vendored buf/validate/validate.proto is pinned at v1.2.2, Apache-2.0 licensed and attributed in the module's NOTICE.

Rules compile eagerly when the validator is constructed for a message type, so a malformed rule is a schema error (RuleCompilationException, thrown deterministically) and a rule that fires on a message is a data error (RuleEvaluationException). The violation rule IDs deliberately use protovalidate's naming (string.min_len, repeated.unique, timestamp.within), so error surfaces interoperate. The rule surface covers strings (including email, uuid, hostname, uri, and ip, checked by the zero-dependency protomolt-formats parsers), ints, uints, floats, bool, bytes, enum, repeated, map, Timestamp, Duration, and per-field and per-message cel rules.

dev-tools/protomolt docs/transform/validation.md, the shape of a declared rule

string email = 2 [(ai.pipestream.proto.validate.v1.field) = {
  cel: { id: "email.not_localhost"
         expression: "!this.endsWith('@localhost')" }
}];

The conformance number: 2872 of 2872

Compatibility with protovalidate is measured rather than claimed. The implementation passes the complete protovalidate v1.2.2 conformance suite, 2872 of 2872 cases. There is no skip list: the suite runs bare, so a regression anywhere shows up immediately.

Two harnesses share one runner in protobuf/validation-conformance. A curated subset runs as unit tests, comparing structured field paths, rule IDs, and key flags exactly as the suite does, gated at 100% so any drift is caught. The field paths are reconstructed with a faithful port of the suite's own field-path algorithm, so a match reflects semantic agreement rather than formatting luck. The authoritative harness is ConformanceMain, a stdin/stdout executor speaking buf's language-agnostic protocol, which lets buf's protovalidate-conformance binary itself score the full suite:

protovalidate-conformance \
  protobuf/validation-conformance/build/install/protomolt-protobuf-validation-conformance/bin/protomolt-protobuf-validation-conformance

At the gRPC boundary, protomolt-grpc-validation ships server and client interceptors: every inbound request, unary or streamed, is validated before the handler runs, and violations are refused with INVALID_ARGUMENT naming every violated rule. A malformed rule is the server's problem (INTERNAL, details kept in logs), never the caller's. The same interceptor can measure per-request quality and refuse below a configured quality floor (FAILED_PRECONDITION), with Micrometer meters for requests, rejections, violations, and quality scores.

Sensitivity masking: remove, redact, or seal a field

Sensitivity is declared on the schema with the meta.v1 field option: a free-form string (public, internal, pii, secret are the conventions, nothing enforces the vocabulary), attached right next to the field it describes:

dev-tools/protomolt docs/transform/masking.md

message Order {
  string id = 1;
  string email = 2 [(ai.pipestream.proto.meta.v1.field) = {sensitivity: "pii"}];
  bytes  token = 3 [(ai.pipestream.proto.meta.v1.field) = {sensitivity: "secret"}];
}

The masking primitive is SensitivityMasker (module protomolt-protobuf-metadata), exposed as the mask-message verb and as the RedactMessage Kafka Connect transform. Given a message and a set of sensitivity classes, it walks the message and acts on every matching field:

• remove (the default) clears the field completely.
• redact replaces strings with the fixed literal *** and clears everything else, because a redacted number would still be a plausible value and would read as data.
• encrypt seals string and bytes fields with AES-GCM (AES/GCM/NoPadding, 128-bit tag) inside a versioned envelope: one version byte, a 12-byte nonce from SecureRandom, then the ciphertext and tag. Strings are base64 encoded. decrypt reverses it.

The envelope binds the value's identity as additional authenticated data: the containing message's full name plus the field number. Move a ciphertext to another field and decryption fails, because the authenticated context no longer matches. Keys are entirely the caller's (16, 24, or 32 raw bytes, base64 encoded): never stored, never derived, no key store, no rotation service. The contract declares what is sensitive; the operator holds the means.

Traversal descends into repeated fields, maps (paths reported as field[key].nested), and packed Any payloads, which are resolved, masked, and repacked under the same type URL. Payloads that cannot be opened are reported in MaskResult.unresolvedPaths(), never silently passed:

dev-tools/protomolt docs/transform/masking.md

var result = SensitivityMasker.mask(order, Set.of("pii"), Strategy.REDACT);
result.message();          // the masked message
result.maskedPaths();      // ["email", "contacts[home].email"]
result.unresolvedPaths();  // packed payloads that could not be opened

Masking is a schema-driven transform, not a security boundary: it cannot see fields nobody annotated, copies upstream of the mask still exist, redaction is one-way, ciphertext length leaks, and the key travels in the request. The typed gRPC and REST MaskMessage RPC carries a narrower envelope (no key field, no unresolved_payloads), so only remove and redact are reachable there; encrypt and decrypt are available through the CLI and MCP surfaces.

Quality scoring: how good is the admissible data

Validation is binary: a message passes or it is refused. Quality is the graded companion. A message declares its own quality dimensions as the (ai.pipestream.proto.quality.v1.quality) message option, each dimension a CEL expression that scores one facet of the data:

message Article {
  option (ai.pipestream.proto.quality.v1.quality) = {
    dimension: { id: "titled" cel: "this.title != ''" }
    dimension: { id: "sized"  weight: 3.0
                 cel: "clamp(double(this.body.size()) / 500.0, 0.0, 1.0)" }
    dimension: { id: "fresh"
                 cel: "exp(-double(this.age_days) / 365.0)" }
  };
  ...
}

QualityScorer.score(message) returns a QualityReport with a composite() (the weighted average, 0 to 1), per-dimension scores, and a failed() list. Each CEL result is coerced to a double and clamped to [0, 1]; a boolean counts as 1 or 0. Two helper functions, exp(x) and clamp(x, lo, hi), exist for decay curves and capped ratios like the ones above. Failure semantics mirror the validator exactly: an expression that does not compile is a schema error, thrown deterministically the first time the type is scored; an expression that compiles but fails on one particular message marks that dimension failed and the dimension weighs nothing. A quality measurement should never take the data path down. Compiled scorers are cached per type, up to 256 types.

The positioning line, from the module's own documentation: validation says whether data is admissible; quality says how good the admissible data is. The two compose at the boundary: the gRPC validation interceptor can refuse below a quality floor.

In the Kafka serde, measurement is on by default (protomolt.quality.on.write), free for types that declare no dimensions, and emitted as protomolt.serde.quality.* Micrometer distributions: a quality dashboard of every stream that flows through the serde. Setting protomolt.quality.min=0.5 turns measurement into a gate on writes. Reads are never rejected, because a consumer cannot improve what a producer already wrote.

Schema inference: a proto definition from sample data

Sometimes the contract does not exist yet. SchemaInferrer (the infer-schema verb) reverse-engineers a proto definition from data-rich JSON. The class documentation states the behavior precisely:

Keys are sanitized into legal proto field identifiers; when sanitization changes a key, the field carries json_name holding the original (user-name becomes user_name), so the inferred schema round-trips the very documents it was inferred from. A MAX_DEPTH = 32 guard rejects adversarial nesting, and more samples strictly improve the schema: keys union together and the numeric heuristic sees every occurrence. The output is proto source plus a linked descriptor set, and a runnable demo (GraphInferSchemaSample) ships in the samples.

An illustration of the documented heuristics, from this sample document:

illustrative input: two samples of the same JSON shape

{ "user-name": "ada", "logins": 12,  "tags": ["admin"] }
{ "user-name": "bo",  "logins": 7,   "tags": []       }
{ "user-name": "cy",  "logins": 2.5, "tags": ["ops"]  }

the inferred shape (numbers went double because one sample was fractional; the renamed key keeps json_name)

message Sample {
  string user_name = 1 [json_name = "user-name"];
  double logins = 2;
  repeated string tags = 3;
  // an empty array alone would have fallen back to google.protobuf.Value
}

Shapes and merges: joins that end as real proto types

protomolt-shapes derives new message types from existing ones, three ways: an envelope (one field per named source, lossless), a projection (a flat message whose field types come from scoped source paths, so the SELECT list becomes the schema), and a tagged union (a oneof over the source types). The result is not a synthetic runtime blob: it is built as a FileDescriptorProto that depends on the sources' own files, linked in-process, and emitted as .proto source with true import paths. A join's output contract registers in the git-backed schema registry like a hand-written one, with references, history, diffs, and compatibility gates; change the join definition and check-compat says whether downstream consumers survive.

SchemaMerger (merge-schemas verb) combines schemas in three steps: validate (same name and type coalesces and can become a natural join key; same name with different type or cardinality is a hard clash that blocks emission), resolve (rename, defaulting to <source>_<field>, prefer one source, or coalesce), and emit. The result records both relationships: the defined join (one ruleset reading all sources at once) and the defined union (one ruleset per source, a structural UNION). Map-typed fields are not yet mergeable and are rejected with a clear error.

Under all of it sits one scope model, MessageScope: an ordered set of named messages that every combination surface shares. Scoped text rules read target = name.path, and CEL sees each entry as a typed variable. Joins bind order and customer; workflow steps bind input and steps.<name>; enrich transforms bind input and response; keyed record joins bind key and value. Learned once, used everywhere.

Typed call chains: workflows, pipelines, keyed joins

Everything above reshapes one message. The workflow and pipeline modules compose reshaping with service calls, and the composition is typed end to end: each edge of the graph declares what it accepts and what it produces, checked against the descriptor set before any service is invoked.

Workflows: checked serial gRPC compositions

A Workflow records its input type, the service-profile dependencies each step needs (with endpoint and descriptor fingerprints), an ordered list of gRPC or structured-generation steps, the typed dataflow between them, an output mapping, and a deadline. Every typed edge does the same dance: select sources, map them into a declared type, optionally project, validate, then deliver. The ordering matters: a rejected edge does not invoke the service or inference provider at all. Fan-out has item and concurrency limits, a stable branch order, and a FAIL_FAST or CONTINUE policy per branch.

flowchart LR
          sel["Select sources from the scope"] --> map["Map into the declared edge type"]
          map --> proj["Project, optional"]
          proj --> val["Validate the edge type"]
          val --> deliver["Deliver to the service call"]
          val -. "a rejected edge stops here, the service is never invoked" .-> refuse["Refused, with named rule violations"]

Workflow runs are recorded, and recording is load-bearing: replay-workflow re-runs a recorded workflow offline, verifying hashes, fingerprints, and verdicts with no network or inference calls at all. Tampering with the recording produces a failed replay finding. Workflows carry a family of the action catalog's verbs, from check-workflow and compile-workflow through record-workflow-run and replay-workflow to export-work-record and its verify and evaluate siblings at the receipt layer.

Pipelines: the compiled, streaming-aware form

protomolt-pipeline is the compiled, streaming-aware execution form of a workflow. Its step vocabulary covers gRPC calls in all four streaming shapes (unary and streaming, on both sides), structured generation that produces a validated protobuf message, unnest (a repeated field becomes a stream), and collect (a stream becomes a repeated field). WorkflowPipelineCompiler converts a workflow into a pipeline, and PipelineChecker then verifies the compiled contract independently of the workflow checker that approved the source.

The discipline that keeps streaming compositions sane is cardinality: every binding is declared ONE or MANY, and a MANY binding is linear, meaning exactly one consumer. You cannot silently fork a stream. A max_stream_messages bound caps any materialized stream, so memory stays explicit. PipelineChecker statically checks the pipeline against its descriptor set before any service call, and PipelineExecutor runs it in-process through a host-owned PipelineTransport, keeping endpoint policy and credentials with the host application. Fan-out runs branches on virtual threads under a semaphore. Pipelines refuse external-completion steps at runtime, and a durable job coordinator is not wired yet.

Keyed joins over two live streams

StreamJoiner joins two live gRPC server streams in ZIP mode (pair arrivals in order) or KEYED mode (pair by a scalar key field). Both sides are flow-controlled streams, so a fast producer cannot flood a slow consumer. Unmatched entries wait in bounded per-side buffers whose oldest entries drop on overflow: memory is explicit, never unbounded.

flowchart LR
          orders["orders stream, key id"] --> join["StreamJoiner, KEYED"]
          payments["payments stream, key order_id"] --> join
          join --> out["Joined message, mapped through scoped rules"]
          join -. "oldest unmatched entry drops on overflow" .-> buf["bounded buffer per side"]

In KEYED mode each side names one singular scalar field path, and construction validates that both sides use the same key type. Duplicate keys queue first-in-first-out; an entry whose key cannot be read is dropped. A matched pair is immediately joined into the target type through the same scoped mapping rules everything else uses. The module is equally clear about what it refuses to be: no stateful topic-to-topic joins ("Kafka Streams does this well; a sidecar should not"), no unbounded buffering ("if a join needs unbounded state, it needs a database"), and no query language. Richer unmatched-entry policies, such as emitting partial pairs or failing on drop, and schema-declared identity keys for joins and dedup are planned, not implemented yet.

When a step needs an agent: delegation

Some steps are not service calls. The delegation protocol hands a bounded task to an agent (a coding assistant, a local model, any worker implementing one small interface) and makes the exchange durable enough to survive restarts on either side. The summary, because the protocol earns its own page:

• One bidirectional gRPC stream per worker. The first frame is a WorkerHello with identity, protocol version, provider and model metadata, and bounded capabilities; the coordinator answers with an admission decision before any task frame flows. The model does not own the stream; a worker adapter does.
• A worker can never mark its own task done. It submits a completion candidate with evidence for every required acceptance check; the coordinator accepts the revision or requests another with structured feedback. Leases are explicit, with heartbeats, renewals, and expiry.
• Every frame is idempotent and ordered. A sender-generated frame_id plus a per-lane sequence make duplicate delivery safe and gaps detectable.
• Transcripts are durable and encrypted. Accepted frames persist as an AES-256-GCM snapshot through the repository service's blob calls; the store only ever sees ciphertext, and an offline reducer can replay a transcript with no external calls to audit exactly what happened.
• The protocol dogfoods this page. Every field in the delegation contract carries validate.v1 bounds and meta.v1 sensitivity metadata, including message-level CEL cross-field checks:

dev-tools/protomolt transform delegation, delegation.proto, the DelegateRequest message options

option (ai.pipestream.proto.validate.v1.message) = {
  cel: {
    id: "hello-is-session-scoped"
    message: "a hello frame leaves task_id empty; every other worker frame names its task"
    expression: "has(this.hello) ? size(this.task_id) == 0 : size(this.task_id) > 0"
  }
  ...
};
string frame_id = 1 [
  (ai.pipestream.proto.validate.v1.field) = { required: true string: {uuid: true} },
  (ai.pipestream.proto.meta.v1.field).sensitivity = "internal"
];

Twelve delegation-* verbs (register, offer, accept, progress, checkpoint, candidate, review, cancel, message, watch, transcript, and worker listing) expose the lifecycle, and protomolt-serve mounts one coordinator in-process. The full protocol, including the reconnect story and the evidence rules, is on the Mesh page.