1.2.0 • Published 2 weeks agoCLI

@exellix/jobs

Licence

exellix-license

Version

1.2.0

Deps

Size

710 kB

Vulns

Weekly

Summary Dependency Versions

@exellix/jobs

Jobs manager — the programmatic runtime for Exellix’s durable job queue: enqueue / claim / complete / fail, plus runWorker, source pollers, dependency wiring, retry-with-backoff, and lease recovery. Import this package from workers, scripts, and other services.

Package	Role
`@exellix/jobs` (this)	Jobs manager — queue logic, worker loop, CLI (`exellix-jobs`)
`@exellix/jobs-api`	Jobs HTTP API — Fastify REST + programmatic queries
`@exellix/jobs-ui`	Jobs dashboard — React operator SPA
`@exellix/jobs-db`	Jobs data tier — Mongo `JobRunStore` (only package that imports `mongodb`)

Single-graph execution comes from @exellix/graph-engine. This package never imports mongodb.

A JobDef says what to run; a JobRun is one queued execution of one graph on one item. "Task" stays reserved for graph task nodes and ai-tasks (@exellix/ai-tasks). Full design in temp/jobs/.

Install

npm install @exellix/jobs @exellix/jobs-db

Quick start

import { createJobRunStore } from '@exellix/jobs-db';
import { enqueue, runWorker, createWorkerDeps, createGraphEngineWorkerFns } from '@exellix/jobs';

const { store } = await createJobRunStore({ mongoUri: process.env.MONGO_URI });

// 1. produce job runs from items (one per graph in the job def)
await enqueue(store, [{ id: 'doc-1', data: { question: 'hi' } }], jobDef);

// 2. wire the graph engine and drain the queue
const { compile, executeGraph } = createGraphEngineWorkerFns({ graphLoader });
const deps = createWorkerDeps(store, 'worker-1', { compile, executeGraph });

await runWorker(deps, { concurrency: 4, pollIntervalMs: 60_000, store, signal });

For sources, a thin producer loop polls and enqueues:

import { runSourcePoller } from '@exellix/jobs';
runSourcePoller(store, jobDef, { signal });   // skips when jobDef.enabled === false

Memorix record tracking (EXLX-CRS-001)

Poll Memorix records, enqueue job runs, execute graphs, stamp _graphRuns, and write graph results per response.persistency:

cd jobs && npm run build
# Requires MONGO_URI, CATALOX_APP_ID, MEMORIX_* env (see graph-engine/.env.example)
exellix-jobs run --job-def-id=your-job-def-id

Commands: worker (consume only), poller (produce only), run (both).

Activix observability

exellix-jobs run / worker wire Activix graph/node lifecycle to Mongo (same as exellix-jobs-api queue-worker). Set ACTIVIX_STORAGE_MODE=database in .env — see jobs/.env.example and jobs-api/.env.example. The Live Console in jobs-ui reads activities from Mongo via jobs-api, not from any local playground/ folder.

`_memorixRef` on enqueue

Every Memorix-backed run needs _memorixRef on JobRun.input so the graph-run tracker can call @x12i/memorix-writer (markGraphRunStarted, writeGraphRunResult).

Source	How ref is attached
`createMemorixSource` poll	Full ref on item data
`enqueue()`	Auto-attaches from memorix-shaped item ids (`entity:…:contentType:recordId`) when absent
jobs-ui work factory	On-demand single/batch, continuous evaluate, `/jobs/enqueue`
`runWorkFactoryEvaluationTick`	Via centralized `enqueue()`

Helpers (exported):

import {
  buildMemorixRecordRef,
  parseMemorixItemId,
  itemDataWithMemorixRef,
  memorixItemId,
  createMemorixGraphRunTracker,
  createDynamicMemorixGraphRunTracker,
  resolveGraphRunContract,
  resolvePersistencyTarget,
} from '@exellix/jobs';

const ref = buildMemorixRecordRef('entity', 'assets', 'core', 'doc-1');
await enqueue(store, [{
  id: memorixItemId(ref),
  data: itemDataWithMemorixRef({ recordId: 'doc-1' }, ref),
}], jobDef);

The worker strips _memorixRef from graph runtime input; the tracker reads it from the stored JobRun.input.

Programmatic wiring:

import { createMemorixJobsRuntime, runMemorixJobs } from '@exellix/jobs';

await runMemorixJobs({ jobDefId: 'assets-scoped-job', mode: 'both', concurrency: 4 });

Live E2E validation:

npm run bind:memorix-catalox   # once: Catalox bindings for CATALOX_APP_ID
npm run seed:memorix-e2e
npm run check:memorix-e2e
npm run test:memorix-live

Mongo: replica set gives atomic result+stamp via @x12i/memorix-writer transactions. Standalone Mongo is supported — the worker retries with best-effort sequential writes. Set MEMORIX_GRAPH_RUN_TRANSACTIONS=0 to skip transactions entirely.

Graph-run persistency (MRX-FRS-002)

Graphs declare writeback intent in graph.response.persistency (Graphenix 2.7.3). On successful runs, createMemorixGraphRunTracker / createDynamicMemorixGraphRunTracker call writeGraphRunResult when the graph contract includes persistency and _memorixRef is present.

Export	Role
`resolveGraphRunContract(graphId, doc)`	Read `graph.response.persistency` from graph JSON (re-exported from `@exellix/graph-engine`)
`resolvePersistencyTarget(contract, ref)`	Map to `@x12i/memorix-descriptors` `PersistencyTarget` (`newRecord: true` → `{ generate: true }`)
`createMemorixGraphRunTracker`	Static contract list (CLI / memorix jobs runtime)
`createDynamicMemorixGraphRunTracker`	Load contract per run via `graphLoader` (jobs-ui worker)

See docs/handoff/graph-run-persistency-upstream.md.

Processing layer — linked context and associated enrichment

The processing layer resolves linked object definitions before graph dispatch. It has two modes that share the same linked-record engine (resolveLinkedRecordSet in context/linked-object-resolution.ts):

Mode	When	Output
Runtime context enrichment	Before a skill/graph run (worker, sync execute, Graph Studio)	Extends injected `jobMemory.context["linked-information"]` — not persisted to source records
Persistent associated enrichment	Preview / operational writeback (generic product code)	Smart-merges linked record `.data` into `associatedData`, `associatedInferred`, `associatedAnalysis`, or `associated<CustomName>` on source snapshots

@exellix/job-dispatcher does not read Memorix or resolve linked definitions. Hosts (@exellix/jobs, @exellix/jobs-api, Graph Studio BFF) resolve first, then pass runtime.context.jobMemory into dispatch.

Runtime linked-information

Configured via ContextSource[] on the job (linked + same-object sources). The worker and executeGraphJob call resolveContextEnrichment → mergeContextEnrichment:

import { resolveJobMemoryForDispatch, mergeContextEnrichment } from '@exellix/jobs';

const jobMemory = await resolveJobMemoryForDispatch(retrieval, runtimeInput, contextSources, {
  sourceEntity: 'assets',
  existingJobMemory: injectedJobMemory,
  recordId: 'asset-1',
});
// jobMemory.context['linked-information'] → [{ context: '...', data: [...] }]

Output rules (locked by tests):

Linked items contain only { context, data } — no resolution metadata in jobMemory.
Empty linked results do not create linked-information items or an empty context object.
Existing jobMemory.context from other providers is preserved.

Sync execute and Graph Studio

executeGraphJob accepts optional contextSources, sourceEntity, and recordId. When contextSources is set, options.retrieval is required; resolution runs before dispatchGraphRun:

import { executeGraphJob } from '@exellix/jobs';

await executeGraphJob(
  {
    graphId: 'cyber-analysis.assets.extended',
    input: assembledInput,
    contextSources: [
      {
        kind: 'linked',
        objectType: 'subnets',
        contentType: 'inferences',
        linkingProperty: 'data.subnetIp',
        targetLinkingProperty: 'data.subnetIp',
        matchContentType: 'snapshots',
        mandatory: true,
      },
    ],
    sourceEntity: 'assets',
    recordId: 'asset-1',
  },
  { graphLoader, retrieval },
);

buildContextProbeRun builds a synthetic JobRun for read-only probes (preview, coverage, pre-dispatch resolution).

Persistent associated enrichment (preview / write helpers)

Generic config (AssociatedEnrichmentConfig): source, linked, output, optional filter and failure. Preview resolves and smart-merges without writing:

import { previewAssociatedEnrichment, suggestAssociatedTargetProperty } from '@exellix/jobs';

suggestAssociatedTargetProperty('inferences'); // → 'associatedInferred'

const preview = await previewAssociatedEnrichment(retrieval, {
  recordId: 'asset-1',
  sourceInput: { recordId: 'asset-1', data: { subnetIp: '10.0.0.0/24' } },
  existingAssociatedValue: [{ riskLevel: 'medium' }],
  config: {
    source: { objectType: 'assets', contentType: 'snapshots' },
    linked: {
      objectType: 'subnets',
      contentType: 'inferences',
      linkingProperty: 'data.subnetIp',
      targetLinkingProperty: 'data.subnetIp',
      matchContentType: 'snapshots',
    },
    output: { mode: 'persist-to-source', targetProperty: 'associatedInferred', writeMode: 'smart-merge' },
  },
});
// preview.ok, preview.merge, preview.willWrite — or preview.failure for mandatory misses

Smart merge: preserve existing arrays, normalize single objects to arrays, append only canonically unique objects, skip empty writes. Mandatory failures are modeled as generic <objectType>-failed records (buildAssociatedEnrichmentFailure).

Preflight — input coverage

sampleGraphInputCoverage samples batch record ids and reports per-record assembly gaps plus per-context-source coverage (resolved counts, filter/mandatory failure reasons):

import { sampleGraphInputCoverage } from '@exellix/jobs';

const coverage = await sampleGraphInputCoverage({
  retrieval,
  graph,
  selection,
  recordIds,
  contextSources,
  sourceEntity: 'assets',
  sampleSize: 20,
});

HTTP equivalents live on @exellix/jobs-api (POST /api/graph/input-coverage, POST /api/context/preview, POST /api/associated-enrichment/preview).

Associated properties on snapshots (`associated*`)

Snapshot records can carry associated* root properties that hold linked or enriched data accumulated over time. Naming follows the linked content type:

Linked content type	Property on source snapshot
`snapshots`	`associatedData`
`inferences`	`associatedInferred`
`analysis`	`associatedAnalysis`
custom	`associated<CustomName>`

At execution time, assembly promotes snapshot root fields whose names start with associated out of runtime.input and onto direct jobMemory.associated* fields (for example jobMemory.associatedData or jobMemory.associatedRiskScores). promotedProperties can also opt explicit top-level fields into the same job-memory path. Skills should read enriched associated data from jobMemory, not from flat graph input.

Persistent associated enrichment (previewAssociatedEnrichment) resolves linked records and smart-merges their .data into the configured associated* array on the source snapshot (preview-only in product code; applying to real records is a separate operational step). Smart merge preserves existing arrays, normalizes single objects to arrays, appends only canonically unique objects, and skips empty writes.

Memorix integration (via graph-engine)

@exellix/jobs does not import Memorix directly. When job graphs use local skills such as scoped-data-reader or scoped-answer-writer, Memorix I/O happens inside @exellix/graph-engine at execute time.

Wire the worker the same way as above — createGraphEngineWorkerFns passes runtime options into graph-engine, which lazily initializes Memorix clients from process env:

Package	Version	Role
`@x12i/memorix-descriptors`	1.10.0	`PersistencyTarget`, `PersistencyLinkIntent`
`@x12i/memorix-retrieval`	1.14.0	`scoped-data-reader` reads, list/poll
`@x12i/memorix-writer`	1.3.0	`_graphRuns` stamp + MRX-FRS-002 `writeGraphRunResult`

Required host env (see graph-engine/.env.example):

MONGO_URI — job runs and Memorix stores
MEMORIX_ENTITIES_DB, MEMORIX_EVENTS_DB
CATALOX_APP_ID (and related Catalox config for descriptor lookup)
LLM keys when graphs include ai-tasks nodes

On shutdown, close the job store and release Memorix clients:

import { shutdownMemorixRuntime } from '@exellix/graph-engine';

signal.addEventListener('abort', async () => {
  await runWorkerPromise; // or abort runWorker via shared signal
  await store.close();
  await shutdownMemorixRuntime();
});

API

Function	Role
`enqueue(store, items, jobDef)`	items → `JobRun[]`; auto-attaches `_memorixRef` from memorix item ids
`buildMemorixRecordRef` / `parseMemorixItemId` / `itemDataWithMemorixRef`	Memorix ref helpers for producers
`createMemorixGraphRunTracker` / `createDynamicMemorixGraphRunTracker`	Post-run `_graphRuns` + persistency writeback
`resolveGraphRunContract` / `resolvePersistencyTarget`	Graph persistency → writer target
`claim(store, workerId, opts?)`	atomic claim → `JobRun \| null`
`complete(store, jobRunId, result)`	mark done; release dependents (merging upstream output)
`fail(store, jobRunId, error)`	retry with backoff while attempts remain, else terminal-fail + cascade
`releaseDependents` / `mergeUpstream`	dependency release and upstream-output-as-input
`sweepExpiredLeases(store, { leaseMs })`	reclaim job runs from dead workers
`runWorker(deps, opts)`	the only loop (concurrency, idle sleep, periodic sweep)
`drainWorker(deps, opts)`	run once until the queue is empty (tests / batch)
`createGraphEngineWorkerFns(options)`	wire `@exellix/graph-engine` compile + execute
`runQueueAdmissionLoop` / `promoteWaitingRuns`	app-wide admission — promote held runs when slots free up
`runWorkFactoryEvaluationTick`	continuous work poller (active work defs → enqueue eligible records)
`resolveJobMemoryForDispatch`	Host-facing pre-dispatch context resolution → `jobMemory` for dispatch
`resolveContextEnrichment` / `mergeContextEnrichment`	Runtime linked/same-object context → `linked-information` / `extendedInformation`
`resolveLinkedRecordSet`	Shared linked-record fetch, join, filter engine (runtime + associated modes)
`previewAssociatedEnrichment`	Preview persistent associated* smart-merge for one source record (no write)
`sampleGraphInputCoverage`	Batch preflight: assembly + per-context-source coverage sample
`executeGraphJob`	Sync graph run; optional `contextSources` + `retrieval` for pre-dispatch resolution

Parallel execution and capacity

Exellix runs many graph executions concurrently, but that does not mean one OS process per job. A typical deployment is one exellix-jobs Node process with an in-memory async pool; admission control decides how many runs may be active app-wide.

Two layers of concurrency

Layer	What it limits	Default / config
App admission	How many job runs are admitted (pending + running, claimable) across all work, batches, and on-demand runs	`maxConcurrentJobs` in Mongo `jobs_app_settings` — 10 (`DEFAULT_JOBS_APP_SETTINGS`). Edit in jobs-ui → Settings.
Worker pool	How many runs one worker process executes at once	`runWorker({ concurrency })`, `--concurrency`, or `JOBS_WORKER_CONCURRENCY` (falls back to `maxConcurrentJobs`)
Graph nodes	Parallel task nodes within one graph run	`WOREX_GRAPH_CONCURRENCY` env or graph metadata — 4 per runnable batch (graph-engine)

When admission is full, extra runs are held (availableAt sentinel) until a slot frees; the admission loop promotes them on queueAdmissionIntervalMs (default 15s).

enqueue → admission (maxConcurrentJobs) → claim → runOne → executeGraph
                                              ↑
                         promoteWaitingRuns when a run completes

Environment

Variable	Role
`JOBS_WORKER_CONCURRENCY`	Override worker pool size for this process
`JOBS_POLL_INTERVAL_MS`	Idle poll when no claimable runs
`JOBS_LEASE_MS`	Claim lease; expired leases are swept back to pending
`WOREX_GRAPH_CONCURRENCY`	Cap parallel nodes inside each graph (lower this when running many admitted jobs)

CLI example — 20 admitted jobs but only 2 parallel nodes per graph:

JOBS_WORKER_CONCURRENCY=20 WOREX_GRAPH_CONCURRENCY=2 exellix-jobs run --job-def-id=your-job-def-id

Sizing guidance

There is no hard upper cap on maxConcurrentJobs in code — capacity is bounded by the host and downstream services.

Target	Typical host	Notes
8–12	Dev laptop (8 cores, 16 GB RAM)	Comfortable default when Cursor, Mongo, and jobs-ui also run locally
10	Any	Shipped default — good balance for I/O-bound graphs (Mongo, LLM APIs)
20	16 GB+ RAM, dedicated worker	Workable if graphs are mostly waiting on network; watch RAM and Mongo connection pools
50–100+	Dedicated worker fleet (32–64 GB+ RAM)	Run multiple worker processes (distinct `WORKER_ID`s), each with modest `JOBS_WORKER_CONCURRENCY` (8–16); not one Node process on a dev machine

Bottlenecks to watch: Node heap (one process holding many in-flight graphs), MongoDB pool/contention, LLM provider rate limits, and lease expiry if the event loop stalls under load.

For high throughput, scale workers horizontally rather than pushing a single process to 100 concurrent runs.

What it adds over the Execution Matrix

Automatic retry with backoff — maxAttempts + availableAt.
Cross-item dependencies — a job run may depend on a job run from a different item.
Upstream-output-as-input in the queue layer — mergeUpstream, out of engine internals.

Scripts

npm run build      # tsc → dist/
npm test           # unit + live (live runs only when MONGO_URI is set)
npm run test:live  # live Mongo integration tests only

License

exellix-license

Keywords

exellix jobs queue worker