@netscript/plugin-workers

The NetScript background-workers plugin: background jobs, task execution, workflow orchestration, a Workers API service, CLI commands, scaffolding, durable streams, and Aspire process wiring. This page is written against the package public surface reported by deno doc. For the full index of packages and plugins return to the reference overview.

The deployable plugin (@netscript/plugin-workers) binds the host plugin system to the reusable worker primitives that live in @netscript/plugin-workers-core. The plugin root entrypoint exposes the plugin manifest and a pure inspection helper; framework integrations are carried by sub-path exports documented below.

Plugin manifest (@netscript/plugin-workers)

The root entrypoint exposes the plugin manifest consumed by the NetScript host. Shared manifest inspection is provided by inspectPlugin from @netscript/plugin.

Sub-path exports

The following entrypoints are published alongside the plugin root export. Each is generated from its own deno doc surface.

@netscript/plugin-workers/aspire

@netscript/plugin-workers/cli

@netscript/plugin-workers/contracts

Versioned oRPC service contracts plus the request/response schemas and types used to generate worker service clients.

Trigger procedures (RPC routes)

workersContract exposes two enqueue procedures. A generated typed client (createServiceClient<typeof workersContract>({ contract: workersContract, serviceName: 'workers-api', routerName: 'workers' })) reaches them over /api/rpc/*; the same routes answer REST callers under /api/v1/workers/.... For each, the {id} path segment is the single source of truth for the target — a request-body id is optional and ignored when sent, and a missing id short-circuits to a typed VALIDATION_ERROR (HTTP 422) via the centralized validationFailed contract helper before any KV write.

See the background-jobs capability page for the typed-client walkthrough.

@netscript/plugin-workers/scaffolding

@netscript/plugin-workers/services

@netscript/plugin-workers/streams

Browser-facing StreamDB factory and the entity schemas mirrored from worker execution state.

@netscript/plugin-workers/streams/server

Server-side wiring that mirrors execution-state changes into the durable stream producer.

@netscript/plugin-workers/worker

The long-running runtime processes: the Worker job/task consumer and the cron Scheduler.

Core package

The separately published @netscript/plugin-workers-core page is canonical for worker definition builders, runtime composition, registries, executors, streams, configuration, and testing exports. This page stays focused on the deployable plugin's manifest and integration entrypoints. The workflow example below uses core APIs intentionally; exhaustive core entrypoint and symbol tables live only on its reference page.

Resumable Workflows and Idempotent Resume

Durable workflows are defined using defineWorkflow and executed step-by-step using WorkflowExecutor (both from @netscript/plugin-workers-core/workflow). In the event of a step failure, operators can resume the workflow execution from the failed step while skipping already completed steps.

import { assertEquals } from "@std/assert";
import {
  defineWorkflow,
  WorkflowExecutor,
  MemoryWorkflowStateStore,
} from "@netscript/plugin-workers-core/workflow";
import type { TaskId } from "@netscript/plugin-workers-core/workflow";

Deno.test("WorkflowExecutor infrastructure-free testing and idempotent resume", async () => {
  const workflow = defineWorkflow("data-processing-pipeline")
    .taskStep("download", { taskId: "download-file" as TaskId })
    .taskStep("transform", { taskId: "transform-data" as TaskId })
    .build();

  const stateStore = new MemoryWorkflowStateStore();
  const executedSteps: string[] = [];
  let shouldFailTransform = true;

  // 1. Create a WorkflowExecutor with mock handlers for infrastructure-free testing
  const executor = new WorkflowExecutor({
    stateStore,
    runTaskStep: (step, _state) => {
      const taskId = step.taskId;
      executedSteps.push(taskId!);
      if (taskId === "transform-data" && shouldFailTransform) {
        return Promise.reject(new Error("Transform step failed!"));
      }
      return Promise.resolve({ success: true });
    },
  });

  // 2. Execute the workflow (it will fail on the transform step)
  const firstRunState = await executor.execute(workflow, {
    executionId: "pipeline-run-1",
  });

  assertEquals(firstRunState.status, "failed");
  assertEquals(firstRunState.currentStepIndex, 2); // Failed step is recorded, advancing cursor
  assertEquals(executedSteps, ["download-file", "transform-data"]);

  // 3. Reset the failure flag
  shouldFailTransform = false;
  executedSteps.length = 0; // Clear execution log

  // 4. Simulate operator intervention: reset failed step to replay/resume it
  const savedState = await stateStore.findState(workflow.id, "pipeline-run-1");
  if (savedState) {
    const resultsCopy = { ...savedState.results };
    delete resultsCopy.transform; // Remove the failed step result

    await stateStore.saveState({
      ...savedState,
      status: "pending",
      currentStepIndex: 1, // Set cursor back to the transform step index
      results: resultsCopy,
    });
  }

  // 5. Resume the workflow using the executor
  const resumedState = await executor.execute(workflow, {
    executionId: "pipeline-run-1",
  });

  assertEquals(resumedState.status, "completed");
  assertEquals(resumedState.currentStepIndex, 2); // Completed all 2 steps
  
  // Idempotency check: "download" step is skipped on resume, only "transform" is run
  assertEquals(executedSteps, ["transform-data"]);
});

Back to the reference overview.