A polyglot transform task

In Chapter 2 you imported a SAP export as-is. But the SAP export is not a Dynamics import: the legacy system writes material_no where Dynamics wants sku, description where Dynamics wants name, and — the one that really hurts — prices as integer cents where Dynamics wants decimals. Load a legacy row into Dynamics untransformed and every price in the new system is wrong by a factor of one hundred. The pipeline needs a transform stage, and in NetScript that stage is a task: a standalone script defined with a builder, spawned as a subprocess, its result captured. In this chapter you build one and run it, using the deno runtime — the one task runtime NetScript sandboxes.

1. 1 · Scaffold
2. 2 · Import job
3. 3 · Polyglot transform
4. 4 · Queue & cron
5. 5 · Deploy

What you will build

By the end of this chapter you will have a runnable normalize-sap task: a transform script that reads the SAP export you dropped in Chapter 2, rewrites its legacy columns into Dynamics' shape, and writes the normalized file to a staging folder — executed through the workers task executor as a sandboxed subprocess whose filesystem access you granted explicitly. You will watch its output stream into your terminal, read its structured JSON result, and cat the normalized file it produced. You will also see how the same builder chain targets Python or shell when a transform belongs in another language — as a clearly-marked forward step for your own host, since those runtimes are not sandboxed and need their interpreter installed.

Before you begin

You need the my-erp/ workspace from Chapter 2 with the workers plugin installed and the SAP export still on disk from that chapter's file drop:

netscript plugin list
cat .data/incoming/products/products_2024.csv

Expected: workers appears in the plugin list, and the CSV prints the SAP legacy shape — material_no,description,price_cents and two rows. If the file is missing, re-create it exactly as in Chapter 2's verify step. Aspire does not need to be running for this chapter — the task executor runs the transform directly.

Step 1 — Write the transform script

A subprocess task is a contract about two streams. Input goes in as argv and environment variables — never stdin. The result comes back as exactly one JSON object printed as the last line of stdout; everything else on stdout/stderr is captured as logs. Write the transform as a plain Deno script honoring that contract:

// plugins/workers/scripts/normalize-sap.ts
// Legacy SAP export rows in -> Microsoft Dynamics-shaped rows out. Runs as a sandboxed subprocess.

// Input arrives as argv + env (NOT stdin).
const input = Deno.args[Deno.args.indexOf('--input') + 1];
const outDir = Deno.env.get('STAGING_DIR') ?? '.data/staging';

const raw = await Deno.readTextFile(input);
const lines = raw.trim().split('\n').filter((line) => line.trim().length > 0);
const headers = lines[0].split(',').map((h) => h.trim().toLowerCase());
const col = (name: string) => headers.indexOf(name);

// legacy -> target: material_no -> sku, description -> name, price_cents -> price (decimal).
const out: string[] = ['sku,name,price'];
let skipped = 0;
for (const line of lines.slice(1)) {
  const values = line.split(',').map((v) => v.trim());
  const sku = values[col('material_no')] ?? '';
  const name = values[col('description')] ?? '';
  const cents = Number.parseInt(values[col('price_cents')] ?? '', 10);
  if (sku === '' || Number.isNaN(cents)) {
    skipped++;
    continue;
  }
  out.push(`${sku},${name},${(cents / 100).toFixed(2)}`);
}

await Deno.mkdir(outDir, { recursive: true });
const fileName = input.split('/').pop() ?? 'export.csv';
const output = `${outDir}/${fileName.replace(/\.csv$/, '')}.normalized.csv`;
await Deno.writeTextFile(output, out.join('\n') + '\n');

// Diagnostics go to stderr; the RESULT is the last stdout line and must be a
// single JSON OBJECT (not an array) to populate result.result.
console.error(`normalize-sap: ${lines.length - 1} rows in, ${out.length - 1} written`);
console.log(JSON.stringify({ input, output, read: lines.length - 1, written: out.length - 1, skipped }));

Nothing here imports NetScript — that is the point. The script is an ordinary program with a narrow I/O contract, which is what lets the same execution model run TypeScript today and Python tomorrow.

Step 2 — Define the task, permissions included

Now wrap the script in a task definition. defineTask(id) from @netscript/plugin-workers-core/builders returns a typestate builder: .runtime(type) selects the runtime (default 'deno'), .entrypoint(path) points at the script and unlocks .build(), and input crosses as .args(...) plus .env({...}). For a deno task, .permissions({...}) is the sandbox — each key compiles directly into an --allow-* flag on the spawned deno run command line:

// plugins/workers/tasks/normalize-sap.ts
import { defineTask } from '@netscript/plugin-workers-core/builders';

export const normalizeSap = defineTask('normalize-sap')
  .runtime('deno') // the default — and the only sandboxed runtime
  .entrypoint('./plugins/workers/scripts/normalize-sap.ts')
  .args('--input', '.data/incoming/products/products_2024.csv')
  .env({ STAGING_DIR: '.data/staging' })
  .permissions({
    read: ['.data'],           // -> --allow-read=.data
    write: ['.data/staging'],  // -> --allow-write=.data/staging
    env: ['STAGING_DIR'],      // -> --allow-env=STAGING_DIR
  })
  .timeout(30_000) // ms; defaults to 300_000
  .build();

export default normalizeSap;

Read the permission set as a statement about the transform: it may read the incoming drop folder, write only to staging, and see one environment variable — nothing else. If the script ever tries to phone home or touch a file outside those grants, the Deno sandbox refuses at the subprocess boundary, not in your code review.

Step 3 — Run it through the executor

createDefaultTaskExecutor() from @netscript/plugin-workers-core/executor builds the multi-runtime executor wired with every built-in runtime adapter. executor.execute(task) resolves the adapter for the task's runtime, spawns the subprocess, streams its output, and returns one TaskResult. Write a small runner:

// plugins/workers/run-normalize.ts
import { createDefaultTaskExecutor } from '@netscript/plugin-workers-core/executor';
import { normalizeSap } from './tasks/normalize-sap.ts';

const executor = createDefaultTaskExecutor();

const result = await executor.execute(normalizeSap, {
  onStdout: (line) => console.log('[normalize]', line),
  onStderr: (line) => console.warn('[normalize:err]', line),
});

if (result.success) {
  // result.result is the parsed JSON object from the LAST stdout line, or null.
  console.log('normalized', result.result, `in ${result.duration}ms`);
} else {
  // status is 'failed' | 'timeout' | 'cancelled'; exitCode is -1 when the process never ran.
  console.error('task failed', result.status, result.exitCode, result.error);
  Deno.exit(1);
}

Run it from the workspace root (the task's relative paths — entrypoint, input, staging — resolve from where you launch the runner):

deno run -A plugins/workers/run-normalize.ts

The runner itself is trusted host code, so -A is fine here — the sandbox that matters is the subprocess: the executor spawns deno run --allow-read=.data --allow-write=.data/staging --allow-env=STAGING_DIR … with exactly the flags your permission set compiled to. You should see:

[normalize:err] normalize-sap: 2 rows in, 2 written
[normalize] {"input":".data/incoming/products/products_2024.csv","output":".data/staging/products_2024.normalized.csv","read":2,"written":2,"skipped":0}
normalized {
  input: ".data/incoming/products/products_2024.csv",
  output: ".data/staging/products_2024.normalized.csv",
  read: 2,
  written: 2,
  skipped: 0
} in 187ms

(Your duration will differ.) Now read the file the task produced:

cat .data/staging/products_2024.normalized.csv

sku,name,price
WID-1,Widget,9.99
GAD-2,Gadget,19.99

The cents are decimals, the legacy columns are Dynamics' names, and the off-by-100 price bug never gets a chance to exist. This is the transform stage of the pipeline: the import job stages the raw SAP rows, normalize-sap rewrites them for Dynamics, and a follow-up job would upsert the staged file. Tasks run through the same workers runtime as jobs and propagate W3C trace context (TRACEPARENT/TRACESTATE) into the subprocess, so a cross-runtime span still stitches together in the Aspire dashboard.

Step 4 — The same chain in another language

normalize-sap is TypeScript because a column rename needs nothing more. But some transforms live more naturally elsewhere — a pandas dedupe across historical SAP exports, a shell pipeline through jq, a .NET routine you already own. The builder chain is identical; only the .runtime(...) argument and the entrypoint change:

// The Python variant of the same stage — a forward step for your own host.
export const dedupeSap = defineTask('dedupe-sap')
  .runtime('python')
  .entrypoint('./plugins/workers/scripts/dedupe_sap.py')
  .args('--input', '.data/staging')
  .timeout(120_000)
  .build();

// Spawns: python3 -u ./plugins/workers/scripts/dedupe_sap.py --input .data/staging

The process contract is unchanged — argv + env in, one JSON object on the last stdout line out (the Python runtime runs python3 -u, unbuffered, for exactly that reason). Two things do change, and they are why this step is a read-now, run-on-your-own-host capability rather than part of this chapter's checkpoint:

Seven runtime types ship today — the literal members of the TASK_TYPES constant in @netscript/plugin-workers-core:

Verify your progress

Confirm the transform ran and the workspace still type-checks:

cat .data/staging/products_2024.normalized.csv   # sku,name,price + 2 decimal-priced rows
deno task check                                  # clean

• [ ] deno run -A plugins/workers/run-normalize.ts exits 0 and prints normalized { … written: 2 … }.
• [ ] .data/staging/products_2024.normalized.csv exists with headers sku,name,price and prices 9.99 / 19.99.
• [ ] The [normalize:err] diagnostic line and the JSON result line both appeared — logs and result travel on separate streams.
• [ ] deno task check is clean.

What you built

A runnable transform stage for the SAP→Dynamics pipeline: a plain-Deno script honoring the argv/env-in

• JSON-out process contract, a defineTask definition whose .permissions(...) compile into real --allow-* flags on the spawned subprocess, and an executor run you observed end to end — plus the shape of the same stage in Python, clearly marked with the two rules that govern non-Deno runtimes (no sandbox, interpreter must exist). Next, the pipeline learns to absorb bursts and run on a schedule.

Where to go deeper

• Run the Python/shell variants for real → Run a polyglot task — the hands-on recipe: define, write the script, pin the interpreter, read the result.
• The capability → Polyglot tasks — the WHY: what a task is, the subprocess seam, the full TaskResult shape.
• Tune the runtime → Tune the worker runtime — concurrency, the permission presets, and the per-task timeout/retry knobs.