refactor(@probitas/probitas): extract subprocess communication patterns

Reduces code duplication by extracting common IPC patterns into
reusable factories and helpers across parent and subprocess sides.

## Subprocess side (_templates/utils.ts)
- Add runSubprocess() factory for standardized bootstrap lifecycle
- Add createOutputValidator() for type-safe protocol validation
- Add serializeError/deserializeError for unified error handling
- Extract ~90 lines of duplicated bootstrap code from list.ts and run.ts

## Parent side (subprocess.ts)
- Add startSubprocess() for connection setup and stream creation
- Add runSubprocessToCompletion() for simple request-response pattern
- Add cleanupSubprocess() for proper resource cleanup ordering
- Consolidate ~100 lines of duplicated spawn/connect/cleanup logic

## Protocol refactoring
- Migrate list_protocol.ts and run_protocol.ts to use shared validators
- Re-export error utilities from utils.ts for consistency
- Document architectural asymmetry between parent and subprocess

## Testing
- Add utils_test.ts covering all new utility functions
- Verify error serialization, port parsing, and validator logic

Total: ~190 lines of duplication eliminated, improved maintainability.

Author: lambdalisue

deno.json

@@ -89,6 +89,7 @@
     "@probitas/reporter": "jsr:@probitas/reporter@^0.7.0",
     "@probitas/runner": "jsr:@probitas/runner@^0.5.0",
     "@std/assert": "jsr:@std/assert@^1.0.16",
+    "@std/cbor": "jsr:@std/cbor@^0.1.9",
     "@std/cli": "jsr:@std/cli@^1.0.25",
     "@std/collections": "jsr:@std/collections@^1.1.3",
     "@std/dotenv": "jsr:@std/dotenv@^0.225.6",

deno.lock

@@ -2,88 +2,45 @@
  * Subprocess entry point for list command
  *
  * Loads scenarios and outputs metadata via TCP IPC.
- * Communication is via NDJSON over TCP (not stdin/stdout).
+ * Communication is via CBOR over TCP (not stdin/stdout).
  *
  * @module
  * @internal
  */
 
 import { loadScenarios } from "@probitas/core/loader";
 import { applySelectors } from "@probitas/core/selector";
-import { toErrorObject } from "@core/errorutil/error-object";
 import type { ListInput, ListOutput, ScenarioMeta } from "./list_protocol.ts";
-import {
-  closeIpc,
-  connectIpc,
-  parseIpcPort,
-  readInput,
-  writeOutput,
-} from "./utils.ts";
-
-/**
- * Main entry point
- */
-async function main(): Promise<void> {
-  // Parse IPC port from command line arguments
-  const port = parseIpcPort(Deno.args);
-
-  // Connect to parent process IPC server
-  const ipc = await connectIpc(port);
-
-  try {
-    // Read input from IPC (TCP connection establishes readiness)
-    const input = await readInput(ipc) as ListInput;
-
-    // Load scenarios from files
-    const scenarios = await loadScenarios(input.filePaths, {
-      onImportError: (file, err) => {
-        const m = err instanceof Error ? err.message : String(err);
-        throw new Error(`Failed to load scenario from ${file}: ${m}`);
-      },
-    });
-
-    // Apply selectors to filter scenarios
-    const filtered = input.selectors.length > 0
-      ? applySelectors(scenarios, input.selectors)
-      : scenarios;
-
-    // Build output metadata
-    const scenarioMetas: ScenarioMeta[] = filtered.map((s) => ({
-      name: s.name,
-      tags: s.tags,
-      steps: s.steps.filter((e) => e.kind === "step").length,
-      file: s.origin?.path || "unknown",
-    }));
-
-    await writeOutput(
-      ipc,
-      {
-        type: "result",
-        scenarios: scenarioMetas,
-      } satisfies ListOutput,
-    );
-  } catch (error) {
-    const err = error instanceof Error ? error : new Error(String(error));
-    try {
-      await writeOutput(
-        ipc,
-        {
-          type: "error",
-          error: toErrorObject(err),
-        } satisfies ListOutput,
-      );
-    } catch {
-      // Failed to write error to IPC, log to console as fallback
-      console.error("Subprocess error:", error);
-    }
-  } finally {
-    // Await close to ensure all pending writes are flushed
-    await closeIpc(ipc);
-  }
-}
-
-// Run main and exit explicitly to avoid async operations keeping process alive
-main().finally(() => {
-  // Ensure process exits after output is flushed
-  setTimeout(() => Deno.exit(0), 0);
+import { runSubprocess, writeOutput } from "./utils.ts";
+
+// Run subprocess with bootstrap handling
+runSubprocess<ListInput>(async (ipc, input) => {
+  // Load scenarios from files
+  const scenarios = await loadScenarios(input.filePaths, {
+    onImportError: (file, err) => {
+      const m = err instanceof Error ? err.message : String(err);
+      throw new Error(`Failed to load scenario from ${file}: ${m}`);
+    },
+  });
+
+  // Apply selectors to filter scenarios
+  const filtered = input.selectors.length > 0
+    ? applySelectors(scenarios, input.selectors)
+    : scenarios;
+
+  // Build output metadata
+  const scenarioMetas: ScenarioMeta[] = filtered.map((s) => ({
+    name: s.name,
+    tags: s.tags,
+    steps: s.steps.filter((e) => e.kind === "step").length,
+    file: s.origin?.path || "unknown",
+  }));
+
+  await writeOutput(
+    ipc,
+    {
+      type: "result",
+      scenarios: scenarioMetas,
+    } satisfies ListOutput,
+  );
 });

src/cli/_templates/list.ts

@@ -5,6 +5,7 @@
  */
 
 import type { ErrorObject } from "@core/errorutil/error-object";
+import { createOutputValidator } from "./utils.ts";
 
 /**
  * Input sent to subprocess via IPC
@@ -51,20 +52,10 @@ export interface ScenarioMeta {
   readonly file: string;
 }
 
-/**
- * Valid ListOutput type values
- */
-const LIST_OUTPUT_TYPES = new Set(["result", "error"]);
-
 /**
  * Type guard to check if a value is a valid ListOutput
  */
-export function isListOutput(value: unknown): value is ListOutput {
-  return (
-    value !== null &&
-    typeof value === "object" &&
-    "type" in value &&
-    typeof value.type === "string" &&
-    LIST_OUTPUT_TYPES.has(value.type)
-  );
-}
+export const isListOutput = createOutputValidator<ListOutput>([
+  "result",
+  "error",
+]);

src/cli/_templates/list_protocol.ts

@@ -2,7 +2,7 @@
  * Subprocess entry point for run command
  *
  * Executes scenarios and streams progress via TCP IPC.
- * Communication is via NDJSON over TCP (not stdin/stdout).
+ * Communication is via CBOR over TCP (not stdin/stdout).
  *
  * @module
  * @internal
@@ -19,15 +19,7 @@ import {
   serializeError,
   serializeRunResult,
 } from "./run_protocol.ts";
-import {
-  closeIpc,
-  configureLogging,
-  connectIpc,
-  type IpcConnection,
-  parseIpcPort,
-  readInput,
-  writeOutput,
-} from "./utils.ts";
+import { configureLogging, runSubprocess, writeOutput } from "./utils.ts";
 
 const logger = getLogger(["probitas", "cli", "run", "subprocess"]);
 
@@ -60,11 +52,12 @@ globalThis.addEventListener(
 
 /**
  * Execute all scenarios
+ *
+ * This handler manages its own error handling because it needs to:
+ * 1. Clean up the abort controller on error
+ * 2. Write structured error output for scenario execution failures
  */
-async function runScenarios(
-  ipc: IpcConnection,
-  input: RunCommandInput,
-): Promise<void> {
+runSubprocess<RunCommandInput>(async (ipc, input) => {
   const {
     filePaths,
     selectors,
@@ -142,44 +135,4 @@ async function runScenarios(
   } finally {
     globalAbortController = null;
   }
-}
-
-/**
- * Main entry point
- */
-async function main(): Promise<void> {
-  // Parse IPC port from command line arguments
-  const port = parseIpcPort(Deno.args);
-
-  // Connect to parent process IPC server
-  const ipc = await connectIpc(port);
-
-  try {
-    // Read input from IPC (TCP connection establishes readiness)
-    const input = await readInput(ipc) as RunCommandInput;
-
-    await runScenarios(ipc, input);
-  } catch (error) {
-    try {
-      await writeOutput(
-        ipc,
-        {
-          type: "error",
-          error: serializeError(error),
-        } satisfies RunOutput,
-      );
-    } catch {
-      // Failed to write error to IPC, log to console as fallback
-      console.error("Subprocess error:", error);
-    }
-  } finally {
-    // Await close to ensure all pending writes are flushed
-    await closeIpc(ipc);
-  }
-}
-
-// Run main and exit explicitly to avoid LogTape keeping process alive
-main().finally(() => {
-  // Ensure process exits after output is flushed
-  setTimeout(() => Deno.exit(0), 0);
 });