@x12i/memorix-pipeline-services

v1.4.0

Published

20 hours ago

Work-management plane: Catalox registry and HTTP invoke router for Memorix pipeline services

0High
0Medium
0Low

x12i

memorix pipeline services registry catalox

@x12i/memorix-pipeline-services

Work-management plane for registering pipeline service workers and routing invoke calls to them.

This package does not read or write Memorix entity/event/knowledge data. That stays on memorix-explorer-api (data plane). Pipeline services is only:

Registry — Catalox catalog memorix-pipeline-services
Runtime — look up { agents, serviceType, serviceId, serviceURL } and POST a typed envelope to serviceURL
Contracts — shared PipelineServiceRequest / PipelineServiceResponse plus practical per-serviceType payloads

Run

cd memorix-pipeline-services
npm install
npm run dev          # PIPELINE_SERVICES_PORT default 5185
# or
npm run serve

Requires Catalox Mongo env (MONGO_URI, MEMORIX_CATALOX_DB, optional MEMORIX_APP_ID / CATALOX_APP_ID). Optional M2M: PIPELINE_SERVICES_TOKEN.

Related requirements

Worker base URLs + liveness ping — manage shared worker origins and GET {baseURL}/health
Optional worker discovery: GET {baseURL}/metadata (initialization pack; missing endpoint is fine — see Initialization metadata)

Service category docs

Connectivity has two peer kinds (both required for full vendor coverage — one is not a substitute for the other):

| Connectivity kind | Category | Direction | Doc | |-------------------|----------|-----------|-----| | Connectors | connector | inbound (pull / discover / read) | docs/connectors/README.md | | Integrations | integration | outbound (notify / act) | docs/integrations/README.md |

Overview: docs/connectivity/README.md

| Category | Role | Doc | |----------|------|-----| | data | transform / validate / enrich | docs/data/README.md | | ai | LLM / reasoning | docs/ai/README.md |

Naming hierarchy

| Layer | Example | Field | |-------|---------|-------| | Domain | firewall / security-operations | domain | | Capability | topology / xdr | capability | | Domain capability | firewall-topology / xdr-detection-response | domainCapability | | Module / pack | neo-firewall-topology / neo-cortex-xdr / neo-tenable-nessus | module | | Integration | paloalto-cortex-xdr | integration | | Provider (vendor) | paloalto / fortigate | provider | | Service type | entity-provider / normalizer | serviceType | | Service ID | paloalto-firewall / paloalto-cortex-xdr | serviceId | | Object type (shorthand) | network-interfaces | objectType | | Object type binding | consume/produce metadata | objectTypeBinding |

firewall-topology and xdr-detection-response are domain capabilities, not providers and not object types.

Use vendor-neutral object types (network-interfaces, not paloalto-firewall-interface). Distinguish:

objectType = what the thing is
schemaRef  = which shape/version it has
vendor     = where it came from (provider)
target     = entity | event | knowledge

Service record

{
  id: string;           // internal key: `${serviceType}:${serviceId}`
  provider?: string;    // worker/product (e.g. paloalto, fortigate)
  agents: string[];
  serviceType: string;
  serviceId: string;    // external id used with serviceType for invoke
  domain?: string;
  capability?: string;
  domainCapability?: string;
  module?: string;
  integration?: string; // e.g. paloalto-cortex-xdr
  objectType?: string;  // optional shorthand
  objectTypeBinding?: ServiceObjectTypeBinding;
  stubSupport?: PipelineServiceStubSupport; // discoverable fixture packs
  serviceURL: string;
  baseId?: string;
  invokePath?: string;
  contractVersion?: string;
  description?: string;
  details?: string;
  /** Optional seed / discovery pack — omit for neo-tools-style workers */
  initializationMetadata?: PipelineServiceInitializationMetadata;
  registeredAt: string;
  updatedAt: string;
}

Catalox item id: `${serviceType}:${serviceId}` (same as id).

Initialization metadata (optional)

Workers that only serve POST /pipeline (+ GET /health) need not change. Packs that also ship object types, sibling service rows, orchestrator steps/pipelines, or a run-input schema may expose them in two non-breaking ways:

On the registry row — optional initializationMetadata on register/update.
On the worker — optional GET {baseURL}/metadata returning the same JSON pack (no secrets). Missing endpoint is fine.

type PipelineServiceInitializationMetadata = {
  metadataPackage?: string;
  objectTypes?: unknown[];
  serviceRegistrations?: unknown[];
  pipelineSteps?: unknown[];
  pipelines?: unknown[];
  runInputSchema?: unknown;
  compatibility?: {
    metadataPackageVersion: string;
    pipelineContractVersion: string;
    serviceContractVersion: string;
    minimumOrchestratorVersion?: string;
    minimumPipelineServicesVersion?: string;
  };
};

Library helpers (from @x12i/memorix-pipeline-services):

| Helper | Behavior | |--------|----------| | fetchWorkerInitializationMetadata(baseURL) | GET {origin}/metadata → pack or null (404 / miss / invalid) | | registerServiceWithInitializationMetadata(store, input, options?) | Upserts the primary service (and any inlined serviceRegistrations); optionally discovers via discoverFromWorker; returns pending pack arrays for callers |

This package persists the pack on the service row and upserts sibling service registrations. objectTypes / pipelineSteps / pipelines / runInputSchema stay opaque — pass hooks.applyObjectTypes / applyPipelineSteps / applyPipelines, or seed them from result.pending into Catalox / orchestrator yourself. Plain POST /api/pipeline-services/services never auto-fetches /metadata.

import {
  createMemoryPipelineServicesStore,
  registerServiceWithInitializationMetadata,
} from "@x12i/memorix-pipeline-services";

const store = createMemoryPipelineServicesStore();
const { service, pending, warnings } = await registerServiceWithInitializationMetadata(
  store,
  {
    agents: ["agent-demo"],
    serviceType: "health-provider",
    serviceId: "acme-connector",
    serviceURL: "http://127.0.0.1:9100/pipeline",
  },
  {
    discoverFromWorker: true,
    hooks: {
      // optional — wire to your Catalox / orchestrator seeders
      applyPipelineSteps: async (steps) => { /* … */ },
      applyPipelines: async (pipelines) => { /* … */ },
    },
  },
);

Worker contract (alongside /health):

GET /metadata
→ 200 application/json  PipelineServiceInitializationMetadata

Omit the route entirely for workers that do not publish a pack.

Object type binding

Every registration stores objectTypeBinding so the orchestrator/UI can know what a service consumes and produces before it runs. Runtime invoke still follows the sparse-fields rule — not every call needs objectType.

When omitted on register, defaults are applied from the service-type table (e.g. health-provider → output connector-health, entity-provider → inputMode: required / same-as-input).

type ServiceObjectTypeBinding = {
  inputMode: "none" | "required" | "optional" | "multi-required" | "derived-from-refs" | "fixed";
  inputSource?: "payload.objectType" | "payload.objectTypes" | "payload.eventTypes" | …;
  inputObjectTypes?: ObjectTypeRef[];
  outputMode: "none" | "same-as-input" | "fixed" | "multi" | "derived";
  outputObjectTypes?: ObjectTypeRef[];
  defaultInputObjectType?: string;
  defaultOutputObjectType?: string;
};

Catalog: GET /api/pipeline-services/object-types
Defaults: GET /api/pipeline-services/object-type-bindings and …/object-type-bindings/:serviceType

Topology object types include topology-scopes, network-interfaces, topology-routes, topology-security-rules, topology-relationships, topology-changes, topology-insights, topology-narratives, connector-health, indexes, and more (see src/object-types/catalog.ts). Relationships stay target: "entity" for now (no separate relationship target).

Taxonomy

Connectivity (2 kinds — both kept):

| Kind | Category | Direction | serviceType | |------|----------|-----------|----------------| | Connectors | connector | inbound | health-provider, capability-provider, scope-provider, entity-provider, relationship-provider, events-provider, content-provider, entity-resolver, change-provider | | Integrations | integration | outbound | communicate, notify, add-content, assign, execute-action |

A vendor pack typically registers connector workers and integration workers. Integrations do not replace connectors.

Processing:

| Category | Role | serviceType | |----------|------|----------------| | data | process | normalizer, data-validator, deduplicator, data-resolver, entity-associator, events-associator, relationship-builder, data-enrichment, classifier, scorer, rollup, analytics, change-detector, insight, narrative-associator, indexer | | ai | reason | discovery, analysis, decision, narrative-builder, step-narrative-builder, content, suggested-plan-builder, suggested-action-builder |

See docs/connectivity/README.md. Note: category integration ≠ record field integration (vendor pack id like paloalto-cortex-xdr).

Invoke contract

Workers receive a full envelope (not a flattened bag of fields):

PipelineServiceRequest<TPayload>

and should return:

PipelineServiceResponse<TResult>

Sparse fields rule

Do not fill every field every time.
Only fill what the service needs.

Common request sections (all optional except what you need): run, service, scope, auth, schemas, inputs, sync, execution, lineage, stub, payload.

Common response sections: status, result, outputs, sync, lineage, metrics, warnings, error.

Runtime always sets contractVersion (default "1.0") and fills service.category / service.serviceType / service.serviceId from the route + registry before POSTing to the worker. Top-level stub is forwarded unchanged to workers (after normalizing payload.useStubs / payload.stubMode aliases). When stub.enabled is true (or the service’s stubSupport.enabledByDefault applies), registry defaults for vendor / pack are filled from stubSupport if omitted.

Stub / fixture mode

First-class demo and CI path so workers can serve layered fixtures without live vendor APIs. Phase 1: contract + conventions + worker-owned packs. Phase 2 (optional): registry-hosted fixture proxy when serviceURL is absent.

Request: `stub` options

stub?: {
  enabled?: boolean;          // force stub on/off for this invoke
  vendor?: string;            // e.g. paloalto | fortigate
  pack?: string;              // fixture pack id / path key
  layer?: "raw" | "normalized" | "derived" | "auto";
  persist?: boolean;          // write fixture records to Explorer
  objectType?: string;        // optional override
};

| Field | Behavior | |-------|----------| | stub.enabled: true | Worker must serve fixtures (if available) and skip live vendor I/O | | stub.enabled: false | Force live even if worker env has stubs on | | omitted | Worker/env default |

Compat aliases: payload.useStubs / payload.stubMode (and optional payload.stubVendor / payload.stubPack / payload.stubPersist).

Registration: `stubSupport`

stubSupport?: {
  enabledByDefault?: boolean;
  vendors?: string[];
  packs?: Array<{
    id: string;
    vendor: string;
    path?: string;              // worker-local or URI
    layers: Array<"raw"|"normalized"|"derived">;
    objectTypes?: string[];
    serviceTypes?: string[];
  }>;
};

Seeded examples: paloalto-firewall and fortigate-firewall in catalox-seeds/inputs/firewall-topology-services.json. List/get responses surface stubSupport so UIs can offer demo mode.

Fixture pack layout

Workers may ship either form under mock/<vendor>/:

A. Aggregate packs

mock/<vendor>/
  <vendor>.connector.raw.fixtures.json
  <vendor>.normalized.fixtures.json
  <vendor>.derived.fixtures.json
  stub-mapping.json

B. File-per-object packs

mock/<vendor>/
  raw/<objectType>.raw.json
  normalized/<objectType>.normalized.json
  derived/<name>.json

Worker helper

import {
  shouldUseStubs,
  loadStubFixture,
  respondFromStub,
  tryRespondFromStub,
} from "@x12i/memorix-pipeline-services";
// or: import { … } from "@x12i/memorix-pipeline-services/stubs";

const config = {
  envEnabled: process.env.MY_WORKER_USE_STUBS === "1",
  packsRoot: new URL("../mock", import.meta.url).pathname,
  defaultVendor: "paloalto",
};

export async function handle(request: PipelineServiceRequest) {
  const stubbed = tryRespondFromStub(request, config);
  if (stubbed) return stubbed; // warnings include "stub: …"; metrics.apiCalls = 0
  // … live path
}

Payload / result by service type

| Type | Payload essentials | Result essentials | |------|--------------------|-------------------| | health-provider | checks[] | healthy, reachable, authenticated, version, permissions | | capability-provider | include[] | capabilities (objectTypes, syncModes, actions, …) | | scope-provider | scopeTypes[] | scopes[] | | entity-provider | objectType, filters? | rawRef, count | | relationship-provider | relationshipTypes[], source/target object types | relationshipsRef, count | | events-provider | eventTypes[], timeRange? | eventsRef, count | | content-provider | objectType, objectRefs / externalIds | contentRef, items? | | entity-resolver | objectType, unresolved[] | resolved[], unresolved[] | | change-provider | objectTypes[], cursor / since | changesRef, nextCursor, count | | communicate | channel, to?, message?, template?, threadId? | outcome, messageRef?, externalRef? | | notify | channel, to, subject?, body?, template?, templateVars? | outcome, notificationRef?, deliveredTo? | | add-content | destination, target?, content, format?, title? | outcome, contentRef?, externalContentId? | | assign | target, assignee, role?, reason? | outcome, assignmentRef?, assignee? | | execute-action | action, target?, parameters?, dryRun? | outcome, actionId?, externalRef?, auditRef? | | normalizer | inputRefs, source/target schema refs, vendor | normalizedRef, count, rejectedCount | | data-validator | inputRefs, rulesetRef? | validCount, invalidCount, issuesRef? | | deduplicator | inputRefs, identityFields[] | dedupedRef, duplicateGroupsCount | | data-resolver | inputRefs, resolveSpecs[] | resolvedRef, counts | | entity-associator | entityRefs, associationRules[] | relationshipsRef, count | | events-associator | eventRefs, entityRefs, rules | relationshipsRef, count | | relationship-builder | entityRefs, buildSpec | relationshipsRef, count | | data-enrichment | inputRefs, sources, fields[] | enrichedRef, enrichedCount | | classifier | inputRefs, taxonomy / labels | classifiedRef, labelCounts | | scorer | inputRefs, scoreModel | scoredRef, scoreSummary | | rollup | inputRefs, groupBy[], aggregations[] | rollupRef, groupCount | | analytics | inputRefs, analyses[] | analyticsRef, summaries? | | change-detector | currentRef, previousRef, compareSpec | changesRef, added/removed/changed counts | | insight | inputRefs, insightType, question? | insightsRef, topInsights? | | narrative-associator | insight/entity/event refs | narrativesRef, count | | indexer | inputRefs, indexTargets[] | indexRef, indexedCount | | discovery | question / goal, inputRefs? | discoveriesRef / candidates[] | | analysis | question, inputRefs, analysisType? | analysisRef, findings? | | decision | decisionType, options?, inputRefs | decision, confidence? | | narrative-builder | inputRefs, narrativeType? | narrativeRef / narrative | | step-narrative-builder | stepId, refs, templateRef? | stepNarrative | | content | contentType (required), prompt?, inputRefs? | contentRef / content | | suggested-plan-builder | subject, goal, inputRefs | memorix-suggested-plan-result/1 | | suggested-action-builder | subject, inputRefs, planRefs? | memorix-suggested-action-result/1 |

Import types from the package:

import type {
  PipelineServiceRequest,
  PipelineServiceResponse,
  PipelineServiceRequestFor,
  HealthProviderPayload,
  EntityProviderResult,
} from "@x12i/memorix-pipeline-services";

Firewall topology examples (condensed)

health-provider — before the connector runs:

{
  contractVersion: "1.0",
  run: { pipelineRunId: "run_fw_topology_001", stepId: "check-paloalto-access", attempt: 1 },
  service: {
    category: "connector",
    serviceType: "health-provider",
    serviceId: "paloalto-xml-api",
    agentId: "agent-demo",
    vendor: "paloalto",
  },
  scope: { orgId: "customer-a", environment: "prod", externalScope: { target: "panorama-prod" } },
  auth: { mode: "credential-ref", credentialRef: "customer-a:paloalto-prod-readonly" },
  payload: { checks: ["connectivity", "auth", "permissions", "version"] },
}

entity-provider — fetch interfaces:

{
  contractVersion: "1.0",
  service: {
    category: "connector",
    serviceType: "entity-provider",
    serviceId: "paloalto-xml-api",
    objectType: "firewall-interface",
  },
  sync: { mode: "full" },
  payload: {
    objectType: "firewall-interface",
    includeDisabled: true,
    filters: { interfaceTypes: ["ethernet", "aggregate", "vlan"] },
  },
}

normalizer — raw → Memorix schema:

{
  contractVersion: "1.0",
  service: { category: "data", serviceType: "normalizer", serviceId: "firewall-interface-normalizer" },
  inputs: {
    refs: [{ collection: "firewall-interfaces-raw", snapshotId: "snap_interfaces_raw_001" }],
  },
  payload: {
    sourceSchemaRef: "paloalto.firewall-interface.raw.v1",
    targetSchemaRef: "memorix.firewall-interface.normalized.v1",
    vendor: "paloalto",
    objectType: "firewall-interface",
  },
}

insight — topology risk findings:

{
  contractVersion: "1.0",
  service: { category: "data", serviceType: "insight", serviceId: "firewall-topology-insight" },
  payload: {
    insightType: "firewall-topology-risk",
    question: "Which topology changes create new exposure?",
    inputRefs: [{ collection: "firewall-topology-changes", snapshotId: "snap_topology_changes_001" }],
  },
}

See src/contract/ for the full TypeScript shapes for all service types. Per-category docs: connectors, integrations, data, ai.

HTTP

| Method | Path | Role | |--------|------|------| | GET | /api/pipeline-services/health | health | | GET | /api/pipeline-services/agents | Catalox agent registry (memorix-ui-agents item) | | GET | /api/pipeline-services/object-types | object type catalog | | GET | /api/pipeline-services/object-type-bindings | default bindings for all service types | | GET | /api/pipeline-services/object-type-bindings/:serviceType | default binding for one type | | GET | /api/pipeline-services/services?serviceType=&agentId= | list | | GET | /api/pipeline-services/services/:serviceType/:serviceId | get | | POST | /api/pipeline-services/services | register | | PUT | /api/pipeline-services/services/:serviceType/:serviceId | update | | DELETE | /api/pipeline-services/services/:serviceType/:serviceId | delete | | POST | /api/pipeline-services/:serviceType | invoke |

Register

curl -sS -X POST http://127.0.0.1:5185/api/pipeline-services/services \
  -H 'content-type: application/json' \
  -d '{
    "agents": ["agent-demo"],
    "serviceType": "insight",
    "serviceId": "vuln-enricher",
    "serviceURL": "http://127.0.0.1:9100/insight",
    "description": "Enriches vulnerability records",
    "details": "Expects envelope.payload with insight fields; returns PipelineServiceResponse."
  }'

Seed security unification maps

Vendor-neutral object catalog, relationship types, per-integration abstraction maps, and cross-domain identity/bindings live under catalox-seeds/inputs/security-abstraction/ and seed into:

| Catalox catalog | Contents | |-----------------|----------| | memorix-security-unified-objects | 73 unified object types | | memorix-security-relationship-types | 50 relationship types | | memorix-integration-abstraction-maps | Palo Alto FW, FortiGate, Cortex XDR, Nessus, Microsoft Entra ID maps | | memorix-cross-domain-unification | identity-resolution + service-object-bindings |

# Requires MONGO_URI (same as serve)
npm run seed:security-abstraction

Readable inventory (unified objects, abstract classes, relationship types, all integration maps, cross-domain) is regenerated from the same seed JSON — no Mongo:

npm run export:security-abstraction

Output: catalox-seeds/inputs/security-abstraction/SECURITY-ABSTRACTION.md. Re-run after editing seed JSON.

Canonical names follow the pack (network-interfaces, network-traffic-events, …). These catalogs are normalization metadata (source → unified), not Mappix abstract bindings.

Seed firewall topology registry

Git-authored metadata for the firewall topology integration lives in catalox-seeds/inputs/firewall-topology-services.json (37 records: Palo Alto + FortiGate connectors, data workers, AI workers). One Catalox item per serviceType + serviceId in catalog memorix-pipeline-services.

Customer-specific values (firewall IP, credentials, VDOM/VSYS, Panorama target, etc.) stay out of the registry — put them in the runtime invoke envelope (scope, auth, payload, sync, …).

# Requires MONGO_URI (same as serve)
npm run seed:firewall-topology

Re-runs are idempotent (create or update). The same command also upserts agent-firewall-topology into Catalox catalog memorix-inventory-policies (item memorix-ui-agents) from memorix-descriptors/catalox-seeds/inputs/inventory-policies/memorix-ui-agents.json.

Seed Cortex XDR registry

Git-authored metadata for Palo Alto Cortex XDR lives in catalox-seeds/inputs/cortex-xdr-services.json (connector + execute-action outbound + data/AI workers for domainCapability: xdr-detection-response, with full objectTypeBinding on each row). Service URLs point at http://neo-cortex-xdr-services:9501/pipeline.

# Requires MONGO_URI (same as serve)
npm run seed:cortex-xdr

Re-runs are idempotent. The command also upserts agent-xdr-investigation and agent-xdr-response via the shared memorix-ui-agents inventory policy item.

Seed Tenable Nessus registry

Git-authored metadata for Tenable Nessus lives in catalox-seeds/inputs/tenable-nessus-services.json (connector + execute-action outbound + data/AI workers for domainCapability: vulnerability-assessment, with full objectTypeBinding on each row). Service URLs point at http://neo-tenable-nessus-services:9601/pipeline.

# Requires MONGO_URI (same as serve)
npm run seed:tenable-nessus

Re-runs are idempotent. The command also upserts agent-vulnerability-assessment and agent-vulnerability-assessment-actions via the shared memorix-ui-agents inventory policy item.

Seed CrowdStrike Falcon registry

Git-authored metadata for CrowdStrike Falcon lives in catalox-seeds/inputs/crowdstrike-falcon-services.json (health/capability/scope, entity/events providers, normalizer/validator/dedupe/associator, and action executor). Service URLs point at http://neo-crowdstrike-falcon-services:9701/pipeline (draft port; shared with Splunk until dedicated ports are assigned). Abstraction map: catalox-seeds/inputs/security-abstraction/crowdstrike-falcon/.

# Requires MONGO_URI (same as serve)
npm run seed:crowdstrike-falcon

Seed Splunk Platform registry

Git-authored metadata for Splunk Platform lives in catalox-seeds/inputs/splunk-platform-services.json (REST pull + HEC push health, search export events provider, normalizer/validator/dedupe, HEC action executor). Service URLs point at http://neo-splunk-platform-services:9701/pipeline (draft port; shared with CrowdStrike until dedicated ports are assigned). Abstraction map: catalox-seeds/inputs/security-abstraction/splunk-platform/.

# Requires MONGO_URI (same as serve)
npm run seed:splunk-platform

Invoke

API body:

{
  "serviceId": "vuln-enricher",
  "agentId": "agent-demo",
  "request": {
    "run": { "pipelineRunId": "run_1", "stepId": "enrich", "attempt": 1 },
    "payload": { "insightType": "vuln-risk", "question": "What is exposed?" }
  }
}

Worker receives the full PipelineServiceRequest (with service filled). Success response from this API:

{
  "serviceId": "vuln-enricher",
  "serviceType": "insight",
  "response": { "status": "success", "result": { "count": 1 } }
}

If agentId is provided, it must appear in the record’s agents list.

Library

import {
  createMemoryPipelineServicesStore,
  createCataloxPipelineServicesStore,
  invokePipelineService,
  buildWorkerRequest,
  shouldUseStubs,
  tryRespondFromStub,
  fetchWorkerInitializationMetadata,
  registerServiceWithInitializationMetadata,
  PIPELINE_SERVICE_TYPES,
  PIPELINE_SERVICE_CONTRACT_VERSION,
} from "@x12i/memorix-pipeline-services";

See Initialization metadata for the optional seed / discovery helpers.

Boundary vs explorer-api

| Plane | Package | |-------|---------| | Work (who runs what, how to call it) | this package | | Data (records, lists, enrichment apply) | memorix-explorer-api |

Do not mix registry here with explorer’s operational pipeline-registry-parts.

Published

Vulnerabilities

Links

Maintainers

Keywords

Readme

@x12i/memorix-pipeline-services

Run

Related requirements

Service category docs

Naming hierarchy

Service record

Initialization metadata (optional)

Object type binding

Taxonomy

Invoke contract

Sparse fields rule

Stub / fixture mode

Request: stub options

Registration: stubSupport

Fixture pack layout

Worker helper

Payload / result by service type

Firewall topology examples (condensed)

HTTP

Register

Seed security unification maps

Seed firewall topology registry

Seed Cortex XDR registry

Seed Tenable Nessus registry

Seed CrowdStrike Falcon registry

Seed Splunk Platform registry

Invoke

Library

Boundary vs explorer-api

Request: `stub` options

Registration: `stubSupport`