diff --git a/.llm/runs/release-0.0.6-features--orchestration/slices/worklog-1398.md b/.llm/runs/release-0.0.6-features--orchestration/slices/worklog-1398.md new file mode 100644 index 0000000000..04845d0b1c --- /dev/null +++ b/.llm/runs/release-0.0.6-features--orchestration/slices/worklog-1398.md @@ -0,0 +1,163 @@ +# Worklog — #1398 publish job executions to the durable stream + +Implementation lane: Codex · GPT-5.6 Sol · medium (`normal_implementation`). +Branch: `fix/1398-publish-job-executions-to-durable-stream`. +Baseline: `origin/main@01aa12b67`. + +The approved `plan.md`, `plan-eval.md`, and `slices/research-1398.md` were read from the local +orchestration branch `chore/release-0.0.6-features-orchestration` because the fresh leaf branch was +created at the shared baseline before those orchestration-only commits existed. The locked plan was +not re-decided. + +## Design + +- **Public surface:** no new export and no `WorkerExecutionZodSchema` change. The existing + `WorkerExecutionRecord` type acknowledges the already-stored optional W3C trace headers. +- **Runtime state and identity:** every created, updated, and deleted execution mutation retains its + execution id, correlation id, `traceparent`, and optional `tracestate`. +- **Ports:** `WorkersStreamProducer` remains the stream port. The core mutation hook uses + `@netscript/telemetry/context` to make the stored W3C context ambient while that port publishes. +- **Plugin wiring:** the thin workers plugin installs the core-owned mutation hook only in the + worker-owning background entrypoints (`startWorkerProcess` and `startCombinedProcess`). The + scheduler does not own or mutate execution state. +- **Commit slices:** S1 installs and tests the hook/context join; S2 un-defers the two OTEL gates and + updates both registry tests; S3 records the one-pass live `scaffold.runtime` result. +- **Deferred scope:** stream rejection reasons (#1405), undeclared stream-core imports, and worker + execution schema changes remain out of scope. +- **Contributor path:** execution stream policy lives in + `packages/plugin-workers-core/src/streams/producer.ts`; process wiring lives in + `plugins/workers/bin/runtime.ts`; the adjacent stream and runtime tests are the regression map. + +## S1 — mutation hook installation and trace-context join + +Changed: + +- `packages/plugin-workers-core/src/streams/producer.ts` wraps every mutation publication in the + execution's extracted W3C context. +- `plugins/workers/bin/runtime.ts` installs the hook in worker-only and combined background + processes; scheduler-only remains unchanged because its execution-state surface is reserved and + has no mutation methods. +- `packages/plugin-workers-core/tests/streams/workers-streams_test.ts` adds stored-context and + pre-span trace-id guards. +- `plugins/workers/tests/runtime/background-stream-hook_test.ts` spies the real combined entrypoint's + `KvExecutionState.setMutationHook` call while suppressing worker/scheduler loops. + +Gate results before commit: + +| Gate | Result | +| --- | --- | +| workers scoped check | PASS — 101 files, 0 findings | +| plugin-workers-core scoped check | PASS — 111 files, 0 findings | +| workers scoped lint | PASS — 101 files, 0 findings | +| workers scoped format | PASS — 101 files, 0 findings | +| plugin-workers-core package test | PASS — 27 passed, 0 failed | +| combined-runtime installation test | PASS — 1 passed, 0 failed | +| `quality:gate` | PASS — quality scan clean; doctrine `FAIL=0` on touched roots | +| explicit plugin-workers-core quality scan | PASS — 0 findings, 0 allowances | + +Negative guard evidence: + +- Removing only D3 made both trace-context tests fail (exit 1): ambient trace id was retained for + the override case and a new random trace id appeared for the pre-span case. +- Removing only the combined-process hook installation made the installation test fail (exit 1): + actual installation count `0`, expected `1`. + +Reconcile note: issue #1398 is open with milestone `0.0.6` and already carries `type:fix`, +`area:plugins`, `area:telemetry`, `priority:p1`, and exactly one lifecycle label, `status:impl`. + +## S2 — restore the two OTEL runtime gates + +Changed in one slice, per PLAN-EVAL F2: + +- Added `behavior.otel.stream-consumer` and `behavior.otel.traces` to `RUNTIME_GATES`, which also + carries them into the SQLite runtime tier. +- Emptied `SCAFFOLD_RUNTIME_DEFERRED_GATES` while preserving the shared explicit deferral surface. +- Flipped the main runtime-suite presence assertions to `true`. +- Rewrote the exact-deferral test to assert both runtime tiers have no #1398 deferral and execute + both gates. + +First full-package check: RED. An empty `as const satisfies readonly DeferredGate[]` inferred the +constant's element as `never`, so existing runner-test fixture code could not read `issue` or +`reason`. Fixed by declaring the empty value against the explicit `readonly DeferredGate[]` +contract; no runtime behavior or assertion was relaxed. + +Second full-package test: RED with one in-scope and one unrelated failure. The in-scope failure was +a third stale #1398 pin in `suite-runner_test.ts`, which still expected two deferred skipped steps; +it now asserts zero deferrals, steps, skipped summary entries, and deferred reporter events. This +test was not named in the approved plan or PLAN-EVAL F2 and is recorded as minor factual drift. The +unrelated `quickstart-command-drift_test.ts` failure came from the package task changing cwd while +the test reads `docs/site/quickstart.vto` relative to the repository root; verify it from root and +do not absorb that task/path defect into #1398. + +Final S2 evidence: + +| Gate | Result | +| --- | --- | +| suite-runner + registry focused tests | PASS — 27 passed, 0 failed | +| CLI E2E test tree from repository root | PASS — 152 passed, 0 failed | +| CLI E2E scoped check | PASS — 163 files, 0 findings | +| CLI E2E scoped lint | PASS — 163 files, 0 findings | +| CLI E2E scoped format | PASS — 163 files, 0 findings | +| explicit CLI E2E quality scan | PASS — 0 findings, 0 allowances | + +Negative guard evidence: temporarily removing both OTEL gates from `RUNTIME_GATES` made the main +runtime presence test and the both-tier execution test fail (exit 1, actual `false`, expected +`true`). + +## Final implementation gates + +All required non-live gates passed on the S1+S2 head: + +| Gate | Result | +| --- | --- | +| workers scoped check | PASS — 101 files, 0 findings | +| plugin-workers-core scoped check | PASS — 111 files, 0 findings | +| workers scoped lint | PASS — 101 files, 0 findings | +| workers scoped format | PASS — 101 files, 0 findings | +| plugin-workers-core package test | PASS — 27 passed, 0 failed | +| `quality:gate` | PASS — quality scan clean; doctrine `FAIL=0` on touched roots | +| explicit plugin-workers-core quality scan | PASS — 0 findings, 0 allowances | +| explicit CLI E2E quality scan | PASS — 0 findings, 0 allowances | + +The configured quality scan covers `packages/cli/src` and `plugins`, but not +`packages/plugin-workers-core` or `packages/cli/e2e`; the two explicit scans above close that +coverage gap. No lockfile change, schema change, public export, lint suppression, `any`, unsafe +double cast, or `@ts-ignore` was added. + +## S3 — one-pass live runtime evidence + +Serialization was confirmed before launch: `/tmp/netscript-e2e-scaffold-runtime.lease` was absent, +no competing `e2e:cli`/`scaffold.runtime` process was running, and Docker reported zero containers. + +The mandated single invocation was run once and was not retried: + +```text +$ deno task e2e:cli run scaffold.runtime --cleanup --format pretty +> runtime.flow-b-fixture: Wire real Flow-B callback fixture + FAILED 10777ms + Command exited 1; expected 0. + error: Uncaught (in promise) Error: netscript generate plugins failed: Error: fetch failed +> cleanup.aspire-stop: Stop generated Aspire AppHost + PASSED 81ms +Summary: passed=33 failed=1 skipped=0 +raw exit code: 1 +``` + +This failure occurred before Aspire launch and before `behavior.otel.stream-consumer` or +`behavior.otel.traces`, so neither restored gate received a live verdict. The gate therefore does +not provide #1398's required end-to-end acceptance evidence. + +Diagnostics followed the required order: + +- Plugin doctor against the preserved generated fixture reported every configured plugin healthy; + its only warning was that no AppHost was running, expected because the failure preceded launch. +- Aspire OTEL logs/traces were unavailable because the AppHost never started. +- The failure is not the known `Missing plugin reference "streams"` residual; that message did not + occur, and generation failed earlier with Deno's generic network-resolution `fetch failed`. +- A focused rerun of only the failed `generate plugins` command against the preserved fixture then + passed in 0.9 seconds and wrote all three registries. This makes the observed failure consistent + with a transient dependency/registry fetch, but the nested Deno error did not expose a URL, so + the exact endpoint could not be verified. This diagnostic does not alter the authoritative red + E2E verdict and was not a second suite run. +- `agentic:leak-check` reported Aspire and Docker probes `ok` with no survivors; Docker also showed + zero running containers after cleanup. diff --git a/packages/cli/e2e/suites/scaffold/capability-suites.ts b/packages/cli/e2e/suites/scaffold/capability-suites.ts index d33c8f84e7..427e346a36 100644 --- a/packages/cli/e2e/suites/scaffold/capability-suites.ts +++ b/packages/cli/e2e/suites/scaffold/capability-suites.ts @@ -20,19 +20,8 @@ export interface ScaffoldCapabilitySuite { readonly defaults?: Partial; } -/** Runtime gates deferred until workers-combined publishes execution mutations (#1398). */ -export const SCAFFOLD_RUNTIME_DEFERRED_GATES = [ - { - id: GATE.BEHAVIOR_OTEL_STREAM_CONSUMER, - issue: '#1398', - reason: 'workers-combined does not install the stream mutation hook', - }, - { - id: GATE.BEHAVIOR_OTEL_TRACES, - issue: '#1398', - reason: 'TC-14 requires the deferred Flow-B stream-consumer record', - }, -] as const satisfies readonly DeferredGate[]; +/** Runtime gates with explicitly accepted temporary deferrals. */ +export const SCAFFOLD_RUNTIME_DEFERRED_GATES: readonly DeferredGate[] = []; const SERVICE_GATES = [ GATE.PREFLIGHT_DENO, @@ -143,6 +132,8 @@ const RUNTIME_GATES = [ GATE.BEHAVIOR_MCP_WIDGET_ROUNDTRIP, GATE.BEHAVIOR_PLUGINS_HEALTH, GATE.BEHAVIOR_OTEL_WEBHOOK, + GATE.BEHAVIOR_OTEL_STREAM_CONSUMER, + GATE.BEHAVIOR_OTEL_TRACES, GATE.BEHAVIOR_STREAMS_PRODUCER_RECONNECT, GATE.BEHAVIOR_OTEL_TASK_TRACES, GATE.CLEANUP_ASPIRE_STOP, @@ -206,14 +197,12 @@ export const scaffoldCapabilitySuites: readonly ScaffoldCapabilitySuite[] = [ id: SCAFFOLD.RUNTIME, title: SCAFFOLD_TITLE.RUNTIME, gates: RUNTIME_GATES, - // Explicit release deferral: #1398 must restore both gates to RUNTIME_GATES. deferredGates: SCAFFOLD_RUNTIME_DEFERRED_GATES, }, { id: SCAFFOLD.RUNTIME_SQLITE, title: SCAFFOLD_TITLE.RUNTIME_SQLITE, gates: RUNTIME_SQLITE_GATES, - // Keep both runtime tiers honest about the same #1398 coverage gap. deferredGates: SCAFFOLD_RUNTIME_DEFERRED_GATES, defaults: { database: DATABASE.SQLITE, cache: false }, }, diff --git a/packages/cli/e2e/tests/application/runner/suite-runner_test.ts b/packages/cli/e2e/tests/application/runner/suite-runner_test.ts index 24537eb95b..b8403ed66d 100644 --- a/packages/cli/e2e/tests/application/runner/suite-runner_test.ts +++ b/packages/cli/e2e/tests/application/runner/suite-runner_test.ts @@ -29,7 +29,7 @@ import { scaffoldCapabilitySuites, } from '../../../suites/scaffold/capability-suites.ts'; -Deno.test('suite runner reports every #1398 deferral as an explicit skipped step', async () => { +Deno.test('suite runner reports no skipped steps after the #1398 deferrals are removed', async () => { const runtime = createScaffoldRuntimeSuite({ repoRoot: '.', projectName: 'runner-deferral-test', @@ -47,29 +47,14 @@ Deno.test('suite runner reports every #1398 deferral as an explicit skipped step suiteLeaseManager: new RecordingSuiteLeaseManager(), }).run(suite, { suiteId: suite.id, options: suite.defaultOptions }); - assertEquals(report.steps.map((step) => step.id), [ - GATE.BEHAVIOR_OTEL_STREAM_CONSUMER, - GATE.BEHAVIOR_OTEL_TRACES, - ]); - assertEquals(report.steps.every((step) => step.verdict === 'skipped' && !step.critical), true); - assertEquals(report.summary, { passed: 0, failed: 0, skipped: 2 }); - assertEquals( - report.steps.map((step) => step.evidence[0]), - SCAFFOLD_RUNTIME_DEFERRED_GATES.map((deferred) => ({ - kind: 'summary' as const, - label: 'owned suite deferral', - data: { - status: 'deferred', - issue: deferred.issue, - reason: deferred.reason, - }, - })), - ); + assertEquals(SCAFFOLD_RUNTIME_DEFERRED_GATES, []); + assertEquals(report.steps, []); + assertEquals(report.summary, { passed: 0, failed: 0, skipped: 0 }); assertEquals( reporter.events .filter((event) => event.type === 'gate-start') - .map((event) => event.title.startsWith('DEFERRED #1398:')), - [true, true], + .map((event) => event.title), + [], ); }); diff --git a/packages/cli/e2e/tests/presentation/suite-registry_test.ts b/packages/cli/e2e/tests/presentation/suite-registry_test.ts index fdde54eef9..a2c517c62a 100644 --- a/packages/cli/e2e/tests/presentation/suite-registry_test.ts +++ b/packages/cli/e2e/tests/presentation/suite-registry_test.ts @@ -209,35 +209,19 @@ Deno.test('runtime suite includes full scaffold, database, runtime, and behavior runtime.gates.some((gate) => gate.id === GATE.BEHAVIOR_STREAMS_PRODUCER_RECONNECT), true, ); - assertEquals(runtime.gates.some((gate) => gate.id === GATE.BEHAVIOR_OTEL_STREAM_CONSUMER), false); - assertEquals(runtime.gates.some((gate) => gate.id === GATE.BEHAVIOR_OTEL_TRACES), false); + assertEquals(runtime.gates.some((gate) => gate.id === GATE.BEHAVIOR_OTEL_STREAM_CONSUMER), true); + assertEquals(runtime.gates.some((gate) => gate.id === GATE.BEHAVIOR_OTEL_TRACES), true); assertEquals(runtime.gates.some((gate) => gate.id === GATE.BEHAVIOR_OTEL_TASK_TRACES), true); }); -Deno.test('runtime suites pin the exact #1398 OTEL deferral without widening it', () => { - assertEquals(SCAFFOLD_RUNTIME_DEFERRED_GATES, [ - { - id: GATE.BEHAVIOR_OTEL_STREAM_CONSUMER, - issue: '#1398', - reason: 'workers-combined does not install the stream mutation hook', - }, - { - id: GATE.BEHAVIOR_OTEL_TRACES, - issue: '#1398', - reason: 'TC-14 requires the deferred Flow-B stream-consumer record', - }, - ]); +Deno.test('runtime suites execute the formerly deferred #1398 OTEL gates', () => { + assertEquals(SCAFFOLD_RUNTIME_DEFERRED_GATES, []); for (const suiteId of [SCAFFOLD.RUNTIME, SCAFFOLD.RUNTIME_SQLITE]) { const suite = resolveSuite(suiteId); assertEquals(suite.deferredGates, SCAFFOLD_RUNTIME_DEFERRED_GATES, suiteId); - assertEquals( - suite.gates.some((gate) => - SCAFFOLD_RUNTIME_DEFERRED_GATES.some((deferred) => deferred.id === gate.id) - ), - false, - `${suiteId} must not execute a deferred gate`, - ); + assertEquals(suite.gates.some((gate) => gate.id === GATE.BEHAVIOR_OTEL_STREAM_CONSUMER), true); + assertEquals(suite.gates.some((gate) => gate.id === GATE.BEHAVIOR_OTEL_TRACES), true); } }); diff --git a/packages/plugin-workers-core/src/streams/producer.ts b/packages/plugin-workers-core/src/streams/producer.ts index 8195f13c9d..296e144f0f 100644 --- a/packages/plugin-workers-core/src/streams/producer.ts +++ b/packages/plugin-workers-core/src/streams/producer.ts @@ -1,4 +1,5 @@ import { createDurableStream } from '@netscript/plugin-streams-core'; +import { extractContext, withContext } from '@netscript/telemetry/context'; import { DEFAULT_TOPIC } from '../domain/mod.ts'; import { type WorkerExecution, type WorkerJob, workersStreamSchema } from './schema.ts'; @@ -37,6 +38,8 @@ export type WorkerExecutionRecord = Readonly< readonly result?: Record | null; readonly workerId?: string | null; readonly attempt?: number; + readonly traceparent?: string; + readonly tracestate?: string; } >; @@ -110,11 +113,19 @@ export function createStreamMutationHook( producer: WorkersStreamProducer, ): ExecutionMutationHook { return ({ type, execution }) => { - if (type === 'deleted') { - producer.delete('execution', execution.id); - } else { - producer.upsert('execution', toExecutionStreamEntity(execution)); - } + withContext( + extractContext({ + ...(execution.traceparent ? { traceparent: execution.traceparent } : {}), + ...(execution.tracestate ? { tracestate: execution.tracestate } : {}), + }), + () => { + if (type === 'deleted') { + producer.delete('execution', execution.id); + return; + } + producer.upsert('execution', toExecutionStreamEntity(execution)); + }, + ); }; } diff --git a/packages/plugin-workers-core/tests/streams/workers-streams_test.ts b/packages/plugin-workers-core/tests/streams/workers-streams_test.ts index 5a6ce2aa06..ee5318626a 100644 --- a/packages/plugin-workers-core/tests/streams/workers-streams_test.ts +++ b/packages/plugin-workers-core/tests/streams/workers-streams_test.ts @@ -1,11 +1,26 @@ -import { assertEquals } from '@std/assert'; +import { context, propagation, trace, type Tracer } from 'npm:@opentelemetry/api@^1.9.1'; +import { AsyncLocalStorageContextManager } from 'npm:@opentelemetry/context-async-hooks@^2.9.0'; +import { W3CTraceContextPropagator } from 'npm:@opentelemetry/core@^2.5.0'; +import { + BasicTracerProvider, + InMemorySpanExporter, + SimpleSpanProcessor, +} from 'npm:@opentelemetry/sdk-trace-base@^2.5.0'; +import { assert, assertEquals, assertNotEquals } from '@std/assert'; +import { extractContext, withContext } from '@netscript/telemetry/context'; import { createStreamMutationHook, toExecutionStreamEntity, WorkerJobSchema, + type WorkersStreamProducer, } from '../../src/streams/mod.ts'; import type { ExecutionRecord } from '../../src/domain/mod.ts'; +const STORED_TRACE_ID = '11111111111111111111111111111111'; +const AMBIENT_TRACE_ID = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'; +const STORED_TRACEPARENT = `00-${STORED_TRACE_ID}-2222222222222222-01`; +const AMBIENT_TRACEPARENT = `00-${AMBIENT_TRACE_ID}-bbbbbbbbbbbbbbbb-01`; + Deno.test('toExecutionStreamEntity maps execution records to stream entities', () => { const execution: ExecutionRecord = { id: '6e029d36-e1bc-4a75-a0d8-58f24e33f6a5', @@ -86,6 +101,45 @@ Deno.test('createStreamMutationHook upserts and deletes execution entities', () ]); }); +Deno.test('createStreamMutationHook publishes mutations on the stored trace context', async () => { + const telemetry = createTelemetryHarness(); + const hook = createStreamMutationHook(recordingProducer(telemetry.tracer)); + const execution = executionFixture({ traceparent: STORED_TRACEPARENT }); + + try { + withContext(extractContext({ traceparent: AMBIENT_TRACEPARENT }), () => { + hook({ type: 'updated', execution }); + }); + await telemetry.provider.forceFlush(); + + const publishSpan = telemetry.exporter.getFinishedSpans()[0]; + assertEquals(publishSpan?.name, 'stream.publish'); + assertEquals(publishSpan?.spanContext().traceId, STORED_TRACE_ID); + assertNotEquals(publishSpan?.spanContext().traceId, AMBIENT_TRACE_ID); + } finally { + await telemetry.close(); + } +}); + +Deno.test( + 'createStreamMutationHook publishes pre-span mutations on the stored trace id', + async () => { + const telemetry = createTelemetryHarness(); + const hook = createStreamMutationHook(recordingProducer(telemetry.tracer)); + + try { + hook({ type: 'created', execution: executionFixture({ traceparent: STORED_TRACEPARENT }) }); + await telemetry.provider.forceFlush(); + + const publishTraceId = telemetry.exporter.getFinishedSpans()[0]?.spanContext().traceId; + assertEquals(publishTraceId, STORED_TRACE_ID); + assertNotEquals(publishTraceId, '00000000000000000000000000000000'); + } finally { + await telemetry.close(); + } + }, +); + Deno.test('WorkerJobSchema keeps the public job stream surface thin', () => { assertEquals( WorkerJobSchema.parse({ @@ -102,3 +156,65 @@ Deno.test('WorkerJobSchema keeps the public job stream surface thin', () => { }, ); }); + +function executionFixture(overrides: Partial = {}): ExecutionRecord { + return { + id: '6e029d36-e1bc-4a75-a0d8-58f24e33f6a5', + concept: 'job', + jobId: 'health-check', + topic: 'default', + status: 'completed', + triggeredBy: 'api', + triggeredAt: '2026-05-11T00:00:00.000Z', + startedAt: null, + completedAt: null, + exitCode: 0, + duration: 10, + error: null, + result: null, + workerId: null, + attempt: 0, + maxAttempts: 3, + ...overrides, + }; +} + +function recordingProducer(tracer: Tracer): WorkersStreamProducer { + return { + delete: () => {}, + upsert: () => { + const span = tracer.startSpan('stream.publish'); + span.end(); + }, + }; +} + +function createTelemetryHarness(): Readonly<{ + tracer: Tracer; + exporter: InMemorySpanExporter; + provider: BasicTracerProvider; + close(): Promise; +}> { + trace.disable(); + context.disable(); + propagation.disable(); + const contextManager = new AsyncLocalStorageContextManager().enable(); + assert(context.setGlobalContextManager(contextManager)); + assert(propagation.setGlobalPropagator(new W3CTraceContextPropagator())); + const exporter = new InMemorySpanExporter(); + const provider = new BasicTracerProvider({ + spanProcessors: [new SimpleSpanProcessor(exporter)], + }); + assert(trace.setGlobalTracerProvider(provider)); + return { + tracer: provider.getTracer('workers-stream-hook-test'), + exporter, + provider, + async close(): Promise { + await provider.shutdown(); + trace.disable(); + context.disable(); + propagation.disable(); + }, + }; +} diff --git a/plugins/workers/bin/runtime.ts b/plugins/workers/bin/runtime.ts index 0fd40a158e..ea2d652739 100644 --- a/plugins/workers/bin/runtime.ts +++ b/plugins/workers/bin/runtime.ts @@ -31,6 +31,7 @@ import { type StaticJobDefinitionRegistry, } from '../src/runtime/generated-jobs.ts'; import { createWorkersServiceRuntime } from '../services/src/service-runtime.ts'; +import { createStreamMutationHook } from '../streams/server.ts'; import { Scheduler, Worker } from '../worker/mod.ts'; /** @@ -88,6 +89,7 @@ export type StartCombinedProcessOptions = StartWorkerProcessOptions & StartSched /** Start the plugin worker process. */ export async function startWorkerProcess(options: StartWorkerProcessOptions = {}): Promise { const runtime = await createWorkersServiceRuntime(); + runtime.executionState.setMutationHook(createStreamMutationHook()); const generated = await registerProjectJobs(runtime, options.definitions); const poolRegistry = options.registry ?? generated?.registry; const worker = new Worker({ @@ -126,6 +128,7 @@ export async function startCombinedProcess( options: StartCombinedProcessOptions = {}, ): Promise> { const runtime = await createWorkersServiceRuntime(); + runtime.executionState.setMutationHook(createStreamMutationHook()); const generated = await registerProjectJobs(runtime, options.definitions); const poolRegistry = options.registry ?? generated?.registry; const taskExecutor = createDefaultTaskExecutor(); diff --git a/plugins/workers/tests/runtime/background-stream-hook_test.ts b/plugins/workers/tests/runtime/background-stream-hook_test.ts new file mode 100644 index 0000000000..d3790c32eb --- /dev/null +++ b/plugins/workers/tests/runtime/background-stream-hook_test.ts @@ -0,0 +1,46 @@ +import { assert, assertEquals, assertNotEquals } from '@std/assert'; +import { type ExecutionMutationHook, KvExecutionState } from '@netscript/plugin-workers-core/state'; +import { startCombinedProcess } from '../../bin/runtime.ts'; +import { Scheduler, Worker } from '../../worker/mod.ts'; + +Deno.test('startCombinedProcess installs the execution stream mutation hook', async () => { + const originalSetMutationHook = KvExecutionState.prototype.setMutationHook; + const originalSchedulerStart = Scheduler.prototype.start; + const originalWorkerStart = Worker.prototype.start; + const previousStreamsUrl = Deno.env.get('DURABLE_STREAMS_URL'); + let installedHook: ExecutionMutationHook | undefined; + let installationCount = 0; + + KvExecutionState.prototype.setMutationHook = function ( + this: KvExecutionState, + hook: ExecutionMutationHook, + ): void { + installationCount++; + installedHook = hook; + originalSetMutationHook.call(this, hook); + }; + Scheduler.prototype.start = function (): Promise { + return Promise.resolve(); + }; + Worker.prototype.start = function (): Promise { + return Promise.resolve(); + }; + Deno.env.set('DURABLE_STREAMS_URL', 'http://127.0.0.1:1'); + + try { + await startCombinedProcess({ definitions: new Map() }); + + assertEquals(installationCount, 1); + assertNotEquals(installationCount, 0); + assert(typeof installedHook === 'function'); + } finally { + KvExecutionState.prototype.setMutationHook = originalSetMutationHook; + Scheduler.prototype.start = originalSchedulerStart; + Worker.prototype.start = originalWorkerStart; + if (previousStreamsUrl === undefined) { + Deno.env.delete('DURABLE_STREAMS_URL'); + } else { + Deno.env.set('DURABLE_STREAMS_URL', previousStreamsUrl); + } + } +});