@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_IDs), 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