feat(ru-code): restore WS connection stability + process/IO hardening

apps/server/src/cli/project.ts

@@ -28,6 +28,7 @@ import { AuthControlPlaneRuntimeLive } from "../auth/Layers/AuthControlPlane.ts"
 import { AuthControlPlane } from "../auth/Services/AuthControlPlane.ts";
 import type { AuthControlPlaneShape } from "../auth/Services/AuthControlPlane.ts";
 import { ServerConfig, type ServerConfigShape } from "../config.ts";
+import { PROJECT_CLI_LIVE_SERVER_TIMEOUT_MS } from "../timeouts.ts";
 import { OrchestrationEngineService } from "../orchestration/Services/OrchestrationEngine.ts";
 import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSnapshotQuery.ts";
 import { OrchestrationLayerLive } from "../orchestration/runtimeLayer.ts";
@@ -66,7 +67,6 @@ const ProjectCliRuntimeLive = Layer.mergeAll(
   ),
 );
 
-const PROJECT_CLI_LIVE_SERVER_TIMEOUT = Duration.seconds(1);
 const OrchestrationHttpErrorResponse = Schema.Struct({
   error: Schema.String,
 });
@@ -85,7 +85,7 @@ const withProjectCliSessionToken = <A, E, R>(
   );
 
 const withProjectCliLiveServerTimeout = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
-  effect.pipe(Effect.timeout(PROJECT_CLI_LIVE_SERVER_TIMEOUT));
+  effect.pipe(Effect.timeout(Duration.millis(PROJECT_CLI_LIVE_SERVER_TIMEOUT_MS)));
 
 const runLiveServerRequest = <A, E extends Error, R>(
   request: HttpClientRequest.HttpClientRequest,

apps/server/src/config.ts

@@ -105,39 +105,6 @@ export const MODE_CHANGE_METHOD: AbortMethod = "end-force";
  */
 export const MAINTENANCE_METHOD: AbortMethod = "end-force";
 
-/**
- * ACP_WIRE_STALL_WARN_MS — soft threshold for "CLI seems stuck"
- * detection. While a turn is active, the adapter watches incoming
- * JSON-RPC frames on the ACP wire (every notification, response, or
- * frame from CLI). If no frame arrives for this many milliseconds,
- * a WARN log is emitted once per stall. No teardown — just visibility.
- */
-export const ACP_WIRE_STALL_WARN_MS = 3_600_000;
-
-/**
- * ACP_WIRE_STALL_KILL_MS — hard threshold. After this much wire
- * silence during an active turn the adapter declares the session
- * stuck and runs `abortSession(ctx, "end-force")`. The next user
- * message starts a fresh CLI child and `session/load`s the prior
- * transcript via the persisted resumeCursor. Backstop so a stuck
- * agent can never strand the chat indefinitely.
- */
-export const ACP_WIRE_STALL_KILL_MS = 7_200_000;
-
-/**
- * POST_ANSWER_RESUME_TIMEOUT_MS — one-shot timeout used by the
- * post-answer resume probe (`armPostAnswerResumeProbe` in
- * `AcpPendingRequests.ts`). After the adapter resolves a pending
- * user-input / approval / plan-approval Deferred, a fiber sleeps this
- * long and then checks `ctx.wireActivity.lastIncomingAt`. If CLI has
- * not produced a single inbound frame in that window, the session is
- * declared wedged and `abortSession(ctx, MAINTENANCE_METHOD)` is
- * called. Probe measurement showed the typical resume latency is
- * ~4 ms, so 10 s is ~2500× margin — generous but tight enough to
- * recover snappily.
- */
-export const POST_ANSWER_RESUME_TIMEOUT_MS = 3_600_000;
-
 /**
  * CONTEXT_WINDOW_TOKENS — total context window size advertised to the
  * UI for the current CLI binary. Hardcoded rather than extracted from

apps/server/src/daemonLauncher.ts

@@ -15,6 +15,7 @@ import { resolveStartupCli, runStartupChecks } from "./ru-fork/preflight/preflig
 import { ARROW_DIM, ARROW_OK, ARROW_WARN, paint } from "./ru-fork/local-startup/cliPaint.ts";
 import { printDaemonReadyBanner } from "./ru-fork/local-startup/daemonReadyBanner.ts";
 import { deriveServerPaths } from "./config.ts";
+import { DAEMON_HEALTH_PROBE_TIMEOUT_MS, DAEMON_SPAWN_TIMEOUT_MS } from "./timeouts.ts";
 import { expandHomePath, resolveBaseDir } from "./os-jank.ts";
 import { readPersistedServerRuntimeState } from "./serverRuntimeState.ts";
 import { Open } from "./open.ts";
@@ -51,20 +52,18 @@ const lineKV = (key: string, value: string): string =>
   `    ${paint.dim(key.padEnd(6))}  ${paint.bold(paint.magenta(value))}`;
 const headline = (arrow: string, text: string): string => `  ${arrow} ${paint.bold(text)}`;
 
-const HEALTH_PROBE_TIMEOUT_MS = 1_000;
 const SPAWN_HEALTH_POLL_INTERVAL_MS = 200;
 // `Effect.retry({ times: N })` means 1 initial attempt + N retries.
 // 75 retries × 200ms = 15s of additional polling on top of the first try.
 const SPAWN_HEALTH_POLL_RETRIES = 75;
-const SPAWN_TIMEOUT_MS = 5_000;
 const STOP_DRAIN_INTERVAL_MS = 100;
 const STOP_DRAIN_RETRIES = 50; // ~5s drain budget after POST /shutdown
 
 const probeHealth = (origin: string) =>
   Effect.tryPromise({
     try: async () => {
       const controller = new AbortController();
-      const timer = setTimeout(() => controller.abort(), HEALTH_PROBE_TIMEOUT_MS);
+      const timer = setTimeout(() => controller.abort(), DAEMON_HEALTH_PROBE_TIMEOUT_MS);
       try {
         const response = await fetch(`${origin}/health`, {
           signal: controller.signal,
@@ -195,10 +194,10 @@ const spawnDetachedServer = (input: {
 }) =>
   spawnDetachedServerCallback(input).pipe(
     Effect.timeoutOrElse({
-      duration: Duration.millis(SPAWN_TIMEOUT_MS),
+      duration: Duration.millis(DAEMON_SPAWN_TIMEOUT_MS),
       orElse: () =>
         Effect.fail(
-          new Error(`spawn did not signal 'spawn' or 'error' within ${SPAWN_TIMEOUT_MS}ms`),
+          new Error(`spawn did not signal 'spawn' or 'error' within ${DAEMON_SPAWN_TIMEOUT_MS}ms`),
         ),
     }),
   );
@@ -215,7 +214,7 @@ const fetchPairingStartupUrl = (origin: string) =>
   Effect.tryPromise({
     try: async () => {
       const controller = new AbortController();
-      const timer = setTimeout(() => controller.abort(), HEALTH_PROBE_TIMEOUT_MS);
+      const timer = setTimeout(() => controller.abort(), DAEMON_HEALTH_PROBE_TIMEOUT_MS);
       try {
         const response = await fetch(`${origin}/pair/startup`, {
           method: "POST",
@@ -362,7 +361,7 @@ export const runStopCommand = (input: StopCommandInput) =>
     const sent = yield* Effect.tryPromise({
       try: async () => {
         const controller = new AbortController();
-        const timer = setTimeout(() => controller.abort(), HEALTH_PROBE_TIMEOUT_MS);
+        const timer = setTimeout(() => controller.abort(), DAEMON_HEALTH_PROBE_TIMEOUT_MS);
         try {
           const response = await fetch(`${state.value.origin}/shutdown`, {
             method: "POST",

apps/server/src/git/GitManager.ts

@@ -51,6 +51,7 @@ import type { GitManagerServiceError } from "@t3tools/contracts";
 import { GitVcsDriver, type GitStatusDetails } from "../vcs/GitVcsDriver.ts";
 import { SourceControlProviderRegistry } from "../sourceControl/SourceControlProviderRegistry.ts";
 import type { ChangeRequest } from "@t3tools/contracts";
+import { GIT_COMMIT_TIMEOUT_MS } from "../timeouts.ts";
 
 export interface GitActionProgressReporter {
   readonly publish: (event: GitActionProgressEvent) => Effect.Effect<void, never>;
@@ -90,7 +91,6 @@ export class GitManager extends Context.Service<GitManager, GitManagerShape>()(
   "t3/git/GitManager",
 ) {}
 
-const COMMIT_TIMEOUT_MS = 10 * 60_000;
 const MAX_PROGRESS_TEXT_LENGTH = 500;
 const SHORT_SHA_LENGTH = 7;
 const TOAST_DESCRIPTION_MAX = 72;
@@ -1224,7 +1224,7 @@ export const makeGitManager = Effect.fn("makeGitManager")(function* () {
           }
         : null;
     const { commitSha } = yield* gitCore.commit(cwd, suggestion.subject, suggestion.body, {
-      timeoutMs: COMMIT_TIMEOUT_MS,
+      timeoutMs: GIT_COMMIT_TIMEOUT_MS,
       ...(commitProgress ? { progress: commitProgress } : {}),
     });
     if (currentHookName !== null) {

apps/server/src/processRunner.ts

@@ -14,6 +14,8 @@ import {
 } from "./stream/collectUint8StreamText.ts";
 // ru-fork: shared spawn policy (shell:false default for exe + bash-routing).
 import { resolveSpawn } from "./ru-fork/spawn/policy.ts";
+import { haltOnExit } from "./ru-fork/spawn/haltOnExit.ts";
+import { PROCESS_RUNNER_DEFAULT_TIMEOUT_MS } from "./timeouts.ts";
 
 export interface ProcessRunInput {
   readonly command: string;
@@ -95,7 +97,6 @@ export class ProcessRunner extends Context.Service<ProcessRunner, ProcessRunnerS
   "t3/processRunner",
 ) {}
 
-const DEFAULT_TIMEOUT = "60 seconds";
 const DEFAULT_MAX_OUTPUT_BYTES = 8 * 1024 * 1024;
 
 const WINDOWS_COMMAND_NOT_FOUND_PATTERNS = [
@@ -194,7 +195,7 @@ function finalizeRunProcess<R>(
   effect: Effect.Effect<ProcessRunOutput, ProcessRunError, R | Scope.Scope>,
   input: ProcessRunInput,
 ): Effect.Effect<ProcessRunOutput, ProcessRunError, Exclude<R, Scope.Scope>> {
-  const timeout = Duration.fromInputUnsafe(input.timeout ?? DEFAULT_TIMEOUT);
+  const timeout = Duration.fromInputUnsafe(input.timeout ?? PROCESS_RUNNER_DEFAULT_TIMEOUT_MS);
   const timeoutBehavior = input.timeoutBehavior ?? "error";
 
   return effect.pipe(
@@ -289,7 +290,7 @@ const runProcessCore = Effect.fn("processRunner.runProcessCore")(function* (
         args: input.args,
         cwd: input.cwd,
         streamName: "stdout",
-        stream: child.stdout,
+        stream: child.stdout.pipe(haltOnExit(child.exitCode)),
         maxOutputBytes,
         outputMode,
         truncatedMarker,
@@ -299,7 +300,7 @@ const runProcessCore = Effect.fn("processRunner.runProcessCore")(function* (
         args: input.args,
         cwd: input.cwd,
         streamName: "stderr",
-        stream: child.stderr,
+        stream: child.stderr.pipe(haltOnExit(child.exitCode)),
         maxOutputBytes,
         outputMode,
         truncatedMarker,

apps/server/src/provider/Layers/CliAdapter.ts

@@ -47,16 +47,18 @@ import type * as EffectAcpSchema from "effect-acp/schema";
 
 import { resolveAttachmentPath } from "../../attachmentStore.ts";
 import {
-  ACP_WIRE_STALL_KILL_MS,
-  ACP_WIRE_STALL_WARN_MS,
   CONTEXT_WINDOW_TOKENS,
   MAINTENANCE_METHOD,
   MODE_CHANGE_METHOD,
-  POST_ANSWER_RESUME_TIMEOUT_MS,
   ServerConfig,
   SLASH_COMMAND_NOTIFICATION_METHODS,
   STOP_BUTTON_METHOD,
 } from "../../config.ts";
+import {
+  ACP_WIRE_STALL_KILL_MS,
+  ACP_WIRE_STALL_WARN_MS,
+  POST_ANSWER_RESUME_TIMEOUT_MS,
+} from "../../timeouts.ts";
 import {
   buildPostAnswerResumeProbe,
   settleAndDelete,

apps/server/src/provider/Layers/CliProvider.ts

@@ -31,6 +31,8 @@ import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
 import { buildCliSpawn } from "../../ru-fork/spawn/policy.ts";
 import { APP_NAME } from "@ru-fork/branding";
 import { CLI_BINARY_NAME, CLI_NAME } from "../../config.ts";
+import { CLI_VERSION_PROBE_TIMEOUT_MS } from "../../timeouts.ts";
+import { haltOnExit } from "../../ru-fork/spawn/haltOnExit.ts";
 import {
   buildServerProvider,
   collectStreamAsString,
@@ -40,8 +42,6 @@ import {
   type ServerProviderDraft,
 } from "../providerSnapshot.ts";
 
-const VERSION_TIMEOUT_MS = 8_000;
-
 const CLI_PRESENTATION = {
   displayName: CLI_NAME,
   showInteractionModeToggle: true,
@@ -157,8 +157,8 @@ const runCliVersionCommand = (
     const child = yield* spawner.spawn(command);
     const [stdout, stderr, exitCode] = yield* Effect.all(
       [
-        collectStreamAsString(child.stdout),
-        collectStreamAsString(child.stderr),
+        collectStreamAsString(child.stdout.pipe(haltOnExit(child.exitCode))),
+        collectStreamAsString(child.stderr.pipe(haltOnExit(child.exitCode))),
         child.exitCode.pipe(Effect.map(Number)),
       ],
       { concurrency: "unbounded" },
@@ -191,7 +191,7 @@ export const checkCliProviderStatus = Effect.fn("checkCliProviderStatus")(functi
   }
 
   const versionProbe = yield* runCliVersionCommand(cliJs, environment).pipe(
-    Effect.timeoutOption(VERSION_TIMEOUT_MS),
+    Effect.timeoutOption(CLI_VERSION_PROBE_TIMEOUT_MS),
     Effect.result,
   );
 
@@ -216,13 +216,13 @@ export const checkCliProviderStatus = Effect.fn("checkCliProviderStatus")(functi
 
   if (Option.isNone(versionProbe.success)) {
     // ru-fork: cold-start of cli-code (Node ESM + heavy import graph)
-    // regularly approaches the 8 s budget. Treat a slow `--version` as a
+    // regularly approaches the 3 s budget. Treat a slow `--version` as a
     // soft signal — the binary is installed (spawn succeeded), only the
     // version string is unknown. Log to server, keep the provider usable.
     // The previous `status: "error"` + UI message are commented below in
     // case we want to surface this again later.
     yield* Effect.logWarning("cli-version-probe-timeout", {
-      timeoutMs: VERSION_TIMEOUT_MS,
+      timeoutMs: CLI_VERSION_PROBE_TIMEOUT_MS,
     });
     return buildServerProvider({
       presentation: CLI_PRESENTATION,

apps/server/src/provider/providerMaintenance.ts

@@ -12,8 +12,9 @@ import * as Option from "effect/Option";
 import * as Schema from "effect/Schema";
 import { HttpClient, HttpClientRequest } from "effect/unstable/http";
 
+import { PROVIDER_LATEST_VERSION_TIMEOUT_MS } from "../timeouts.ts";
+
 const LATEST_VERSION_CACHE_TTL_MS = 60 * 60 * 1_000;
-const LATEST_VERSION_TIMEOUT_MS = 4_000;
 const PROVIDER_UPDATE_ACTION_TOAST_MESSAGE = "Install the update now or review provider settings.";
 
 export interface ProviderMaintenanceCapabilities {
@@ -405,7 +406,7 @@ const fetchNpmLatestVersion = Effect.fn("fetchNpmLatestVersion")(function* (pack
     `https://registry.npmjs.org/${encodeURIComponent(packageName)}/latest`,
   ).pipe(HttpClientRequest.setHeader("accept", "application/json"));
   const response = yield* client.execute(request).pipe(
-    Effect.timeoutOption(LATEST_VERSION_TIMEOUT_MS),
+    Effect.timeoutOption(PROVIDER_LATEST_VERSION_TIMEOUT_MS),
     Effect.catch(() => Effect.succeed(Option.none())),
   );
   if (Option.isNone(response)) {

apps/server/src/provider/providerMaintenanceRunner.ts

@@ -29,9 +29,10 @@ import type { ProviderMaintenanceCapabilities } from "./providerMaintenance.ts";
 // bash-route on opt-in. See ru-fork/spawn/policy.ts.
 import { resolveSpawn } from "../ru-fork/spawn/policy.ts";
 import { collectUint8StreamText } from "../stream/collectUint8StreamText.ts";
+import { haltOnExit } from "../ru-fork/spawn/haltOnExit.ts";
+import { PROVIDER_UPDATE_TIMEOUT_MS } from "../timeouts.ts";
 const isServerProviderUpdateError = Schema.is(ServerProviderUpdateError);
 
-const UPDATE_TIMEOUT_MS = 5 * 60_000;
 const UPDATE_OUTPUT_MAX_BYTES = 10_000;
 
 export interface ProviderMaintenanceCommandResult {
@@ -103,11 +104,11 @@ const runProviderMaintenanceCommandWithSpawner = Effect.fn("ProviderMaintenanceR
         const [stdout, stderr, exitCode] = yield* Effect.all(
           [
             collectUint8StreamText({
-              stream: child.stdout,
+              stream: child.stdout.pipe(haltOnExit(child.exitCode)),
               maxBytes: UPDATE_OUTPUT_MAX_BYTES,
             }),
             collectUint8StreamText({
-              stream: child.stderr,
+              stream: child.stderr.pipe(haltOnExit(child.exitCode)),
               maxBytes: UPDATE_OUTPUT_MAX_BYTES,
             }),
             child.exitCode,
@@ -136,7 +137,7 @@ const runProviderMaintenanceCommandWithSpawner = Effect.fn("ProviderMaintenanceR
 
     return yield* collectCommandResult().pipe(
       Effect.scoped,
-      Effect.timeoutOption(Duration.millis(UPDATE_TIMEOUT_MS)),
+      Effect.timeoutOption(Duration.millis(PROVIDER_UPDATE_TIMEOUT_MS)),
       Effect.map((result) =>
         Option.match(result, {
           onSome: (value) => value,

apps/server/src/provider/providerSnapshot.ts

@@ -11,15 +11,16 @@ import type {
 import * as Effect from "effect/Effect";
 import * as Data from "effect/Data";
 import * as Stream from "effect/Stream";
-import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
+// ru-fork: only used by the commented-out spawnAndCollect below.
+// import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
 import { normalizeModelSlug } from "@t3tools/shared/model";
-import { isWindowsCommandNotFound } from "../processRunner.ts";
+// import { isWindowsCommandNotFound } from "../processRunner.ts";
 import { createProviderVersionAdvisory } from "./providerMaintenance.ts";
 import { collectUint8StreamText } from "../stream/collectUint8StreamText.ts";
 
-export const DEFAULT_TIMEOUT_MS = 4_000;
-// Auth status checks involve disk/network lookups and can be slow on first run (especially Windows)
-export const AUTH_PROBE_TIMEOUT_MS = 10_000;
+// ru-fork: dead — no consumers in the repo. Not relocated to timeouts.ts.
+// export const DEFAULT_TIMEOUT_MS = 4_000;
+// export const AUTH_PROBE_TIMEOUT_MS = 10_000;
 
 export interface CommandResult {
   readonly stdout: string;
@@ -60,25 +61,26 @@ export function isCommandMissingCause(error: { readonly message: string }): bool
   return lower.includes("enoent") || lower.includes("notfound");
 }
 
-export const spawnAndCollect = (binaryPath: string, command: ChildProcess.Command) =>
-  Effect.gen(function* () {
-    const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
-    const child = yield* spawner.spawn(command);
-    const [stdout, stderr, exitCode] = yield* Effect.all(
-      [
-        collectStreamAsString(child.stdout),
-        collectStreamAsString(child.stderr),
-        child.exitCode.pipe(Effect.map(Number)),
-      ],
-      { concurrency: "unbounded" },
-    );
-
-    const result: CommandResult = { stdout, stderr, code: exitCode };
-    if (isWindowsCommandNotFound(exitCode, stderr)) {
-      return yield* new ProviderCommandExecutionError({ message: `spawn ${binaryPath} ENOENT` });
-    }
-    return result;
-  }).pipe(Effect.scoped);
+// ru-fork: no callers in the repo; kept for reference, not exit-gated.
+// export const spawnAndCollect = (binaryPath: string, command: ChildProcess.Command) =>
+//   Effect.gen(function* () {
+//     const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
+//     const child = yield* spawner.spawn(command);
+//     const [stdout, stderr, exitCode] = yield* Effect.all(
+//       [
+//         collectStreamAsString(child.stdout),
+//         collectStreamAsString(child.stderr),
+//         child.exitCode.pipe(Effect.map(Number)),
+//       ],
+//       { concurrency: "unbounded" },
+//     );
+//
+//     const result: CommandResult = { stdout, stderr, code: exitCode };
+//     if (isWindowsCommandNotFound(exitCode, stderr)) {
+//       return yield* new ProviderCommandExecutionError({ message: `spawn ${binaryPath} ENOENT` });
+//     }
+//     return result;
+//   }).pipe(Effect.scoped);
 
 export function detailFromResult(
   result: CommandResult & { readonly timedOut?: boolean },