@x12i/multistep-connector

Durable multi-step session runtime for connector pipelines. Hosts MGC vendor pulls and pure/worker stages with Explorer-backed state in memorix-connectors-data.

• Package name is multistep-connector (not memorix-multistep): Memorix is the platform; this library owns stage/session contracts and execution.
• Production state is database-only via injected SessionStore (use ExplorerSessionStore). No production in-memory store.
• Does not reimplement MGC hydrate/compose/poll/cleanup — inject an MgcHostAdapter for mgc-pull stages.

Design FRs: docs/multistep-connector/.

Install

npm install @x12i/multistep-connector

Optional peers at the app boundary:

• @x12i/memorix-generic-connector / @x12i/memorix-source-connector — wire into createInjectableMgcHostAdapter
• Explorer HTTP API for durable sessions

Quick start (pure stages)

import {
  createMultistepRuntime,
  createPipelineRegistry,
  type StageHandler,
} from "@x12i/multistep-connector";
import { MemorySessionStore } from "@x12i/multistep-connector/testing"; // tests only
import {
  ExplorerSessionStore,
  createExplorerClient,
  resolveExplorerConfig,
} from "@x12i/multistep-connector/explorer";

const registry = createPipelineRegistry({
  handlers: [
    {
      id: "normalize",
      kind: "pure",
      async execute(ctx) {
        return { outputs: { items: ctx.inputs.raw } };
      },
    } satisfies StageHandler,
  ],
  pipelines: [
    {
      pipelineId: "example",
      version: "1.0.0",
      inputPorts: [{ name: "raw" }],
      outputPorts: [{ name: "items" }],
      stages: [
        {
          stepId: "normalize",
          kind: "pure",
          handlerId: "normalize",
          inputPorts: [{ name: "raw" }],
          outputPorts: [{ name: "items" }],
        },
      ],
    },
  ],
});

// Production:
// const store = new ExplorerSessionStore(createExplorerClient(resolveExplorerConfig()));
// Tests only:
const store = new MemorySessionStore();

const runtime = createMultistepRuntime({ store, registry });
const session = await runtime.createSession({
  orgId: "org-1",
  pipelineId: "example",
  inputs: { raw: [{ id: 1 }] },
});
const done = await runtime.run(session.sessionId, "org-1");

Subpath exports

| Import | Contents | | --- | --- | | @x12i/multistep-connector | Runtime, contracts, registry, Explorer types, credentials, MGC adapter | | @x12i/multistep-connector/explorer | SessionStore, ExplorerSessionStore, collections | | @x12i/multistep-connector/credentials | Credential SPI (vendor / plane / dataplane) | | @x12i/multistep-connector/mgc | Injectable MGC host | | @x12i/multistep-connector/testing | Export/replay fixtures + test-only MemorySessionStore | | @x12i/multistep-connector/observability | Problem Details + redaction helpers |

Environment (Explorer)

MEMORIX_CONNECTORS_DATA_DB=memorix-connectors-data
EXPLORER_BASE_URL=https://explorer.example
EXPLORER_API_TOKEN=...

Optional for dev vendor credentials: MULTISTEP_ALLOW_ENV_CREDENTIALS=1 and MULTISTEP_VENDOR_<REF>=….

MGC host wiring

import { createInjectableMgcHostAdapter } from "@x12i/multistep-connector/mgc";
import { createPausableAcquisitionContext } from "@x12i/memorix-source-connector";

const mgc = createInjectableMgcHostAdapter({
  http: yourHttpPort,
  collectStream: yourCollectStream,
  resolveAuth: (ref) => credentials.resolve({ plane: "vendor", ref, purpose: "mgc-pull" }),
  // Optional: after auth is resolved, build an SDK pull context for the port.
  createPullContext: ({ session, authHeaders }) => createPausableAcquisitionContext({
    definition,
    orgId: session.orgId,
    pullId: session.sessionId,
    http: yourHttpPort,
    authHeaders,
  }),
  executeWorkflow: async (args) => {
    // Wire executePausableWorkflow — the void executeWorkflow wrapper throws on pause.
    // Port checkpoint is the MGC acquisition resume object (not CheckpointPatch);
    // the adapter maps it onto StepResult.acquisition.
    // If you omit createPullContext, call createPausableAcquisitionContext here
    // (the default thin { http, collectStream, resolveAuth } is not an SDK context).
    const pullContext = createPausableAcquisitionContext({
      definition,
      orgId: "org",
      pullId: "pull",
      http: yourHttpPort,
      authHeaders: args.authHeaders,
    });
    const result = await sourceRuntime.executePausableWorkflow(definition, pullContext, {
      signal: args.pauseSignal,
      resumeFrom: args.resumeFrom,
      resumeOutputs: args.resumeOutputs,
      authHeaders: args.authHeaders,
    });
    return result;
  },
});

const runtime = createMultistepRuntime({ store, registry, mgc, credentials });

Collections (memorix-connectors-data)

msc-sessions, msc-step-runs, msc-events, msc-artifacts, msc-checkpoints, msc-idempotency, msc-lineage — every document includes sessionId.

Discovery APIs (contracts only)

HTTP route DTOs live under discovery contracts (MSC-009). Hosts implement routes such as POST /api/multistep/sessions and POST …/run.

License

Proprietary — x12i.