From cea38efa4032b6cd07e7df9bae3bcfb84afdc6d8 Mon Sep 17 00:00:00 2001 From: ru-code-dev <53821477+ru-code-dev@users.noreply.github.com> Date: Fri, 12 Jun 2026 18:00:29 +0300 Subject: [PATCH] feat(ru-code): restore WS connection stability + process/IO hardening MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Restores the WS connection-stability feature that a cleanup commit deleted, plus process I/O hardening. Success paths unchanged. - **WS stability:** restore effect RpcClient heartbeat patch + intentional-close tracking — a deliberate reconnect/dispose close is no longer misread as a disconnect, so no retry/connection leak. - **Process I/O:** end child stdio on process exit (+250ms drain), not EOF — fixes hangs when AV/git daemon holds the pipe open. - **Terminal:** show real open-failure reason in UI (was generic "Failed to open terminal"); log the cause server-side. - **RPC:** restore per-RPC failure logging lost with telemetry (defect→ERROR, typed/interrupt→DEBUG). - **Timeouts:** consolidate all server timeout literals into timeouts.ts. - **Tests:** haltOnExit, preflight probe, terminal errors. ### Enable WS debug logging (web) In the browser DevTools console: localStorage.debugging = "ws" // then reload Combine layers with commas: "ws,mcp". Disable: delete localStorage.debugging // then reload Off by default — zero console noise, no cost on hot paths. --- apps/server/src/cli/project.ts | 4 +- apps/server/src/config.ts | 33 --- apps/server/src/daemonLauncher.ts | 13 +- apps/server/src/git/GitManager.ts | 4 +- apps/server/src/processRunner.ts | 9 +- apps/server/src/provider/Layers/CliAdapter.ts | 8 +- .../server/src/provider/Layers/CliProvider.ts | 14 +- .../src/provider/providerMaintenance.ts | 5 +- .../src/provider/providerMaintenanceRunner.ts | 9 +- apps/server/src/provider/providerSnapshot.ts | 50 ++-- .../src/ru-fork/preflight/common/checks.ts | 26 +- .../src/ru-fork/preflight/common/constants.ts | 10 +- .../src/ru-fork/preflight/common/messages.ts | 5 +- .../src/ru-fork/preflight/common/probe.ts | 55 +++- .../ru-fork/preflight/preflight-install.ts | 6 +- .../ru-fork/preflight/preflight-startup.ts | 2 +- apps/server/src/ru-fork/spawn/haltOnExit.ts | 24 ++ .../terminal/logTerminalStartFailure.ts | 35 +++ apps/server/src/sourceControl/GitHubCli.ts | 5 +- .../sourceControl/SourceControlDiscovery.ts | 8 +- .../SourceControlProviderDiscovery.ts | 8 +- .../SourceControlRepositoryService.ts | 3 +- apps/server/src/terminal/Layers/Manager.ts | 21 +- .../src/textGeneration/CliTextGeneration.ts | 10 +- apps/server/src/timeouts.ts | 81 +++++ apps/server/src/vcs/GitVcsDriver.ts | 15 +- apps/server/src/vcs/GitVcsDriverCore.ts | 54 ++-- apps/server/src/vcs/VcsProcess.ts | 9 +- apps/server/src/ws.ts | 34 ++- apps/server/tests/haltOnExit.test.ts | 61 ++++ apps/server/tests/preflightCheckCli.test.ts | 45 +++ .../server/tests/preflightProbeByExit.test.ts | 59 ++++ .../ru-fork/logTerminalStartFailure.test.ts | 50 ++++ .../tests/ru-fork/terminalErrors.test.ts | 48 +++ apps/web/src/components/CommandPalette.tsx | 22 +- .../src/components/ThreadTerminalDrawer.tsx | 13 +- apps/web/src/environments/runtime/service.ts | 60 ++++ apps/web/src/rpc/protocol.ts | 97 +++++- apps/web/src/rpc/transportError.ts | 1 + apps/web/src/rpc/wsRpcClient.ts | 2 + apps/web/src/rpc/wsTransport.ts | 44 +++ apps/web/src/ru-fork/debugging/index.ts | 60 ++++ packages/contracts/src/index.ts | 1 + .../contracts/src/ru-fork/terminalErrors.ts | 51 ++++ patches/effect@4.0.0-beta.59.patch | 277 ++++++++++++++++++ pnpm-workspace.yaml | 7 + 46 files changed, 1269 insertions(+), 189 deletions(-) create mode 100644 apps/server/src/ru-fork/spawn/haltOnExit.ts create mode 100644 apps/server/src/ru-fork/terminal/logTerminalStartFailure.ts create mode 100644 apps/server/src/timeouts.ts create mode 100644 apps/server/tests/haltOnExit.test.ts create mode 100644 apps/server/tests/preflightCheckCli.test.ts create mode 100644 apps/server/tests/preflightProbeByExit.test.ts create mode 100644 apps/server/tests/ru-fork/logTerminalStartFailure.test.ts create mode 100644 apps/server/tests/ru-fork/terminalErrors.test.ts create mode 100644 apps/web/src/ru-fork/debugging/index.ts create mode 100644 packages/contracts/src/ru-fork/terminalErrors.ts create mode 100644 patches/effect@4.0.0-beta.59.patch diff --git a/apps/server/src/cli/project.ts b/apps/server/src/cli/project.ts index 39772fcd061..706f2fb505b 100644 --- a/apps/server/src/cli/project.ts +++ b/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 = ( ); const withProjectCliLiveServerTimeout = (effect: Effect.Effect) => - effect.pipe(Effect.timeout(PROJECT_CLI_LIVE_SERVER_TIMEOUT)); + effect.pipe(Effect.timeout(Duration.millis(PROJECT_CLI_LIVE_SERVER_TIMEOUT_MS))); const runLiveServerRequest = ( request: HttpClientRequest.HttpClientRequest, diff --git a/apps/server/src/config.ts b/apps/server/src/config.ts index ce6db0589d9..dc2f0d3df8a 100644 --- a/apps/server/src/config.ts +++ b/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 diff --git a/apps/server/src/daemonLauncher.ts b/apps/server/src/daemonLauncher.ts index 241584ec1a7..948144bf5bc 100644 --- a/apps/server/src/daemonLauncher.ts +++ b/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,12 +52,10 @@ 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 @@ -64,7 +63,7 @@ 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", diff --git a/apps/server/src/git/GitManager.ts b/apps/server/src/git/GitManager.ts index 8c4d00101d9..18b50c081ac 100644 --- a/apps/server/src/git/GitManager.ts +++ b/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; @@ -90,7 +91,6 @@ export class GitManager extends Context.Service()( "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) { diff --git a/apps/server/src/processRunner.ts b/apps/server/src/processRunner.ts index de547cf57cd..575ace2dd6e 100644 --- a/apps/server/src/processRunner.ts +++ b/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( effect: Effect.Effect, input: ProcessRunInput, ): Effect.Effect> { - 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, diff --git a/apps/server/src/provider/Layers/CliAdapter.ts b/apps/server/src/provider/Layers/CliAdapter.ts index 2329f829ae1..0052408da78 100644 --- a/apps/server/src/provider/Layers/CliAdapter.ts +++ b/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, diff --git a/apps/server/src/provider/Layers/CliProvider.ts b/apps/server/src/provider/Layers/CliProvider.ts index cca32f59a39..eb46708ec74 100644 --- a/apps/server/src/provider/Layers/CliProvider.ts +++ b/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, diff --git a/apps/server/src/provider/providerMaintenance.ts b/apps/server/src/provider/providerMaintenance.ts index 7f5e9d94dc5..bfa34fe4f16 100644 --- a/apps/server/src/provider/providerMaintenance.ts +++ b/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)) { diff --git a/apps/server/src/provider/providerMaintenanceRunner.ts b/apps/server/src/provider/providerMaintenanceRunner.ts index 0940986568b..d08155d69d4 100644 --- a/apps/server/src/provider/providerMaintenanceRunner.ts +++ b/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, diff --git a/apps/server/src/provider/providerSnapshot.ts b/apps/server/src/provider/providerSnapshot.ts index c40903e1b45..086005988d3 100644 --- a/apps/server/src/provider/providerSnapshot.ts +++ b/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 }, diff --git a/apps/server/src/ru-fork/preflight/common/checks.ts b/apps/server/src/ru-fork/preflight/common/checks.ts index cbb231d714d..afdb3ea45fb 100644 --- a/apps/server/src/ru-fork/preflight/common/checks.ts +++ b/apps/server/src/ru-fork/preflight/common/checks.ts @@ -8,7 +8,7 @@ import { NODE_ENGINE_RANGE, } from "./constants.ts"; import { MESSAGES } from "./messages.ts"; -import { probeVersion } from "./probe.ts"; +import { probeVersion, probeVersionByExit } from "./probe.ts"; import { render } from "./render.ts"; import type { CheckResult } from "./types.ts"; import { isAtLeast, satisfiesRange } from "./version.ts"; @@ -35,11 +35,23 @@ export const checkGit = (): CheckResult => { }; /** Probe the resolved cli.js directly with the running node interpreter. */ -export const checkCli = (cliJs: string): CheckResult => { - const cliProbe = probeVersion(process.execPath, [cliJs, "--version"], CLI_PROBE_TIMEOUT_MS); - if (!cliProbe.ok) return { ok: false, line: MESSAGES.CLI_BROKEN }; - if (CLI_MIN_VERSION && !isAtLeast(cliProbe.version, CLI_MIN_VERSION)) { - return { ok: false, line: render(MESSAGES.CLI_LOW, { found: cliProbe.version }) }; +export const checkCli = async (cliJs: string): Promise => { + const cliProbe = await probeVersionByExit( + process.execPath, + [cliJs, "--version"], + CLI_PROBE_TIMEOUT_MS, + ); + if (cliProbe.ok) { + if (CLI_MIN_VERSION && !isAtLeast(cliProbe.version, CLI_MIN_VERSION)) { + return { ok: false, line: render(MESSAGES.CLI_LOW, { found: cliProbe.version }) }; + } + return { ok: true, line: render(MESSAGES.CLI_OK, { found: cliProbe.version }) }; } - return { ok: true, line: render(MESSAGES.CLI_OK, { found: cliProbe.version }) }; + if (cliProbe.reason !== "timeout") { + return { ok: false, line: MESSAGES.CLI_BROKEN }; + } + // ru-fork: timeout = the process is too slow to EXIT. Held-pipe machines exit + // fast (probeVersionByExit measures the "exit" event) and pass; only genuinely + // slow-to-exit hardware reaches here. Gate the app off it. + return { ok: false, line: MESSAGES.CLI_TOO_SLOW }; }; diff --git a/apps/server/src/ru-fork/preflight/common/constants.ts b/apps/server/src/ru-fork/preflight/common/constants.ts index ba79284dd2a..64298fead54 100644 --- a/apps/server/src/ru-fork/preflight/common/constants.ts +++ b/apps/server/src/ru-fork/preflight/common/constants.ts @@ -17,6 +17,10 @@ export const NODE_ENGINE_RANGE = "^22.16 || ^23.11 || >=24.10"; /** Minimum CLI version; "" disables the version check (presence only). */ export const CLI_MIN_VERSION = "0.13.1"; -export const NODE_PROBE_TIMEOUT_MS = 5_000; -export const CLI_PROBE_TIMEOUT_MS = 15_000; -export const GIT_PROBE_TIMEOUT_MS = 5_000; +// ru-fork: this is a standalone, import-free preflight bundle — it CANNOT import +// apps/server/src/timeouts.ts, so these probe budgets are DUPLICATED there and +// must be kept in sync by hand: +// CLI_PROBE_TIMEOUT_MS ↔ timeouts.ts CLI_VERSION_PROBE_TIMEOUT_MS +// GIT_PROBE_TIMEOUT_MS ↔ timeouts.ts SOURCE_CONTROL_VERSION_PROBE_TIMEOUT_MS +export const CLI_PROBE_TIMEOUT_MS = 3_000; +export const GIT_PROBE_TIMEOUT_MS = 2_000; diff --git a/apps/server/src/ru-fork/preflight/common/messages.ts b/apps/server/src/ru-fork/preflight/common/messages.ts index ad90b83426d..bfc097b1352 100644 --- a/apps/server/src/ru-fork/preflight/common/messages.ts +++ b/apps/server/src/ru-fork/preflight/common/messages.ts @@ -1,6 +1,6 @@ // User-facing Russian strings. `{x}` placeholders are filled via render.ts. -import { CLI_NPM_PACKAGE } from "@ru-fork/branding"; +import { CLI_NAME, CLI_NPM_PACKAGE } from "@ru-fork/branding"; import { CLI_MIN_VERSION, NODE_ENGINE_RANGE } from "./constants.ts"; export const MESSAGES = { @@ -17,7 +17,8 @@ export const MESSAGES = { CLI_OK: "CLI {found} ✓", CLI_BROKEN: "CLI установлен, но `cli.js --version` завершилась с ошибкой или превысила тайм-аут.", CLI_LOW: `CLI {found} установлен, требуется ≥ ${CLI_MIN_VERSION}. Обновите: npm install -g ${CLI_NPM_PACKAGE}@latest`, - FOOTER_FAIL: "Установите недостающие компоненты и перезапустите.", + CLI_TOO_SLOW: `${CLI_NAME} работает слишком медленно на данном оборудовании, приложение будет работать не стабильно.`, + FOOTER_FAIL: "Устраните указанные выше проблемы и перезапустите.", USING_BACKUP_PRIORITY: "Используется резервный CLI (TRY_TO_FIND_CLI имеет приоритет над основным).", } as const; diff --git a/apps/server/src/ru-fork/preflight/common/probe.ts b/apps/server/src/ru-fork/preflight/common/probe.ts index 89ae4ec720d..a2b85fba7f8 100644 --- a/apps/server/src/ru-fork/preflight/common/probe.ts +++ b/apps/server/src/ru-fork/preflight/common/probe.ts @@ -1,11 +1,12 @@ // @effect-diagnostics nodeBuiltinImport:off +// @effect-diagnostics globalTimers:off // Preflight is deliberately Effect-free + node-builtins-only so it bundles into // one standalone file (dist/preflight.mjs) that runs before any deps exist. // // The only subprocess in the resolver: ` ` with a timeout, used // for node/git/cli version probes. Distinguishes missing / broken / timeout. -import { spawnSync } from "node:child_process"; +import { spawn, spawnSync } from "node:child_process"; import { extractVersion } from "./version.ts"; import type { ProbeResult } from "./types.ts"; @@ -35,3 +36,55 @@ export const probeVersion = ( const version = extractVersion(combinedOutput); return version ? { ok: true, version } : { ok: false, reason: "broken" }; }; + +// ru-fork: measure PROCESS-EXIT time (the "exit" event), NOT pipe close. A child +// that exits but whose stdout is held open (AV / daemon) still resolves here +// fast; only a process genuinely slow to exit trips the timeout. +export const probeVersionByExit = ( + command: string, + args: ReadonlyArray, + timeoutMs: number, +): Promise => + new Promise((resolve) => { + let stdout = ""; + let stderr = ""; + let settled = false; + let child: ReturnType | undefined; + const finish = (result: ProbeResult): void => { + if (settled) return; + settled = true; + clearTimeout(timer); + resolve(result); + }; + const timer = setTimeout(() => { + try { + child?.kill(); + } catch { + /* already gone */ + } + finish({ ok: false, reason: "timeout" }); + }, timeoutMs); + try { + const c = spawn(command, [...args], { windowsHide: true }); + child = c; + c.stdout?.on("data", (chunk: Buffer) => { + stdout += chunk.toString("utf8"); + }); + c.stderr?.on("data", (chunk: Buffer) => { + stderr += chunk.toString("utf8"); + }); + c.on("error", (error: NodeJS.ErrnoException) => { + finish({ ok: false, reason: error.code === "ENOENT" ? "missing" : "broken" }); + }); + c.on("exit", (status: number | null) => { + if (status !== 0) { + finish({ ok: false, reason: "broken" }); + return; + } + const version = extractVersion(`${stdout}${stderr}`); + finish(version ? { ok: true, version } : { ok: false, reason: "broken" }); + }); + } catch { + finish({ ok: false, reason: "broken" }); + } + }); diff --git a/apps/server/src/ru-fork/preflight/preflight-install.ts b/apps/server/src/ru-fork/preflight/preflight-install.ts index 26b21901b23..68b17e6ee99 100644 --- a/apps/server/src/ru-fork/preflight/preflight-install.ts +++ b/apps/server/src/ru-fork/preflight/preflight-install.ts @@ -33,7 +33,7 @@ const writeError = (line: string): void => const exitFailure = (): never => process.exit(1); -const main = (): void => { +const main = async (): Promise => { // Diagnostics first — so even an early STOP carries the full environment. writeInfo("Диагностика:"); for (const line of collectDiagnostics()) writeInfo(` ${line}`); @@ -81,7 +81,7 @@ const main = (): void => { // the failure reads as "node", not a confusing cli error. (nodeCheck computed // above for NODE_OK.) const cliCheck: CheckResult = nodeCheck.ok - ? checkCli(cliJs) + ? await checkCli(cliJs) : { ok: false, line: "CLI: пропущено — Node.js не соответствует требованиям" }; const gitCheck = checkGit(); @@ -98,4 +98,4 @@ const main = (): void => { } }; -main(); +void main(); diff --git a/apps/server/src/ru-fork/preflight/preflight-startup.ts b/apps/server/src/ru-fork/preflight/preflight-startup.ts index dd5ceaff6ef..b900fd377cd 100644 --- a/apps/server/src/ru-fork/preflight/preflight-startup.ts +++ b/apps/server/src/ru-fork/preflight/preflight-startup.ts @@ -94,7 +94,7 @@ export const resolveStartupCli: Effect.Effect => Effect.gen(function* () { - const results = [checkNodeEngine(), checkGit(), checkCli(cliJs)]; + const results = [checkNodeEngine(), checkGit(), yield* Effect.promise(() => checkCli(cliJs))]; for (const result of results) { if (result.ok) yield* Effect.logInfo(result.line); else yield* Effect.logError(result.line); diff --git a/apps/server/src/ru-fork/spawn/haltOnExit.ts b/apps/server/src/ru-fork/spawn/haltOnExit.ts new file mode 100644 index 00000000000..efb73bc0dc8 --- /dev/null +++ b/apps/server/src/ru-fork/spawn/haltOnExit.ts @@ -0,0 +1,24 @@ +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Stream from "effect/Stream"; + +import { EXIT_DRAIN_GRACE_MS } from "../../timeouts.ts"; + +// ru-fork: end a child's stdio stream when the PROCESS EXITS (+ a short drain), +// instead of waiting for stdio EOF — which corporate AV / a git daemon / a stuck +// async `end` can hold open. exitCode resolves on the Node "exit" event, +// independent of the pipe (verified: probe-interruptwhen.mjs). On a healthy +// machine EOF ends the stream first → zero added latency, identical output. +export const haltOnExit = + (exitCode: Effect.Effect) => + (stream: Stream.Stream): Stream.Stream => + stream.pipe( + Stream.interruptWhen( + // Effect.ignore: settle on exit success OR failure (we only want the + // timing); never let the signal fail the stream. + exitCode.pipe( + Effect.ignore, + Effect.flatMap(() => Effect.sleep(Duration.millis(EXIT_DRAIN_GRACE_MS))), + ), + ), + ); diff --git a/apps/server/src/ru-fork/terminal/logTerminalStartFailure.ts b/apps/server/src/ru-fork/terminal/logTerminalStartFailure.ts new file mode 100644 index 00000000000..1c33ec52026 --- /dev/null +++ b/apps/server/src/ru-fork/terminal/logTerminalStartFailure.ts @@ -0,0 +1,35 @@ +// ru-fork: log terminal open()/restart() failures to the server at ERROR level. +// +// Upstream's RPC observability wrapper (`observeRpcEffect` in `ws.ts`) was +// stripped to a no-op when telemetry was removed, so a failed terminal open was +// logged NOWHERE — the real reason (bad cwd, unreadable history, a defect) was +// invisible on the server while the UI showed a generic "Failed to open +// terminal". This pipe-able transform taps the failure cause, logs the real +// reason once, and RE-RAISES it unchanged — the typed `TerminalError` still +// propagates, so every caller (the run-a-command callers, the RPC handler, the +// drawer) keeps its existing error handling. A cancelled request +// (interrupt-only cause) is not logged as an error. + +import { DEFAULT_TERMINAL_ID } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Effect from "effect/Effect"; + +export const logTerminalStartFailure = + (input: { + readonly threadId: string; + readonly terminalId?: string | undefined; + readonly cwd: string; + }) => + (work: Effect.Effect): Effect.Effect => + work.pipe( + Effect.tapCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.void + : Effect.logError("terminal start failed", { + threadId: input.threadId, + terminalId: input.terminalId ?? DEFAULT_TERMINAL_ID, + cwd: input.cwd, + cause: Cause.pretty(cause), + }), + ), + ); diff --git a/apps/server/src/sourceControl/GitHubCli.ts b/apps/server/src/sourceControl/GitHubCli.ts index 14d01aab2ed..b2ffc97f4d7 100644 --- a/apps/server/src/sourceControl/GitHubCli.ts +++ b/apps/server/src/sourceControl/GitHubCli.ts @@ -13,8 +13,7 @@ import { import * as VcsProcess from "../vcs/VcsProcess.ts"; import * as GitHubPullRequests from "./gitHubPullRequests.ts"; - -const DEFAULT_TIMEOUT_MS = 30_000; +import { SOURCE_CONTROL_DEFAULT_TIMEOUT_MS } from "../timeouts.ts"; export class GitHubCliError extends Schema.TaggedErrorClass()("GitHubCliError", { operation: Schema.String, @@ -236,7 +235,7 @@ export const make = Effect.fn("makeGitHubCli")(function* () { command: "gh", args: input.args, cwd: input.cwd, - timeoutMs: input.timeoutMs ?? DEFAULT_TIMEOUT_MS, + timeoutMs: input.timeoutMs ?? SOURCE_CONTROL_DEFAULT_TIMEOUT_MS, }) .pipe(Effect.mapError((error) => normalizeGitHubCliError("execute", error))); diff --git a/apps/server/src/sourceControl/SourceControlDiscovery.ts b/apps/server/src/sourceControl/SourceControlDiscovery.ts index 8aa07fa5f8c..12921502ac5 100644 --- a/apps/server/src/sourceControl/SourceControlDiscovery.ts +++ b/apps/server/src/sourceControl/SourceControlDiscovery.ts @@ -13,6 +13,10 @@ import * as Option from "effect/Option"; import { ServerConfig } from "../config.ts"; import * as VcsProcess from "../vcs/VcsProcess.ts"; +import { + SOURCE_CONTROL_AUTH_PROBE_TIMEOUT_MS, + SOURCE_CONTROL_VERSION_PROBE_TIMEOUT_MS, +} from "../timeouts.ts"; interface DiscoveryProbe { readonly label: string; @@ -324,7 +328,7 @@ export const layer = Layer.effect( command: input.executable, args: input.versionArgs, cwd: config.cwd, - timeoutMs: 5_000, + timeoutMs: SOURCE_CONTROL_VERSION_PROBE_TIMEOUT_MS, maxOutputBytes: 8_000, truncateOutputAtMaxBytes: true, }) @@ -375,7 +379,7 @@ export const layer = Layer.effect( args: input.authArgs, cwd: config.cwd, allowNonZeroExit: true, - timeoutMs: 5_000, + timeoutMs: SOURCE_CONTROL_AUTH_PROBE_TIMEOUT_MS, maxOutputBytes: 8_000, truncateOutputAtMaxBytes: true, }) diff --git a/apps/server/src/sourceControl/SourceControlProviderDiscovery.ts b/apps/server/src/sourceControl/SourceControlProviderDiscovery.ts index 425c4f61778..766bebb3ced 100644 --- a/apps/server/src/sourceControl/SourceControlProviderDiscovery.ts +++ b/apps/server/src/sourceControl/SourceControlProviderDiscovery.ts @@ -7,6 +7,10 @@ import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import type * as VcsProcess from "../vcs/VcsProcess.ts"; +import { + SOURCE_CONTROL_AUTH_PROBE_TIMEOUT_MS, + SOURCE_CONTROL_VERSION_PROBE_TIMEOUT_MS, +} from "../timeouts.ts"; export interface SourceControlAuthProbeInput { readonly stdout: string; @@ -137,7 +141,7 @@ function probeCli(input: { command: input.spec.executable, args: input.spec.versionArgs, cwd: input.cwd, - timeoutMs: 5_000, + timeoutMs: SOURCE_CONTROL_VERSION_PROBE_TIMEOUT_MS, maxOutputBytes: 8_000, truncateOutputAtMaxBytes: true, }) @@ -214,7 +218,7 @@ export function probeSourceControlProvider(input: { args: spec.authArgs, cwd: input.cwd, allowNonZeroExit: true, - timeoutMs: 5_000, + timeoutMs: SOURCE_CONTROL_AUTH_PROBE_TIMEOUT_MS, maxOutputBytes: 8_000, truncateOutputAtMaxBytes: true, }) diff --git a/apps/server/src/sourceControl/SourceControlRepositoryService.ts b/apps/server/src/sourceControl/SourceControlRepositoryService.ts index f1ee1940390..a7e3d685d8c 100644 --- a/apps/server/src/sourceControl/SourceControlRepositoryService.ts +++ b/apps/server/src/sourceControl/SourceControlRepositoryService.ts @@ -20,6 +20,7 @@ import { } from "@t3tools/contracts"; import { ServerConfig } from "../config.ts"; +import { SOURCE_CONTROL_REPO_OP_TIMEOUT_MS } from "../timeouts.ts"; import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import * as SourceControlProviderRegistry from "./SourceControlProviderRegistry.ts"; const isSourceControlRepositoryError = Schema.is(SourceControlRepositoryError); @@ -235,7 +236,7 @@ export const make = Effect.fn("makeSourceControlRepositoryService")(function* () operation: "SourceControlRepositoryService.cloneRepository", cwd: preparedDestination.parentPath, args: ["clone", remoteUrl, preparedDestination.directoryName], - timeoutMs: 120_000, + timeoutMs: SOURCE_CONTROL_REPO_OP_TIMEOUT_MS, maxOutputBytes: 256 * 1024, }); diff --git a/apps/server/src/terminal/Layers/Manager.ts b/apps/server/src/terminal/Layers/Manager.ts index 8585aca30e1..33680cfbd07 100644 --- a/apps/server/src/terminal/Layers/Manager.ts +++ b/apps/server/src/terminal/Layers/Manager.ts @@ -21,12 +21,20 @@ import * as Semaphore from "effect/Semaphore"; import * as SynchronizedRef from "effect/SynchronizedRef"; import { ServerConfig } from "../../config.ts"; +// ru-fork: log terminal start failures to the server (the RPC observability +// seam is a no-op since telemetry was removed). See the module for details. +import { logTerminalStartFailure } from "../../ru-fork/terminal/logTerminalStartFailure.ts"; import { increment, terminalRestartsTotal, terminalSessionsTotal, } from "../../observability/Metrics.ts"; import * as ProcessRunner from "../../processRunner.ts"; +import { + TERMINAL_BUSY_CHECK_TIMEOUT_MS, + TERMINAL_KILL_GRACE_MS, + TERMINAL_SUBPROCESS_CHECK_TIMEOUT_MS, +} from "../../timeouts.ts"; import { TerminalCwdError, TerminalHistoryError, @@ -46,7 +54,6 @@ import { const DEFAULT_HISTORY_LINE_LIMIT = 5_000; const DEFAULT_PERSIST_DEBOUNCE_MS = 40; const DEFAULT_SUBPROCESS_POLL_INTERVAL_MS = 1_000; -const DEFAULT_PROCESS_KILL_GRACE_MS = 1_000; const DEFAULT_MAX_RETAINED_INACTIVE_SESSIONS = 128; const DEFAULT_OPEN_COLS = 120; const DEFAULT_OPEN_ROWS = 30; @@ -397,7 +404,7 @@ const checkWindowsSubprocessActivity = Effect.fn("terminal.checkWindowsSubproces .run({ command: "powershell.exe", args: ["-NoProfile", "-NonInteractive", "-Command", command], - timeout: 1_500, + timeout: TERMINAL_BUSY_CHECK_TIMEOUT_MS, maxOutputBytes: 32_768, outputMode: "truncate", timeoutBehavior: "timedOutResult", @@ -425,7 +432,7 @@ const checkPosixSubprocessActivity = Effect.fn("terminal.checkPosixSubprocessAct .run({ command: "pgrep", args: ["-P", String(terminalPid)], - timeout: 1_000, + timeout: TERMINAL_SUBPROCESS_CHECK_TIMEOUT_MS, maxOutputBytes: 32_768, outputMode: "truncate", timeoutBehavior: "timedOutResult", @@ -446,7 +453,7 @@ const checkPosixSubprocessActivity = Effect.fn("terminal.checkPosixSubprocessAct .run({ command: "ps", args: ["-eo", "pid=,ppid="], - timeout: 1_000, + timeout: TERMINAL_SUBPROCESS_CHECK_TIMEOUT_MS, maxOutputBytes: 262_144, outputMode: "truncate", timeoutBehavior: "timedOutResult", @@ -780,7 +787,7 @@ export const makeTerminalManagerWithOptions = Effect.fn("makeTerminalManagerWith const subprocessChecker = options.subprocessChecker ?? defaultSubprocessChecker; const subprocessPollIntervalMs = options.subprocessPollIntervalMs ?? DEFAULT_SUBPROCESS_POLL_INTERVAL_MS; - const processKillGraceMs = options.processKillGraceMs ?? DEFAULT_PROCESS_KILL_GRACE_MS; + const processKillGraceMs = options.processKillGraceMs ?? TERMINAL_KILL_GRACE_MS; const maxRetainedInactiveSessions = options.maxRetainedInactiveSessions ?? DEFAULT_MAX_RETAINED_INACTIVE_SESSIONS; @@ -1803,7 +1810,7 @@ export const makeTerminalManagerWithOptions = Effect.fn("makeTerminalManagerWith } return snapshot(liveSession); - }), + }).pipe(logTerminalStartFailure(input)), ); const write: TerminalManagerShape["write"] = Effect.fn("terminal.write")(function* (input) { @@ -1935,7 +1942,7 @@ export const makeTerminalManagerWithOptions = Effect.fn("makeTerminalManagerWith "restarted", ); return snapshot(session); - }), + }).pipe(logTerminalStartFailure(input)), ); const close: TerminalManagerShape["close"] = (input) => diff --git a/apps/server/src/textGeneration/CliTextGeneration.ts b/apps/server/src/textGeneration/CliTextGeneration.ts index dc600e8eaef..73c864d3259 100644 --- a/apps/server/src/textGeneration/CliTextGeneration.ts +++ b/apps/server/src/textGeneration/CliTextGeneration.ts @@ -27,6 +27,8 @@ import { TextGenerationError } from "@t3tools/contracts"; // ru-fork: CLI is launched as `node …` directly. See // ru-fork/spawn/policy.ts buildCliSpawn. import { buildCliSpawn } from "../ru-fork/spawn/policy.ts"; +import { CLI_TEXT_GENERATION_TIMEOUT_MS } from "../timeouts.ts"; +import { haltOnExit } from "../ru-fork/spawn/haltOnExit.ts"; import { expandHomePath } from "../pathExpansion.ts"; import { type TextGenerationShape } from "./TextGeneration.ts"; import { @@ -67,8 +69,6 @@ const buildCliSingleStringPrompt = (input: { return sections.join("\n"); }; -const CLI_TIMEOUT_MS = 180_000; - // Precompiled JSON decoders for Cli's structured-output operations. Hoisted to // module scope so neither the schema literal nor the compiled decoder is // rebuilt per call (see oxlint t3code/no-inline-schema-compile and the same @@ -173,8 +173,8 @@ export const makeCliTextGeneration = Effect.fn("makeCliTextGeneration")(function const [stdout, stderr, exitCode] = yield* Effect.all( [ - readStreamAsString(input.operation, child.stdout), - readStreamAsString(input.operation, child.stderr), + readStreamAsString(input.operation, child.stdout.pipe(haltOnExit(child.exitCode))), + readStreamAsString(input.operation, child.stderr.pipe(haltOnExit(child.exitCode))), child.exitCode.pipe( Effect.mapError((cause) => normalizeCliError(CLI_NAME, input.operation, cause, "Failed to read Cli CLI exit code"), @@ -203,7 +203,7 @@ export const makeCliTextGeneration = Effect.fn("makeCliTextGeneration")(function const runCliWithTimeout = (input: { operation: CliOperation; cwd: string; prompt: string }) => runCliCommand(input).pipe( Effect.scoped, - Effect.timeoutOption(CLI_TIMEOUT_MS), + Effect.timeoutOption(CLI_TEXT_GENERATION_TIMEOUT_MS), Effect.flatMap( Option.match({ onNone: () => diff --git a/apps/server/src/timeouts.ts b/apps/server/src/timeouts.ts new file mode 100644 index 00000000000..12592930a79 --- /dev/null +++ b/apps/server/src/timeouts.ts @@ -0,0 +1,81 @@ +// SINGLE SOURCE OF TRUTH for every server-side timeout. All values in ms. +// ru-fork: no hardcoded timeout literal anywhere else in apps/server/src +// (intervals/TTLs are a different concept and stay where they are). +// Preflight keeps its own ru-fork/preflight/common/constants.ts (standalone bundle). + +// subprocess collect (the exit-gated-collect fix) +export const EXIT_DRAIN_GRACE_MS = 250; + +export const PROCESS_RUNNER_DEFAULT_TIMEOUT_MS = 60_000; + +// CLI provider / text-gen / provider maintenance +// ru-fork: KEEP IN SYNC with ru-fork/preflight/common/constants.ts → +// CLI_PROBE_TIMEOUT_MS. Same ` --version` probe; preflight is a standalone +// import-free bundle, so the value is duplicated there, not shared. +export const CLI_VERSION_PROBE_TIMEOUT_MS = 3_000; + +export const CLI_TEXT_GENERATION_TIMEOUT_MS = 180_000; +export const PROVIDER_UPDATE_TIMEOUT_MS = 300_000; +export const PROVIDER_LATEST_VERSION_TIMEOUT_MS = 4_000; // npm-registry "latest" fetch + +// git (by current value — behavior-preserving tiers) +export const GIT_TIMEOUT_FAST_MS = 5_000; +export const GIT_TIMEOUT_STANDARD_MS = 10_000; +export const GIT_TIMEOUT_FETCH_MS = 15_000; +export const GIT_TIMEOUT_NETWORK_MS = 20_000; +export const GIT_TIMEOUT_DEFAULT_MS = 30_000; +export const GIT_COMMIT_TIMEOUT_MS = 600_000; +export const GIT_STATUS_UPSTREAM_REFRESH_TIMEOUT_MS = 15_000; + +// source control +export const SOURCE_CONTROL_DEFAULT_TIMEOUT_MS = 30_000; +// ru-fork: SOURCE_CONTROL_VERSION_PROBE_TIMEOUT_MS — KEEP IN SYNC with +// ru-fork/preflight/common/constants.ts → GIT_PROBE_TIMEOUT_MS. Same ` +// --version` probe; preflight duplicates the value (can't import this file). +export const SOURCE_CONTROL_VERSION_PROBE_TIMEOUT_MS = 2_000; // ` --version` (git/gh/…), local +export const SOURCE_CONTROL_AUTH_PROBE_TIMEOUT_MS = 5_000; // ` auth status`, network +export const SOURCE_CONTROL_REPO_OP_TIMEOUT_MS = 120_000; + +// terminal +export const TERMINAL_BUSY_CHECK_TIMEOUT_MS = 1_500; +export const TERMINAL_SUBPROCESS_CHECK_TIMEOUT_MS = 1_000; +export const TERMINAL_KILL_GRACE_MS = 1_000; + +// daemon / CLI live-server probe +export const DAEMON_HEALTH_PROBE_TIMEOUT_MS = 1_000; +export const DAEMON_SPAWN_TIMEOUT_MS = 5_000; +export const PROJECT_CLI_LIVE_SERVER_TIMEOUT_MS = 1_000; + +// ACP wire-stall / post-answer resume (moved verbatim from config.ts). +/** + * 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; diff --git a/apps/server/src/vcs/GitVcsDriver.ts b/apps/server/src/vcs/GitVcsDriver.ts index 2bd2d5c00da..8d788b360eb 100644 --- a/apps/server/src/vcs/GitVcsDriver.ts +++ b/apps/server/src/vcs/GitVcsDriver.ts @@ -25,6 +25,11 @@ import { import * as GitVcsDriverCore from "./GitVcsDriverCore.ts"; import * as VcsDriver from "./VcsDriver.ts"; import * as VcsProcess from "./VcsProcess.ts"; +import { + GIT_TIMEOUT_FAST_MS, + GIT_TIMEOUT_NETWORK_MS, + GIT_TIMEOUT_STANDARD_MS, +} from "../timeouts.ts"; export interface ExecuteGitInput { readonly operation: string; @@ -357,7 +362,7 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* ( ["rev-parse", "--is-inside-work-tree"], { allowNonZeroExit: true, - timeoutMs: 5_000, + timeoutMs: GIT_TIMEOUT_FAST_MS, maxOutputBytes: 4_096, }, ).pipe(Effect.map((result) => result.exitCode === 0 && result.stdout.trim() === "true")); @@ -415,7 +420,7 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* ( ], { allowNonZeroExit: true, - timeoutMs: 20_000, + timeoutMs: GIT_TIMEOUT_NETWORK_MS, maxOutputBytes: WORKSPACE_FILES_MAX_OUTPUT_BYTES, truncateOutputAtMaxBytes: true, }, @@ -446,7 +451,7 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* ( function* (cwd) { const result = yield* gitCommand(process, "GitVcsDriver.listRemotes", cwd, ["remote", "-v"], { allowNonZeroExit: true, - timeoutMs: 5_000, + timeoutMs: GIT_TIMEOUT_FAST_MS, maxOutputBytes: 64 * 1024, }); @@ -501,7 +506,7 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* ( { stdin: `${chunk.join("\0")}\0`, allowNonZeroExit: true, - timeoutMs: 20_000, + timeoutMs: GIT_TIMEOUT_NETWORK_MS, maxOutputBytes: WORKSPACE_FILES_MAX_OUTPUT_BYTES, truncateOutputAtMaxBytes: true, }, @@ -531,7 +536,7 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* ( const initRepository: VcsDriver.VcsDriverShape["initRepository"] = (input) => gitCommand(process, "GitVcsDriver.initRepository", input.cwd, ["init"], { - timeoutMs: 10_000, + timeoutMs: GIT_TIMEOUT_STANDARD_MS, maxOutputBytes: 64 * 1024, }).pipe(Effect.asVoid); diff --git a/apps/server/src/vcs/GitVcsDriverCore.ts b/apps/server/src/vcs/GitVcsDriverCore.ts index 2c7ed73013d..a65c12af57e 100644 --- a/apps/server/src/vcs/GitVcsDriverCore.ts +++ b/apps/server/src/vcs/GitVcsDriverCore.ts @@ -31,6 +31,7 @@ import { // resolved shell value stays false unless --windows-use-bash-for // explicitly lists "git". See ru-fork/spawn/policy.ts. import { resolveSpawn } from "../ru-fork/spawn/policy.ts"; +import { haltOnExit } from "../ru-fork/spawn/haltOnExit.ts"; // ru-fork: exclude Windows reserved-name junk (e.g. a stray `nul`) from // staging so it can't abort `git add -A`. See git-issues.md. import { WINDOWS_RESERVED_EXCLUDES } from "../ru-fork/vcs/reservedNames.ts"; @@ -38,9 +39,15 @@ import { ServerConfig } from "../config.ts"; // ru-fork: needed so `refreshStatusUpstreamIfStale` can short-circuit when // the user sets `automaticGitFetchInterval` to 0. import { ServerSettingsService } from "../serverSettings.ts"; +import { + GIT_STATUS_UPSTREAM_REFRESH_TIMEOUT_MS, + GIT_TIMEOUT_DEFAULT_MS, + GIT_TIMEOUT_FAST_MS, + GIT_TIMEOUT_FETCH_MS, + GIT_TIMEOUT_STANDARD_MS, +} from "../timeouts.ts"; const isGitCommandError = Schema.is(GitCommandError); -const DEFAULT_TIMEOUT_MS = 30_000; const DEFAULT_MAX_OUTPUT_BYTES = 1_000_000; const OUTPUT_TRUNCATED_MARKER = "\n\n[truncated]"; const PREPARED_COMMIT_PATCH_MAX_OUTPUT_BYTES = 49_000; @@ -48,9 +55,6 @@ const RANGE_COMMIT_SUMMARY_MAX_OUTPUT_BYTES = 19_000; const RANGE_DIFF_SUMMARY_MAX_OUTPUT_BYTES = 19_000; const RANGE_DIFF_PATCH_MAX_OUTPUT_BYTES = 59_000; const STATUS_UPSTREAM_REFRESH_INTERVAL = Duration.seconds(15); -// ru-fork: 5s was too tight — proxy HTTPS fetch can legitimately -// exceed that on the first round trip. 15s gives the network real headroom. -const STATUS_UPSTREAM_REFRESH_TIMEOUT = Duration.seconds(15); // ru-fork: on failure, back off for 15s (matching the success cadence) // instead of 5s so we don't hot-loop and spam the log every few seconds. const STATUS_UPSTREAM_REFRESH_FAILURE_COOLDOWN = Duration.seconds(15); @@ -634,7 +638,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* ...input, args: [...input.args], } as const; - const timeoutMs = input.timeoutMs ?? DEFAULT_TIMEOUT_MS; + const timeoutMs = input.timeoutMs ?? GIT_TIMEOUT_DEFAULT_MS; const maxOutputBytes = input.maxOutputBytes ?? DEFAULT_MAX_OUTPUT_BYTES; const truncateOutputAtMaxBytes = input.truncateOutputAtMaxBytes ?? false; @@ -666,14 +670,14 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* [ collectOutput( commandInput, - child.stdout, + child.stdout.pipe(haltOnExit(child.exitCode)), maxOutputBytes, truncateOutputAtMaxBytes, input.progress?.onStdoutLine, ), collectOutput( commandInput, - child.stderr, + child.stderr.pipe(haltOnExit(child.exitCode)), maxOutputBytes, truncateOutputAtMaxBytes, input.progress?.onStderrLine, @@ -834,7 +838,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* ["show-ref", "--verify", "--quiet", `refs/heads/${refName}`], { allowNonZeroExit: true, - timeoutMs: 5_000, + timeoutMs: GIT_TIMEOUT_FAST_MS, }, ).pipe(Effect.map((result) => result.exitCode === 0)); @@ -898,7 +902,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* { allowNonZeroExit: true, env: STATUS_UPSTREAM_REFRESH_ENV, - timeoutMs: Duration.toMillis(STATUS_UPSTREAM_REFRESH_TIMEOUT), + timeoutMs: GIT_STATUS_UPSTREAM_REFRESH_TIMEOUT_MS, }, ).pipe(Effect.asVoid); }; @@ -1178,7 +1182,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* "refs/remotes", ], { - timeoutMs: 15_000, + timeoutMs: GIT_TIMEOUT_FETCH_MS, allowNonZeroExit: true, }, ); @@ -1621,7 +1625,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* true, ).pipe(Effect.map((stdout) => stdout.trim())); yield* executeGit("GitVcsDriver.pullCurrentBranch.pull", cwd, ["pull", "--ff-only"], { - timeoutMs: 30_000, + timeoutMs: GIT_TIMEOUT_DEFAULT_MS, fallbackErrorMessage: "git pull failed", }); const afterSha = yield* runGitStdout( @@ -1699,7 +1703,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* input.cwd, ["branch", "--no-color", "--no-column"], { - timeoutMs: 10_000, + timeoutMs: GIT_TIMEOUT_STANDARD_MS, allowNonZeroExit: true, }, ).pipe( @@ -1738,7 +1742,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* input.cwd, ["branch", "--no-color", "--no-column", "--remotes"], { - timeoutMs: 10_000, + timeoutMs: GIT_TIMEOUT_STANDARD_MS, allowNonZeroExit: true, }, ).pipe( @@ -1762,7 +1766,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* input.cwd, ["remote"], { - timeoutMs: 5_000, + timeoutMs: GIT_TIMEOUT_FAST_MS, allowNonZeroExit: true, }, ).pipe( @@ -1789,7 +1793,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* input.cwd, ["symbolic-ref", "refs/remotes/origin/HEAD"], { - timeoutMs: 5_000, + timeoutMs: GIT_TIMEOUT_FAST_MS, allowNonZeroExit: true, }, ), @@ -1798,7 +1802,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* input.cwd, ["worktree", "list", "--porcelain"], { - timeoutMs: 5_000, + timeoutMs: GIT_TIMEOUT_FAST_MS, allowNonZeroExit: true, }, ), @@ -2014,7 +2018,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* } args.push(input.path); yield* executeGit("GitVcsDriver.removeWorktree", input.cwd, args, { - timeoutMs: 15_000, + timeoutMs: GIT_TIMEOUT_FETCH_MS, fallbackErrorMessage: "git worktree remove failed", }).pipe( Effect.mapError((error) => @@ -2041,7 +2045,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* input.cwd, ["branch", "-m", "--", input.oldBranch, targetBranch], { - timeoutMs: 10_000, + timeoutMs: GIT_TIMEOUT_STANDARD_MS, fallbackErrorMessage: "git branch rename failed", }, ); @@ -2059,7 +2063,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* input.cwd, ["show-ref", "--verify", "--quiet", `refs/heads/${input.refName}`], { - timeoutMs: 5_000, + timeoutMs: GIT_TIMEOUT_FAST_MS, allowNonZeroExit: true, }, ).pipe(Effect.map((result) => result.exitCode === 0)), @@ -2068,7 +2072,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* input.cwd, ["show-ref", "--verify", "--quiet", `refs/remotes/${input.refName}`], { - timeoutMs: 5_000, + timeoutMs: GIT_TIMEOUT_FAST_MS, allowNonZeroExit: true, }, ).pipe(Effect.map((result) => result.exitCode === 0)), @@ -2082,7 +2086,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* input.cwd, ["for-each-ref", "--format=%(refname:short)\t%(upstream:short)", "refs/heads"], { - timeoutMs: 5_000, + timeoutMs: GIT_TIMEOUT_FAST_MS, allowNonZeroExit: true, }, ).pipe( @@ -2102,7 +2106,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* input.cwd, ["show-ref", "--verify", "--quiet", `refs/heads/${localTrackedBranchCandidate}`], { - timeoutMs: 5_000, + timeoutMs: GIT_TIMEOUT_FAST_MS, allowNonZeroExit: true, }, ).pipe(Effect.map((result) => result.exitCode === 0)) @@ -2119,7 +2123,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* : ["checkout", input.refName]; yield* executeGit("GitVcsDriver.switchRef.checkout", input.cwd, checkoutArgs, { - timeoutMs: 10_000, + timeoutMs: GIT_TIMEOUT_STANDARD_MS, fallbackErrorMessage: "git checkout failed", }); @@ -2135,7 +2139,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* const createRef: GitVcsDriver.GitVcsDriverShape["createRef"] = Effect.fn("createRef")( function* (input) { yield* executeGit("GitVcsDriver.createRef", input.cwd, ["branch", input.refName], { - timeoutMs: 10_000, + timeoutMs: GIT_TIMEOUT_STANDARD_MS, fallbackErrorMessage: "git branch create failed", }); if (input.switchRef) { @@ -2148,7 +2152,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* const initRepo: GitVcsDriver.GitVcsDriverShape["initRepo"] = (input) => executeGit("GitVcsDriver.initRepo", input.cwd, ["init"], { - timeoutMs: 10_000, + timeoutMs: GIT_TIMEOUT_STANDARD_MS, fallbackErrorMessage: "git init failed", }).pipe(Effect.asVoid); diff --git a/apps/server/src/vcs/VcsProcess.ts b/apps/server/src/vcs/VcsProcess.ts index e708b09f216..240d4d41b8f 100644 --- a/apps/server/src/vcs/VcsProcess.ts +++ b/apps/server/src/vcs/VcsProcess.ts @@ -20,6 +20,8 @@ import { // --windows-use-bash-for. 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 { GIT_TIMEOUT_DEFAULT_MS } from "../timeouts.ts"; export interface VcsProcessInput { readonly operation: string; @@ -68,7 +70,6 @@ export class VcsProcess extends Context.Service()( "t3/vcs/VcsProcess", ) {} -const DEFAULT_TIMEOUT_MS = 30_000; const DEFAULT_MAX_OUTPUT_BYTES = 1_000_000; const OUTPUT_TRUNCATED_MARKER = "\n\n[truncated]"; @@ -168,7 +169,7 @@ export const make = Effect.fn("makeVcsProcess")(function* () { Effect.scoped(spawn(input).pipe(Effect.flatMap(use))); const run = Effect.fn("VcsProcess.run")(function* (input: VcsProcessInput) { - const timeoutMs = input.timeoutMs ?? DEFAULT_TIMEOUT_MS; + const timeoutMs = input.timeoutMs ?? GIT_TIMEOUT_DEFAULT_MS; const maxOutputBytes = input.maxOutputBytes ?? DEFAULT_MAX_OUTPUT_BYTES; const label = commandLabel(input.command, input.args); @@ -180,7 +181,7 @@ export const make = Effect.fn("makeVcsProcess")(function* () { operation: input.operation, command: label, cwd: input.cwd, - stream: child.stdout, + stream: child.stdout.pipe(haltOnExit(child.exitCode)), maxOutputBytes, truncateOutputAtMaxBytes: input.truncateOutputAtMaxBytes ?? false, }), @@ -188,7 +189,7 @@ export const make = Effect.fn("makeVcsProcess")(function* () { operation: input.operation, command: label, cwd: input.cwd, - stream: child.stderr, + stream: child.stderr.pipe(haltOnExit(child.exitCode)), maxOutputBytes, truncateOutputAtMaxBytes: input.truncateOutputAtMaxBytes ?? false, }), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index f95a424170a..2793752150a 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -46,27 +46,39 @@ import { Open, resolveAvailableEditors } from "./open.ts"; import { normalizeDispatchCommand } from "./orchestration/Normalizer.ts"; import { OrchestrationEngineService } from "./orchestration/Services/OrchestrationEngine.ts"; import { ProjectionSnapshotQuery } from "./orchestration/Services/ProjectionSnapshotQuery.ts"; -// Telemetry/observability removed: keep the call shape but pass through. -// `observeRpcStreamEffect` MUST `Stream.unwrap` the effect — the RPC -// handler signature for streaming methods expects a `Stream`, not an -// `Effect`. Returning the Effect raw breaks every stream -// subscription (subscribeServerConfig, orchestration shell, etc.). +// ru-fork: telemetry/observability (spans + metrics) was removed, and the +// per-RPC FAILURE CAPTURE that lived in this wrapper went with it — so every +// RPC failed silently server-side. These wrappers restore just the failure +// log (no OTel): a real bug (defect/die) → ERROR; an expected typed failure +// or a cancellation → DEBUG, so the error log stays signal-rich. `_attributes` +// is kept for call-shape compatibility. See changes/rpc-failure-logging.md. +// `observeRpcStreamEffect` MUST `Stream.unwrap` the effect — the RPC handler +// signature for streaming methods expects a `Stream`, not an `Effect`. +const logRpcCause = (method: string, cause: Cause.Cause): Effect.Effect => + Cause.hasInterruptsOnly(cause) + ? Effect.logDebug("RPC прерван", { method }) + : Cause.hasDies(cause) + ? Effect.logError("Сбой RPC (defect)", { method, cause: Cause.pretty(cause) }) + : Effect.logDebug("Сбой RPC", { method, cause: Cause.pretty(cause) }); + const observeRpcEffect = ( - _method: string, + method: string, effect: Effect.Effect, _attributes?: Readonly>, -): Effect.Effect => effect; +): Effect.Effect => effect.pipe(Effect.tapCause((cause) => logRpcCause(method, cause))); const observeRpcStream = ( - _method: string, + method: string, stream: Stream.Stream, _attributes?: Readonly>, -): Stream.Stream => stream; +): Stream.Stream => stream.pipe(Stream.tapCause((cause) => logRpcCause(method, cause))); const observeRpcStreamEffect = ( - _method: string, + method: string, effect: Effect.Effect, EffectError, EffectContext>, _attributes?: Readonly>, ): Stream.Stream => - Stream.unwrap(effect); + Stream.unwrap(effect.pipe(Effect.tapCause((cause) => logRpcCause(method, cause)))).pipe( + Stream.tapCause((cause) => logRpcCause(method, cause)), + ); import { ProviderRegistry } from "./provider/Services/ProviderRegistry.ts"; import * as ProviderMaintenanceRunner from "./provider/providerMaintenanceRunner.ts"; // ru-fork: filesystem skill scanner — owned RPCs (server.listSkillsForCwd, diff --git a/apps/server/tests/haltOnExit.test.ts b/apps/server/tests/haltOnExit.test.ts new file mode 100644 index 00000000000..e5a0f5ebd92 --- /dev/null +++ b/apps/server/tests/haltOnExit.test.ts @@ -0,0 +1,61 @@ +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Stream from "effect/Stream"; + +import { haltOnExit } from "../src/ru-fork/spawn/haltOnExit.ts"; + +// A stream that emits a known prefix then blocks forever (Stream.never), +// modelling a child whose pipe never reaches EOF. +const prefixThenHang = Stream.concat(Stream.make("a", "b", "c"), Stream.never); + +it.live("ends on exit + drain and keeps the prefix (runFold)", () => + Effect.gen(function* () { + const out = yield* prefixThenHang.pipe( + haltOnExit(Effect.void), + Stream.runFold( + () => "", + (acc, chunk) => acc + chunk, + ), + ); + assert.strictEqual(out, "abc"); + }), +); + +it.live("works with a mutable accumulator (runForEach)", () => + Effect.gen(function* () { + let acc = ""; + const halted = prefixThenHang.pipe(haltOnExit(Effect.void)); + yield* Stream.runForEach(halted, (chunk) => + Effect.sync(() => { + acc += chunk; + }), + ); + assert.strictEqual(acc, "abc"); + }), +); + +it.live("works with an effectful fold (runFoldEffect)", () => + Effect.gen(function* () { + const out = yield* prefixThenHang.pipe( + haltOnExit(Effect.void), + Stream.runFoldEffect( + () => "", + (acc, chunk) => Effect.succeed(acc + chunk), + ), + ); + assert.strictEqual(out, "abc"); + }), +); + +it.live("a failing exit signal still settles the stream (Effect.ignore)", () => + Effect.gen(function* () { + const out = yield* prefixThenHang.pipe( + haltOnExit(Effect.fail("boom")), + Stream.runFold( + () => "", + (acc, chunk) => acc + chunk, + ), + ); + assert.strictEqual(out, "abc"); + }), +); diff --git a/apps/server/tests/preflightCheckCli.test.ts b/apps/server/tests/preflightCheckCli.test.ts new file mode 100644 index 00000000000..1c6c7189ffd --- /dev/null +++ b/apps/server/tests/preflightCheckCli.test.ts @@ -0,0 +1,45 @@ +// @effect-diagnostics nodeBuiltinImport:off +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; + +import { checkCli } from "../src/ru-fork/preflight/common/checks.ts"; +import { MESSAGES } from "../src/ru-fork/preflight/common/messages.ts"; + +let dir: string; +const cli = (name: string): string => join(dir, name); + +beforeAll(() => { + dir = mkdtempSync(join(tmpdir(), "ru-fork-checkcli-")); + writeFileSync(cli("ok.js"), "process.stdout.write('9.9.9')"); // >= CLI_MIN_VERSION + writeFileSync(cli("low.js"), "process.stdout.write('0.1.0')"); // < CLI_MIN_VERSION + writeFileSync(cli("broken.js"), "process.exit(2)"); + writeFileSync(cli("slow.js"), "setTimeout(function(){}, 10000)"); // never exits in time +}); + +afterAll(() => { + rmSync(dir, { recursive: true, force: true }); +}); + +describe("checkCli", () => { + it("ok for a fast cli at/above the minimum version", async () => { + const result = await checkCli(cli("ok.js")); + expect(result.ok).toBe(true); + }); + + it("fails (CLI_LOW) for a version below the minimum", async () => { + const result = await checkCli(cli("low.js")); + expect(result.ok).toBe(false); + }); + + it("fails (CLI_BROKEN) for a non-zero exit", async () => { + const result = await checkCli(cli("broken.js")); + expect(result).toEqual({ ok: false, line: MESSAGES.CLI_BROKEN }); + }); + + it("fails (CLI_TOO_SLOW) when the cli is too slow to exit", async () => { + const result = await checkCli(cli("slow.js")); + expect(result).toEqual({ ok: false, line: MESSAGES.CLI_TOO_SLOW }); + }); +}); diff --git a/apps/server/tests/preflightProbeByExit.test.ts b/apps/server/tests/preflightProbeByExit.test.ts new file mode 100644 index 00000000000..827e59b2ec6 --- /dev/null +++ b/apps/server/tests/preflightProbeByExit.test.ts @@ -0,0 +1,59 @@ +// @effect-diagnostics nodeBuiltinImport:off +import { execPath } from "node:process"; +import { describe, expect, it } from "vitest"; + +import { probeVersionByExit } from "../src/ru-fork/preflight/common/probe.ts"; + +const TIMEOUT = 3_000; + +describe("probeVersionByExit", () => { + it("ok+version when the process exits fast with a version", async () => { + const result = await probeVersionByExit( + execPath, + ["-e", "process.stdout.write('1.2.3')"], + TIMEOUT, + ); + expect(result).toEqual({ ok: true, version: "1.2.3" }); + }); + + it("broken when the process exits non-zero", async () => { + const result = await probeVersionByExit(execPath, ["-e", "process.exit(1)"], TIMEOUT); + expect(result).toEqual({ ok: false, reason: "broken" }); + }); + + it("broken when the process exits 0 but prints no version", async () => { + const result = await probeVersionByExit( + execPath, + ["-e", "process.stdout.write('hello')"], + TIMEOUT, + ); + expect(result).toEqual({ ok: false, reason: "broken" }); + }); + + it("missing when the command does not exist (ENOENT)", async () => { + const result = await probeVersionByExit("definitely-not-a-real-binary-xyz", ["--version"], TIMEOUT); + expect(result).toEqual({ ok: false, reason: "missing" }); + }); + + // CASE A — the gate must NOT reject this: process exits fast but leaks stdout + // to a detached grandchild that holds the pipe open. Measuring "exit" → ok. + it("ok when the process exits fast but a child holds stdout open", async () => { + const held = [ + "process.stdout.write('2.0.0');", + "require('child_process').spawn(process.execPath, ['-e', 'setTimeout(function(){}, 3000)'], { stdio: ['ignore', 1, 'ignore'], detached: true }).unref();", + "process.exit(0);", + ].join(" "); + const result = await probeVersionByExit(execPath, ["-e", held], TIMEOUT); + expect(result).toEqual({ ok: true, version: "2.0.0" }); + }); + + // CASE B — the gate's reject path: genuinely slow to EXIT trips the timeout. + it("timeout when the process is slow to exit", async () => { + const result = await probeVersionByExit( + execPath, + ["-e", "setTimeout(function(){}, 5000)"], + 500, + ); + expect(result).toEqual({ ok: false, reason: "timeout" }); + }); +}); diff --git a/apps/server/tests/ru-fork/logTerminalStartFailure.test.ts b/apps/server/tests/ru-fork/logTerminalStartFailure.test.ts new file mode 100644 index 00000000000..b607e89caaf --- /dev/null +++ b/apps/server/tests/ru-fork/logTerminalStartFailure.test.ts @@ -0,0 +1,50 @@ +// ru-fork: proof that a failed terminal open is LOGGED to the server (the RPC +// observability seam is a no-op since telemetry was removed) AND that the +// failure is re-raised unchanged — so every caller keeps its error handling and +// the typed TerminalError still propagates. +import { describe, expect, it } from "vitest"; +import * as Effect from "effect/Effect"; +import * as Logger from "effect/Logger"; + +import { logTerminalStartFailure } from "../../src/ru-fork/terminal/logTerminalStartFailure.ts"; + +function captureLogs() { + const lines: string[] = []; + const logger = Logger.make(({ message }) => { + lines.push(Array.isArray(message) ? message.map((part) => String(part)).join(" ") : String(message)); + }); + return { lines, layer: Logger.layer([logger], { mergeWithExisting: false }) }; +} + +describe("logTerminalStartFailure", () => { + it("logs the real reason and re-raises the failure unchanged", async () => { + const { lines, layer } = captureLogs(); + const cause = new Error("Terminal cwd does not exist: C:/x"); + + const exit = await Effect.runPromiseExit( + Effect.fail(cause).pipe( + logTerminalStartFailure({ threadId: "thread-1", cwd: "C:/x" }), + Effect.provide(layer), + ), + ); + + // re-raised, not swallowed + expect(exit._tag).toBe("Failure"); + // logged at the server + expect(lines.some((line) => line.includes("terminal start failed"))).toBe(true); + }); + + it("does not log on success", async () => { + const { lines, layer } = captureLogs(); + + const value = await Effect.runPromise( + Effect.succeed(42).pipe( + logTerminalStartFailure({ threadId: "thread-1", cwd: "C:/x" }), + Effect.provide(layer), + ), + ); + + expect(value).toBe(42); + expect(lines).toHaveLength(0); + }); +}); diff --git a/apps/server/tests/ru-fork/terminalErrors.test.ts b/apps/server/tests/ru-fork/terminalErrors.test.ts new file mode 100644 index 00000000000..eba12bafac5 --- /dev/null +++ b/apps/server/tests/ru-fork/terminalErrors.test.ts @@ -0,0 +1,48 @@ +// ru-fork: the terminal-error formatter rebuilds the localized, user-facing +// message from a WIRE-DECODED error object (the tagged error's `message` getter +// is lost over the RPC boundary). This is the proof that a failed terminal open +// surfaces the REAL reason in the UI instead of a generic fallback. +import { describe, expect, it } from "vitest"; +import { terminalErrorMessage } from "@t3tools/contracts"; + +describe("terminalErrorMessage", () => { + it("formats a missing cwd (notFound)", () => { + expect( + terminalErrorMessage({ _tag: "TerminalCwdError", cwd: "C:/x", reason: "notFound" }), + ).toBe("Рабочий каталог терминала не существует: C:/x"); + }); + + it("formats a cwd that is not a directory", () => { + expect( + terminalErrorMessage({ _tag: "TerminalCwdError", cwd: "C:/x", reason: "notDirectory" }), + ).toBe("Рабочий каталог терминала не является папкой: C:/x"); + }); + + it("formats a stat failure, including the underlying cause message", () => { + expect( + terminalErrorMessage({ + _tag: "TerminalCwdError", + cwd: "C:/x", + reason: "statFailed", + cause: new Error("PermissionDenied"), + }), + ).toBe("Не удалось открыть рабочий каталог терминала: C:/x (PermissionDenied)"); + }); + + it("formats a stat failure without a usable cause message", () => { + expect( + terminalErrorMessage({ _tag: "TerminalCwdError", cwd: "C:/x", reason: "statFailed" }), + ).toBe("Не удалось открыть рабочий каталог терминала: C:/x"); + }); + + it("formats a history error", () => { + expect(terminalErrorMessage({ _tag: "TerminalHistoryError", operation: "read" })).toBe( + "Не удалось загрузить историю терминала (операция: read).", + ); + }); + + it("returns undefined for an unknown tag so callers fall back", () => { + expect(terminalErrorMessage({ _tag: "SomethingElse" })).toBeUndefined(); + expect(terminalErrorMessage({})).toBeUndefined(); + }); +}); diff --git a/apps/web/src/components/CommandPalette.tsx b/apps/web/src/components/CommandPalette.tsx index 2294dafbb0f..70a976260ed 100644 --- a/apps/web/src/components/CommandPalette.tsx +++ b/apps/web/src/components/CommandPalette.tsx @@ -35,6 +35,7 @@ import { import { useShallow } from "zustand/react/shallow"; import { useCommandPaletteStore } from "../commandPaletteStore"; import { readEnvironmentApi } from "../environmentApi"; +import { wsDebug } from "../ru-fork/debugging"; import { readPrimaryEnvironmentDescriptor, usePrimaryEnvironmentId } from "../environments/primary"; import { useSavedEnvironmentRegistryStore, @@ -727,9 +728,16 @@ function OpenCommandPaletteDialog() { const handleAddProject = useCallback( async (rawCwd: string) => { - if (!browseEnvironmentId) return; + wsDebug("addProject: click", { rawCwd, browseEnvironmentId }); + if (!browseEnvironmentId) { + wsDebug("addProject: abort no-env"); + return; + } const api = readEnvironmentApi(browseEnvironmentId); - if (!api) return; + if (!api) { + wsDebug("addProject: abort no-api"); + return; + } if (isUnsupportedWindowsProjectPath(rawCwd.trim(), browseEnvironmentPlatform)) { toastManager.add( @@ -754,7 +762,10 @@ function OpenCommandPaletteDialog() { } const cwd = resolveProjectPathForDispatch(rawCwd, currentProjectCwdForBrowse); - if (cwd.length === 0) return; + if (cwd.length === 0) { + wsDebug("addProject: abort empty-cwd"); + return; + } const existing = findProjectByPath( projects.filter((project) => project.environmentId === browseEnvironmentId), @@ -784,6 +795,7 @@ function OpenCommandPaletteDialog() { try { const projectId = newProjectId(); + wsDebug("addProject: dispatching project.create", { projectId, cwd }); await api.orchestration.dispatchCommand({ type: "project.create", commandId: newCommandId(), @@ -797,11 +809,15 @@ function OpenCommandPaletteDialog() { }, createdAt: new Date().toISOString(), }); + wsDebug("addProject: dispatch RESOLVED", { projectId }); await handleNewThread(scopeProjectRef(browseEnvironmentId, projectId), { envMode: settings.defaultThreadEnvMode, }).catch(() => undefined); setOpen(false); } catch (error) { + wsDebug("addProject: dispatch REJECTED", { + error: error instanceof Error ? error.message : String(error), + }); toastManager.add( stackedThreadToast({ type: "error", diff --git a/apps/web/src/components/ThreadTerminalDrawer.tsx b/apps/web/src/components/ThreadTerminalDrawer.tsx index 166ad3b240f..c0da1b4b146 100644 --- a/apps/web/src/components/ThreadTerminalDrawer.tsx +++ b/apps/web/src/components/ThreadTerminalDrawer.tsx @@ -4,6 +4,7 @@ import { type ResolvedKeybindingsConfig, type ScopedThreadRef, type TerminalEvent, + terminalErrorMessage, type TerminalSessionSnapshot, type ThreadId, } from "@t3tools/contracts"; @@ -702,10 +703,14 @@ export function TerminalViewport({ } } catch (err) { if (disposed) return; - writeSystemMessage( - terminal, - err instanceof Error ? err.message : "Failed to open terminal", - ); + // ru-fork: rebuild the real reason from the wire-decoded terminal error + // (the tagged error's `message` getter is lost over the RPC boundary), + // falling back to a plain Error message, then a generic string. + const detail = + terminalErrorMessage(err as { _tag?: string }) ?? + (err instanceof Error ? err.message : undefined) ?? + "Не удалось открыть терминал."; + writeSystemMessage(terminal, detail); } }; diff --git a/apps/web/src/environments/runtime/service.ts b/apps/web/src/environments/runtime/service.ts index affca269bdf..b8539207d7f 100644 --- a/apps/web/src/environments/runtime/service.ts +++ b/apps/web/src/environments/runtime/service.ts @@ -97,6 +97,8 @@ const lastAppliedProjectionVersionByEnvironment = new Map< let activeService: EnvironmentServiceState | null = null; let needsProviderInvalidation = false; +let lastBrowserHiddenAt: number | null = null; +let lastBrowserResumeReconnectAt = Number.NEGATIVE_INFINITY; // Thread detail subscription cache policy: // - Active consumers keep a subscription retained via refCount. @@ -108,6 +110,7 @@ let needsProviderInvalidation = false; const THREAD_DETAIL_SUBSCRIPTION_IDLE_EVICTION_MS = 15 * 60 * 1000; const MAX_CACHED_THREAD_DETAIL_SUBSCRIPTIONS = 32; const INITIAL_SERVER_CONFIG_SNAPSHOT_WAIT_MS = 150; +const BROWSER_RESUME_RECONNECT_COOLDOWN_MS = 2_000; const NOOP = () => undefined; function createDeferredPromise() { @@ -1074,6 +1077,58 @@ function stopActiveService() { activeService = null; } +function reconnectEnvironmentConnectionsAfterBrowserResume(reason: string): void { + const now = Date.now(); + if (now - lastBrowserResumeReconnectAt < BROWSER_RESUME_RECONNECT_COOLDOWN_MS) { + return; + } + + for (const connection of environmentConnections.values()) { + if (connection.client.isHeartbeatFresh()) { + continue; + } + lastBrowserResumeReconnectAt = now; + void connection.reconnect().catch((error) => { + console.warn("Environment reconnect after browser resume failed", { + environmentId: connection.environmentId, + reason, + error: error instanceof Error ? error.message : String(error), + }); + }); + } +} + +function subscribeBrowserResumeReconnects(): () => void { + if (typeof document === "undefined" || typeof window === "undefined") { + return NOOP; + } + + const handleVisibilityChange = () => { + if (document.visibilityState === "hidden") { + lastBrowserHiddenAt = Date.now(); + return; + } + if (document.visibilityState === "visible" && lastBrowserHiddenAt !== null) { + lastBrowserHiddenAt = null; + reconnectEnvironmentConnectionsAfterBrowserResume("visibilitychange"); + } + }; + + const handlePageShow = (event: PageTransitionEvent) => { + if (event.persisted || lastBrowserHiddenAt !== null) { + lastBrowserHiddenAt = null; + reconnectEnvironmentConnectionsAfterBrowserResume("pageshow"); + } + }; + + document.addEventListener("visibilitychange", handleVisibilityChange); + window.addEventListener("pageshow", handlePageShow); + return () => { + document.removeEventListener("visibilitychange", handleVisibilityChange); + window.removeEventListener("pageshow", handlePageShow); + }; +} + export function subscribeEnvironmentConnections(listener: () => void): () => void { environmentConnectionListeners.add(listener); return () => { @@ -1252,12 +1307,15 @@ export function startEnvironmentConnectionService(queryClient: QueryClient): () .then(() => requestSavedEnvironmentSync()) .catch(() => undefined); + const unsubscribeBrowserResumeReconnects = subscribeBrowserResumeReconnects(); + activeService = { queryClient, queryInvalidationThrottler, refCount: 1, stop: () => { unsubscribeSavedEnvironments(); + unsubscribeBrowserResumeReconnects(); queryInvalidationThrottler.cancel(); }, }; @@ -1275,6 +1333,8 @@ export function startEnvironmentConnectionService(queryClient: QueryClient): () export async function resetEnvironmentServiceForTests(): Promise { stopActiveService(); + lastBrowserHiddenAt = null; + lastBrowserResumeReconnectAt = Number.NEGATIVE_INFINITY; lastAppliedProjectionVersionByEnvironment.clear(); for (const key of Array.from(threadDetailSubscriptions.keys())) { disposeThreadDetailSubscriptionByKey(key); diff --git a/apps/web/src/rpc/protocol.ts b/apps/web/src/rpc/protocol.ts index 124b40d7d55..de416afb404 100644 --- a/apps/web/src/rpc/protocol.ts +++ b/apps/web/src/rpc/protocol.ts @@ -19,13 +19,31 @@ import { recordWsConnectionOpened, WS_RECONNECT_MAX_RETRIES, } from "./wsConnectionState"; +import { wsDebug } from "../ru-fork/debugging"; + +// ru-fork: connection-robustness restored from 8fc31793 (it was removed by +// c36945d8 "…+ cleanup"). The intentional-close flag lets the transport mark +// deliberate reconnect/dispose closes so they are NOT recorded as a disconnect +// (a deliberate close being misread re-triggered the retry and leaked a zombie +// socket). The telemetry/metadata that the cleanup also removed (connection +// label, version-mismatch hint, client tracing) is intentionally NOT restored. +export interface WsProtocolCloseContext { + readonly intentional: boolean; +} export interface WsProtocolLifecycleHandlers { readonly isActive?: () => boolean; + readonly isCloseIntentional?: () => boolean; readonly onAttempt?: (socketUrl: string) => void; readonly onOpen?: () => void; + readonly onHeartbeatPing?: () => void; + readonly onHeartbeatPong?: () => void; + readonly onHeartbeatTimeout?: () => void; readonly onError?: (message: string) => void; - readonly onClose?: (details: { readonly code: number; readonly reason: string }) => void; + readonly onClose?: ( + details: { readonly code: number; readonly reason: string }, + context: WsProtocolCloseContext, + ) => void; } export const makeWsRpcProtocolClient = RpcClient.make(WsRpcGroup); @@ -56,7 +74,11 @@ function resolveWsRpcSocketUrl(rawUrl: string): string { return resolved.toString(); } -function defaultLifecycleHandlers(): Required { +type ComposedWsProtocolLifecycleHandlers = Required< + Pick +>; + +function defaultLifecycleHandlers(): ComposedWsProtocolLifecycleHandlers { return { isActive: () => true, onAttempt: recordWsConnectionAttempt, @@ -65,8 +87,13 @@ function defaultLifecycleHandlers(): Required { clearAllTrackedRpcRequests(); recordWsConnectionErrored(message); }, - onClose: (details) => { + onClose: (details, context) => { clearAllTrackedRpcRequests(); + // ru-fork: a deliberate close (reconnect/dispose) is not a disconnect — + // don't record it (else it feeds the retry and leaves a zombie socket). + if (context.intentional) { + return; + } recordWsConnectionClosed(details); }, }; @@ -74,7 +101,7 @@ function defaultLifecycleHandlers(): Required { function composeLifecycleHandlers( handlers?: WsProtocolLifecycleHandlers, -): Required { +): ComposedWsProtocolLifecycleHandlers { const defaults = defaultLifecycleHandlers(); const isActive = handlers?.isActive ?? (() => true); @@ -101,12 +128,12 @@ function composeLifecycleHandlers( defaults.onError(message); handlers?.onError?.(message); }, - onClose: (details) => { + onClose: (details, context) => { if (!isActive()) { return; } - defaults.onClose(details); - handlers?.onClose?.(details); + defaults.onClose(details, context); + handlers?.onClose?.(details, context); }, }; } @@ -132,12 +159,14 @@ export function createWsRpcProtocolLayer( const trackingWebSocketConstructorLayer = Layer.succeed( Socket.WebSocketConstructor, (socketUrl, protocols) => { + wsDebug("ws attempt", { socketUrl }); lifecycle.onAttempt(socketUrl); const socket = new globalThis.WebSocket(socketUrl, protocols); socket.addEventListener( "open", () => { + wsDebug("ws open", { socketUrl }); lifecycle.onOpen(); }, { once: true }, @@ -145,6 +174,7 @@ export function createWsRpcProtocolLayer( socket.addEventListener( "error", () => { + wsDebug("ws error", { socketUrl }); lifecycle.onError("Unable to connect to the T3 server WebSocket."); }, { once: true }, @@ -152,10 +182,17 @@ export function createWsRpcProtocolLayer( socket.addEventListener( "close", (event) => { - lifecycle.onClose({ - code: event.code, - reason: event.reason, - }); + const intentional = handlers?.isCloseIntentional?.() ?? false; + wsDebug("ws close", { code: event.code, reason: event.reason, intentional }); + lifecycle.onClose( + { + code: event.code, + reason: event.reason, + }, + { + intentional, + }, + ); }, { once: true }, ); @@ -181,14 +218,17 @@ export function createWsRpcProtocolLayer( run: (clientId, writeResponse) => protocol.run(clientId, (response) => { if (response._tag === "Chunk" || response._tag === "Exit") { + wsDebug("rpc ←", { tag: response._tag, requestId: response.requestId }); acknowledgeRpcRequest(response.requestId); } else if (response._tag === "ClientProtocolError" || response._tag === "Defect") { + wsDebug("rpc ← error", { tag: response._tag }); clearAllTrackedRpcRequests(); } return writeResponse(response); }), send: (clientId, request, transferables) => { if (request._tag === "Request") { + wsDebug("rpc →", { id: request.id, tag: request.tag }); trackRpcRequestSent(request.id, request.tag); } return protocol.send(clientId, request, transferables); @@ -196,6 +236,39 @@ export function createWsRpcProtocolLayer( }), ), ); + // ru-fork: heartbeat hooks restored from 8fc31793. The RPC client pings the + // server every ~5s; on a missed pong it fires onPingTimeout. We surface + // ping/pong so the transport can track heartbeat freshness (used to skip + // needless reconnects) and clear in-flight requests on a real timeout. + const connectionHooksLayer = Layer.succeed( + RpcClient.ConnectionHooks, + RpcClient.ConnectionHooks.of({ + onConnect: Effect.void, + onDisconnect: Effect.void, + onPing: Effect.sync(() => { + wsDebug("hb ping", { active: lifecycle.isActive() }); + if (lifecycle.isActive()) { + handlers?.onHeartbeatPing?.(); + } + }), + onPong: Effect.sync(() => { + wsDebug("hb pong", { active: lifecycle.isActive() }); + if (lifecycle.isActive()) { + handlers?.onHeartbeatPong?.(); + } + }), + onPingTimeout: Effect.sync(() => { + wsDebug("hb TIMEOUT", { active: lifecycle.isActive() }); + if (lifecycle.isActive()) { + clearAllTrackedRpcRequests(); + recordWsConnectionErrored("WebSocket heartbeat timed out."); + handlers?.onHeartbeatTimeout?.(); + } + }), + }), + ); - return protocolLayer.pipe(Layer.provide(Layer.mergeAll(socketLayer, RpcSerialization.layerJson))); + return protocolLayer.pipe( + Layer.provide(Layer.mergeAll(socketLayer, RpcSerialization.layerJson, connectionHooksLayer)), + ); } diff --git a/apps/web/src/rpc/transportError.ts b/apps/web/src/rpc/transportError.ts index edc90a5a3c9..83f8aa86d4d 100644 --- a/apps/web/src/rpc/transportError.ts +++ b/apps/web/src/rpc/transportError.ts @@ -1,6 +1,7 @@ const TRANSPORT_ERROR_PATTERNS = [ /\bSocketCloseError\b/i, /\bSocketOpenError\b/i, + /\bSocket is not connected\b/i, /Unable to connect to the T3 server WebSocket\./i, /\bping timeout\b/i, ] as const; diff --git a/apps/web/src/rpc/wsRpcClient.ts b/apps/web/src/rpc/wsRpcClient.ts index 13bf1f09ee7..409f9e3b16c 100644 --- a/apps/web/src/rpc/wsRpcClient.ts +++ b/apps/web/src/rpc/wsRpcClient.ts @@ -56,6 +56,7 @@ interface GitRunStackedActionOptions { export interface WsRpcClient { readonly dispose: () => Promise; readonly reconnect: () => Promise; + readonly isHeartbeatFresh: () => boolean; readonly terminal: { readonly open: RpcUnaryMethod; readonly write: RpcUnaryMethod; @@ -159,6 +160,7 @@ export function createWsRpcClient(transport: WsTransport): WsRpcClient { resetWsReconnectBackoff(); await transport.reconnect(); }, + isHeartbeatFresh: () => transport.isHeartbeatFresh(), terminal: { open: (input) => transport.request((client) => client[WS_METHODS.terminalOpen](input)), write: (input) => transport.request((client) => client[WS_METHODS.terminalWrite](input)), diff --git a/apps/web/src/rpc/wsTransport.ts b/apps/web/src/rpc/wsTransport.ts index 582c8096b26..8e88582f907 100644 --- a/apps/web/src/rpc/wsTransport.ts +++ b/apps/web/src/rpc/wsTransport.ts @@ -17,6 +17,7 @@ import { type WsRpcProtocolSocketUrlProvider, } from "./protocol"; import { isTransportConnectionErrorMessage } from "./transportError"; +import { wsDebug } from "../ru-fork/debugging"; interface SubscribeOptions { readonly retryDelay?: Duration.Input; @@ -52,6 +53,11 @@ export class WsTransport { private reconnectChain: Promise = Promise.resolve(); private nextSessionId = 0; private activeSessionId = 0; + // ru-fork: restored from 8fc31793 (removed in c36945d8). intentionalCloseDepth + // marks a deliberate close (reconnect/dispose) so it isn't misread as a + // disconnect; lastHeartbeatPongAt powers isHeartbeatFresh. + private intentionalCloseDepth = 0; + private lastHeartbeatPongAt = 0; private session: TransportSession; constructor( @@ -131,6 +137,10 @@ export class WsTransport { } } + wsDebug("subscribe: stream start", { + tag: options?.tag, + sessionId: this.activeSessionId, + }); const runningStream = this.runStreamOnSession( session, connect, @@ -144,13 +154,20 @@ export class WsTransport { cancelCurrentStream = runningStream.cancel; await runningStream.completed; cancelCurrentStream = NOOP; + wsDebug("subscribe: stream completed", { tag: options?.tag }); } catch (error) { cancelCurrentStream = NOOP; if (!active || this.disposed) { return; } + wsDebug("subscribe: stream ended", { + tag: options?.tag, + error: formatErrorMessage(error), + }); + if (session !== this.session) { + wsDebug("subscribe: RE-ATTACH", { tag: options?.tag, to: this.activeSessionId }); continue; } @@ -168,6 +185,10 @@ export class WsTransport { }); } this.hasReportedTransportDisconnect = true; + wsDebug("subscribe: retry same session", { + tag: options?.tag, + sessionId: this.activeSessionId, + }); await sleep(retryDelayMs); } } @@ -190,8 +211,11 @@ export class WsTransport { } clearAllTrackedRpcRequests(); + this.lastHeartbeatPongAt = 0; const previousSession = this.session; + const previousSessionId = this.activeSessionId; this.session = this.createSession(); + wsDebug("transport.reconnect", { from: previousSessionId, to: this.activeSessionId }); await this.closeSession(previousSession); }); @@ -207,8 +231,17 @@ export class WsTransport { await this.closeSession(this.session); } + isHeartbeatFresh(maxAgeMs = 15_000): boolean { + return this.lastHeartbeatPongAt > 0 && Date.now() - this.lastHeartbeatPongAt <= maxAgeMs; + } + private closeSession(session: TransportSession) { + // ru-fork: mark this teardown intentional so the socket's close event is + // not recorded as a disconnect (which would re-trigger the retry / leak). + this.intentionalCloseDepth += 1; + wsDebug("closeSession", { intentionalCloseDepth: this.intentionalCloseDepth }); return session.runtime.runPromise(Scope.close(session.clientScope, Exit.void)).finally(() => { + this.intentionalCloseDepth = Math.max(0, this.intentionalCloseDepth - 1); session.runtime.dispose(); }); } @@ -217,10 +250,21 @@ export class WsTransport { const sessionId = this.nextSessionId + 1; this.nextSessionId = sessionId; this.activeSessionId = sessionId; + wsDebug("session created", { sessionId }); const runtime = ManagedRuntime.make( createWsRpcProtocolLayer(this.url, { ...this.lifecycleHandlers, isActive: () => !this.disposed && this.activeSessionId === sessionId, + // ru-fork: a close is intentional if we're disposing, mid-reconnect + // teardown, or the caller says so — restored from 8fc31793. + isCloseIntentional: () => + this.disposed || + this.intentionalCloseDepth > 0 || + this.lifecycleHandlers?.isCloseIntentional?.() === true, + onHeartbeatPong: () => { + this.lastHeartbeatPongAt = Date.now(); + this.lifecycleHandlers?.onHeartbeatPong?.(); + }, }), ); const clientScope = runtime.runSync(Scope.make()); diff --git a/apps/web/src/ru-fork/debugging/index.ts b/apps/web/src/ru-fork/debugging/index.ts new file mode 100644 index 00000000000..e31367abbed --- /dev/null +++ b/apps/web/src/ru-fork/debugging/index.ts @@ -0,0 +1,60 @@ +// ru-fork: lightweight, gated client-side debug logging shared across layers. +// +// Enable in the browser DevTools console with: +// localStorage.debugging = "ws" (then reload) +// The value is a comma/space-separated list of layers, so combine as needed: +// localStorage.debugging = "ws,mcp" +// Disable with: +// delete localStorage.debugging (then reload) +// +// When a layer is not listed, its logs are a no-op with zero console noise. The +// value is read once and cached, so toggling requires a reload — intentional, so +// hot paths (e.g. per-RPC `rpc →`/`rpc ←`) cost nothing in normal use. +// +// WS instrumentation context: ru-fork-instrumental/connection-stability/networking.md §7. + +export const DebuggingLayer = { + ws: "ws", + mcp: "mcp", +} as const; + +export type DebuggingLayer = (typeof DebuggingLayer)[keyof typeof DebuggingLayer]; + +let activeLayers: ReadonlySet | undefined; + +function getActiveLayers(): ReadonlySet { + if (activeLayers === undefined) { + try { + const raw = typeof localStorage !== "undefined" ? localStorage.getItem("debugging") : null; + activeLayers = new Set(raw ? raw.split(/[\s,]+/).filter(Boolean) : []); + } catch { + activeLayers = new Set(); + } + } + return activeLayers; +} + +export function isDebugEnabled(layer: DebuggingLayer): boolean { + return getActiveLayers().has(layer); +} + +export function debugLog( + layer: DebuggingLayer, + message: string, + data?: Record, +): void { + if (!isDebugEnabled(layer)) { + return; + } + const prefix = `[${layer}] ${message}`; + if (data === undefined) { + console.debug(prefix); + } else { + console.debug(prefix, data); + } +} + +// Convenience wrapper for the WS layer (used by the rpc transport instrumentation). +export function wsDebug(message: string, data?: Record): void { + debugLog(DebuggingLayer.ws, message, data); +} diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 05ef4a12142..21265c8b177 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -14,6 +14,7 @@ export * from "./server.ts"; // don't trip on these additions. export * from "./ru-fork/skills.ts"; export * from "./ru-fork/subagents.ts"; +export * from "./ru-fork/terminalErrors.ts"; export * from "./settings.ts"; export * from "./git.ts"; export * from "./vcs.ts"; diff --git a/packages/contracts/src/ru-fork/terminalErrors.ts b/packages/contracts/src/ru-fork/terminalErrors.ts new file mode 100644 index 00000000000..97c2cf24e9d --- /dev/null +++ b/packages/contracts/src/ru-fork/terminalErrors.ts @@ -0,0 +1,51 @@ +// ru-fork: localized (Russian) user-facing messages for terminal errors. +// +// Lives here — NOT inline in the upstream `terminal.ts` — so re-syncs from +// upstream t3code don't conflict, and so the WEB can rebuild the message from a +// wire-decoded error object: the class `message` getters in `terminal.ts` are +// lost over the RPC boundary (only the schema data fields survive), which is +// why a failed terminal open showed a generic "Failed to open terminal" +// instead of the real reason. +// +// Returns `undefined` for an unrecognized tag so callers can fall back to a +// generic message (or the raw `Error.message`). + +export interface TerminalErrorShape { + readonly _tag?: string; + readonly cwd?: string; + readonly reason?: string; + readonly operation?: string; + readonly cause?: unknown; +} + +export function terminalErrorMessage(error: TerminalErrorShape): string | undefined { + switch (error._tag) { + case "TerminalCwdError": { + const cwd = error.cwd ?? ""; + if (error.reason === "notDirectory") { + return `Рабочий каталог терминала не является папкой: ${cwd}`; + } + if (error.reason === "notFound") { + return `Рабочий каталог терминала не существует: ${cwd}`; + } + const causeMessage = + error.cause !== undefined && + error.cause !== null && + typeof error.cause === "object" && + "message" in error.cause + ? (error.cause as { readonly message?: unknown }).message + : undefined; + return typeof causeMessage === "string" && causeMessage.length > 0 + ? `Не удалось открыть рабочий каталог терминала: ${cwd} (${causeMessage})` + : `Не удалось открыть рабочий каталог терминала: ${cwd}`; + } + case "TerminalHistoryError": + return `Не удалось загрузить историю терминала (операция: ${error.operation ?? "?"}).`; + case "TerminalSessionLookupError": + return "Сессия терминала не найдена."; + case "TerminalNotRunningError": + return "Терминал не запущен."; + default: + return undefined; + } +} diff --git a/patches/effect@4.0.0-beta.59.patch b/patches/effect@4.0.0-beta.59.patch new file mode 100644 index 00000000000..479004acf33 --- /dev/null +++ b/patches/effect@4.0.0-beta.59.patch @@ -0,0 +1,277 @@ +diff --git a/dist/unstable/rpc/RpcClient.d.ts b/dist/unstable/rpc/RpcClient.d.ts +index 7c5970042df67a25b4145df017960b3e6febed9c..7af3374632d53ad59f24e80835aa6741853b1f2b 100644 +--- a/dist/unstable/rpc/RpcClient.d.ts ++++ b/dist/unstable/rpc/RpcClient.d.ts +@@ -2,6 +2,7 @@ import * as Cause from "../../Cause.ts"; + import * as Context from "../../Context.ts"; + import type * as Duration from "../../Duration.ts"; + import * as Effect from "../../Effect.ts"; ++import * as Exit from "../../Exit.ts"; + import * as Layer from "../../Layer.ts"; + import * as Queue from "../../Queue.ts"; + import * as Schedule from "../../Schedule.ts"; +@@ -192,9 +193,40 @@ export declare const layerProtocolWorker: (options: { + readonly targetUtilization?: number | undefined; + readonly timeToLive: Duration.Input; + }) => Layer.Layer; ++declare const RequestHooks_base: Context.ServiceClass Effect.Effect) | undefined; ++ readonly onRequestChunk?: ((info: { ++ readonly id: RequestId; ++ readonly tag: string; ++ readonly chunkCount: number; ++ }) => Effect.Effect) | undefined; ++ readonly onRequestExit?: ((info: { ++ readonly id: RequestId; ++ readonly tag: string; ++ readonly stream: boolean; ++ readonly exit: Exit.Exit; ++ }) => Effect.Effect) | undefined; ++ readonly onRequestInterrupt?: ((info: { ++ readonly id: RequestId; ++ readonly tag?: string | undefined; ++ }) => Effect.Effect) | undefined; ++}>; ++/** ++ * @since 4.0.0 ++ * @category RequestHooks ++ */ ++export declare class RequestHooks extends RequestHooks_base { ++} + declare const ConnectionHooks_base: Context.ServiceClass; + readonly onDisconnect: Effect.Effect; ++ readonly onPing?: Effect.Effect | undefined; ++ readonly onPong?: Effect.Effect | undefined; ++ readonly onPingTimeout?: Effect.Effect | undefined; + }>; + /** + * @since 4.0.0 +diff --git a/dist/unstable/rpc/RpcClient.js b/dist/unstable/rpc/RpcClient.js +index 951dfc2d667a9136cfa714dcd3195e6575917fde..599c0345873bedd72bc93afe80eab0c676319421 100644 +--- a/dist/unstable/rpc/RpcClient.js ++++ b/dist/unstable/rpc/RpcClient.js +@@ -42,6 +42,7 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro + const services = yield* Effect.context(); + const scope = Context.get(services, Scope.Scope); + const entries = new Map(); ++ const hooks = yield* Effect.serviceOption(RequestHooks); + let isShutdown = false; + yield* Scope.addFinalizer(scope, Effect.withFiber(parent => { + isShutdown = true; +@@ -79,6 +80,11 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro + return Effect.interrupt; + } + const id = generateRequestId(); ++ const onStart = Option.isSome(hooks) && hooks.value.onRequestStart ? hooks.value.onRequestStart({ ++ id, ++ tag: rpc._tag, ++ stream: false ++ }) : Effect.void; + const send = middleware(message => options.onFromClient({ + message, + context, +@@ -96,7 +102,7 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro + headers: Headers.merge(parentFiber.getRef(CurrentHeaders), headers) + }); + if (discard) { +- return send; ++ return onStart.pipe(Effect.andThen(send)); + } + let fiber; + return Effect.onInterrupt(Effect.callback(resume => { +@@ -114,7 +120,7 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro + } + }; + entries.set(id, entry); +- fiber = send.pipe(span ? Effect.withParentSpan(span, { ++ fiber = onStart.pipe(Effect.andThen(send), span ? Effect.withParentSpan(span, { + captureStackTrace: false + }) : identity, Effect.runForkWith(parentFiber.context)); + fiber.addObserver(exit => { +@@ -123,8 +129,9 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro + } + }); + }), interruptors => { ++ const entry = entries.get(id); + entries.delete(id); +- return Effect.andThen(Fiber.interrupt(fiber), sendInterrupt(id, Array.from(interruptors), context)); ++ return Effect.andThen(Fiber.interrupt(fiber), sendInterrupt(id, Array.from(interruptors), context, entry?.rpc._tag)); + }); + }); + const onStreamRequest = Effect.fnUntraced(function* (rpc, middleware, payload, headers, streamBufferSize, context) { +@@ -136,11 +143,17 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro + }); + const fiber = Fiber.getCurrent(); + const id = generateRequestId(); ++ const onStart = Option.isSome(hooks) && hooks.value.onRequestStart ? hooks.value.onRequestStart({ ++ id, ++ tag: rpc._tag, ++ stream: true ++ }) : Effect.void; + const scope = Context.getUnsafe(fiber.context, Scope.Scope); + yield* Scope.addFinalizerExit(scope, exit => { +- if (!entries.has(id)) return Effect.void; ++ const entry = entries.get(id); ++ if (!entry) return Effect.void; + entries.delete(id); +- return sendInterrupt(id, Exit.isFailure(exit) ? Array.from(Cause.interruptors(exit.cause)) : [], context); ++ return sendInterrupt(id, Exit.isFailure(exit) ? Array.from(Cause.interruptors(exit.cause)) : [], context, entry.rpc._tag); + }); + const queue = yield* Queue.bounded(streamBufferSize); + entries.set(id, { +@@ -148,9 +161,10 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro + rpc, + queue, + scope, +- context ++ context, ++ chunkCount: 0 + }); +- yield* middleware(message => options.onFromClient({ ++ yield* onStart.pipe(Effect.andThen(middleware(message => options.onFromClient({ + message, + context, + discard: false +@@ -165,7 +179,7 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro + sampled: span.sampled + } : {}), + headers: Headers.merge(fiber.getRef(CurrentHeaders), headers) +- }).pipe(span ? Effect.withParentSpan(span, { ++ }))).pipe(span ? Effect.withParentSpan(span, { + captureStackTrace: false + }) : identity, Effect.catchCause(error => Queue.failCause(queue, error)), Effect.interruptible, Effect.forkIn(scope, { + startImmediately: true +@@ -195,7 +209,10 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro + }); + }; + }; +- const sendInterrupt = (requestId, interruptors, context) => Effect.callback(resume => { ++ const sendInterrupt = (requestId, interruptors, context, tag) => (Option.isSome(hooks) && hooks.value.onRequestInterrupt ? hooks.value.onRequestInterrupt({ ++ id: requestId, ++ tag ++ }) : Effect.void).pipe(Effect.andThen(Effect.callback(resume => { + const parentFiber = Fiber.getCurrent(); + const fiber = options.onFromClient({ + message: { +@@ -209,7 +226,7 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro + fiber.addObserver(() => { + resume(Effect.void); + }); +- }); ++ }))); + const write = message => { + switch (message._tag) { + case "Chunk": +@@ -217,7 +234,13 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro + const requestId = message.requestId; + const entry = entries.get(requestId); + if (!entry || entry._tag !== "Queue") return Effect.void; +- return Queue.offerAll(entry.queue, message.values).pipe(supportsAck ? Effect.flatMap(() => options.onFromClient({ ++ const chunkCount = entry.chunkCount += message.values.length; ++ const onChunk = Option.isSome(hooks) && hooks.value.onRequestChunk ? hooks.value.onRequestChunk({ ++ id: requestId, ++ tag: entry.rpc._tag, ++ chunkCount ++ }) : Effect.void; ++ return Queue.offerAll(entry.queue, message.values).pipe(Effect.andThen(onChunk), supportsAck ? Effect.flatMap(() => options.onFromClient({ + message: { + _tag: "Ack", + requestId: message.requestId +@@ -232,11 +255,18 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro + const entry = entries.get(requestId); + if (!entry) return Effect.void; + entries.delete(requestId); ++ const onExit = Option.isSome(hooks) && hooks.value.onRequestExit ? hooks.value.onRequestExit({ ++ id: requestId, ++ tag: entry.rpc._tag, ++ stream: entry._tag === "Queue", ++ exit: message.exit ++ }) : Effect.void; + if (entry._tag === "Effect") { +- entry.resume(message.exit); +- return Effect.void; ++ return onExit.pipe(Effect.andThen(Effect.sync(() => { ++ entry.resume(message.exit); ++ }))); + } +- return message.exit._tag === "Success" ? Queue.end(entry.queue) : Queue.failCause(entry.queue, message.exit.cause); ++ return onExit.pipe(Effect.andThen(message.exit._tag === "Success" ? Queue.end(entry.queue) : Queue.failCause(entry.queue, message.exit.cause))); + } + case "Defect": + { +@@ -530,7 +560,7 @@ export const makeProtocolSocket = options => Protocol.make(Effect.fnUntraced(fun + const requestClientMap = new Map(); + const write = yield* socket.writer; + let parser = serialization.makeUnsafe(); +- const pinger = yield* makePinger(write(parser.encode(constPing))); ++ const pinger = yield* makePinger(write(parser.encode(constPing)), Option.isSome(hooks) ? hooks.value : undefined); + let currentError; + const onOpen = Effect.suspend(() => { + currentError = undefined; +@@ -550,8 +580,7 @@ export const makeProtocolSocket = options => Protocol.make(Effect.fnUntraced(fun + body: () => { + const response = responses[i++]; + if (response._tag === "Pong") { +- pinger.onPong(); +- return Effect.void; ++ return pinger.onPong; + } + if ("requestId" in response) { + const clientId = requestClientMap.get(response.requestId); +@@ -579,12 +608,12 @@ export const makeProtocolSocket = options => Protocol.make(Effect.fnUntraced(fun + } + }, { + onOpen +- }).pipe(Effect.raceFirst(Effect.flatMap(pinger.timeout, () => Effect.fail(new Socket.SocketError({ ++ }).pipe(Effect.raceFirst(Effect.flatMap(pinger.timeout, () => (Option.isSome(hooks) && hooks.value.onPingTimeout ? hooks.value.onPingTimeout : Effect.void).pipe(Effect.andThen(Effect.fail(new Socket.SocketError({ + reason: new Socket.SocketOpenError({ + kind: "Timeout", + cause: new Error("ping timeout") + }) +- }))))); ++ }))))))); + }).pipe(Effect.flatMap(() => Effect.fail(new Socket.SocketError({ + reason: new Socket.SocketCloseError({ + code: 1000 +@@ -626,20 +655,20 @@ export const makeProtocolSocket = options => Protocol.make(Effect.fnUntraced(fun + }; + })); + const defaultRetryPolicy = /*#__PURE__*/Schedule.exponential(500, 1.5).pipe(/*#__PURE__*/Schedule.either(/*#__PURE__*/Schedule.spaced(5000))); +-const makePinger = /*#__PURE__*/Effect.fnUntraced(function* (writePing) { ++const makePinger = /*#__PURE__*/Effect.fnUntraced(function* (writePing, hooks) { + let recievedPong = true; + const latch = Latch.makeUnsafe(); + const reset = () => { + recievedPong = true; + latch.closeUnsafe(); + }; +- const onPong = () => { ++ const onPong = Effect.sync(() => { + recievedPong = true; +- }; ++ }).pipe(Effect.andThen(hooks?.onPong ?? Effect.void)); + yield* Effect.suspend(() => { + if (!recievedPong) return latch.open; + recievedPong = false; +- return writePing; ++ return (hooks?.onPing ?? Effect.void).pipe(Effect.andThen(writePing)); + }).pipe(Effect.delay("5 seconds"), Effect.ignore, Effect.forever, Effect.interruptible, Effect.forkScoped); + return { + timeout: latch.await, +@@ -773,6 +802,11 @@ export const makeProtocolWorker = options => Protocol.make(Effect.fnUntraced(fun + * @category protocol + */ + export const layerProtocolWorker = /*#__PURE__*/flow(makeProtocolWorker, /*#__PURE__*/Layer.effect(Protocol)); ++/** ++ * @since 4.0.0 ++ * @category RequestHooks ++ */ ++export class RequestHooks extends /*#__PURE__*/Context.Service()("effect/rpc/RpcClient/RequestHooks") {} + /** + * @since 4.0.0 + * @category ConnectionHooks diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index f655cf25dfb..66b245f9c7b 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -21,3 +21,10 @@ catalog: typescript: 5.9.3 vitest: 3.2.4 "@vitest/coverage-v8": 3.2.4 + +# ru-fork: restored (removed in c36945d8). Patches effect's RpcClient to expose +# ConnectionHooks ping/pong/timeout hooks used by the WS transport heartbeat. +# Functionally identical to t3code's patches/effect@4.0.0-beta.78.patch, pinned +# to our beta.59. Recovered from 8fc31793. +patchedDependencies: + effect@4.0.0-beta.59: patches/effect@4.0.0-beta.59.patch