fix(plugin-workers): derive trigger id from {id} path, fail loudly on missing id (#279)

packages/contracts/src/domain/errors.ts

@@ -1,4 +1,4 @@
-import type { NotFoundError } from './schemas.ts';
+import type { NotFoundError, ValidationError } from './schemas.ts';
 
 const DEFAULT_RESOURCE_TYPE = 'resource';
 const VERSION_SEGMENT_PATTERN = /^v\d+$/i;
@@ -54,3 +54,44 @@ export function notFound(options: NotFoundOptions): never {
 function capitalize(value: string): string {
   return value.charAt(0).toUpperCase() + value.slice(1);
 }
+
+/** Options for throwing a contract `VALIDATION_ERROR` oRPC error. */
+export type ValidationFailedOptions = Readonly<{
+  /** oRPC errors object from a handler context. */
+  errors: unknown;
+  /** Human-readable summary of what failed validation. */
+  message: string;
+  /** Field-level validation messages keyed by field name. */
+  fieldErrors?: Record<string, string[]>;
+  /** Form-level (non-field) validation messages. */
+  formErrors?: string[];
+}>;
+
+type ValidationErrorFactory = (options: {
+  message: string;
+  data: ValidationError;
+}) => unknown;
+
+type ValidationErrorContainer = Readonly<{
+  VALIDATION_ERROR: ValidationErrorFactory;
+}>;
+
+/**
+ * Throws the contract `VALIDATION_ERROR` oRPC error.
+ *
+ * Mirrors {@link notFound}: it accepts the handler's `errors` object as
+ * `unknown` and narrows it to the shared error vocabulary via the single
+ * sanctioned centralized-contract boundary cast, so callers fail loudly with a
+ * typed 422 envelope instead of proceeding with invalid input.
+ */
+export function validationFailed(options: ValidationFailedOptions): never {
+  const constructors = options.errors as ValidationErrorContainer;
+
+  throw constructors.VALIDATION_ERROR({
+    message: options.message,
+    data: {
+      formErrors: options.formErrors ?? [],
+      fieldErrors: options.fieldErrors ?? {},
+    },
+  });
+}

packages/contracts/src/public/mod.ts

@@ -28,7 +28,13 @@ export {
   DEFAULT_PAGINATION_LIMIT_MAX,
   DEFAULT_PAGINATION_OFFSET,
 } from '../domain/constants.ts';
-export { getResourceType, notFound, type NotFoundOptions } from '../domain/errors.ts';
+export {
+  getResourceType,
+  notFound,
+  type NotFoundOptions,
+  validationFailed,
+  type ValidationFailedOptions,
+} from '../domain/errors.ts';
 export type { ErrorResult, OkResult, Result } from '../domain/result.ts';
 export type {
   ContractDefaultableSchema,

packages/contracts/tests/errors_test.ts

@@ -1,4 +1,4 @@
-import { getResourceType } from '../src/domain/errors.ts';
+import { getResourceType, validationFailed } from '../src/domain/errors.ts';
 
 function assertEquals(actual: string, expected: string): void {
   if (actual !== expected) {
@@ -17,3 +17,40 @@ Deno.test('getResourceType skips version segments and singularizes resources', (
   assertEquals(getResourceType({ path: ['v2', 'sagas', 'getById'] }), 'saga');
   assertEquals(getResourceType({ path: ['jobs'] }), 'job');
 });
+
+Deno.test('validationFailed throws a VALIDATION_ERROR envelope via the errors object', () => {
+  const calls: Array<{ message: string; data: unknown }> = [];
+  const errors = {
+    VALIDATION_ERROR(options: { message: string; data: unknown }): Error {
+      calls.push(options);
+      return new Error(options.message);
+    },
+  };
+
+  let thrown: unknown;
+  try {
+    validationFailed({
+      errors,
+      message: 'Job id is required in the {id} path segment.',
+      fieldErrors: { id: ['Job id is required in the {id} path segment.'] },
+    });
+  } catch (error) {
+    thrown = error;
+  }
+
+  if (!(thrown instanceof Error)) {
+    throw new Error('validationFailed must throw the constructed oRPC error');
+  }
+  assertEquals(thrown.message, 'Job id is required in the {id} path segment.');
+  if (calls.length !== 1) {
+    throw new Error(`Expected 1 VALIDATION_ERROR call, received ${calls.length}`);
+  }
+  assertEquals(calls[0].message, 'Job id is required in the {id} path segment.');
+  assertEquals(
+    JSON.stringify(calls[0].data),
+    JSON.stringify({
+      formErrors: [],
+      fieldErrors: { id: ['Job id is required in the {id} path segment.'] },
+    }),
+  );
+});

packages/plugin-workers-core/src/contracts/v1/workers.contract-schemas.ts

@@ -205,7 +205,7 @@ type JobTriggerInputShape =
     'delay' | 'payload' | 'priority' | 'traceparent' | 'tracestate'
   >
   & {
-    id: z.ZodString;
+    id: z.ZodOptional<z.ZodString>;
     correlationId: z.ZodOptional<z.ZodString>;
   };
 
@@ -217,7 +217,11 @@ const JobTriggerInputShape: JobTriggerInputShape = {
     traceparent: true,
     tracestate: true,
   }).shape,
-  id: z.string(),
+  // The `{id}` path segment is the single source of truth for the target job:
+  // oRPC merges the OpenAPI path param into `input.id`, so the body value is only
+  // an optional fallback for RPC transports. The handler fails loudly when no id
+  // resolves rather than persisting an `undefined` KV key.
+  id: z.string().optional().describe('Job id (resolved from the {id} path segment)'),
   correlationId: z.string().optional().describe('Correlation ID for tracing'),
 };
 
@@ -232,7 +236,11 @@ export const JobTriggerInputSchema: ContractSchema<JobTriggerInput> =
   JobTriggerInputZodSchema as unknown as ContractSchema<JobTriggerInput>;
 
 export const TaskTriggerInputZodSchema: z.ZodType<TaskTriggerInput> = z.object({
-  id: z.string(),
+  // The `{id}` path segment is the single source of truth for the target task:
+  // oRPC merges the OpenAPI path param into `input.id`, so the body value is only
+  // an optional fallback for RPC transports. The handler fails loudly when no id
+  // resolves rather than persisting an `undefined` KV key.
+  id: z.string().optional().describe('Task id (resolved from the {id} path segment)'),
   payload: z.record(z.string(), z.unknown()).optional().describe('Task payload'),
   priority: z.number().int().min(0).max(100).default(50).optional(),
   delay: z.number().int().nonnegative().optional().describe('Delay in ms'),

packages/plugin-workers-core/src/contracts/v1/workers.contract-types.ts

@@ -31,7 +31,8 @@ export type SSEEvent = Readonly<Record<string, unknown>>;
 
 /** Input accepted by the trigger-job procedure. */
 export type JobTriggerInput = Readonly<{
-  id: string;
+  /** Job id, resolved from the `{id}` path segment; optional in the body. */
+  id?: string;
   payload?: Record<string, unknown>;
   priority?: number;
   delay?: number;
@@ -45,7 +46,8 @@ export type JobTriggerOutput = Readonly<{ jobId: string; triggered: boolean }>;
 
 /** Input accepted by the trigger-task procedure. */
 export type TaskTriggerInput = Readonly<{
-  id: string;
+  /** Task id, resolved from the `{id}` path segment; optional in the body. */
+  id?: string;
   payload?: Record<string, unknown>;
   priority?: number;
   delay?: number;

packages/sdk/tests/integration/workers-trigger-rpc_test.ts

@@ -0,0 +1,85 @@
+import { assertEquals, assertRejects } from '@std/assert';
+import { ORPCError } from '@orpc/contract';
+import { os } from '@orpc/server';
+import { createService } from '../../../service/mod.ts';
+import { createServiceClient } from '../../src/client/service-client.ts';
+import { createServerServiceEnvKey } from '../../src/discovery/service-url.ts';
+
+// Gap 1 regression (issue #279): the eis-chat consumer saw a live 404 on
+// `GET /api/rpc/v1/workers/triggerJob`, i.e. the type-safe RPC endpoint appeared
+// unmounted. This proves that a `createServiceClient({ serviceName, routerName })`
+// call over the RPC transport reaches a plugin-workers-style `triggerJob` route
+// (route is mounted and routed), and that a request that resolves no task id
+// fails loudly with a `VALIDATION_ERROR` envelope (Gap 2 behavior over RPC)
+// rather than 404-ing or silently succeeding with an `undefined` id.
+
+const SERVICE_NAME = 'workers';
+const ROUTER_NAME = 'workers';
+const RPC_PATH = `/api/rpc/v1/${ROUTER_NAME}`;
+
+interface TriggerJobInput {
+  readonly id?: string;
+}
+
+// A plugin-workers-style router: `triggerJob` mirrors the real contract route
+// (`POST /jobs/{id}/trigger`) with an optional body id, failing loudly when no
+// id resolves — exactly the shape the #279 fix gives the real handler.
+function createWorkersStyleRouter() {
+  return {
+    triggerJob: os.route({ method: 'POST', path: '/jobs/{id}/trigger' }).handler(
+      ({ input }: { input: unknown }) => {
+        const id = (input as TriggerJobInput).id;
+        if (!id) {
+          throw new ORPCError('VALIDATION_ERROR', {
+            status: 422,
+            message: 'Job id is required in the {id} path segment.',
+            data: {
+              formErrors: [],
+              fieldErrors: { id: ['Job id is required in the {id} path segment.'] },
+            },
+          });
+        }
+        return { jobId: id, triggered: true };
+      },
+    ),
+  };
+}
+
+function clientOrigin(hostname: string, port: number): string {
+  const host = hostname === '0.0.0.0' ? '127.0.0.1' : hostname;
+  return `http://${host}:${port}`;
+}
+
+Deno.test('createServiceClient RPC path reaches a plugin-workers triggerJob route', async () => {
+  const router = createWorkersStyleRouter();
+  const running = await createService(router, { name: SERVICE_NAME })
+    .withRPC({ rpcPath: RPC_PATH })
+    .serve({ port: 0 });
+  const envKey = createServerServiceEnvKey(SERVICE_NAME);
+  const previous = Deno.env.get(envKey);
+  Deno.env.set(envKey, clientOrigin(running.addr.hostname, running.addr.port));
+
+  try {
+    const client = createServiceClient({
+      contract: router,
+      serviceName: SERVICE_NAME,
+      routerName: ROUTER_NAME,
+    });
+
+    // Route is mounted and routed over RPC: a body-carried id round-trips.
+    const ok = await client.triggerJob({ id: 'job-1' });
+    assertEquals(ok, { jobId: 'job-1', triggered: true });
+
+    // No id resolves -> loud validation failure, not a 404 and not a silent
+    // success with an undefined id.
+    const error = await assertRejects(() => client.triggerJob({}), ORPCError);
+    assertEquals((error as ORPCError<string, unknown>).code, 'VALIDATION_ERROR');
+  } finally {
+    if (previous === undefined) {
+      Deno.env.delete(envKey);
+    } else {
+      Deno.env.set(envKey, previous);
+    }
+    await running.stop();
+  }
+});

plugins/workers/services/src/routers/jobs.ts

@@ -1,5 +1,5 @@
 import { DEFAULT_TOPIC, type JobMessage } from '@netscript/plugin-workers-core/runtime';
-import { notFound } from '@netscript/contracts';
+import { notFound, validationFailed } from '@netscript/contracts';
 import { getJobQueue, getWorkersRuntime, router, type WorkersHandlers } from './router-context.ts';
 
 export const jobHandlers: WorkersHandlers<
@@ -85,18 +85,31 @@ export const jobHandlers: WorkersHandlers<
   }),
 
   triggerJob: router.triggerJob.handler(async ({ input, errors, path, context }) => {
+    // The `{id}` path segment is the single source of truth for the target job.
+    // oRPC merges the OpenAPI path param into `input.id`; the RPC transport carries
+    // it in the body. If neither resolves an id, fail loudly with a validation
+    // error rather than persisting an `undefined` KV key.
+    const jobId = input.id;
+    if (!jobId) {
+      validationFailed({
+        errors,
+        message: 'Job id is required in the {id} path segment.',
+        fieldErrors: { id: ['Job id is required in the {id} path segment.'] },
+      });
+    }
+
     const { jobRegistry: registry } = getWorkersRuntime(context);
-    const job = await registry.get(input.id);
+    const job = await registry.get(jobId);
 
     if (!job) {
-      notFound({ errors, path, resourceId: input.id });
+      notFound({ errors, path, resourceId: jobId });
     }
 
     const traceHeaders =
       (context as { traceHeaders?: { traceparent?: string; tracestate?: string } })?.traceHeaders;
 
     const jobMessage: JobMessage = {
-      jobId: input.id,
+      jobId,
       topic: job.topic ?? DEFAULT_TOPIC,
       triggeredBy: 'api',
       triggeredAt: new Date().toISOString(),
@@ -119,6 +132,6 @@ export const jobHandlers: WorkersHandlers<
         : undefined,
     });
 
-    return { jobId: input.id, triggered: true };
+    return { jobId, triggered: true };
   }),
 };

plugins/workers/services/src/routers/tasks.ts

@@ -3,7 +3,7 @@ import {
   type ExecutionRecord,
   type TaskMessage,
 } from '@netscript/plugin-workers-core/runtime';
-import { notFound } from '@netscript/contracts';
+import { notFound, validationFailed } from '@netscript/contracts';
 import { getTaskQueue, getWorkersRuntime, router, type WorkersHandlers } from './router-context.ts';
 
 export const taskHandlers: WorkersHandlers<
@@ -43,15 +43,28 @@ export const taskHandlers: WorkersHandlers<
   }),
 
   triggerTask: router.triggerTask.handler(async ({ input, errors, path, context }) => {
+    // The `{id}` path segment is the single source of truth for the target task.
+    // oRPC merges the OpenAPI path param into `input.id`; the RPC transport carries
+    // it in the body. If neither resolves an id, fail loudly with a validation
+    // error rather than persisting an `undefined` KV key.
+    const taskId = input.id;
+    if (!taskId) {
+      validationFailed({
+        errors,
+        message: 'Task id is required in the {id} path segment.',
+        fieldErrors: { id: ['Task id is required in the {id} path segment.'] },
+      });
+    }
+
     const { taskRegistry: registry } = getWorkersRuntime(context);
-    const task = await registry.get(input.id);
+    const task = await registry.get(taskId);
 
     if (!task) {
-      notFound({ errors, path, resourceId: input.id });
+      notFound({ errors, path, resourceId: taskId });
     }
 
     const taskMessage: TaskMessage = {
-      taskId: input.id,
+      taskId,
       topic: task.topic ?? DEFAULT_TOPIC,
       triggeredBy: 'api',
       triggeredAt: new Date().toISOString(),
@@ -67,7 +80,7 @@ export const taskHandlers: WorkersHandlers<
       priority: input.priority,
     });
 
-    return { taskId: input.id, triggered: true };
+    return { taskId, triggered: true };
   }),
 
   listTaskExecutions: router.listTaskExecutions.handler(async ({ input, context }) => {