@semiont/jobs

Job queue, worker infrastructure, and annotation workers for Semiont.

Architecture Context

Workers run in a separate process and connect to the Knowledge System (KS) over HTTP/SSE using a SemiontSession (from @semiont/sdk) driven by a JobClaimAdapter. Workers receive job assignments via an SSE job:queued subscription, claim jobs atomically, and emit domain events back to the KS via session.client.transport.emit(...). The KS ingests these events onto its EventBus for SSE delivery to the frontend.

Installation

npm install @semiont/jobs

Dependencies:

• @semiont/core — Core types, SemiontProject, EventBus
• @semiont/sdk — SemiontSession, WorkerBus (worker process)
• @semiont/http-transport — HTTP transport, OpenAPI types
• @semiont/inference — InferenceClient for AI operations
• @semiont/content — Content storage URI derivation
• @semiont/event-sourcing — Annotation id generation
• @semiont/observability — Spans and job-outcome metrics

Quick Start

import { FsJobQueue, type PendingJob, type GenerationParams } from '@semiont/jobs';
import { EventBus, userId, resourceId, annotationId, jobId } from '@semiont/core';
import { SemiontProject } from '@semiont/core/node';

// Initialize — jobs are stored under project.jobsDir
const eventBus = new EventBus();
const project = new SemiontProject('/path/to/project');
const jobQueue = new FsJobQueue(project, logger, eventBus);
await jobQueue.initialize();

// Create a job
const job: PendingJob<GenerationParams> = {
  status: 'pending',
  metadata: {
    id: jobId('job-abc123'),
    type: 'generation',
    userId: userId('[email protected]'),
    userName: 'Jane Doe',
    userEmail: '[email protected]',
    userDomain: 'example.com',
    created: new Date().toISOString(),
    retryCount: 0,
    maxRetries: 3,
  },
  params: {
    referenceId: annotationId('ref-123'),
    sourceResourceId: resourceId('doc-456'),
    sourceResourceName: 'Source Document',
    annotation: { /* full W3C Annotation */ },
    title: 'Generated Article',
    prompt: 'Write about AI',
    language: 'en-US',
  },
};

await jobQueue.createJob(job);

Job Types

type JobType =
  | 'reference-annotation'     // Entity reference detection
  | 'generation'               // AI content generation
  | 'highlight-annotation'     // Key passage highlighting
  | 'assessment-annotation'    // Evaluative assessments
  | 'comment-annotation'       // Explanatory comments
  | 'tag-annotation'           // Structural role tagging

Job Metadata

All jobs share common metadata:

interface JobMetadata {
  id: JobId;
  type: JobType;
  userId: UserId;
  userName: string;       // Audit-only snapshot of the requesting user
  userEmail: string;      // Audit-only snapshot of the requesting user
  userDomain: string;     // Audit-only snapshot of the requesting user
  created: string;
  retryCount: number;
  maxRetries: number;
}

The userName, userEmail, and userDomain fields are an audit-only snapshot of the requesting user, persisted in the on-disk job file. Workers derive annotation creator attribution from userId via didToAgent().

Annotation Workers

The worker process (worker-main.ts → startWorkerProcess in worker-process.ts) claims jobs over the bus via a JobClaimAdapter and dispatches by jobType to a processor function. There are no per-type worker classes; each job type maps to one process*Job function:

| Job Type | Processor | |----------|-----------| | reference-annotation | processReferenceJob | | generation | processGenerationJob | | highlight-annotation | processHighlightJob | | assessment-annotation | processAssessmentJob | | comment-annotation | processCommentJob | | tag-annotation | processTagJob |

Detection logic lives in the AnnotationDetection class (src/workers/annotation-detection.ts); generation synthesis in generateResourceFromTopic() (src/workers/generation/resource-generation.ts). Processors never fetch content themselves — the worker process fetches it via session.client.browse.resourceContent(resourceId) and passes it in.

Workers emit bus events via session.client.transport.emit('mark:create' | 'job:start' | 'job:report-progress' | 'job:complete' | 'job:fail', payload) — the Stower actor in @semiont/make-meaning handles persistence to the event log, and the job command handlers mirror the same events into the queue files (completion, retry-on-failure with maxRetries, progress-as-heartbeat).

Adding a Job Type

Workers are not subclassed. To add a job type:

1. Add the new JobType and its params/result/progress types in src/types.ts.
2. Add a process*Job function in src/processors.ts that runs the inference and returns the annotations/result.
3. Dispatch the new jobType to that processor in handleJobInner() in src/worker-process.ts.

Processors are transport-agnostic: they take content, an InferenceClient, the job params, the user id, the generator (W3C SoftwareAgent), and an onProgress callback, and return annotations plus a result. The worker process handles claiming, content fetching, and lifecycle event emission.

Discriminated Unions

Jobs use TypeScript discriminated unions for type safety:

function handleJob(job: AnyJob) {
  if (job.status === 'running') {
    console.log(job.progress);    // Available
    // console.log(job.result);   // Compile error
  }
  if (job.status === 'complete') {
    console.log(job.result);      // Available
    // console.log(job.progress); // Compile error
  }
}

Storage Format

Jobs are stored as individual JSON files organized by status:

{project.jobsDir}/
  pending/job-abc123.json
  running/job-def456.json
  complete/job-ghi789.json
  failed/job-jkl012.json
  cancelled/job-mno345.json

Documentation

• Job Queue Guide — JobQueue API and job management
• Workers Guide — Building custom workers
• Job Types Guide — All job type definitions
• Type System Guide — Discriminated unions and type safety
• Configuration Guide — Setup and options
• API Reference — Complete API reference

License

Apache-2.0

Related Packages

• @semiont/core — Domain types, SemiontProject, EventBus
• @semiont/sdk — SemiontSession, WorkerBus
• @semiont/http-transport — HTTP transport, OpenAPI types
• @semiont/inference — AI inference client
• @semiont/make-meaning — Actor model, Knowledge Base, service orchestration