ProtoMolt

Modules — Agent Communication

ProtoMolt's processing modules are language-agnostic gRPC services that pull work from the engine instead of being called by it. A module implements a small, fixed contract — one descriptor RPC, one processing RPC, standard gRPC health — and the platform supplies everything else: work distribution, catalog registration, health gating, and retries.

Demand-pull: the engine never dials out

The engine owns the entire data path. It consumes per-module Kafka topics (pipestream.module.<module_id>) that carry claim-check pointers, hydrates document bodies from the repository, runs all CEL filter/mapping/routing logic, stages results, and commits offsets. Modules are pure compute: they never touch Kafka, Redis, or object storage. A module worker asks for work by opening a short-lived bidirectional stream to ai.pipestream.module.work.v1.ModuleWorkService/Work — one stream per work unit, not one long-lived multiplexed connection.

core-services/pipestream-engine/docs/architecture/15-demand-driven-flow-control.md

Module                          Engine
  │  Hello(module_id, instance_id)│
  │◄─ WorkUnit(Any<PipeStream>) ──│  (hydrated payload; ~5s no-work wait, then NoWorkAvailable)
  │  Heartbeat (every ~15s)       │
  │  WorkAck(status, result)      │  result oneof: updated_payload | no_op | reject
  │◄─ AckConfirmed ───────────────│  (offset committed; stream closes)

The protocol's properties come straight from the proto (module_work_service.proto):

The WorkAck result model is unusually explicit: Reject carries a typed RejectReason (EMPTY_PARSE, PARSE_ERROR, POLICY, DUPLICATE, UNPROCESSABLE), NoOpResult marks an intentional no-change, and soft_error records a non-fatal error on a document that continues. The invariant stated in the proto is "never drop, move forward": a no-op forwards the document unchanged; only Reject stops it.

The module contract

What a module must serve is three small services in ai.pipestream.data.module.v1, plus standard gRPC health. The descriptor door is mandatory; the processor doors come in unary and streaming flavors, and a module declares which it serves in Capabilities.processor_transports. That declaration is load-bearing: callers select a door from it instead of probing, and an empty declaration on a module claiming a processor capability is refused at registration.

core-services/pipestream-protos/pipeline-module/proto/ai/pipestream/data/module/v1/module_service.proto

// Module identity and registration. MANDATORY: every module serves this,
// whatever processor transport it implements.
service ModuleDescriptorService {
  rpc DescribeModule(DescribeModuleRequest) returns (DescribeModuleResponse);
}

// Unary processor door: one Process call per document. …
// The engine never calls it: engine-fed modules pull work over the
// ModuleWorkService demand-pull stream.
service ModuleUnaryProcessorService {
  rpc Process(ProcessRequest) returns (ProcessResponse);
}

// Streaming processor door: many documents multiplexed over one long-lived
// bidirectional stream, each tagged with a caller-assigned correlation id…
service ModuleStreamProcessorService {
  rpc ProcessStream(stream ProcessStreamRequest) returns (stream ProcessStreamResponse);
}

Health is the stock grpc.health.v1.Health/Check, and it must answer SERVING on the empty service name — the proxy probes "", and registering health only under your own service name is the classic mistake.

Doc drift. Older documentation (including the workspace AGENTS.md and several module READMEs) describes a combined PipeStepProcessorService with ProcessData, ProcessDataStream, and GetServiceRegistration RPCs. That service no longer exists in the protos; it was split into the three services above. GetServiceRegistrationResponse survives as the descriptor message wrapped in DescribeModuleResponse, but the RPC door is gone. The proxy's code is explicit that it "has no legacy leg and never dials the combined PipeStepProcessorService" — code wins where the docs disagree.

The JVM module runtime: two beans is a whole module

JVM modules build on the pipestream-module-runtime Quarkus extension. The reference module module-echo demonstrates the entire authoring surface: its whole business logic is one bean implementing ModuleProcessor, plus one descriptor bean returning name, version, JSON config schema, tags, and PipeDoc part masks.

modules/module-echo/src/main/java/ai/pipestream/echo/EchoProcessor.java

@ApplicationScoped
public class EchoProcessor implements ModuleProcessor<PipeStream> {
    @Override
    public PipeStream process(PipeStream input) {
        return input;
    }
}

From those two beans the runtime auto-wires the ModuleWorkerLoop (engine dialing, stream lifecycle, heartbeats, reconnect with exponential backoff from 500ms to 30s), the generic gRPC processor doors, self-registration with the catalog, readiness, a live ramp-control endpoint (GET/POST /module-runtime/v1/ramp) for runtime concurrency ceilings, and a dev/test door (POST /module-runtime/v1/process-once). The worker pool self-sizes: it starts at one worker, adds one per successful unit up to a concurrency ceiling (default 8), and retires surplus workers after three consecutive idle rounds. The idle cost is one poller opening a stream every 3 seconds.

The ModuleProcessor contract is documented on the interface itself: calls are synchronous on a virtual thread, must be idempotent, and must be explicit about failure — throwing PermanentFailure means no redelivery, while any other exception is retryable and Kafka redelivers, likely to a different module instance.

module-proxy: the polyglot adapter

A module whose implementation lives in another language runs behind module-proxy, a JVM adapter that owns everything platform-specific: it runs the demand-pull loop against the engine and forwards each work unit to a configured downstream (PROXY_DOWNSTREAM_HOST/PORT, mode unary or streaming via PROXY_DOWNSTREAM_MODE). Its rules are deliberately strict:

Reference modules in three languages implement the identical contract with identical title_suffix and simulate_failure config hooks — examples/python/reference_module.py (153 lines), examples/node/reference_module.mjs (154), and examples/go/reference_module.go (177). The Python one, in essence:

modules/module-proxy/examples/python/reference_module.py

class ReferenceProcessor(pb_grpc.ModuleUnaryProcessorServiceServicer):
    def Process(self, request, context):
        return pb.ProcessResponse(response=_process(request.request))
…
health_servicer = health.HealthServicer()
health_servicer.set("", health_pb2.HealthCheckResponse.SERVING)   # EMPTY name is what the proxy probes
Reference deployment. In the compose stack, eight module-proxy-* containers run from one image — one per first-party module — but all with the worker loop disabled: every first-party module is JVM and pulls its own work, so the proxies sit there as a catalog and health boundary only, held in reserve for modules that cannot run a pull loop. The polyglot forwarding path itself is real and continuously tested.

The certification battery

Polyglot claims are enforced by ModuleContractIT, an abstract, language-agnostic integration test with seven tests: health SERVING on the empty name; the descriptor accepted by the real catalog-publisher mapping; the descriptor declaring its transports; an unconfigured document echoed untouched; title_suffix config reaching the module and transforming output; simulate_failure returning PROCESSING_OUTCOME_FAILURE with structured error details; and work units flowing through a real ModuleWorkerLoop against an in-process engine. Subclasses certify each language — JavaModuleContractIT runs always in-process; PythonModuleContractIT, NodeModuleContractIT, and GoModuleContractIT are enabled via environment flags and self-skip without the toolchain. A separate smoke test runs the same contract against a real published container image.

The leased module catalog

The platform-registration-service (port 18101) is the live module catalog, exposed as ai.pipestream.platform.registration.v1.PlatformRegistrationService. It is deliberately narrow: it accepts SERVICE_TYPE_MODULE and nothing else — core services do not register; their addresses are deployment configuration. Registration is by lease, not permanent row:

core-services/pipestream-protos/…/platform_registration.proto

rpc PublishModule(PublishModuleRequest) returns (PublishModuleResponse);   // route_id + lease_token + expiry
rpc RenewModuleLease(RenewModuleLeaseRequest) returns (RenewModuleLeaseResponse);
rpc ReleaseModuleLease(ReleaseModuleLeaseRequest) returns (ReleaseModuleLeaseResponse);
// "Best-effort early release… Correctness does not depend on this RPC because
//  an abandoned lease expires automatically."

The design treats liveness as something that must be continuously re-earned:

Two publisher clients exist. JVM modules use ModuleSelfRegistrar, which dials its own gRPC server over loopback for the descriptor — deliberately, to prove the door is actually served before advertising it. Polyglot downstreams are published by the proxy's ModuleCatalogPublisher.

Limitations. The registration service has no authentication — by design, as an internal control plane; anyone who can reach port 18101 can publish a module route, so security is network posture only. The streaming Register RPC from the older contract is kept implemented for wire compatibility but no current module calls it. And CapabilityType is minimal: only PARSER and SINK are active enum values, the rest commented out in the proto until needed.

Module web UIs

A module can ship a bundled admin web interface and advertise it through ordinary descriptor metadata — no schema change needed. The frontend's backend-for-frontend mounts a reverse proxy at /modules/<name>/<prefix>* that pipes requests verbatim (SSE-safe, Host rewritten) to the module, but only for paths under the module's declared ui_proxy_prefixes, with path-root boundary matching: /api matches /api/x, never /apish. Unadvertised paths fall through to the SPA, and the lab shell mounts the module's UI in an iframe.

modules/module-testing-sidecar/…/TestingSidecarDescriptor.java

private static final Map<String, String> UI_METADATA = Map.of(
        "ui_path", "/admin",
        "ui_title", "Module Testing Sidecar",
        "ui_proxy_prefixes", "/admin,/test-sidecar,/mock-engine,/sample-data,/api");

Related machinery: descriptor metadata keys prefixed pipestream.schema-artifact. are promoted at publication into named catalog schema artifacts (for example a parser's UI schema), fetched by the BFF and rendered as schema-driven config forms.

The testing sidecar

module-testing-sidecar is the platform's end-to-end test harness — and itself a demand-pull module, which is how it observes: the graph it deploys carries tap-<node> fan-out nodes and a terminal node pointing at its own processor, so the counts it asserts are the counts the engine actually served it. One server-streaming RPC, ai.pipestream.testing.harness.v1.PipelineCrawlerService/RunPipelineCrawl, drives a full run: validate, provision an account and datasource, deploy a graph through the same engine RPCs the designer UI uses, dispatch real documents through real connectors, wait on taps, drain to index receipts, summarize with the actual missing document ids, and clean up everything it created. It also ships a mock engine implementing ModuleWorkService (gated by MOCK_ENGINE_ENABLED) so any module's pull loop can be exercised with no real engine, Kafka, Redis, or S3. Its run journal is in-memory by design, and it is explicitly not for always-on production use.

Related