# Files changed · PR \#2679 · pingdotgg/t3code [View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/2679/diffs) Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code) Visibility: public [Reduce idle work and disk churn with native resource diagnostics](https://git.cafe/pingdotgg/t3code/pull/2679.md) Version: 95 Head commit: 6b8a5163be9bac7c3b8b73d8b59a0d7f4dc6ad48 Base commit: 3137c2b162109640343a6ff88120609e061f3782 Comparison base: 3137c2b162109640343a6ff88120609e061f3782 Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## \.github/workflows/ci.yml Status: modified; +14; −0 ```diff @@ -22,6 +22,11 @@ cache: true run-install: true + - name: Setup Rust + uses: dtolnay/rust-toolchain@stable + with: + components: rustfmt + - name: Ensure Electron runtime is installed run: vp run --filter @t3tools/desktop ensure:electron @@ -31,6 +36,9 @@ - name: Typecheck run: vpr typecheck + - name: Check resource monitor formatting + run: cargo fmt --manifest-path native/resource-monitor/Cargo.toml -- --check + - name: Build desktop pipeline run: vp run build:desktop @@ -55,12 +63,18 @@ cache: true run-install: true + - name: Setup Rust + uses: dtolnay/rust-toolchain@stable + - name: Ensure Electron runtime is installed run: vp run --filter @t3tools/desktop ensure:electron - name: Test run: vp run test + - name: Test resource monitor + run: cargo test --locked --manifest-path native/resource-monitor/Cargo.toml + mobile_native_static_analysis: name: Mobile Native Static Analysis runs-on: blacksmith-12vcpu-macos-26 ``` ## \.github/workflows/release.yml Status: modified; +51; −0 ```diff @@ -326,21 +326,29 @@ platform: mac target: dmg arch: arm64 + rust_target: aarch64-apple-darwin + resource_key: darwin-arm64 - label: macOS x64 runner: blacksmith-12vcpu-macos-26 platform: mac target: dmg arch: x64 + rust_target: x86_64-apple-darwin + resource_key: darwin-x64 - label: Linux x64 runner: blacksmith-32vcpu-ubuntu-2404 platform: linux target: AppImage arch: x64 + rust_target: x86_64-unknown-linux-gnu + resource_key: linux-x64 - label: Windows x64 runner: blacksmith-32vcpu-windows-2025 platform: win target: nsis arch: x64 + rust_target: x86_64-pc-windows-msvc + resource_key: win32-x64 # - label: Windows arm64 # runner: windows-11-arm # platform: win @@ -359,6 +367,11 @@ node-version-file: package.json cache: true run-install: true + + - name: Setup Rust + uses: dtolnay/rust-toolchain@stable + with: + targets: ${{ matrix.rust_target }} - name: Download relay client tracing config uses: actions/download-artifact@v8 @@ -598,6 +611,19 @@ # done # fi + - name: Collect resource monitor + shell: bash + run: | + set -euo pipefail + binary_name="t3-resource-monitor" + if [[ "${{ matrix.platform }}" == "win" ]]; then + binary_name="${binary_name}.exe" + fi + source_path="native/resource-monitor/target/${{ matrix.rust_target }}/release/${binary_name}" + target_dir="resource-monitor-publish/${{ matrix.resource_key }}" + mkdir -p "$target_dir" + cp "$source_path" "$target_dir/$binary_name" + - name: Upload build artifacts uses: actions/upload-artifact@v7 with: @@ -605,6 +631,13 @@ path: release-publish/* if-no-files-found: error + - name: Upload resource monitor + uses: actions/upload-artifact@v7 + with: + name: resource-monitor-${{ matrix.resource_key }} + path: resource-monitor-publish/${{ matrix.resource_key }}/* + if-no-files-found: error + publish_cli: name: Publish CLI to npm needs: [preflight, relay_public_config, build] @@ -659,6 +692,24 @@ - name: Build CLI package run: vp run --filter t3 build + - name: Download resource monitors + uses: actions/download-artifact@v8 + with: + pattern: resource-monitor-* + path: ${{ runner.temp }}/resource-monitors + + - name: Bundle resource monitors into CLI package + shell: bash + run: | + set -euo pipefail + for artifact_dir in "$RUNNER_TEMP"/resource-monitors/resource-monitor-*; do + resource_key="${artifact_dir##*/resource-monitor-}" + target_dir="apps/server/dist/resource-monitor/${resource_key}" + mkdir -p "$target_dir" + cp "$artifact_dir"/t3-resource-monitor* "$target_dir/" + chmod +x "$target_dir"/t3-resource-monitor 2>/dev/null || true + done + - name: Publish CLI package run: node apps/server/scripts/cli.ts publish --tag "${{ needs.preflight.outputs.cli_dist_tag }}" --app-version "${{ needs.preflight.outputs.version }}" --verbose ``` ## \.gitignore Status: modified; +1; −0 ```diff @@ -30,6 +30,7 @@ .showcase/ apps/mobile/.showcase/ artifacts/app-store/screenshots/ +native/**/target/ node_modules/ .alchemy/ *.log ``` ## apps/desktop/src/app/DesktopAppIdentity.test.ts Status: modified; +1; −0 ```diff @@ -55,6 +55,7 @@ }), setAppUserModelId: () => Effect.void, requestSingleInstanceLock: Effect.succeed(true), + getAppMetrics: Effect.succeed([]), isDefaultProtocolClient: () => Effect.succeed(false), setAsDefaultProtocolClient: () => Effect.succeed(true), setDesktopName: () => Effect.void, ``` ## apps/desktop/src/app/DesktopLifecycle.test.ts Status: modified; +1; −0 ```diff @@ -30,6 +30,7 @@ setAboutPanelOptions: () => Effect.void, setAppUserModelId: () => Effect.void, requestSingleInstanceLock: Effect.succeed(true), + getAppMetrics: Effect.succeed([]), isDefaultProtocolClient: () => Effect.succeed(false), setAsDefaultProtocolClient: () => Effect.succeed(true), setDesktopName: () => Effect.void, ``` ## apps/desktop/src/app/DesktopObservability.test.ts Status: modified; +180; −14 ```diff @@ -49,20 +49,41 @@ runningUnderArm64Translation: false, }) satisfies DesktopEnvironment.MakeDesktopEnvironmentInput; -const makeEnvironmentLayer = (baseDir: string) => +const makeEnvironmentLayer = (baseDir: string, isDevelopment = true) => DesktopEnvironment.layer(environmentInput(baseDir)).pipe( Layer.provide( Layer.mergeAll( NodeServices.layer, DesktopConfig.layerTest({ T3CODE_HOME: baseDir, - VITE_DEV_SERVER_URL: "http://127.0.0.1:5733", + VITE_DEV_SERVER_URL: isDevelopment ? "http://127.0.0.1:5733" : undefined, }), ), ), ); describe("DesktopObservability", () => { + it("advances a retained output offset instead of repeatedly copying a full head chunk", () => { + const maxBufferedBytes = 1024 * 1024; + const initial = DesktopObservability.appendBoundedOutputChunk( + { + runId: "test-run", + startDetails: "pid=123", + chunks: [], + byteLength: 0, + }, + "stderr", + new Uint8Array(maxBufferedBytes), + ); + const initialBackingBuffer = initial.chunks[0]?.chunk.buffer; + + const next = DesktopObservability.appendBoundedOutputChunk(initial, "stderr", Uint8Array.of(1)); + + assert.equal(next.chunks[0]?.chunk.buffer, initialBackingBuffer); + assert.equal(next.chunks[0]?.offset, 1); + assert.equal(next.byteLength, maxBufferedBytes); + }); + it.effect("persists desktop Effect logs as span events in desktop.trace.ndjson", () => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; @@ -112,26 +133,34 @@ ), ); - it.effect("persists backend child output as structured JSON records in development", () => + it.effect("buffers backend child output and persists it only when a failure is reported", () => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; const baseDir = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-desktop-backend-output-log-test-", }); - const environmentLayer = makeEnvironmentLayer(baseDir); + const environmentLayer = makeEnvironmentLayer(baseDir, false); const logPath = yield* Effect.gen(function* () { const environment = yield* DesktopEnvironment.DesktopEnvironment; return environment.path.join(environment.logDir, "server-child.log"); }).pipe(Effect.provide(environmentLayer)); + const tracePath = yield* Effect.gen(function* () { + const environment = yield* DesktopEnvironment.DesktopEnvironment; + return environment.path.join(environment.logDir, "desktop.trace.ndjson"); + }).pipe(Effect.provide(environmentLayer)); yield* Effect.gen(function* () { const factory = yield* DesktopObservability.DesktopBackendOutputLogFactory; const outputLog = yield* factory.forInstance("primary"); - yield* outputLog.writeSessionBoundary({ - phase: "START", + yield* outputLog.beginSession({ details: "pid=123 port=3773 cwd=/repo", }); yield* outputLog.writeOutputChunk("stdout", new TextEncoder().encode("hello server\n")); + assert.isFalse(yield* fileSystem.exists(logPath)); + yield* outputLog.persistFailure({ details: "code=1" }); + yield* outputLog.beginSession({ details: "pid=456" }); + yield* outputLog.writeOutputChunk("stderr", new TextEncoder().encode("normal shutdown\n")); + yield* outputLog.discardSession; }).pipe( Effect.annotateLogs({ runId: "test-run" }), Effect.provide(DesktopObservability.layer.pipe(Layer.provideMerge(environmentLayer))), @@ -139,16 +168,18 @@ const log = yield* fileSystem.readFileString(logPath); const lines = log.trimEnd().split("\n"); + const start = yield* decodeDesktopBackendChildLogRecord(lines[0] ?? ""); - const boundary = yield* decodeDesktopBackendChildLogRecord(lines[0] ?? ""); const output = yield* decodeDesktopBackendChildLogRecord(lines[1] ?? ""); + const end = yield* decodeDesktopBackendChildLogRecord(lines[2] ?? ""); + assert.equal(lines.length, 3); + assert.equal(start.message, "backend child process failure output start"); + assert.equal(start.level, "ERROR"); + assert.equal(start.annotations.component, "desktop-backend-child"); + assert.equal(start.annotations.runId, "test-run"); - assert.equal(boundary.message, "backend child process session start"); - assert.equal(boundary.level, "INFO"); - assert.equal(boundary.annotations.component, "desktop-backend-child"); - assert.equal(boundary.annotations.runId, "test-run"); - assert.equal(boundary.annotations.instanceId, "primary"); - assert.equal(boundary.annotations.phase, "START"); + assert.equal(start.annotations.instanceId, "primary"); + assert.equal(start.annotations.phase, "START"); - assert.equal(boundary.annotations.details, "pid=123 port=3773 cwd=/repo"); + assert.equal(start.annotations.details, "pid=123 port=3773 cwd=/repo"); assert.equal(output.message, "backend child process output"); assert.equal(output.level, "INFO"); @@ -157,9 +188,144 @@ assert.equal(output.annotations.instanceId, "primary"); assert.equal(output.annotations.stream, "stdout"); assert.equal(output.annotations.text, "hello server\n"); + + assert.equal(end.message, "backend child process failure output end"); + assert.equal(end.level, "ERROR"); + assert.equal(end.annotations.instanceId, "primary"); + assert.equal(end.annotations.phase, "END"); + assert.equal(end.annotations.details, "code=1"); + + const traceRecords = (yield* fileSystem.readFileString(tracePath)) + .trim() + .split("\n") + .filter((line) => line.length > 0) + .map((line) => decodeTraceRecordLine(line)); + assert.isFalse( + traceRecords.some( + (record) => record.name === "desktop.observability.backendOutput.writeOutputChunk", + ), + ); }).pipe( Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, NodeHttpClient.layerUndici)), ), ); + + it.effect("keeps buffering output after a non-terminal failure snapshot", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const baseDir = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-desktop-backend-output-snapshot-test-", + }); + const environmentLayer = makeEnvironmentLayer(baseDir, false); + const logPath = yield* Effect.gen(function* () { + const environment = yield* DesktopEnvironment.DesktopEnvironment; + return environment.path.join(environment.logDir, "server-child.log"); + }).pipe(Effect.provide(environmentLayer)); + + yield* Effect.gen(function* () { + const factory = yield* DesktopObservability.DesktopBackendOutputLogFactory; + const outputLog = yield* factory.forInstance("primary"); + yield* outputLog.beginSession({ details: "pid=123" }); + yield* outputLog.writeOutputChunk("stdout", new TextEncoder().encode("before timeout\n")); + yield* outputLog.persistFailureSnapshot({ details: "readiness timeout" }); + yield* outputLog.writeOutputChunk("stderr", new TextEncoder().encode("after timeout\n")); + yield* outputLog.persistFailure({ details: "code=1" }); + }).pipe( + Effect.annotateLogs({ runId: "test-run" }), + Effect.provide(DesktopObservability.layer.pipe(Layer.provideMerge(environmentLayer))), + ); + + const records = yield* Effect.forEach( + (yield* fileSystem.readFileString(logPath)).trimEnd().split("\n"), + (line) => decodeDesktopBackendChildLogRecord(line), + ); + assert.equal( + records.some((record) => record.annotations.text === "after timeout\n"), + true, + ); + assert.equal(records.at(-1)?.annotations.details, "code=1"); + }).pipe( + Effect.scoped, + Effect.provide(Layer.mergeAll(NodeServices.layer, NodeHttpClient.layerUndici)), + ), + ); + + it.effect("retains only the last mebibyte of backend child output", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const baseDir = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-desktop-backend-output-bound-test-", + }); + const environmentLayer = makeEnvironmentLayer(baseDir, false); + const logPath = yield* Effect.gen(function* () { + const environment = yield* DesktopEnvironment.DesktopEnvironment; + return environment.path.join(environment.logDir, "server-child.log"); + }).pipe(Effect.provide(environmentLayer)); + const maxBufferedBytes = 1024 * 1024; + const discardedPrefixBytes = 128; + const output = new Uint8Array(maxBufferedBytes + discardedPrefixBytes); + output.fill("x".charCodeAt(0)); + output.fill("y".charCodeAt(0), 0, discardedPrefixBytes); + + yield* Effect.scoped( + Effect.gen(function* () { + const factory = yield* DesktopObservability.DesktopBackendOutputLogFactory; + const outputLog = yield* factory.forInstance("primary"); + yield* outputLog.beginSession({ details: "pid=123" }); + yield* outputLog.writeOutputChunk("stderr", output); + yield* outputLog.persistFailure({ details: "code=1" }); + }).pipe( + Effect.provide(DesktopObservability.layer.pipe(Layer.provideMerge(environmentLayer))), + ), + ); + + const lines = (yield* fileSystem.readFileString(logPath)).trimEnd().split("\n"); + const record = yield* decodeDesktopBackendChildLogRecord(lines[1] ?? ""); + const text = record.annotations.text; + assert.equal(typeof text, "string"); + if (typeof text !== "string") { + return; + } + assert.equal(new TextEncoder().encode(text).byteLength, maxBufferedBytes); + assert.isFalse(text.includes("y")); + }).pipe( + Effect.scoped, + Effect.provide(Layer.mergeAll(NodeServices.layer, NodeHttpClient.layerUndici)), + ), + ); + + it.effect("bounds the number of retained backend child output chunks", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const baseDir = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-desktop-backend-output-chunks-test-", + }); + const environmentLayer = makeEnvironmentLayer(baseDir, false); + const logPath = yield* Effect.gen(function* () { + const environment = yield* DesktopEnvironment.DesktopEnvironment; + return environment.path.join(environment.logDir, "server-child.log"); + }).pipe(Effect.provide(environmentLayer)); + + yield* Effect.scoped( + Effect.gen(function* () { + const factory = yield* DesktopObservability.DesktopBackendOutputLogFactory; + const outputLog = yield* factory.forInstance("primary"); + yield* outputLog.beginSession({ details: "pid=123" }); + for (let index = 0; index < 300; index += 1) { + yield* outputLog.writeOutputChunk("stderr", Uint8Array.of(index % 128)); + } + yield* outputLog.persistFailure({ details: "code=1" }); + }).pipe( + Effect.provide(DesktopObservability.layer.pipe(Layer.provideMerge(environmentLayer))), + ), + ); + + const lines = (yield* fileSystem.readFileString(logPath)).trimEnd().split("\n"); + assert.equal(lines.length, 258); + }).pipe( + Effect.scoped, + Effect.provide(Layer.mergeAll(NodeServices.layer, NodeHttpClient.layerUndici)), + ), + ); }); ``` ## apps/desktop/src/app/DesktopObservability.ts Status: modified; +164; −39 ```diff @@ -24,7 +24,9 @@ const DESKTOP_LOG_FILE_MAX_BYTES = 10 * 1024 * 1024; const DESKTOP_LOG_FILE_MAX_FILES = 10; const DESKTOP_BACKEND_CHILD_LOG_FIBER_ID = "#backend-child"; -const DESKTOP_TRACE_BATCH_WINDOW_MS = 200; +const DESKTOP_TRACE_BATCH_WINDOW_MS = 1_000; +const DESKTOP_BACKEND_OUTPUT_BUFFER_MAX_BYTES = 1024 * 1024; +const DESKTOP_BACKEND_OUTPUT_BUFFER_MAX_CHUNKS = 256; export interface RotatingLogFileWriter { readonly writeBytes: (chunk: Uint8Array) => Effect.Effect; @@ -32,14 +34,14 @@ } export interface DesktopBackendOutputLogShape { - readonly writeSessionBoundary: (input: { + readonly beginSession: (input: { readonly details: string }) => Effect.Effect; - readonly phase: "START" | "END"; - readonly details: string; - }) => Effect.Effect; readonly writeOutputChunk: ( streamName: "stdout" | "stderr", chunk: Uint8Array, ) => Effect.Effect; + readonly persistFailureSnapshot: (input: { readonly details: string }) => Effect.Effect; + readonly persistFailure: (input: { readonly details: string }) => Effect.Effect; + readonly discardSession: Effect.Effect; } // Factory for per-instance backend output logs. `forInstance(id)` returns @@ -127,10 +129,80 @@ ); const DesktopBackendOutputLogNoop: DesktopBackendOutputLogShape = { - writeSessionBoundary: () => Effect.void, + beginSession: () => Effect.void, writeOutputChunk: () => Effect.void, + persistFailureSnapshot: () => Effect.void, + persistFailure: () => Effect.void, + discardSession: Effect.void, }; +interface BufferedBackendOutputChunk { + readonly streamName: "stdout" | "stderr"; + readonly chunk: Uint8Array; + readonly offset: number; +} + +interface BackendOutputSession { + readonly runId: string; + readonly startDetails: string; + readonly chunks: ReadonlyArray; + readonly byteLength: number; +} + +export function appendBoundedOutputChunk( + session: BackendOutputSession, + streamName: "stdout" | "stderr", + chunk: Uint8Array, +): BackendOutputSession { + if (chunk.byteLength === 0) { + return session; + } + + const retainedChunk = + chunk.byteLength > DESKTOP_BACKEND_OUTPUT_BUFFER_MAX_BYTES + ? chunk.slice(chunk.byteLength - DESKTOP_BACKEND_OUTPUT_BUFFER_MAX_BYTES) + : chunk.slice(); + const chunks = [...session.chunks, { streamName, chunk: retainedChunk, offset: 0 }]; + let byteLength = session.byteLength + retainedChunk.byteLength; + let overflow = Math.max(0, byteLength - DESKTOP_BACKEND_OUTPUT_BUFFER_MAX_BYTES); + let firstRetainedIndex = 0; + + while (overflow > 0) { + const first = chunks[firstRetainedIndex]; + if (!first) break; + const retainedByteLength = first.chunk.byteLength - first.offset; + if (retainedByteLength <= overflow) { + overflow -= retainedByteLength; + byteLength -= retainedByteLength; + firstRetainedIndex += 1; + continue; + } + + chunks[firstRetainedIndex] = { + ...first, + offset: first.offset + overflow, + }; + byteLength -= overflow; + overflow = 0; + } + + const excessChunks = Math.max( + 0, + chunks.length - firstRetainedIndex - DESKTOP_BACKEND_OUTPUT_BUFFER_MAX_CHUNKS, + ); + for (let index = firstRetainedIndex; index < firstRetainedIndex + excessChunks; index += 1) { + const chunk = chunks[index]; + byteLength -= chunk ? chunk.chunk.byteLength - chunk.offset : 0; + } + firstRetainedIndex += excessChunks; + + return { + ...session, + chunks: chunks.slice(firstRetainedIndex), + byteLength, + }; +} + const currentDesktopRunId = Effect.gen(function* () { const annotations = yield* References.CurrentLogAnnotations; const runId = annotations.runId; @@ -345,47 +417,93 @@ environment: DesktopEnvironment.DesktopEnvironment["Service"], id: string, sink: Option.Option, -): DesktopBackendOutputLogShape => +): Effect.Effect => Option.match(sink, { - onNone: () => DesktopBackendOutputLogNoop, + onNone: () => Effect.succeed(DesktopBackendOutputLogNoop), onSome: (logFile) => + Effect.gen(function* () { + const sessionRef = yield* Ref.make(Option.none()); - ({ - writeSessionBoundary: Effect.fn("desktop.observability.backendOutput.writeSessionBoundary")( - function* ({ phase, details }) { + const writeFailure = Effect.fn("desktop.observability.backendOutput.writeFailure")( + function* (session: BackendOutputSession, details: string) { - const runId = yield* currentDesktopRunId; yield* writeBackendChildLogRecord(logFile, { - message: `backend child process session ${phase.toLowerCase()}`, + message: "backend child process failure output start", - level: "INFO", + level: "ERROR", annotations: { component: "desktop-backend-child", + runId: session.runId, - runId, instanceId: id, - phase, - details: sanitizeLogValue(details), + phase: "START", + details: session.startDetails, }, }); + for (const output of session.chunks) { + yield* writeBackendChildLogRecord(logFile, { + message: "backend child process output", - }, - ), - writeOutputChunk: Effect.fn("desktop.observability.backendOutput.writeOutputChunk")( - function* (streamName, chunk) { - if (environment.isDevelopment) { - yield* writeDevelopmentConsoleOutput(streamName, chunk); + level: output.streamName === "stderr" ? "ERROR" : "INFO", + annotations: { + component: "desktop-backend-child", + runId: session.runId, + instanceId: id, + stream: output.streamName, + text: textDecoder.decode(output.chunk.subarray(output.offset)), + }, + }); } - const runId = yield* currentDesktopRunId; yield* writeBackendChildLogRecord(logFile, { + message: "backend child process failure output end", + level: "ERROR", - message: "backend child process output", - level: streamName === "stderr" ? "ERROR" : "INFO", annotations: { component: "desktop-backend-child", - runId, + runId: session.runId, instanceId: id, + phase: "END", - stream: streamName, + details: sanitizeLogValue(details), - text: textDecoder.decode(chunk), }, }); }, + ); + return { + beginSession: Effect.fn("desktop.observability.backendOutput.beginSession")(function* ({ + details, + }) { + const runId = yield* currentDesktopRunId; + yield* Ref.set( + sessionRef, + Option.some({ + runId, + startDetails: sanitizeLogValue(details), + chunks: [], + byteLength: 0, + }), + ); + }), + writeOutputChunk: Effect.fnUntraced(function* (streamName, chunk) { + if (environment.isDevelopment) { + yield* writeDevelopmentConsoleOutput(streamName, chunk); + } + yield* Ref.update( + sessionRef, + Option.map((session) => appendBoundedOutputChunk(session, streamName, chunk)), + ); + }), + persistFailureSnapshot: Effect.fn( + "desktop.observability.backendOutput.persistFailureSnapshot", + )(function* ({ details }) { + const session = yield* Ref.get(sessionRef); + if (Option.isSome(session)) { + yield* writeFailure(session.value, details); - ), + } - }) satisfies DesktopBackendOutputLogShape, + }), + persistFailure: Effect.fn("desktop.observability.backendOutput.persistFailure")( + function* ({ details }) { + const session = yield* Ref.modify(sessionRef, (current) => [current, Option.none()]); + if (Option.isNone(session)) return; + yield* writeFailure(session.value, details); + }, + ), + discardSession: Ref.set(sessionRef, Option.none()), + } satisfies DesktopBackendOutputLogShape; + }), }); const backendOutputLogFactoryLayer = Layer.effect( @@ -412,10 +530,9 @@ const cacheKey = backendLogFilePathForInstance(environment, id); const cached = cache.get(cacheKey); if (cached !== undefined) { + return makeBackendOutputLogShape(environment, id, cached).pipe( + Effect.map((outputLog) => [outputLog, cache] as const), - return Effect.succeed([ + ); - makeBackendOutputLogShape(environment, id, cached), - cache, - ] as const); } return makeBackendOutputSinkForInstance(environment, id).pipe( Effect.provideService(FileSystem.FileSystem, fileSystem), @@ -424,11 +541,19 @@ Effect.map((sink) => { const next = new Map(cache); next.set(cacheKey, sink); - return [ - makeBackendOutputLogShape(environment, id, sink), + return { sink, next }; - next as ReadonlyMap>, - ] as const; }), + Effect.flatMap(({ sink, next }) => + makeBackendOutputLogShape(environment, id, sink).pipe( + Effect.map( + (outputLog) => + [ + outputLog, + next as ReadonlyMap>, + ] as const, + ), + ), + ), ); }); ``` ## apps/desktop/src/backend/DesktopBackendConfiguration.test.ts Status: modified; +95; −1 ```diff @@ -52,6 +52,7 @@ baseDir: string, options?: { readonly appPath?: string; + readonly dirname?: string; readonly isPackaged?: boolean; readonly devServerUrl?: string; readonly platform?: NodeJS.Platform; @@ -59,7 +60,7 @@ }, ) { return DesktopEnvironment.layer({ + dirname: options?.dirname ?? "/repo/apps/desktop/src", - dirname: "/repo/apps/desktop/src", homeDirectory: baseDir, platform: options?.platform ?? "darwin", processArch: "x64", @@ -464,6 +465,8 @@ // both wslhost-forwarded localhost and the distro's eth0 IP. assert.equal(config.bootstrap.host, "0.0.0.0"); assert.equal(config.bootstrap.tailscaleServeEnabled, false); + assert.notProperty(config.bootstrap, "desktopTelemetryFd"); + assert.notProperty(config.bootstrap, "resourceMonitorPath"); // httpBaseUrl uses the resolved distro IP from the test stub, // not localhost — the renderer reaches the backend directly to // avoid relying on wslhost forwarding. @@ -713,6 +716,97 @@ }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), ); + it.effect("prefers the external packaged resource monitor over the copy inside the asar", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const baseDir = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-desktop-backend-config-test-", + }); + const resourcesPath = `${baseDir}/resources`; + const dirname = `${resourcesPath}/app.asar/apps/desktop/dist-electron`; + const embeddedMonitorPath = `${resourcesPath}/app.asar/apps/desktop/prod-resources/resource-monitor/t3-resource-monitor`; + const monitorPath = `${resourcesPath}/resource-monitor/t3-resource-monitor`; + yield* fileSystem.makeDirectory( + `${resourcesPath}/app.asar/apps/desktop/prod-resources/resource-monitor`, + { recursive: true }, + ); + yield* fileSystem.makeDirectory(`${resourcesPath}/resource-monitor`, { + recursive: true, + }); + yield* fileSystem.writeFileString(embeddedMonitorPath, "embedded"); + yield* fileSystem.writeFileString(monitorPath, "binary"); + yield* fileSystem.chmod(monitorPath, 0o755); + + yield* Effect.gen(function* () { + const configuration = yield* DesktopBackendConfiguration.DesktopBackendConfiguration; + const config = yield* configuration.resolvePrimary; + assert.equal(config.bootstrap.resourceMonitorPath, monitorPath); + assert.equal(config.bootstrap.desktopTelemetryFd, 4); + assert.equal(config.bootstrap.desktopTelemetryControlFd, 5); + }).pipe( + Effect.provide( + DesktopBackendConfiguration.layer.pipe( + Layer.provideMerge(serverExposureLayer), + Layer.provideMerge(DesktopAppSettings.layerTest()), + Layer.provideMerge(DesktopWslEnvironment.layerTest()), + Layer.provideMerge( + makeEnvironmentLayer(baseDir, { + appPath: `${resourcesPath}/app.asar`, + dirname, + isPackaged: true, + resourcesPath, + }), + ), + ), + ), + ); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("prefers the release resource monitor when both development builds exist", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const baseDir = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-desktop-backend-config-test-", + }); + const dirname = path.join(baseDir, "apps/desktop/src"); + const releaseMonitorPath = path.join( + baseDir, + "native/resource-monitor/target/release/t3-resource-monitor", + ); + const debugMonitorPath = path.join( + baseDir, + "native/resource-monitor/target/debug/t3-resource-monitor", + ); + yield* fileSystem.makeDirectory(path.dirname(releaseMonitorPath), { recursive: true }); + yield* fileSystem.makeDirectory(path.dirname(debugMonitorPath), { recursive: true }); + yield* fileSystem.writeFileString(releaseMonitorPath, "release"); + yield* fileSystem.writeFileString(debugMonitorPath, "debug"); + + yield* Effect.gen(function* () { + const configuration = yield* DesktopBackendConfiguration.DesktopBackendConfiguration; + const config = yield* configuration.resolvePrimary; + assert.equal(config.bootstrap.resourceMonitorPath, releaseMonitorPath); + }).pipe( + Effect.provide( + DesktopBackendConfiguration.layer.pipe( + Layer.provideMerge(serverExposureLayer), + Layer.provideMerge(DesktopAppSettings.layerTest()), + Layer.provideMerge(DesktopWslEnvironment.layerTest()), + Layer.provideMerge( + makeEnvironmentLayer(baseDir, { + dirname, + devServerUrl: "http://127.0.0.1:5733", + isPackaged: false, + }), + ), + ), + ), + ); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + it.effect("resolvePrimaryLabel reports the local environment on non-Windows platforms", () => withHarness( Effect.gen(function* () { ``` ## apps/desktop/src/backend/DesktopBackendConfiguration.ts Status: modified; +56; −2 ```diff @@ -140,7 +140,45 @@ }), ); }; + +function resourceMonitorBinaryName(platform: NodeJS.Platform): string { + return platform === "win32" ? "t3-resource-monitor.exe" : "t3-resource-monitor"; +} + +const resolveResourceMonitorPath = Effect.fn( + "desktop.backendConfiguration.resolveResourceMonitorPath", +)(function* () { + const environment = yield* DesktopEnvironment.DesktopEnvironment; + const fileSystem = yield* FileSystem.FileSystem; + const binaryName = resourceMonitorBinaryName(environment.platform); + const candidates = environment.isDevelopment + ? [ + environment.path.join( + environment.rootDir, + "native/resource-monitor/target/release", + binaryName, + ), + environment.path.join( + environment.rootDir, + "native/resource-monitor/target/debug", + binaryName, + ), + ] + : environment.isPackaged + ? [environment.path.join(environment.resourcesPath, "resource-monitor", binaryName)] + : environment.resolveResourcePathCandidates( + environment.path.join("resource-monitor", binaryName), + ); + for (const candidate of candidates) { + if (yield* fileSystem.exists(candidate).pipe(Effect.orElseSucceed(() => false))) { + return Option.some(candidate); + } + } + + return Option.none(); +}); + const readPersistedBackendObservabilitySettings = Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; const environment = yield* DesktopEnvironment.DesktopEnvironment; @@ -326,7 +364,9 @@ const resolvePrimaryStartConfig = Effect.fn("desktop.backendConfiguration.resolvePrimary")( function* ( + input: SharedBootstrapInput & { + readonly resourceMonitorPath: Option.Option; + }, - input: SharedBootstrapInput, ): Effect.fn.Return< DesktopBackendManager.DesktopBackendStartConfig, never, @@ -345,6 +385,12 @@ desktopBootstrapToken: input.bootstrapToken, tailscaleServeEnabled: backendExposure.tailscaleServeEnabled, tailscaleServePort: backendExposure.tailscaleServePort, + desktopTelemetryFd: 4, + desktopTelemetryControlFd: 5, + ...Option.match(input.resourceMonitorPath, { + onNone: () => ({}), + onSome: (resourceMonitorPath) => ({ resourceMonitorPath }), + }), ...buildObservabilityFragment(input.observabilitySettings), }; @@ -411,6 +457,10 @@ // inert. tailscaleServeEnabled: false, tailscaleServePort: 443, + // The packaged sidecar is a Windows executable and cannot run inside the + // Linux WSL backend. Keep the field absent instead of passing an unusable + // `/mnt/.../*.exe` path; WSL resource telemetry is reported unavailable. + // See docs/architecture/resource-telemetry.md. ...buildObservabilityFragment(input.observabilitySettings), }; @@ -621,7 +671,11 @@ const buildWindowsPrimaryConfig = Effect.gen(function* () { const shared = yield* sharedInputs; + const resourceMonitorPath = yield* resolveResourceMonitorPath().pipe( + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(DesktopEnvironment.DesktopEnvironment, environment), + ); + return yield* resolvePrimaryStartConfig({ ...shared, resourceMonitorPath }).pipe( - return yield* resolvePrimaryStartConfig(shared).pipe( Effect.provideService(DesktopEnvironment.DesktopEnvironment, environment), Effect.provideService(DesktopServerExposure.DesktopServerExposure, serverExposure), ); ``` ## apps/desktop/src/backend/DesktopBackendManager.test.ts Status: modified; +700; −16 ```diff @@ -1,14 +1,17 @@ import { DesktopBackendBootstrap, type DesktopBackendBootstrap as DesktopBackendBootstrapValue, + DesktopTelemetryControlMessage, } from "@t3tools/contracts"; import { assert, describe, it } from "@effect/vitest"; import * as Deferred from "effect/Deferred"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as PlatformError from "effect/PlatformError"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; @@ -21,10 +24,15 @@ import * as DesktopBackendManager from "./DesktopBackendManager.ts"; import * as DesktopObservability from "../app/DesktopObservability.ts"; +import * as DesktopTelemetryPublisher from "../telemetry/DesktopTelemetryPublisher.ts"; const decodeDesktopBackendBootstrap = Schema.decodeEffect( Schema.fromJsonString(DesktopBackendBootstrap), ); +const isBackendProcessError = Schema.is(DesktopBackendManager.BackendProcessError); +const encodeDesktopTelemetryControl = Schema.encodeSync( + Schema.fromJsonString(DesktopTelemetryControlMessage), +); const baseConfig: DesktopBackendManager.DesktopBackendStartConfig = { executablePath: "/electron", @@ -41,6 +49,8 @@ desktopBootstrapToken: "token", tailscaleServeEnabled: false, tailscaleServePort: 443, + desktopTelemetryFd: 4, + desktopTelemetryControlFd: 5, }, bootstrapDelivery: "fd3", extendEnv: true, @@ -52,14 +62,16 @@ const configWithObservability: DesktopBackendBootstrapValue = { ...baseConfig.bootstrap, tailscaleServeEnabled: true, + desktopTelemetryFd: 4, otlpTracesUrl: "http://127.0.0.1:4318/v1/traces", }; function makeProcess(options?: { + readonly stdout?: Stream.Stream; - readonly stdout?: Stream.Stream; - readonly stderr?: Stream.Stream; - readonly exitCode?: Effect.Effect; + readonly stderr?: Stream.Stream; + readonly exitCode?: Effect.Effect; readonly kill?: ChildProcessSpawner.ChildProcessHandle["kill"]; + readonly getOutputFd?: ChildProcessSpawner.ChildProcessHandle["getOutputFd"]; }): ChildProcessSpawner.ChildProcessHandle { return ChildProcessSpawner.makeHandle({ pid: ChildProcessSpawner.ProcessId(123), @@ -71,7 +83,7 @@ kill: options?.kill ?? (() => Effect.void), stdin: Sink.drain, getInputFd: () => Sink.drain, - getOutputFd: () => Stream.empty, + getOutputFd: options?.getOutputFd ?? (() => Stream.empty), unref: Effect.succeed(Effect.void), }); } @@ -112,7 +124,14 @@ failure: DesktopBackendManager.PreflightFailure, ) => Effect.Effect; readonly config?: DesktopBackendManager.DesktopBackendStartConfig; + readonly configResolve?: Effect.Effect< + DesktopBackendManager.DesktopBackendStartConfig, + PlatformError.PlatformError + >; + readonly desktopTelemetryStream?: Stream.Stream; - readonly configResolve?: Effect.Effect; + readonly desktopTelemetryPublisher?: Partial< + DesktopTelemetryPublisher.DesktopTelemetryPublisher["Service"] + >; } // Helper that constructs a primary backend instance using the factory @@ -122,8 +141,11 @@ // to drive the instance's lifecycle. function makeTestInstance(input: MakeInstanceInput) { const stubLog: DesktopObservability.DesktopBackendOutputLogShape = { - writeSessionBoundary: () => Effect.void, + beginSession: () => Effect.void, writeOutputChunk: () => Effect.void, + persistFailureSnapshot: () => Effect.void, + persistFailure: () => Effect.void, + discardSession: Effect.void, ...input.backendOutputLog, }; const servicesLayer = Layer.mergeAll( @@ -135,6 +157,16 @@ Layer.succeed(DesktopObservability.DesktopBackendOutputLogFactory, { forInstance: () => Effect.succeed(stubLog), } satisfies DesktopObservability.DesktopBackendOutputLogFactory["Service"]), + Layer.succeed(DesktopTelemetryPublisher.DesktopTelemetryPublisher, { + latest: Effect.succeed(Option.none()), + changes: Stream.empty, + encoded: input.desktopTelemetryStream ?? Stream.empty, + handleControl: () => Effect.void, + handleControlForSource: (_sourceId, message) => + (input.desktopTelemetryPublisher?.handleControl ?? (() => Effect.void))(message), + removeControlSource: () => Effect.void, + ...input.desktopTelemetryPublisher, + }), ); const instance = DesktopBackendManager.makeBackendInstance({ @@ -150,11 +182,12 @@ } describe("DesktopBackendManager", () => { - it.effect("spawns the backend with fd3 bootstrap JSON and reports HTTP readiness", () => + it.effect("spawns the backend with fd3 bootstrap and fd4 telemetry", () => Effect.scoped( Effect.gen(function* () { let spawnedCommand: ChildProcess.Command | undefined; let bootstrapJson = ""; + let telemetryJson = ""; let readyCount = 0; const ready = yield* Deferred.make(); const exited = yield* Queue.unbounded(); @@ -168,6 +201,10 @@ const fd3 = command.options.additionalFds?.fd3; if (fd3?.type === "input" && fd3.stream) { bootstrapJson = yield* fd3.stream.pipe(Stream.decodeText(), Stream.mkString); + } + const fd4 = command.options.additionalFds?.fd4; + if (fd4?.type === "input" && fd4.stream) { + telemetryJson = yield* fd4.stream.pipe(Stream.decodeText(), Stream.mkString); } } @@ -184,12 +221,14 @@ bootstrap: configWithObservability, }, spawnerLayer, + desktopTelemetryStream: Stream.encodeText( + Stream.make('{"version":1,"type":"desktopTelemetryHello","electronPid":123}\n'), + ), onReady: Effect.sync(() => { readyCount += 1; }).pipe(Effect.andThen(Deferred.succeed(ready, void 0)), Effect.asVoid), backendOutputLog: { - writeSessionBoundary: ({ phase }) => + persistFailure: () => Queue.offer(exited, void 0).pipe(Effect.asVoid), - phase === "END" ? Queue.offer(exited, void 0).pipe(Effect.asVoid) : Effect.void, }, }); @@ -210,16 +249,400 @@ assert.equal(spawnedCommand.options.stderr, "pipe"); assert.equal(spawnedCommand.options.killSignal, "SIGTERM"); assert.isDefined(spawnedCommand.options.forceKillAfter); + assert.equal(spawnedCommand.options.additionalFds?.fd4?.type, "input"); + assert.equal(spawnedCommand.options.additionalFds?.fd5?.type, "output"); assert.equal( Duration.toMillis(Duration.fromInputUnsafe(spawnedCommand.options.forceKillAfter)), 2_000, ); assert.deepEqual(yield* decodeBootstrap(bootstrapJson), configWithObservability); + assert.equal( + telemetryJson, + '{"version":1,"type":"desktopTelemetryHello","electronPid":123}\n', + ); }), ), ); + it.effect("preserves the readiness timeout cause and process context", () => + Effect.gen(function* () { + const requested = yield* Deferred.make(); + const layer = Layer.merge( + TestClock.layer(), + httpClientLayer((request) => + Deferred.succeed(requested, request).pipe(Effect.andThen(Effect.never)), + ), + ); + + yield* Effect.gen(function* () { + const readiness = yield* DesktopBackendManager.waitForHttpReady({ + executablePath: baseConfig.executablePath, + entryPath: baseConfig.entryPath, + cwd: baseConfig.cwd, + httpBaseUrl: baseConfig.httpBaseUrl, + timeout: Duration.millis(50), + }).pipe(Effect.flip, Effect.forkChild); + + const request = yield* Deferred.await(requested); + assert.equal(request.url, "http://127.0.0.1:3773/.well-known/t3/environment"); + + yield* TestClock.adjust(Duration.millis(50)); + const error = yield* Fiber.join(readiness); + + assert.instanceOf(error, DesktopBackendManager.BackendReadinessTimeoutError); + assert.equal(error.executablePath, "/electron"); + assert.equal(error.entryPath, "/server/bin.mjs"); + assert.equal(error.cwd, "/server"); + assert.equal(error.httpBaseUrl.href, "http://127.0.0.1:3773/"); + assert.equal(error.readinessUrl.href, "http://127.0.0.1:3773/.well-known/t3/environment"); + assert.equal(error.timeoutMs, 50); + assert.isDefined(error.cause); + assert.equal( + error.message, + "Timed out after 50ms waiting for desktop backend readiness at http://127.0.0.1:3773/.well-known/t3/environment.", + ); + }).pipe(Effect.provide(layer)); + }), + ); + + it.effect("reports bootstrap encoding failures with stable process context", () => + Effect.gen(function* () { + const spawnerLayer = Layer.succeed( + ChildProcessSpawner.ChildProcessSpawner, + ChildProcessSpawner.make(() => Effect.die("unexpected backend spawn")), + ); + const error = yield* DesktopBackendManager.runBackendProcess({ + ...baseConfig, + desktopTelemetryStream: Stream.empty, + bootstrap: { + ...baseConfig.bootstrap, + port: 0, + }, + }).pipe( + Effect.flip, + Effect.scoped, + Effect.provide(Layer.merge(spawnerLayer, healthyHttpClientLayer)), + ); + + if (error._tag !== "BackendProcessBootstrapEncodeError") { + return assert.fail(`Expected bootstrap encode error, received ${error._tag}`); + } + assert.equal(error.executablePath, "/electron"); + assert.equal(error.entryPath, "/server/bin.mjs"); + assert.equal(error.cwd, "/server"); + assert.equal(error.httpBaseUrl.href, "http://127.0.0.1:3773/"); + assert.isDefined(error.cause); + assert.equal( + error.message, + "Failed to encode the desktop backend bootstrap payload for /server/bin.mjs.", + ); + assert.isTrue(isBackendProcessError(error)); + }), + ); + + it.effect("preserves spawn failures without deriving their message from the cause", () => + Effect.gen(function* () { + const spawnCause = PlatformError.systemError({ + _tag: "PermissionDenied", + module: "ChildProcessSpawner", + method: "spawn", + pathOrDescriptor: baseConfig.executablePath, + description: "low-level detail that must not become the public message", + }); + const spawnerLayer = Layer.succeed( + ChildProcessSpawner.ChildProcessSpawner, + ChildProcessSpawner.make(() => Effect.fail(spawnCause)), + ); + const error = yield* DesktopBackendManager.runBackendProcess({ + ...baseConfig, + desktopTelemetryStream: Stream.empty, + }).pipe( + Effect.flip, + Effect.scoped, + Effect.provide(Layer.merge(spawnerLayer, healthyHttpClientLayer)), + ); + + if (error._tag !== "BackendProcessSpawnError") { + return assert.fail(`Expected backend spawn error, received ${error._tag}`); + } + assert.equal(error.executablePath, "/electron"); + assert.equal(error.entryPath, "/server/bin.mjs"); + assert.equal(error.cwd, "/server"); + assert.equal(error.httpBaseUrl.href, "http://127.0.0.1:3773/"); + assert.strictEqual(error.cause, spawnCause); + assert.equal( + error.message, + "Failed to spawn desktop backend entry /server/bin.mjs with /electron.", + ); + assert.notInclude(error.message, spawnCause.message); + assert.isTrue(isBackendProcessError(error)); + }), + ); + + it.effect("preserves exit-status failures without copying their detail into the message", () => + Effect.gen(function* () { + const exitCause = PlatformError.systemError({ + _tag: "PermissionDenied", + module: "ChildProcess", + method: "exitCode", + description: "exit-status-secret-sentinel", + }); + const spawnerLayer = Layer.succeed( + ChildProcessSpawner.ChildProcessSpawner, + ChildProcessSpawner.make(() => + Effect.succeed( + makeProcess({ + exitCode: Effect.fail(exitCause), + }), + ), + ), + ); + const error = yield* DesktopBackendManager.runBackendProcess({ + ...baseConfig, + desktopTelemetryStream: Stream.empty, + }).pipe( + Effect.flip, + Effect.scoped, + Effect.provide(Layer.merge(spawnerLayer, healthyHttpClientLayer)), + ); + + if (error._tag !== "BackendProcessExitStatusError") { + return assert.fail(`Expected backend exit-status error, received ${error._tag}`); + } + assert.equal(error.pid, 123); + assert.equal(error.executablePath, "/electron"); + assert.equal(error.entryPath, "/server/bin.mjs"); + assert.equal(error.cwd, "/server"); + assert.equal(error.httpBaseUrl.href, "http://127.0.0.1:3773/"); + assert.strictEqual(error.cause, exitCause); + assert.equal(error.message, "Failed to read the exit status of desktop backend process 123."); + assert.notInclude(error.message, "exit-status-secret-sentinel"); + assert.isTrue(isBackendProcessError(error)); + }), + ); + + it.effect("reports output stream failures with process and stream context", () => + Effect.gen(function* () { + const outputCause = PlatformError.systemError({ + _tag: "BadResource", + module: "ChildProcess", + method: "stdout", + description: "output-stream-secret-sentinel", + }); + const reported = yield* Deferred.make(); + const spawnerLayer = Layer.succeed( + ChildProcessSpawner.ChildProcessSpawner, + ChildProcessSpawner.make(() => + Effect.succeed( + makeProcess({ + stdout: Stream.fail(outputCause), + exitCode: Deferred.await(reported).pipe(Effect.as(ChildProcessSpawner.ExitCode(0))), + }), + ), + ), + ); + + const exit = yield* DesktopBackendManager.runBackendProcess({ + ...baseConfig, + desktopTelemetryStream: Stream.empty, + onOutputFailure: (error) => Deferred.succeed(reported, error).pipe(Effect.asVoid), + }).pipe(Effect.scoped, Effect.provide(Layer.merge(spawnerLayer, healthyHttpClientLayer))); + const error = yield* Deferred.await(reported); + + assert.equal(exit.code.pipe(Option.getOrUndefined), 0); + if (error._tag !== "BackendProcessOutputReadError") { + return assert.fail(`Expected output read error, received ${error._tag}`); + } + assert.equal(error.executablePath, "/electron"); + assert.equal(error.entryPath, "/server/bin.mjs"); + assert.equal(error.cwd, "/server"); + assert.equal(error.httpBaseUrl.href, "http://127.0.0.1:3773/"); + assert.equal(error.pid, 123); + assert.equal(error.streamName, "stdout"); + assert.strictEqual(error.cause, outputCause); + assert.equal(error.message, "Failed to read stdout from desktop backend process 123."); + assert.notInclude(error.message, "output-stream-secret-sentinel"); + }), + ); + + it.effect("reports output handler failures separately from stream read failures", () => + Effect.gen(function* () { + const chunk = new TextEncoder().encode("backend output"); + const nextChunk = new TextEncoder().encode("still draining"); + const outputCause = new Error("output-handler-secret-sentinel"); + const reported = yield* Deferred.make(); + const drained = yield* Deferred.make(); + let outputCount = 0; + const spawnerLayer = Layer.succeed( + ChildProcessSpawner.ChildProcessSpawner, + ChildProcessSpawner.make(() => + Effect.succeed( + makeProcess({ + stdout: Stream.make(chunk, nextChunk), + exitCode: Deferred.await(drained).pipe(Effect.as(ChildProcessSpawner.ExitCode(0))), + }), + ), + ), + ); + + const exit = yield* DesktopBackendManager.runBackendProcess({ + ...baseConfig, + desktopTelemetryStream: Stream.empty, + onOutput: () => { + outputCount += 1; + return outputCount === 1 + ? Effect.fail(outputCause) + : Deferred.succeed(drained, void 0).pipe(Effect.asVoid); + }, + onOutputFailure: (error) => Deferred.succeed(reported, error).pipe(Effect.asVoid), + }).pipe(Effect.scoped, Effect.provide(Layer.merge(spawnerLayer, healthyHttpClientLayer))); + const error = yield* Deferred.await(reported); + + assert.equal(exit.code.pipe(Option.getOrUndefined), 0); + if (error._tag !== "BackendProcessOutputHandlingError") { + return assert.fail(`Expected output handling error, received ${error._tag}`); + } + assert.equal(error.executablePath, "/electron"); + assert.equal(error.entryPath, "/server/bin.mjs"); + assert.equal(error.cwd, "/server"); + assert.equal(error.httpBaseUrl.href, "http://127.0.0.1:3773/"); + assert.equal(error.pid, 123); + assert.equal(error.streamName, "stdout"); + assert.equal(error.chunkByteLength, chunk.byteLength); + assert.strictEqual(error.cause, outputCause); + assert.equal( + error.message, + `Failed to handle ${chunk.byteLength} bytes from stdout of desktop backend process 123.`, + ); + assert.notInclude(error.message, "output-handler-secret-sentinel"); + assert.equal(outputCount, 2); + }), + ); + + it.effect("reports child exit before waiting for trailing output to drain", () => + Effect.scoped( + Effect.gen(function* () { + const exitObserved = yield* Deferred.make(); + const finishOutputDrain = yield* Deferred.make(); + const spawnerLayer = Layer.succeed( + ChildProcessSpawner.ChildProcessSpawner, + ChildProcessSpawner.make(() => + Effect.succeed( + makeProcess({ + stdout: Stream.fromEffect( + Deferred.await(finishOutputDrain).pipe( + Effect.as(new TextEncoder().encode("trailing output\n")), + ), + ), + exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(1)), + }), + ), + ), + ); + + const runFiber = yield* DesktopBackendManager.runBackendProcess({ + ...baseConfig, + desktopTelemetryStream: Stream.empty, + onExitObserved: () => Deferred.succeed(exitObserved, void 0).pipe(Effect.asVoid), + }).pipe( + Effect.provide(Layer.merge(spawnerLayer, healthyHttpClientLayer)), + Effect.forkChild, + ); + + yield* Deferred.await(exitObserved); + assert.isUndefined(runFiber.pollUnsafe()); + + yield* Deferred.succeed(finishOutputDrain, void 0); + assert.equal((yield* Fiber.join(runFiber)).code.pipe(Option.getOrUndefined), 1); + }), + ), + ); + + it.effect("continues routing desktop telemetry control messages after an invalid line", () => + Effect.scoped( + Effect.gen(function* () { + const handled = yield* Deferred.make(); + const controlMessage = encodeDesktopTelemetryControl({ + version: 1, + type: "setDiagnosticsDemand", + enabled: true, + }); + const spawnerLayer = Layer.succeed( + ChildProcessSpawner.ChildProcessSpawner, + ChildProcessSpawner.make(() => + Effect.succeed( + makeProcess({ + getOutputFd: (fd) => + fd === 5 + ? Stream.encodeText(Stream.make(`not-json\n${controlMessage}\n`)) + : Stream.empty, + exitCode: Deferred.await(handled).pipe(Effect.as(ChildProcessSpawner.ExitCode(0))), + }), + ), + ), + ); + const instance = yield* makeTestInstance({ + spawnerLayer, + desktopTelemetryPublisher: { + handleControl: (message) => + message.type === "setDiagnosticsDemand" + ? Deferred.succeed(handled, message.enabled).pipe(Effect.asVoid) + : Effect.void, + }, + }); + + yield* instance.start; + assert.isTrue(yield* Deferred.await(handled)); + }), + ), + ); + + it.effect("drains trailing child output before persisting an unexpected exit", () => + Effect.scoped( + Effect.gen(function* () { + const persistedOutput = yield* Deferred.make>(); + const outputDrainStarted = yield* Deferred.make(); + const outputChunks = yield* Ref.make>([]); + const spawnerLayer = Layer.succeed( + ChildProcessSpawner.ChildProcessSpawner, + ChildProcessSpawner.make(() => + Effect.succeed( + makeProcess({ + stdout: Stream.fromEffect( + Deferred.succeed(outputDrainStarted, void 0).pipe( + Effect.andThen(Effect.sleep(Duration.seconds(1))), + Effect.as(new TextEncoder().encode("trailing output\n")), + ), + ), + exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(1)), + }), + ), + ), + ); + const instance = yield* makeTestInstance({ + spawnerLayer, + httpClientLayer: httpClientLayer(() => Effect.never), + backendOutputLog: { + writeOutputChunk: (_streamName, chunk) => + Ref.update(outputChunks, (current) => [...current, new TextDecoder().decode(chunk)]), + persistFailure: () => + Ref.get(outputChunks).pipe( + Effect.flatMap((chunks) => Deferred.succeed(persistedOutput, chunks)), + Effect.asVoid, + ), + }, + }); + + yield* instance.start; + yield* Deferred.await(outputDrainStarted); + yield* TestClock.adjust(Duration.seconds(1)); + + assert.deepEqual(yield* Deferred.await(persistedOutput), ["trailing output\n"]); + }).pipe(Effect.provide(TestClock.layer())), + ), + ); + it.effect("retries HTTP readiness before reporting the backend ready", () => Effect.scoped( Effect.gen(function* () { @@ -256,8 +679,7 @@ readyCount += 1; }).pipe(Effect.andThen(Deferred.succeed(ready, void 0)), Effect.asVoid), backendOutputLog: { + persistFailure: () => Queue.offer(exited, void 0).pipe(Effect.asVoid), - writeSessionBoundary: ({ phase }) => - phase === "END" ? Queue.offer(exited, void 0).pipe(Effect.asVoid) : Effect.void, }, }); @@ -285,9 +707,15 @@ let startCount = 0; let closedCount = 0; const closed = yield* Deferred.make(); + const teardownStarted = yield* Deferred.make(); + const finishTeardown = yield* Deferred.make(); const startedPids = yield* Queue.unbounded(); const ready = yield* Deferred.make(); const backendReadyFlag = yield* Ref.make(false); + let shutdownCount = 0; + let persistedFailureCount = 0; + let discardedSessionCount = 0; + let removedTelemetrySources = 0; const spawnerLayer = Layer.succeed( ChildProcessSpawner.ChildProcessSpawner, @@ -296,9 +724,16 @@ const scope = yield* Scope.Scope; startCount += 1; yield* Queue.offer(startedPids, 123); + const close = Deferred.succeed(teardownStarted, undefined).pipe( + Effect.andThen(Deferred.await(finishTeardown)), + Effect.andThen( + Effect.sync(() => { + closedCount += 1; + }), + ), + Effect.andThen(Deferred.succeed(closed, void 0)), + Effect.asVoid, + ); - const close = Effect.sync(() => { - closedCount += 1; - }).pipe(Effect.andThen(Deferred.succeed(closed, void 0)), Effect.asVoid); yield* Scope.addFinalizer(scope, close); @@ -316,7 +751,28 @@ Effect.andThen(Deferred.succeed(ready, void 0)), Effect.asVoid, ), + onShutdown: Ref.set(backendReadyFlag, false).pipe( + Effect.andThen( + Effect.sync(() => { + shutdownCount += 1; + }), + ), + ), + backendOutputLog: { + persistFailure: () => + Effect.sync(() => { + persistedFailureCount += 1; + }), + discardSession: Effect.sync(() => { + discardedSessionCount += 1; + }), + }, + desktopTelemetryPublisher: { - onShutdown: Ref.set(backendReadyFlag, false), + removeControlSource: () => + Effect.sync(() => { + removedTelemetrySources += 1; + }), + }, }); assert.isTrue(Option.isNone(yield* instance.currentConfig)); @@ -330,12 +786,21 @@ assert.equal(runningSnapshot.ready, true); assert.deepEqual(runningSnapshot.activePid, Option.some(123)); + const stopFiber = yield* instance.stop().pipe(Effect.forkChild); + yield* Deferred.await(teardownStarted).pipe(Effect.timeout("1 second")); + assert.isFalse(yield* Ref.get(backendReadyFlag)); + assert.equal(shutdownCount, 1); + yield* Deferred.succeed(finishTeardown, undefined); - yield* instance.stop(); + yield* Fiber.join(stopFiber).pipe(Effect.timeout("1 second")); assert.equal(startCount, 1); assert.equal(closedCount, 1); + assert.equal(persistedFailureCount, 0); + assert.equal(discardedSessionCount, 1); + assert.equal(removedTelemetrySources, 1); const stoppedSnapshot = yield* instance.snapshot; assert.isFalse(yield* Ref.get(backendReadyFlag)); + assert.equal(shutdownCount, 1); assert.equal(stoppedSnapshot.desiredRunning, false); assert.equal(stoppedSnapshot.ready, false); assert.equal(Option.isNone(stoppedSnapshot.activePid), true); @@ -343,6 +808,213 @@ ), ); + it.effect("restarts when start is requested during stop teardown", () => + Effect.scoped( + Effect.gen(function* () { + const starts = yield* Queue.unbounded(); + const teardownStarted = yield* Deferred.make(); + const finishTeardown = yield* Deferred.make(); + let startCount = 0; + + const spawnerLayer = Layer.succeed( + ChildProcessSpawner.ChildProcessSpawner, + ChildProcessSpawner.make(() => + Effect.gen(function* () { + const scope = yield* Scope.Scope; + const closed = yield* Deferred.make(); + startCount += 1; + yield* Queue.offer(starts, startCount); + if (startCount === 1) { + yield* Scope.addFinalizer( + scope, + Deferred.succeed(teardownStarted, undefined).pipe( + Effect.andThen(Deferred.await(finishTeardown)), + Effect.andThen(Deferred.succeed(closed, undefined)), + Effect.asVoid, + ), + ); + } else { + yield* Scope.addFinalizer( + scope, + Deferred.succeed(closed, undefined).pipe(Effect.asVoid), + ); + } + return makeProcess({ + exitCode: Deferred.await(closed).pipe(Effect.as(ChildProcessSpawner.ExitCode(0))), + }); + }), + ), + ); + + const instance = yield* makeTestInstance({ + spawnerLayer, + httpClientLayer: httpClientLayer(() => Effect.never), + }); + + yield* instance.start; + assert.equal(yield* Queue.take(starts), 1); + + const stopFiber = yield* instance.stop().pipe(Effect.forkChild); + yield* Deferred.await(teardownStarted).pipe(Effect.timeout("1 second")); + yield* instance.start; + assert.equal((yield* instance.snapshot).desiredRunning, true); + + yield* Deferred.succeed(finishTeardown, undefined); + yield* Fiber.join(stopFiber).pipe(Effect.timeout("1 second")); + yield* TestClock.adjust(Duration.millis(500)); + + assert.equal(yield* Queue.take(starts).pipe(Effect.timeout("1 second")), 2); + const restarted = yield* instance.snapshot; + assert.equal(restarted.desiredRunning, true); + assert.deepEqual(restarted.activePid, Option.some(123)); + }).pipe(Effect.provide(TestClock.layer())), + ), + ); + + it.effect("retries config resolution after a start request during stop teardown", () => + Effect.scoped( + Effect.gen(function* () { + const starts = yield* Queue.unbounded(); + const teardownStarted = yield* Deferred.make(); + const finishTeardown = yield* Deferred.make(); + const configAttempts = yield* Ref.make(0); + let startCount = 0; + + const configFailure = PlatformError.systemError({ + _tag: "Unknown", + module: "DesktopBackendManager", + method: "configResolve", + description: "transient configuration failure", + }); + const spawnerLayer = Layer.succeed( + ChildProcessSpawner.ChildProcessSpawner, + ChildProcessSpawner.make(() => + Effect.gen(function* () { + const scope = yield* Scope.Scope; + const closed = yield* Deferred.make(); + startCount += 1; + yield* Queue.offer(starts, startCount); + if (startCount === 1) { + yield* Scope.addFinalizer( + scope, + Deferred.succeed(teardownStarted, undefined).pipe( + Effect.andThen(Deferred.await(finishTeardown)), + Effect.andThen(Deferred.succeed(closed, undefined)), + Effect.asVoid, + ), + ); + } else { + yield* Scope.addFinalizer( + scope, + Deferred.succeed(closed, undefined).pipe(Effect.asVoid), + ); + } + return makeProcess({ + exitCode: Deferred.await(closed).pipe(Effect.as(ChildProcessSpawner.ExitCode(0))), + }); + }), + ), + ); + const configResolve = Ref.updateAndGet(configAttempts, (attempt) => attempt + 1).pipe( + Effect.flatMap((attempt) => + attempt === 2 ? Effect.fail(configFailure) : Effect.succeed(baseConfig), + ), + ); + const instance = yield* makeTestInstance({ + spawnerLayer, + configResolve, + httpClientLayer: httpClientLayer(() => Effect.never), + }); + + yield* instance.start; + assert.equal(yield* Queue.take(starts), 1); + + const stopFiber = yield* instance.stop().pipe(Effect.forkChild); + yield* Deferred.await(teardownStarted).pipe(Effect.timeout("1 second")); + yield* instance.start; + yield* Deferred.succeed(finishTeardown, undefined); + yield* Fiber.join(stopFiber).pipe(Effect.timeout("1 second")); + + const pendingRestart = yield* instance.snapshot; + assert.equal(pendingRestart.desiredRunning, true); + assert.equal(pendingRestart.restartScheduled, true); + + yield* TestClock.adjust(Duration.seconds(2)); + + assert.equal(yield* Queue.take(starts).pipe(Effect.timeout("1 second")), 2); + assert.equal(yield* Ref.get(configAttempts), 3); + assert.equal((yield* instance.snapshot).desiredRunning, true); + }).pipe(Effect.provide(TestClock.layer())), + ), + ); + + it.effect("keeps a timed-out run active until its process exits", () => + Effect.scoped( + Effect.gen(function* () { + const starts = yield* Queue.unbounded(); + const teardownStarted = yield* Deferred.make(); + const finishTeardown = yield* Deferred.make(); + let startCount = 0; + + const spawnerLayer = Layer.succeed( + ChildProcessSpawner.ChildProcessSpawner, + ChildProcessSpawner.make(() => + Effect.gen(function* () { + const scope = yield* Scope.Scope; + const closed = yield* Deferred.make(); + startCount += 1; + yield* Queue.offer(starts, startCount); + if (startCount === 1) { + yield* Scope.addFinalizer( + scope, + Deferred.succeed(teardownStarted, undefined).pipe( + Effect.andThen(Deferred.await(finishTeardown)), + Effect.andThen(Deferred.succeed(closed, undefined)), + Effect.asVoid, + ), + ); + } else { + yield* Scope.addFinalizer( + scope, + Deferred.succeed(closed, undefined).pipe(Effect.asVoid), + ); + } + return makeProcess({ + exitCode: Deferred.await(closed).pipe(Effect.as(ChildProcessSpawner.ExitCode(0))), + }); + }), + ), + ); + + const instance = yield* makeTestInstance({ + spawnerLayer, + httpClientLayer: httpClientLayer(() => Effect.never), + }); + + yield* instance.start; + assert.equal(yield* Queue.take(starts), 1); + + const stopFiber = yield* instance + .stop({ timeout: Duration.millis(100) }) + .pipe(Effect.forkChild); + yield* Deferred.await(teardownStarted).pipe(Effect.timeout("1 second")); + yield* instance.start; + yield* TestClock.adjust(Duration.millis(100)); + yield* Fiber.join(stopFiber).pipe(Effect.timeout("1 second")); + + assert.equal(startCount, 1); + const timedOut = yield* instance.snapshot; + assert.equal(timedOut.desiredRunning, true); + assert.deepEqual(timedOut.activePid, Option.some(123)); + + yield* Deferred.succeed(finishTeardown, undefined); + yield* TestClock.adjust(Duration.millis(500)); + + assert.equal(yield* Queue.take(starts).pipe(Effect.timeout("1 second")), 2); + }).pipe(Effect.provide(TestClock.layer())), + ), + ); + it.effect("does not notify shutdown before the first start has prior state", () => Effect.scoped( Effect.gen(function* () { @@ -383,6 +1055,7 @@ Effect.scoped( Effect.gen(function* () { const starts = yield* Queue.unbounded(); + const failures = yield* Queue.unbounded(); let startCount = 0; const spawnerLayer = Layer.succeed( @@ -402,11 +1075,15 @@ const instance = yield* makeTestInstance({ spawnerLayer, httpClientLayer: httpClientLayer(() => Effect.never), + backendOutputLog: { + persistFailure: ({ details }) => Queue.offer(failures, details).pipe(Effect.asVoid), + }, }); yield* instance.start; assert.equal(yield* Queue.take(starts), 1); + assert.equal(yield* Queue.take(failures), "pid=123 code=1"); yield* TestClock.adjust(Duration.millis(499)); assert.equal(yield* Queue.size(starts), 0); @@ -657,6 +1334,13 @@ yield* instance.start; assert.equal(yield* Queue.take(starts), 1); + let restartScheduled = false; + while (!restartScheduled) { + restartScheduled = (yield* instance.snapshot).restartScheduled; + if (!restartScheduled) { + yield* Effect.yieldNow; + } + } yield* instance.start; assert.equal(yield* Queue.take(starts), 2); ``` ## apps/desktop/src/backend/DesktopBackendManager.ts Status: modified; +466; −144 ```diff @@ -33,7 +33,6 @@ import * as Option from "effect/Option"; import * as PlatformError from "effect/PlatformError"; import * as Ref from "effect/Ref"; -import * as Result from "effect/Result"; import * as Schedule from "effect/Schedule"; import * as Schema from "effect/Schema"; import * as Semaphore from "effect/Semaphore"; @@ -46,10 +45,13 @@ DesktopBackendBootstrap, type DesktopBackendBootstrap as DesktopBackendBootstrapValue, PRIMARY_LOCAL_ENVIRONMENT_ID, + DesktopTelemetryControlMessage, + type DesktopTelemetryControlMessage as DesktopTelemetryControlMessageValue, } from "@t3tools/contracts"; import { waitForHttpReady as waitForHttpReadyShared } from "@t3tools/shared/httpReadiness"; import * as DesktopObservability from "../app/DesktopObservability.ts"; +import * as DesktopTelemetryPublisher from "../telemetry/DesktopTelemetryPublisher.ts"; const INITIAL_RESTART_DELAY = Duration.millis(500); const MAX_RESTART_DELAY = Duration.seconds(10); @@ -62,7 +64,10 @@ const DEFAULT_BACKEND_READINESS_INTERVAL = Duration.millis(100); const DEFAULT_BACKEND_READINESS_REQUEST_TIMEOUT = Duration.seconds(1); const DEFAULT_BACKEND_TERMINATE_GRACE = Duration.seconds(2); +const DEFAULT_BACKEND_OUTPUT_DRAIN_TIMEOUT = Duration.seconds(5); const BACKEND_READINESS_PATH = "/.well-known/t3/environment"; +const { logWarning: logBackendProcessWarning } = + DesktopObservability.makeComponentLogger("desktop-backend-process"); type BackendProcessLayerServices = ChildProcessSpawner.ChildProcessSpawner | HttpClient.HttpClient; @@ -70,13 +75,17 @@ export type BackendProcessOutputStream = "stdout" | "stderr"; -export type DesktopBackendBootstrapDelivery = "fd3" | "stdin"; - -export interface DesktopBackendStartConfig { +export interface BackendProcessContext { readonly executablePath: string; - readonly args: ReadonlyArray; readonly entryPath: string; readonly cwd: string; + readonly httpBaseUrl: URL; +} + +export type DesktopBackendBootstrapDelivery = "fd3" | "stdin"; + +export interface DesktopBackendStartConfig extends BackendProcessContext { + readonly args: ReadonlyArray; readonly env: Record; // When true the spawner merges the desktop process.env on top of `env`; // when false `env` is passed verbatim. WSL mode opts out so a leaking @@ -106,55 +115,122 @@ interface BackendProcessExit { readonly code: Option.Option; readonly reason: string; - readonly result: Result.Result; } +const backendProcessContextSchema = { + executablePath: Schema.String, + entryPath: Schema.String, -export class BackendTimeoutError extends Schema.TaggedErrorClass()( + cwd: Schema.String, + httpBaseUrl: Schema.URL, +}; + +export class BackendReadinessTimeoutError extends Schema.TaggedErrorClass()( + "BackendReadinessTimeoutError", - "BackendTimeoutError", { + ...backendProcessContextSchema, + readinessUrl: Schema.URL, + timeoutMs: Schema.Number, + cause: Schema.Defect(), - url: Schema.instanceOf(URL), }, ) { - override get message() { + override get message(): string { + return `Timed out after ${this.timeoutMs}ms waiting for desktop backend readiness at ${this.readinessUrl.href}.`; - return `Timed out waiting for backend readiness at ${this.url.href}.`; } } +export class BackendProcessBootstrapEncodeError extends Schema.TaggedErrorClass()( -class BackendProcessBootstrapEncodeError extends Schema.TaggedErrorClass()( "BackendProcessBootstrapEncodeError", { - entryPath: Schema.String, + ...backendProcessContextSchema, cause: Schema.Defect(), }, ) { - override get message() { + override get message(): string { return `Failed to encode the desktop backend bootstrap payload for ${this.entryPath}.`; } } +export class BackendProcessSpawnError extends Schema.TaggedErrorClass()( -class BackendProcessSpawnError extends Schema.TaggedErrorClass()( "BackendProcessSpawnError", { + ...backendProcessContextSchema, + cause: Schema.Defect(), + }, +) { + override get message(): string { + return `Failed to spawn desktop backend entry ${this.entryPath} with ${this.executablePath}.`; + } +} + +export class BackendProcessOutputReadError extends Schema.TaggedErrorClass()( + "BackendProcessOutputReadError", + { + ...backendProcessContextSchema, + pid: Schema.Number, + streamName: Schema.Literals(["stdout", "stderr"]), + cause: Schema.Defect(), + }, +) { + override get message(): string { + return `Failed to read ${this.streamName} from desktop backend process ${this.pid}.`; + } +} + +export class BackendProcessOutputHandlingError extends Schema.TaggedErrorClass()( + "BackendProcessOutputHandlingError", + { + ...backendProcessContextSchema, + pid: Schema.Number, + streamName: Schema.Literals(["stdout", "stderr"]), - executablePath: Schema.String, + chunkByteLength: Schema.Number, cause: Schema.Defect(), }, ) { + override get message(): string { - override get message() { - return `Failed to spawn the desktop backend process at ${this.executablePath}.`; + return `Failed to handle ${this.chunkByteLength} bytes from ${this.streamName} of desktop backend process ${this.pid}.`; } } +export type BackendProcessOutputError = + | BackendProcessOutputReadError -type BackendProcessError = BackendProcessBootstrapEncodeError | BackendProcessSpawnError; + | BackendProcessOutputHandlingError; +export class BackendProcessExitStatusError extends Schema.TaggedErrorClass()( + "BackendProcessExitStatusError", + { + ...backendProcessContextSchema, + pid: Schema.Number, + cause: Schema.Defect(), + }, +) { + override get message(): string { + return `Failed to read the exit status of desktop backend process ${this.pid}.`; + } +} + +export const BackendProcessError = Schema.Union([ + BackendProcessBootstrapEncodeError, + BackendProcessSpawnError, + BackendProcessExitStatusError, +]); +export type BackendProcessError = typeof BackendProcessError.Type; + interface RunBackendProcessOptions extends DesktopBackendStartConfig { + readonly desktopTelemetryStream: Stream.Stream; + readonly onDesktopTelemetryControl?: ( + message: DesktopTelemetryControlMessageValue, + ) => Effect.Effect; readonly readinessTimeout?: Duration.Duration; + readonly outputDrainTimeout?: Duration.Duration; readonly onStarted?: (pid: number) => Effect.Effect; + readonly onExitObserved?: () => Effect.Effect; readonly onReady?: () => Effect.Effect; - readonly onReadinessFailure?: (error: BackendTimeoutError) => Effect.Effect; + readonly onReadinessFailure?: (error: BackendReadinessTimeoutError) => Effect.Effect; readonly onOutput?: ( streamName: BackendProcessOutputStream, chunk: Uint8Array, + ) => Effect.Effect; + readonly onOutputFailure?: (error: BackendProcessOutputError) => Effect.Effect; - ) => Effect.Effect; } export interface DesktopBackendSnapshot { @@ -225,6 +301,8 @@ readonly scope: Scope.Closeable; readonly fiber: Option.Option>; readonly pid: Option.Option; + readonly exitObserved: boolean; + readonly stopRequested: boolean; } interface BackendManagerState { @@ -266,76 +344,130 @@ const closeRun = ( run: ActiveBackendRun, + parentScope: Scope.Scope, options?: { readonly timeout?: Duration.Duration }, +): Effect.Effect => { -): Effect.Effect => { const waitForFiber = Option.match(run.fiber, { onNone: () => Effect.void, onSome: (fiber) => Fiber.await(fiber).pipe(Effect.asVoid), }); const close = Scope.close(run.scope, Exit.void).pipe(Effect.andThen(waitForFiber)); + const timeout = options?.timeout; + if (!timeout) { + return close.pipe(Effect.as(true)); + } + + return Effect.forkIn(close, parentScope).pipe( + Effect.flatMap((closeFiber) => - return ( - options?.timeout ? close.pipe(Effect.timeoutOption(options.timeout), Effect.asVoid) : close - ).pipe(Effect.ignore); + Fiber.await(closeFiber).pipe(Effect.timeoutOption(timeout), Effect.map(Option.isSome)), + ), + ); }; -const waitForHttpReady = ( - baseUrl: URL, - timeout: Duration.Duration, +export const waitForHttpReady = ( + options: BackendProcessContext & { readonly timeout: Duration.Duration }, +): Effect.Effect => { -): Effect.Effect => { + const readinessUrl = new URL(BACKEND_READINESS_PATH, options.httpBaseUrl); - const readinessUrl = new URL(BACKEND_READINESS_PATH, baseUrl); return waitForHttpReadyShared({ + baseUrl: options.httpBaseUrl.href, - baseUrl: baseUrl.href, path: BACKEND_READINESS_PATH, - timeoutMs: Duration.toMillis(timeout), + timeoutMs: Duration.toMillis(options.timeout), intervalMs: Duration.toMillis(DEFAULT_BACKEND_READINESS_INTERVAL), probeTimeoutMs: Duration.toMillis(DEFAULT_BACKEND_READINESS_REQUEST_TIMEOUT), + makeError: ({ cause }) => + new BackendReadinessTimeoutError({ + executablePath: options.executablePath, + entryPath: options.entryPath, + cwd: options.cwd, + httpBaseUrl: options.httpBaseUrl, + readinessUrl, - makeError: () => new BackendTimeoutError({ url: readinessUrl }), + timeoutMs: Duration.toMillis(options.timeout), + cause, + }), }); }; -function describeProcessExit( - result: Result.Result, -): BackendProcessExit { - if (Result.isSuccess(result)) { - return { - code: Option.some(result.success), - reason: `code=${result.success}`, - result, - }; - } - - return { - code: Option.none(), - reason: result.failure.message, - result, - }; -} - function drainBackendOutput( + context: BackendProcessContext & { readonly pid: number }, streamName: BackendProcessOutputStream, stream: Stream.Stream, + onOutput: ( + streamName: BackendProcessOutputStream, + chunk: Uint8Array, + ) => Effect.Effect, + onOutputFailure: (error: BackendProcessOutputError) => Effect.Effect, - onOutput: (streamName: BackendProcessOutputStream, chunk: Uint8Array) => Effect.Effect, ): Effect.Effect { return stream.pipe( + Stream.mapError( + (cause) => - Stream.runForEach((chunk) => onOutput(streamName, chunk)), - Effect.ignore, + new BackendProcessOutputReadError({ + ...context, + streamName, + cause, + }), + ), + Stream.runForEach((chunk) => + onOutput(streamName, chunk).pipe( + Effect.mapError( + (cause) => + new BackendProcessOutputHandlingError({ + ...context, + streamName, + chunkByteLength: chunk.byteLength, + cause, + }), + ), + Effect.catchTag("BackendProcessOutputHandlingError", onOutputFailure), + ), + ), + Effect.catchTags({ + BackendProcessOutputReadError: onOutputFailure, + }), ); } const encodeBootstrapJson = Schema.encodeEffect(Schema.fromJsonString(DesktopBackendBootstrap)); +const decodeDesktopTelemetryControlLine = Schema.decodeUnknownEffect( + Schema.fromJsonString(DesktopTelemetryControlMessage), +); -const runBackendProcess = Effect.fn("runBackendProcess")(function* ( +export const runBackendProcess = Effect.fn("runBackendProcess")(function* ( options: RunBackendProcessOptions, ): Effect.fn.Return { const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; const bootstrapJson = yield* encodeBootstrapJson(options.bootstrap).pipe( Effect.mapError( + (cause) => + new BackendProcessBootstrapEncodeError({ + executablePath: options.executablePath, + entryPath: options.entryPath, + cwd: options.cwd, + httpBaseUrl: options.httpBaseUrl, + cause, + }), - (cause) => new BackendProcessBootstrapEncodeError({ entryPath: options.entryPath, cause }), ), ); const onOutput = options.onOutput ?? (() => Effect.void); const bootstrapStream = Stream.encodeText(Stream.make(`${bootstrapJson}\n`)); + const additionalFds: Record<`fd${number}`, ChildProcess.AdditionalFdConfig> = {}; + if (options.bootstrapDelivery === "fd3") { + additionalFds.fd3 = { + type: "input", + stream: bootstrapStream, + }; + if (options.bootstrap.desktopTelemetryFd !== undefined) { + additionalFds[`fd${options.bootstrap.desktopTelemetryFd}`] = { + type: "input", + stream: options.desktopTelemetryStream, + }; + } + if (options.bootstrap.desktopTelemetryControlFd !== undefined) { + additionalFds[`fd${options.bootstrap.desktopTelemetryControlFd}`] = { + type: "output", + }; + } + } const command = ChildProcess.make(options.executablePath, options.args, { cwd: options.cwd, env: options.env, @@ -350,34 +482,131 @@ // wsl.exe drops additional file descriptors when forwarding to the Linux // side, so the WSL spawn path delivers the bootstrap envelope via stdin // (`--bootstrap-fd 0`) instead. - ...(options.bootstrapDelivery === "fd3" - ? { additionalFds: { fd3: { type: "input" as const, stream: bootstrapStream } } } - : {}), + ...(options.bootstrapDelivery === "fd3" ? { additionalFds } : {}), }); + const handle = yield* spawner.spawn(command).pipe( + Effect.mapError( + (cause) => + new BackendProcessSpawnError({ + executablePath: options.executablePath, + entryPath: options.entryPath, + cwd: options.cwd, + httpBaseUrl: options.httpBaseUrl, + cause, + }), + ), + ); - const handle = yield* spawner - .spawn(command) - .pipe( - Effect.mapError( - (cause) => new BackendProcessSpawnError({ executablePath: options.executablePath, cause }), - ), - ); + const outputFibers: Array> = []; yield* options.onStarted?.(handle.pid) ?? Effect.void; + if ( + options.bootstrap.desktopTelemetryControlFd !== undefined && + options.onDesktopTelemetryControl !== undefined + ) { + const controlFd = options.bootstrap.desktopTelemetryControlFd; + const handleControl = options.onDesktopTelemetryControl; + yield* handle.getOutputFd(controlFd).pipe( + Stream.decodeText(), + Stream.splitLines, + Stream.filter((line) => line.trim().length > 0), + Stream.runForEach((line) => + decodeDesktopTelemetryControlLine(line).pipe( + Effect.flatMap(handleControl), + Effect.catchCause((cause) => + logBackendProcessWarning("ignored invalid desktop telemetry control message", { + fd: controlFd, + cause: Cause.pretty(cause), + }), + ), + ), + ), + Effect.catchCause((cause) => + logBackendProcessWarning("desktop telemetry control stream stopped", { + fd: controlFd, + cause: Cause.pretty(cause), + }), + ), + Effect.ensuring( + handleControl({ + version: 1, + type: "setDiagnosticsDemand", + enabled: false, + }), + ), + Effect.forkScoped, + ); + } if (options.captureOutput) { + const outputContext = { + executablePath: options.executablePath, + entryPath: options.entryPath, + cwd: options.cwd, + httpBaseUrl: options.httpBaseUrl, + pid: Number(handle.pid), + }; + const onOutputFailure = options.onOutputFailure ?? (() => Effect.void); + outputFibers.push( + yield* drainBackendOutput( + outputContext, + "stdout", + handle.stdout, + onOutput, + onOutputFailure, + ).pipe(Effect.forkScoped), + yield* drainBackendOutput( + outputContext, + "stderr", + handle.stderr, + onOutput, + onOutputFailure, + ).pipe(Effect.forkScoped), - yield* drainBackendOutput("stdout", handle.stdout, onOutput).pipe(Effect.forkScoped); - yield* drainBackendOutput("stderr", handle.stderr, onOutput).pipe(Effect.forkScoped); + ); } + yield* waitForHttpReady({ + executablePath: options.executablePath, + entryPath: options.entryPath, + cwd: options.cwd, + httpBaseUrl: options.httpBaseUrl, + timeout: options.readinessTimeout ?? DEFAULT_BACKEND_READINESS_TIMEOUT, + }).pipe( - yield* waitForHttpReady( - options.httpBaseUrl, - options.readinessTimeout ?? DEFAULT_BACKEND_READINESS_TIMEOUT, - ).pipe( Effect.tap(() => options.onReady?.() ?? Effect.void), + Effect.catchTags({ + BackendReadinessTimeoutError: (error) => options.onReadinessFailure?.(error) ?? Effect.void, - Effect.catch((error) => options.onReadinessFailure?.(error) ?? Effect.void), + }), Effect.forkScoped, ); + const exit = yield* handle.exitCode.pipe( + Effect.mapError( + (cause) => + new BackendProcessExitStatusError({ + executablePath: options.executablePath, + entryPath: options.entryPath, + cwd: options.cwd, + httpBaseUrl: options.httpBaseUrl, + pid: Number(handle.pid), + cause, + }), + ), + Effect.exit, - return describeProcessExit(yield* Effect.result(handle.exitCode)); + ); + yield* options.onExitObserved?.() ?? Effect.void; + yield* Effect.forEach(outputFibers, Fiber.await, { + concurrency: "unbounded", + discard: true, + }).pipe( + Effect.timeout(options.outputDrainTimeout ?? DEFAULT_BACKEND_OUTPUT_DRAIN_TIMEOUT), + Effect.ignore, + ); + if (Exit.isFailure(exit)) { + return yield* Effect.failCause(exit.cause); + } + const exitCode = exit.value; + return { + code: Option.some(exitCode), + reason: `code=${exitCode}`, + } satisfies BackendProcessExit; }); // Factory for one pooled backend instance. The returned instance owns @@ -394,12 +623,14 @@ | ChildProcessSpawner.ChildProcessSpawner | HttpClient.HttpClient | DesktopObservability.DesktopBackendOutputLogFactory + | DesktopTelemetryPublisher.DesktopTelemetryPublisher | Scope.Scope > { const parentScope = yield* Scope.Scope; const fileSystem = yield* FileSystem.FileSystem; const backendOutputLogFactory = yield* DesktopObservability.DesktopBackendOutputLogFactory; const backendOutputLog = yield* backendOutputLogFactory.forInstance(spec.id); + const desktopTelemetryPublisher = yield* DesktopTelemetryPublisher.DesktopTelemetryPublisher; const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; const httpClient = yield* HttpClient.HttpClient; const state = yield* Ref.make(initialState); @@ -444,6 +675,12 @@ Effect.gen(function* () { const current = yield* Ref.get(state); if (Option.isSome(current.active)) { + if (!current.desiredRunning) { + yield* Ref.update(state, (latest) => ({ + ...latest, + desiredRunning: true, + })); + } return; } @@ -462,6 +699,9 @@ Effect.option, ); if (Option.isNone(config)) { + if (current.desiredRunning) { + yield* scheduleRestart("failed to generate desktop backend configuration"); + } return; } const entryExists = yield* fileSystem @@ -561,6 +801,8 @@ scope: runScope, fiber: Option.none(), pid: Option.none(), + exitObserved: false, + stopRequested: false, } satisfies ActiveBackendRun), nextRunId: latest.nextRunId + 1, }, @@ -571,54 +813,70 @@ ) { yield* mutex.withPermits(1)( Effect.gen(function* () { - const { isCurrentRun, nextState, pid } = yield* Ref.modify( - state, - ( - latest, - ): readonly [ - { - readonly isCurrentRun: boolean; - readonly nextState: BackendManagerState; - readonly pid: Option.Option; - }, - BackendManagerState, + const { isCurrentRun, nextState, pid, exitObserved, stopRequested, wasReady } = + yield* Ref.modify( + state, + ( + latest, + ): readonly [ + { + readonly isCurrentRun: boolean; + readonly nextState: BackendManagerState; + readonly pid: Option.Option; + readonly exitObserved: boolean; + readonly stopRequested: boolean; + readonly wasReady: boolean; + }, + BackendManagerState, - ] => { - const currentRun = Option.getOrUndefined(latest.active); - if (currentRun?.id !== runId) { + ] => { + const currentRun = Option.getOrUndefined(latest.active); + if (currentRun?.id !== runId) { + return [ + { + isCurrentRun: false, + nextState: latest, + pid: Option.none(), + exitObserved: false, + stopRequested: false, + wasReady: false, + }, + latest, + ] as const; + } + + const next = { + ...latest, + active: Option.none(), + ready: false, + }; return [ { + isCurrentRun: true, - isCurrentRun: false, - nextState: latest, - pid: Option.none(), + nextState: next, + pid: currentRun.pid, + exitObserved: currentRun.exitObserved, + stopRequested: currentRun.stopRequested, + wasReady: latest.ready, }, - latest, + next, ] as const; + }, - } + ); - const next = { - ...latest, - active: Option.none(), - ready: false, - }; - return [ - { - isCurrentRun: true, - nextState: next, - pid: currentRun.pid, - }, - next, - ] as const; - }, - ); - if (isCurrentRun) { + yield* desktopTelemetryPublisher.removeControlSource(spec.id); if (Option.isSome(pid)) { + if (exitObserved && !stopRequested) { + yield* backendOutputLog.persistFailure({ - yield* backendOutputLog.writeSessionBoundary({ - phase: "END", - details: `pid=${pid.value} ${reason}`, - }); + details: `pid=${pid.value} ${reason}`, + }); + } else { + yield* backendOutputLog.discardSession; + } } + if (wasReady) { + yield* spec.onShutdown?.() ?? Effect.void; + } - yield* spec.onShutdown?.() ?? Effect.void; } if (isCurrentRun && nextState.desiredRunning) { @@ -630,16 +888,23 @@ const program = runBackendProcess({ ...config.value, + desktopTelemetryStream: desktopTelemetryPublisher.encoded, + onDesktopTelemetryControl: (message) => + desktopTelemetryPublisher.handleControlForSource(spec.id, message), onStarted: Effect.fn("desktop.backendInstance.onStarted")(function* (pid) { yield* updateActiveRun(runId, (run) => ({ ...run, pid: Option.some(pid), })); + yield* backendOutputLog.beginSession({ - yield* backendOutputLog.writeSessionBoundary({ - phase: "START", details: `pid=${pid} port=${config.value.bootstrap.port} cwd=${config.value.cwd}`, }); }), + onExitObserved: () => + updateActiveRun(runId, (run) => ({ + ...run, + exitObserved: true, + })), onReady: Effect.fn("desktop.backendInstance.onReady")(function* () { const isCurrentRun = yield* Ref.modify(state, (latest) => { const activeRun = Option.getOrUndefined(latest.active); @@ -662,10 +927,16 @@ yield* spec.onReady?.(config.value.httpBaseUrl) ?? Effect.void; }), + onReadinessFailure: Effect.fn("desktop.backendInstance.onReadinessFailure")( + function* (error) { + yield* logInstanceWarning("backend readiness check failed during bootstrap", { + error: error.message, + }); + yield* backendOutputLog.persistFailureSnapshot({ + details: error.message, + }); + }, - onReadinessFailure: (error) => + ), - logInstanceWarning("backend readiness check failed during bootstrap", { - error: error.message, - }), onOutput: (streamName, chunk) => backendOutputLog.writeOutputChunk(streamName, chunk), }).pipe( Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), @@ -750,41 +1021,92 @@ const stop = Effect.fn("desktop.backendInstance.stop")(function* (options?: { readonly timeout?: Duration.Duration; }) { - const { active, restartFiber } = yield* mutex.withPermits(1)( + const { active, restartFiber, notifyShutdown } = yield* mutex.withPermits(1)( Effect.gen(function* () { - const result = yield* Ref.modify(state, (latest) => [ - { - active: latest.active, - restartFiber: latest.restartFiber, - }, - { - ...latest, - desiredRunning: false, - ready: false, - active: Option.none(), - restartFiber: Option.none>(), - }, - ]); - // Ignore failures from spec.onShutdown so a downstream throw - // can't abort the rest of stop(). Ref.modify above already - // flipped state to "no active run / no restart fiber", and the - // physical cleanup (Fiber.interrupt + closeRun) runs after the - // mutex releases. If onShutdown were allowed to propagate, both - // would be skipped and the child process + restart fiber would - // be orphaned while state claimed nothing was running — the - // next start() would then spawn a second backend on top. + const result = yield* Ref.modify(state, (latest) => { + const active = Option.map(latest.active, (run) => + run.exitObserved ? run : { ...run, stopRequested: true }, + ); + return [ + { + active, + restartFiber: latest.restartFiber, + notifyShutdown: latest.ready, + }, + { + ...latest, + desiredRunning: false, + ready: false, + active, - yield* (spec.onShutdown?.() ?? Effect.void).pipe(Effect.ignore); + restartFiber: Option.none>(), + }, + ] as const; + }); return result; }), ); + if (notifyShutdown) { + yield* (spec.onShutdown?.() ?? Effect.void).pipe(Effect.ignore); + } yield* Option.match(restartFiber, { onNone: () => Effect.void, onSome: (fiber) => Fiber.interrupt(fiber).pipe(Effect.asVoid), }); yield* Option.match(active, { onNone: () => Effect.void, + onSome: (run) => + Effect.gen(function* () { + const closed = yield* closeRun(run, parentScope, options); + if (!closed) { + return; + } + const cleanup = yield* mutex.withPermits(1)( + Ref.modify( + state, + ( + latest, - onSome: (run) => closeRun(run, options), + ): readonly [ + { + readonly needsCleanup: boolean; + readonly shouldStart: boolean; + }, + BackendManagerState, + ] => { + const current = Option.getOrUndefined(latest.active); + if (current?.id !== run.id) { + return [ + { + needsCleanup: false, + shouldStart: + latest.desiredRunning && + Option.isNone(latest.active) && + Option.isNone(latest.restartFiber), + }, + latest, + ]; + } + return [ + { + needsCleanup: true, + shouldStart: latest.desiredRunning, + }, + { + ...latest, + active: Option.none(), + }, + ]; + }, + ), + ); + if (cleanup.needsCleanup) { + yield* desktopTelemetryPublisher.removeControlSource(spec.id); + yield* backendOutputLog.discardSession; + } + if (cleanup.shouldStart) { + yield* start; + } + }), }); }); ``` ## apps/desktop/src/backend/DesktopBackendPool.test.ts Status: modified; +14; −1 ```diff @@ -5,11 +5,13 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; import { HttpClient } from "effect/unstable/http"; import { ChildProcessSpawner } from "effect/unstable/process"; import * as DesktopObservability from "../app/DesktopObservability.ts"; import * as DesktopAppSettings from "../settings/DesktopAppSettings.ts"; +import * as DesktopTelemetryPublisher from "../telemetry/DesktopTelemetryPublisher.ts"; import * as ElectronDialog from "../electron/ElectronDialog.ts"; import * as DesktopWindow from "../window/DesktopWindow.ts"; import * as DesktopBackendConfiguration from "./DesktopBackendConfiguration.ts"; @@ -56,10 +58,21 @@ Layer.succeed(DesktopObservability.DesktopBackendOutputLogFactory, { forInstance: () => Effect.succeed({ - writeSessionBoundary: () => Effect.void, + beginSession: () => Effect.void, writeOutputChunk: () => Effect.void, + persistFailureSnapshot: () => Effect.void, + persistFailure: () => Effect.void, + discardSession: Effect.void, } satisfies DesktopObservability.DesktopBackendOutputLogShape), } satisfies DesktopObservability.DesktopBackendOutputLogFactory["Service"]), + Layer.succeed(DesktopTelemetryPublisher.DesktopTelemetryPublisher, { + latest: Effect.succeed(Option.none()), + changes: Stream.empty, + encoded: Stream.empty, + handleControl: () => Effect.void, + handleControlForSource: () => Effect.void, + removeControlSource: () => Effect.void, + }), Layer.succeed(DesktopBackendConfiguration.DesktopBackendConfiguration, { resolvePrimary: Effect.die("unexpected primary config resolve"), resolvePrimaryLabel: Ref.get(labelRef), ``` ## apps/desktop/src/backend/DesktopBackendPool.ts Status: modified; +3; −1 ```diff @@ -97,6 +97,7 @@ import * as DesktopBackendManager from "./DesktopBackendManager.ts"; import * as DesktopObservability from "../app/DesktopObservability.ts"; import * as DesktopAppSettings from "../settings/DesktopAppSettings.ts"; +import * as DesktopTelemetryPublisher from "../telemetry/DesktopTelemetryPublisher.ts"; import * as DesktopWindow from "../window/DesktopWindow.ts"; import * as ElectronDialog from "../electron/ElectronDialog.ts"; @@ -176,7 +177,8 @@ | FileSystem.FileSystem | ChildProcessSpawner.ChildProcessSpawner | HttpClient.HttpClient - | DesktopObservability.DesktopBackendOutputLogFactory; + | DesktopObservability.DesktopBackendOutputLogFactory + | DesktopTelemetryPublisher.DesktopTelemetryPublisher; interface ActiveRegisteredInstance { readonly _tag: "Active"; ``` ## apps/desktop/src/electron/ElectronApp.ts Status: modified; +2; −0 ```diff @@ -57,6 +57,7 @@ ) => Effect.Effect; readonly setAppUserModelId: (id: string) => Effect.Effect; readonly requestSingleInstanceLock: Effect.Effect; + readonly getAppMetrics: Effect.Effect>; readonly isDefaultProtocolClient: (protocol: string) => Effect.Effect; readonly setAsDefaultProtocolClient: ( protocol: string, @@ -153,6 +154,7 @@ Electron.app.setAppUserModelId(id); }), requestSingleInstanceLock: Effect.sync(() => Electron.app.requestSingleInstanceLock()), + getAppMetrics: Effect.sync(() => Electron.app.getAppMetrics()), isDefaultProtocolClient: (protocol) => Effect.sync(() => Electron.app.isDefaultProtocolClient(protocol)), setAsDefaultProtocolClient: (protocol, path, args) => ``` ## apps/desktop/src/electron/ElectronPowerMonitor.ts Status: added; +89; −0 ```diff @@ -1,0 +1,89 @@ +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Scope from "effect/Scope"; + +import * as Electron from "electron"; + +export type ElectronThermalState = ReturnType; +export type ElectronIdleState = ReturnType; + +export class ElectronPowerMonitor extends Context.Service< + ElectronPowerMonitor, + { + readonly isOnBatteryPower: Effect.Effect; + readonly getSystemIdleTime: Effect.Effect; + readonly getSystemIdleState: (idleThresholdSeconds: number) => Effect.Effect; + readonly getCurrentThermalState: Effect.Effect; + readonly onSimpleEvent: ( + eventName: "lock-screen" | "unlock-screen" | "on-ac" | "on-battery" | "suspend" | "resume", + listener: () => void, + ) => Effect.Effect; + readonly onThermalStateChange: ( + listener: (state: ElectronThermalState) => void, + ) => Effect.Effect; + readonly onSpeedLimitChange: ( + listener: (limit: number) => void, + ) => Effect.Effect; + } +>()("@t3tools/desktop/electron/ElectronPowerMonitor") {} + +const onSimpleEvent: ElectronPowerMonitor["Service"]["onSimpleEvent"] = (eventName, listener) => + Effect.acquireRelease( + Effect.sync(() => { + Electron.powerMonitor.on(eventName as any, listener as any); + }), + () => + Effect.sync(() => { + Electron.powerMonitor.removeListener(eventName as any, listener as any); + }), + ).pipe(Effect.asVoid); + +const onThermalStateChange: ElectronPowerMonitor["Service"]["onThermalStateChange"] = ( + listener, +) => { + const wrapped = ( + event: Electron.Event, + ): void => { + listener(event.state); + }; + return Effect.acquireRelease( + Effect.sync(() => { + Electron.powerMonitor.on("thermal-state-change", wrapped); + }), + () => + Effect.sync(() => { + Electron.powerMonitor.removeListener("thermal-state-change", wrapped); + }), + ).pipe(Effect.asVoid); +}; + +const onSpeedLimitChange: ElectronPowerMonitor["Service"]["onSpeedLimitChange"] = (listener) => { + const wrapped = ( + event: Electron.Event, + ): void => { + listener(event.limit); + }; + return Effect.acquireRelease( + Effect.sync(() => { + Electron.powerMonitor.on("speed-limit-change", wrapped); + }), + () => + Effect.sync(() => { + Electron.powerMonitor.removeListener("speed-limit-change", wrapped); + }), + ).pipe(Effect.asVoid); +}; + +export const make = ElectronPowerMonitor.of({ + isOnBatteryPower: Effect.sync(() => Electron.powerMonitor.isOnBatteryPower()), + getSystemIdleTime: Effect.sync(() => Electron.powerMonitor.getSystemIdleTime()), + getSystemIdleState: (idleThresholdSeconds) => + Effect.sync(() => Electron.powerMonitor.getSystemIdleState(idleThresholdSeconds)), + getCurrentThermalState: Effect.sync(() => Electron.powerMonitor.getCurrentThermalState()), + onSimpleEvent, + onThermalStateChange, + onSpeedLimitChange, +}); + +export const layer = Layer.succeed(ElectronPowerMonitor, make); ``` ## apps/desktop/src/main.ts Status: modified; +4; −0 ```diff @@ -24,6 +24,7 @@ import * as ElectronApp from "./electron/ElectronApp.ts"; import * as ElectronDialog from "./electron/ElectronDialog.ts"; import * as ElectronMenu from "./electron/ElectronMenu.ts"; +import * as ElectronPowerMonitor from "./electron/ElectronPowerMonitor.ts"; import * as ElectronProtocol from "./electron/ElectronProtocol.ts"; import * as ElectronSafeStorage from "./electron/ElectronSafeStorage.ts"; import * as ElectronShell from "./electron/ElectronShell.ts"; @@ -52,6 +53,7 @@ import * as DesktopSshEnvironment from "./ssh/DesktopSshEnvironment.ts"; import * as DesktopSshPasswordPrompts from "./ssh/DesktopSshPasswordPrompts.ts"; import * as DesktopState from "./app/DesktopState.ts"; +import * as DesktopTelemetryPublisher from "./telemetry/DesktopTelemetryPublisher.ts"; import * as DesktopUpdates from "./updates/DesktopUpdates.ts"; import * as BrowserSession from "./preview/BrowserSession.ts"; import * as PreviewManager from "./preview/Manager.ts"; @@ -113,6 +115,7 @@ ElectronApp.layer, ElectronDialog.layer, ElectronMenu.layer, + ElectronPowerMonitor.layer, ElectronProtocol.layer, ElectronSafeStorage.layer, ElectronShell.layer, @@ -160,6 +163,7 @@ Layer.provideMerge(DesktopAppIdentity.layer), Layer.provideMerge(DesktopBackendConfiguration.layer), Layer.provideMerge(DesktopWslEnvironment.layer), + Layer.provideMerge(DesktopTelemetryPublisher.layer), Layer.provideMerge(desktopWindowLayer), ); ``` ## apps/desktop/src/preview/Manager.test.ts Status: modified; +86; −0 ```diff @@ -982,6 +982,92 @@ ), ); + effectIt.effect("emits debugger screencast frames only while recording is active", () => + withManager((manager) => + Effect.gen(function* () { + let debuggerMessage: + | ((event: unknown, method: string, params: Record) => void) + | undefined; + const capturePage = vi.fn(async () => ({ + toJPEG: () => Buffer.from("scheduled-recording-frame"), + getSize: () => ({ width: 1280, height: 720 }), + })); + const sendCommand = vi.fn(async (method: string) => + method === "Runtime.evaluate" ? { result: { value: null } } : undefined, + ); + fromId.mockReturnValue({ + id: 42, + isDestroyed: () => false, + getType: () => "webview", + getURL: () => "https://example.com", + getTitle: () => "Example", + isLoading: () => false, + isDevToolsOpened: () => false, + getZoomFactor: () => 1, + setZoomFactor: vi.fn(), + on: vi.fn(), + off: vi.fn(), + ipc: { on: vi.fn(), off: vi.fn() }, + send: webviewSend, + navigationHistory: { canGoBack: () => false, canGoForward: () => false }, + setWindowOpenHandler: vi.fn(), + debugger: { + isAttached: () => false, + attach: vi.fn(), + sendCommand, + on: vi.fn( + ( + event: string, + listener: (event: unknown, method: string, params: Record) => void, + ) => { + if (event === "message") debuggerMessage = listener; + }, + ), + off: vi.fn(), + }, + capturePage, + } as never); + const recordingFrames: DesktopPreviewRecordingFrame[] = []; + + yield* manager.subscribeRecordingFrames((frame) => + Effect.sync(() => { + recordingFrames.push(frame); + }), + ); + yield* manager.createTab("tab_screencast_guard"); + yield* manager.registerWebview("tab_screencast_guard", 42); + yield* manager.automationEvaluate("tab_screencast_guard", { expression: "null" }); + + debuggerMessage?.({}, "Page.screencastFrame", { + sessionId: 1, + data: "inactive-frame", + metadata: { deviceWidth: 1280, deviceHeight: 720 }, + }); + yield* Effect.yieldNow; + expect(recordingFrames).toHaveLength(0); + + yield* manager.startRecording("tab_screencast_guard"); + recordingFrames.length = 0; + debuggerMessage?.({}, "Page.screencastFrame", { + sessionId: 2, + data: "active-frame", + metadata: { deviceWidth: 1280, deviceHeight: 720 }, + }); + yield* Effect.yieldNow; + + expect(recordingFrames).toEqual([ + expect.objectContaining({ + tabId: "tab_screencast_guard", + data: "active-frame", + width: 1280, + height: 720, + }), + ]); + yield* manager.stopRecording("tab_screencast_guard"); + }), + ), + ); + effectIt.effect("shares background frame capture between recording and picture-in-picture", () => withManager((manager) => Effect.gen(function* () { ``` ## apps/desktop/src/preview/Manager.ts Status: modified; +55; −6 ```diff @@ -671,10 +671,17 @@ ), ); + const tabIdForWebContents = Effect.fnUntraced(function* (webContentsId: number) { + const tabs = yield* SynchronizedRef.get(tabsRef); + return ( + Array.from(tabs.entries()).find(([, tab]) => tab.webContentsId === webContentsId)?.[0] ?? null + ); + }); + const pushBounded = (buffer: ReadonlyArray, entry: A): ReadonlyArray => [...buffer, entry].slice(-DIAGNOSTIC_BUFFER_LIMIT); - const captureDiagnosticMessage = Effect.fn("PreviewManager.captureDiagnosticMessage")(function* ( + const captureDiagnosticMessage = Effect.fnUntraced(function* ( webContentsId: number, method: string, params: Record, @@ -852,11 +859,53 @@ const createControlSession = Effect.fn("PreviewManager.createControlSession")(function* () { const semaphore = yield* Semaphore.make(1); const scope = yield* Scope.fork(parentScope, "sequential"); + const handleDebuggerMessage = Effect.fnUntraced(function* ( + method: string, + params: Record, + ) { + if (method === "Page.screencastFrame") { + const sessionId = params["sessionId"]; + if (typeof sessionId === "number") { + yield* attemptPromise( + { + operation: "ackScreencastFrame", + webContentsId: wc.id, + }, + () => wc.debugger.sendCommand("Page.screencastFrameAck", { sessionId }), + ).pipe(Effect.ignore); + } + const tabId = yield* tabIdForWebContents(wc.id); + const metadata = + typeof params["metadata"] === "object" && params["metadata"] !== null + ? (params["metadata"] as Record) + : {}; + if (tabId && typeof params["data"] === "string") { + const captureSession = (yield* SynchronizedRef.get(frameCaptureSessionsRef)).get( + tabId, + ); + if (captureSession?.consumers.has("recording")) { + const receivedAt = yield* currentIso; + const listeners = yield* Ref.get(recordingFrameListenersRef); + const frame: DesktopPreviewRecordingFrame = { + tabId, + data: params["data"], - const handleDebuggerMessage = Effect.fn("PreviewManager.handleDebuggerMessage")( - function* (method: string, params: Record) { + width: + typeof metadata["deviceWidth"] === "number" ? metadata["deviceWidth"] : 0, - yield* captureDiagnosticMessage(wc.id, method, params); - }, - ); + height: + typeof metadata["deviceHeight"] === "number" ? metadata["deviceHeight"] : 0, + receivedAt, + }; + yield* Effect.forEach( + listeners, + (listener) => + deliverEvent("recording-frame", frame.tabId, () => listener(frame)), + { discard: true }, + ); + } + } + } + yield* captureDiagnosticMessage(wc.id, method, params); + }); const onMessage: BrowserControlSession["onMessage"] = (_event, method, params) => { runFork(handleDebuggerMessage(method, params)); }; ``` ## apps/desktop/src/telemetry/DesktopTelemetryPublisher.test.ts Status: added; +388; −0 ```diff @@ -1,0 +1,388 @@ +import { DesktopHostTelemetryMessage } from "@t3tools/contracts"; +import { assert, describe, it } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; +import * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; + +import type * as Electron from "electron"; + +import * as ElectronApp from "../electron/ElectronApp.ts"; +import * as ElectronPowerMonitor from "../electron/ElectronPowerMonitor.ts"; +import * as DesktopTelemetryPublisher from "./DesktopTelemetryPublisher.ts"; + +function makeElectronAppLayer( + metrics: ReadonlyArray, + onMetricsRead: () => void = () => undefined, +) { + return Layer.succeed(ElectronApp.ElectronApp, { + metadata: Effect.die("unexpected metadata read"), + name: Effect.succeed("T3 Code"), + whenReady: Effect.void, + quit: Effect.void, + exit: () => Effect.void, + relaunch: () => Effect.void, + setPath: () => Effect.void, + setName: () => Effect.void, + setAboutPanelOptions: () => Effect.void, + setAppUserModelId: () => Effect.void, + requestSingleInstanceLock: Effect.succeed(true), + getAppMetrics: Effect.sync(() => { + onMetricsRead(); + return metrics; + }), + isDefaultProtocolClient: () => Effect.succeed(false), + setAsDefaultProtocolClient: () => Effect.succeed(true), + setDesktopName: () => Effect.void, + setDockIcon: () => Effect.void, + appendCommandLineSwitch: () => Effect.void, + onBeforeQuitForUpdate: () => Effect.void, + on: () => Effect.void, + } satisfies ElectronApp.ElectronApp["Service"]); +} + +describe("DesktopTelemetryPublisher", () => { + it.effect("stops when its scope closes during an Electron telemetry sample", () => + Effect.gen(function* () { + const pollStarted = yield* Deferred.make(); + const blockPoll = yield* Deferred.make(); + const powerLayer = Layer.succeed( + ElectronPowerMonitor.ElectronPowerMonitor, + ElectronPowerMonitor.ElectronPowerMonitor.of({ + isOnBatteryPower: Effect.succeed(false), + getSystemIdleTime: Deferred.succeed(pollStarted, undefined).pipe( + Effect.andThen(Deferred.await(blockPoll)), + Effect.as(0), + ), + getSystemIdleState: () => Effect.succeed("active"), + getCurrentThermalState: Effect.succeed("nominal"), + onSimpleEvent: () => Effect.void, + onThermalStateChange: () => Effect.void, + onSpeedLimitChange: () => Effect.void, + }), + ); + const layer = DesktopTelemetryPublisher.layer.pipe( + Layer.provide(Layer.mergeAll(makeElectronAppLayer([]), powerLayer)), + ); + const scope = yield* Scope.make(); + + yield* Layer.buildWithScope(layer, scope); + yield* Deferred.await(pollStarted); + + const closeFiber = yield* Scope.close(scope, Exit.void).pipe(Effect.forkDetach); + yield* Effect.yieldNow; + yield* Effect.yieldNow; + + assert.isDefined(closeFiber.pollUnsafe()); + }), + ); + + it.effect("publishes Electron metrics and event-driven power state over NDJSON", () => + Effect.gen(function* () { + const onBattery = yield* Ref.make(false); + const systemIdleState = yield* Ref.make("active"); + let beforeSystemIdleState: Effect.Effect = Effect.void; + let metricsReadCount = 0; + const simpleListeners = new Map void>(); + let thermalListener: ((state: ElectronPowerMonitor.ElectronThermalState) => void) | null = + null; + let speedLimitListener: ((limit: number) => void) | null = null; + const metrics = [ + { + pid: 4_242, + type: "Browser", + creationTime: 1_000.75, + name: "electron", + cpu: { + percentCPUUsage: 12.5, + cumulativeCPUUsage: 3.25, + idleWakeupsPerSecond: 7, + }, + memory: { + workingSetSize: 2_048, + peakWorkingSetSize: 4_096, + }, + } as Electron.ProcessMetric, + ]; + const powerLayer = Layer.succeed( + ElectronPowerMonitor.ElectronPowerMonitor, + ElectronPowerMonitor.ElectronPowerMonitor.of({ + isOnBatteryPower: Ref.get(onBattery), + getSystemIdleTime: Effect.succeed(5), + getSystemIdleState: () => + beforeSystemIdleState.pipe(Effect.andThen(Ref.get(systemIdleState))), + getCurrentThermalState: Effect.succeed("nominal"), + onSimpleEvent: (eventName, listener) => + Effect.sync(() => { + simpleListeners.set(eventName, listener); + }), + onThermalStateChange: (listener) => + Effect.sync(() => { + thermalListener = listener; + }), + onSpeedLimitChange: (listener) => + Effect.sync(() => { + speedLimitListener = listener; + }), + }), + ); + const layer = DesktopTelemetryPublisher.layer.pipe( + Layer.provide( + Layer.mergeAll( + makeElectronAppLayer(metrics, () => { + metricsReadCount += 1; + }), + powerLayer, + ), + ), + ); + + yield* Effect.gen(function* () { + const publisher = yield* DesktopTelemetryPublisher.DesktopTelemetryPublisher; + const encoded = yield* publisher.encoded.pipe(Stream.take(2), Stream.runCollect); + const decoder = new TextDecoder(); + const decodeMessage = Schema.decodeUnknownEffect( + Schema.fromJsonString(DesktopHostTelemetryMessage), + ); + const messages = yield* Effect.forEach(encoded, (bytes) => + decodeMessage(decoder.decode(bytes).trim()), + ); + + assert.equal(messages[0]?.type, "desktopTelemetryHello"); + assert.equal(messages[0]?.electronPid, process.pid); + const initialSnapshot = messages[1]; + if (initialSnapshot?.type !== "desktopTelemetry") { + return assert.fail("Expected the second telemetry message to be a snapshot."); + } + assert.deepEqual(initialSnapshot.electronProcesses, []); + assert.equal(initialSnapshot.electronPid, process.pid); + assert.equal(metricsReadCount, 0); + + const nextSnapshotFiber = yield* Stream.runHead(publisher.changes).pipe(Effect.forkChild); + yield* Effect.yieldNow; + yield* publisher.handleControl({ + version: 1, + type: "setDiagnosticsDemand", + enabled: true, + }); + const demandedSnapshot = Option.getOrThrow(yield* Fiber.join(nextSnapshotFiber)); + assert.equal(demandedSnapshot.electronProcesses[0]?.pid, 4_242); + assert.equal(demandedSnapshot.electronProcesses[0]?.creationTimeMs, 1_001); + assert.equal(demandedSnapshot.electronProcesses[0]?.cpuPercent, 12.5); + assert.equal(demandedSnapshot.electronProcesses[0]?.workingSetBytes, 2_048 * 1_024); + assert.equal(metricsReadCount, 1); + yield* publisher.handleControlForSource("secondary-backend", { + version: 1, + type: "setDiagnosticsDemand", + enabled: true, + }); + yield* publisher.handleControl({ + version: 1, + type: "setDiagnosticsDemand", + enabled: false, + }); + yield* publisher.handleControlForSource("old-backend", { + version: 1, + type: "setDiagnosticsDemand", + enabled: true, + }); + yield* publisher.handleControlForSource("secondary-backend", { + version: 1, + type: "setDiagnosticsDemand", + enabled: false, + }); + yield* Effect.all( + [ + publisher.removeControlSource("old-backend"), + publisher.handleControlForSource("replacement-backend", { + version: 1, + type: "setDiagnosticsDemand", + enabled: true, + }), + ], + { concurrency: "unbounded", discard: true }, + ); + + const batterySnapshotFiber = yield* Stream.runHead(publisher.changes).pipe( + Effect.forkChild, + ); + yield* Effect.yieldNow; + simpleListeners.get("on-battery")?.(); + const batterySnapshot = Option.getOrThrow(yield* Fiber.join(batterySnapshotFiber)); + assert.equal(batterySnapshot.power.onBattery, "true"); + yield* Ref.set(onBattery, true); + + const metricsAfterBatteryEvent = metricsReadCount; + yield* TestClock.adjust(Duration.millis(4_999)); + assert.equal(metricsReadCount, metricsAfterBatteryEvent); + yield* TestClock.adjust(Duration.millis(1)); + assert.equal(metricsReadCount, metricsAfterBatteryEvent + 1); + assert.equal((yield* publisher.latest).pipe(Option.getOrThrow).power.onBattery, "true"); + + yield* Ref.set(onBattery, false); + const metricsBeforePolledAc = metricsReadCount; + yield* TestClock.adjust(Duration.seconds(5)); + assert.equal(metricsReadCount, metricsBeforePolledAc + 1); + assert.equal((yield* publisher.latest).pipe(Option.getOrThrow).power.onBattery, "false"); + yield* TestClock.adjust(Duration.seconds(1)); + assert.equal(metricsReadCount, metricsBeforePolledAc + 2); + + const suspendedSnapshotFiber = yield* Stream.runHead(publisher.changes).pipe( + Effect.forkChild, + ); + yield* Effect.yieldNow; + simpleListeners.get("suspend")?.(); + const suspendedSnapshot = Option.getOrThrow(yield* Fiber.join(suspendedSnapshotFiber)); + assert.isTrue(suspendedSnapshot.power.suspended); + + const constrainedSnapshotFiber = yield* Stream.runHead(publisher.changes).pipe( + Effect.forkChild, + ); + yield* Effect.yieldNow; + thermalListener?.("serious"); + const constrainedSnapshot = Option.getOrThrow(yield* Fiber.join(constrainedSnapshotFiber)); + assert.equal(constrainedSnapshot.power.thermalState, "serious"); + assert.isTrue(constrainedSnapshot.power.suspended); + + const metricsAfterThermalEvent = metricsReadCount; + const recoveredSnapshotFiber = yield* Stream.runHead(publisher.changes).pipe( + Effect.forkChild, + ); + yield* TestClock.adjust(Duration.millis(14_999)); + assert.equal(metricsReadCount, metricsAfterThermalEvent); + yield* TestClock.adjust(Duration.millis(1)); + assert.equal(metricsReadCount, metricsAfterThermalEvent + 1); + const recoveredSnapshot = Option.getOrThrow(yield* Fiber.join(recoveredSnapshotFiber)); + assert.isFalse(recoveredSnapshot.power.suspended); + + const speedLimitSnapshotFiber = yield* Stream.runHead(publisher.changes).pipe( + Effect.forkChild, + ); + yield* Effect.yieldNow; + speedLimitListener?.(65); + const speedLimitSnapshot = Option.getOrThrow(yield* Fiber.join(speedLimitSnapshotFiber)); + assert.equal(Option.getOrNull(speedLimitSnapshot.speedLimitPercent), 65); + + const encodedSpeedLimit = yield* publisher.encoded.pipe(Stream.take(2), Stream.runCollect); + const decodedSpeedLimit = yield* decodeMessage(decoder.decode(encodedSpeedLimit[1]).trim()); + if (decodedSpeedLimit.type !== "desktopTelemetry") { + return assert.fail("Expected the encoded telemetry message to be a snapshot."); + } + assert.equal(Option.getOrNull(decodedSpeedLimit.speedLimitPercent), 65); + assert.equal(decodedSpeedLimit.electronProcesses[0]?.pid, 4_242); + assert.equal(decodedSpeedLimit.electronProcesses[0]?.creationTimeMs, 1_001); + + const metricsBeforeSecondaryOnlySample = metricsReadCount; + yield* TestClock.adjust(Duration.seconds(15)); + assert.equal(metricsReadCount, metricsBeforeSecondaryOnlySample + 1); + assert.equal( + (yield* publisher.latest).pipe(Option.getOrThrow).electronProcesses[0]?.pid, + 4_242, + ); + + const stoppedSnapshotFiber = yield* Stream.runHead(publisher.changes).pipe( + Effect.forkChild, + ); + yield* Effect.yieldNow; + yield* publisher.handleControlForSource("replacement-backend", { + version: 1, + type: "setDiagnosticsDemand", + enabled: false, + }); + const stoppedSnapshot = Option.getOrThrow(yield* Fiber.join(stoppedSnapshotFiber)); + assert.deepEqual(stoppedSnapshot.electronProcesses, []); + const metricsAfterStopping = metricsReadCount; + const configuredSnapshotFiber = yield* Stream.runHead(publisher.changes).pipe( + Effect.forkChild, + ); + yield* Effect.yieldNow; + yield* publisher.handleControl({ + version: 1, + type: "setHostPowerIntervals", + activeIntervalMs: 7_000, + idleIntervalMs: 11_000, + }); + const configuredSequence = Option.getOrThrow( + yield* Fiber.join(configuredSnapshotFiber), + ).sequence; + + yield* TestClock.adjust(Duration.millis(6_999)); + assert.equal( + (yield* publisher.latest).pipe(Option.getOrThrow).sequence, + configuredSequence, + ); + assert.equal(metricsReadCount, metricsAfterStopping); + yield* TestClock.adjust(Duration.millis(1)); + assert.equal( + (yield* publisher.latest).pipe(Option.getOrThrow).sequence, + configuredSequence + 1, + ); + assert.equal(metricsReadCount, metricsAfterStopping); + + yield* Ref.set(systemIdleState, "locked"); + yield* TestClock.adjust(Duration.seconds(7)); + const lockedSequence = (yield* publisher.latest).pipe(Option.getOrThrow).sequence; + assert.equal((yield* publisher.latest).pipe(Option.getOrThrow).power.locked, "true"); + yield* TestClock.adjust(Duration.millis(10_999)); + assert.equal((yield* publisher.latest).pipe(Option.getOrThrow).sequence, lockedSequence); + yield* TestClock.adjust(Duration.millis(1)); + assert.equal( + (yield* publisher.latest).pipe(Option.getOrThrow).sequence, + lockedSequence + 1, + ); + + yield* Ref.set(systemIdleState, "active"); + yield* TestClock.adjust(Duration.seconds(11)); + const unlockedSnapshot = (yield* publisher.latest).pipe(Option.getOrThrow); + assert.equal(unlockedSnapshot.power.locked, "false"); + assert.equal(unlockedSnapshot.power.idle, "false"); + + yield* TestClock.adjust(Duration.millis(6_999)); + assert.equal( + (yield* publisher.latest).pipe(Option.getOrThrow).sequence, + unlockedSnapshot.sequence, + ); + yield* TestClock.adjust(Duration.millis(1)); + assert.equal( + (yield* publisher.latest).pipe(Option.getOrThrow).sequence, + unlockedSnapshot.sequence + 1, + ); + + const pollStarted = yield* Deferred.make(); + const releasePoll = yield* Deferred.make(); + beforeSystemIdleState = Deferred.succeed(pollStarted, undefined).pipe( + Effect.andThen(Deferred.await(releasePoll)), + ); + const concurrentEventSnapshotFiber = yield* Stream.runHead(publisher.changes).pipe( + Effect.forkChild, + ); + yield* Effect.yieldNow; + yield* publisher.handleControl({ + version: 1, + type: "setDiagnosticsDemand", + enabled: true, + }); + yield* Deferred.await(pollStarted); + simpleListeners.get("lock-screen")?.(); + thermalListener?.("critical"); + yield* Effect.yieldNow; + yield* Effect.yieldNow; + yield* Deferred.succeed(releasePoll, undefined); + + const concurrentEventSnapshot = Option.getOrThrow( + yield* Fiber.join(concurrentEventSnapshotFiber), + ); + assert.equal(concurrentEventSnapshot.power.locked, "true"); + assert.equal(concurrentEventSnapshot.power.thermalState, "critical"); + }).pipe(Effect.provide(layer)); + }), + ); +}); ``` ## apps/desktop/src/telemetry/DesktopTelemetryPublisher.ts Status: added; +379; −0 ```diff @@ -1,0 +1,379 @@ +import { + DesktopHostTelemetryMessage, + type DesktopHostTelemetrySnapshot, + type DesktopTelemetryControlMessage, + type HostPowerSnapshot, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as PubSub from "effect/PubSub"; +import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; + +import * as ElectronApp from "../electron/ElectronApp.ts"; +import * as ElectronPowerMonitor from "../electron/ElectronPowerMonitor.ts"; + +const LIVE_SAMPLE_INTERVAL = Duration.seconds(1); +const BATTERY_SAMPLE_INTERVAL = Duration.seconds(5); +const CONSTRAINED_SAMPLE_INTERVAL = Duration.seconds(15); +const DEFAULT_HOST_POWER_ACTIVE_INTERVAL = Duration.seconds(30); +const DEFAULT_HOST_POWER_IDLE_INTERVAL = Duration.minutes(2); +const IDLE_THRESHOLD_SECONDS = 60; +const encodeMessage = Schema.encodeSync(Schema.fromJsonString(DesktopHostTelemetryMessage)); +const textEncoder = new TextEncoder(); + +type PowerEvent = + | { readonly type: "locked"; readonly value: boolean } + | { readonly type: "suspended"; readonly value: boolean } + | { readonly type: "onBattery"; readonly value: boolean } + | { readonly type: "thermal"; readonly value: HostPowerSnapshot["thermalState"] } + | { readonly type: "speedLimit"; readonly value: number }; + +interface PowerState { + readonly idle: HostPowerSnapshot["idle"]; + readonly locked: HostPowerSnapshot["locked"]; + readonly lockedEventPending: boolean; + readonly suspended: boolean; + readonly suspendedEventPending: boolean; + readonly onBattery: HostPowerSnapshot["onBattery"]; + readonly onBatteryEventPending: boolean; + readonly thermalState: HostPowerSnapshot["thermalState"]; + readonly speedLimitPercent: Option.Option; +} + +interface HostPowerIntervals { + readonly active: Duration.Duration; + readonly idle: Duration.Duration; +} + +export class DesktopTelemetryPublisher extends Context.Service< + DesktopTelemetryPublisher, + { + readonly latest: Effect.Effect>; + readonly changes: Stream.Stream; + readonly encoded: Stream.Stream; + readonly handleControl: (message: DesktopTelemetryControlMessage) => Effect.Effect; + readonly handleControlForSource: ( + sourceId: string, + message: DesktopTelemetryControlMessage, + ) => Effect.Effect; + readonly removeControlSource: (sourceId: string) => Effect.Effect; + } +>()("@t3tools/desktop/telemetry/DesktopTelemetryPublisher") {} + +function booleanState(value: boolean): HostPowerSnapshot["onBattery"] { + return value ? "true" : "false"; +} + +function idleState(value: ElectronPowerMonitor.ElectronIdleState): HostPowerSnapshot["idle"] { + switch (value) { + case "active": + return "false"; + case "idle": + case "locked": + return "true"; + case "unknown": + return "unknown"; + } +} + +function updatePowerState(state: PowerState, event: PowerEvent): PowerState { + switch (event.type) { + case "locked": + return { + ...state, + locked: booleanState(event.value), + lockedEventPending: true, + }; + case "suspended": + return { + ...state, + suspended: event.value, + suspendedEventPending: event.value, + }; + case "onBattery": + return { + ...state, + onBattery: booleanState(event.value), + onBatteryEventPending: true, + }; + case "thermal": + return { ...state, thermalState: event.value }; + case "speedLimit": + return { ...state, speedLimitPercent: Option.some(event.value) }; + } +} + +function sampleInterval( + power: PowerState, + diagnosticsDemand: boolean, + hostPowerIntervals: HostPowerIntervals, +): Duration.Duration { + if (!diagnosticsDemand) { + return power.suspended || power.locked === "true" || power.idle === "true" + ? hostPowerIntervals.idle + : hostPowerIntervals.active; + } + if ( + power.suspended || + power.locked === "true" || + power.thermalState === "serious" || + power.thermalState === "critical" + ) { + return CONSTRAINED_SAMPLE_INTERVAL; + } + if (power.onBattery === "true") return BATTERY_SAMPLE_INTERVAL; + return LIVE_SAMPLE_INTERVAL; +} + +export const make = Effect.fn("desktop.telemetryPublisher.make")(function* () { + const electronApp = yield* ElectronApp.ElectronApp; + const powerMonitor = yield* ElectronPowerMonitor.ElectronPowerMonitor; + yield* electronApp.whenReady; + + const initialPowerState: PowerState = { + idle: "unknown", + locked: "unknown", + lockedEventPending: false, + suspended: false, + suspendedEventPending: false, + onBattery: booleanState(yield* powerMonitor.isOnBatteryPower), + onBatteryEventPending: false, + thermalState: yield* powerMonitor.getCurrentThermalState, + speedLimitPercent: Option.none(), + }; + const powerState = yield* Ref.make(initialPowerState); + const hostPowerIntervals = yield* Ref.make({ + active: DEFAULT_HOST_POWER_ACTIVE_INTERVAL, + idle: DEFAULT_HOST_POWER_IDLE_INTERVAL, + }); + const powerEvents = yield* Queue.unbounded(); + const sampleTriggers = yield* Queue.sliding(1); + const diagnosticsDemandSources = yield* Ref.make>(new Set()); + const latest = yield* Ref.make(Option.none()); + const changes = yield* PubSub.sliding(8); + const sequence = yield* Ref.make(0); + + const offer = (event: PowerEvent): void => { + Queue.offerUnsafe(powerEvents, event); + }; + yield* Effect.all( + [ + powerMonitor.onSimpleEvent("lock-screen", () => offer({ type: "locked", value: true })), + powerMonitor.onSimpleEvent("unlock-screen", () => offer({ type: "locked", value: false })), + powerMonitor.onSimpleEvent("suspend", () => offer({ type: "suspended", value: true })), + powerMonitor.onSimpleEvent("resume", () => offer({ type: "suspended", value: false })), + powerMonitor.onSimpleEvent("on-battery", () => offer({ type: "onBattery", value: true })), + powerMonitor.onSimpleEvent("on-ac", () => offer({ type: "onBattery", value: false })), + powerMonitor.onThermalStateChange((value) => offer({ type: "thermal", value })), + powerMonitor.onSpeedLimitChange((value) => offer({ type: "speedLimit", value })), + ], + { concurrency: "unbounded" }, + ); + yield* Effect.forever( + Queue.take(powerEvents).pipe( + Effect.flatMap((event) => Ref.update(powerState, (state) => updatePowerState(state, event))), + Effect.andThen(Queue.offer(sampleTriggers, undefined)), + ), + ).pipe(Effect.forkScoped); + + const sampleOnce = (allowSuspendRecovery: boolean) => + Effect.gen(function* () { + const sampledAt = yield* DateTime.now; + const sampledAtUnixMs = DateTime.toEpochMillis(sampledAt); + const demand = (yield* Ref.get(diagnosticsDemandSources)).size > 0; + const [currentPower, idleSeconds, systemIdleState, onBattery, metrics] = yield* Effect.all( + [ + Ref.get(powerState), + powerMonitor.getSystemIdleTime, + powerMonitor.getSystemIdleState(IDLE_THRESHOLD_SECONDS), + powerMonitor.isOnBatteryPower, + demand ? electronApp.getAppMetrics : Effect.succeed([]), + ], + { concurrency: "unbounded" }, + ); + const polledLocked = + systemIdleState === "unknown" + ? currentPower.locked + : booleanState(systemIdleState === "locked"); + const polledOnBattery = booleanState(onBattery); + const observedPower = yield* Ref.modify(powerState, (latestPower) => { + const preserveEventLocked = + latestPower.lockedEventPending || + latestPower.locked !== currentPower.locked || + latestPower.lockedEventPending !== currentPower.lockedEventPending; + const preserveEventOnBattery = + latestPower.onBatteryEventPending || + latestPower.onBattery !== currentPower.onBattery || + latestPower.onBatteryEventPending !== currentPower.onBatteryEventPending; + const preserveEventSuspended = + latestPower.suspendedEventPending || + latestPower.suspended !== currentPower.suspended || + latestPower.suspendedEventPending !== currentPower.suspendedEventPending; + const next: PowerState = { + ...latestPower, + idle: idleState(systemIdleState), + locked: preserveEventLocked ? latestPower.locked : polledLocked, + lockedEventPending: + latestPower.lockedEventPending && + (systemIdleState === "unknown" || polledLocked !== latestPower.locked), + suspended: + allowSuspendRecovery && !preserveEventSuspended ? false : latestPower.suspended, + suspendedEventPending: false, + onBattery: preserveEventOnBattery ? latestPower.onBattery : polledOnBattery, + onBatteryEventPending: + latestPower.onBatteryEventPending && polledOnBattery !== latestPower.onBattery, + }; + return [next, next] as const; + }); + const nextSequence = yield* Ref.modify(sequence, (current) => [current + 1, current + 1]); + const snapshot: DesktopHostTelemetrySnapshot = { + version: 1, + type: "desktopTelemetry", + sequence: nextSequence, + sampledAtUnixMs, + electronPid: process.pid, + power: { + source: "electron-main", + idle: observedPower.idle, + idleSeconds, + locked: observedPower.locked, + suspended: observedPower.suspended, + onBattery: observedPower.onBattery, + lowPowerMode: "unknown", + thermalState: observedPower.thermalState, + stale: false, + updatedAt: sampledAt, + }, + speedLimitPercent: observedPower.speedLimitPercent, + electronProcesses: metrics.map((metric) => ({ + pid: metric.pid, + creationTimeMs: Math.max(0, Math.round(metric.creationTime)), + type: metric.type, + ...(metric.name === undefined ? {} : { name: metric.name }), + ...(metric.serviceName === undefined ? {} : { serviceName: metric.serviceName }), + cpuPercent: metric.cpu.percentCPUUsage, + ...(metric.cpu.cumulativeCPUUsage === undefined + ? {} + : { cumulativeCpuSeconds: metric.cpu.cumulativeCPUUsage }), + idleWakeupsPerSecond: metric.cpu.idleWakeupsPerSecond, + workingSetBytes: Math.max(0, Math.round(metric.memory.workingSetSize * 1024)), + peakWorkingSetBytes: Math.max(0, Math.round(metric.memory.peakWorkingSetSize * 1024)), + })), + }; + + yield* Ref.set(latest, Option.some(snapshot)); + yield* PubSub.publish(changes, snapshot); + }).pipe( + Effect.catchCause((cause) => + Cause.hasInterrupts(cause) + ? Effect.failCause(cause) + : Effect.logWarning("Failed to sample Electron telemetry", { + cause: String(cause), + }), + ), + ); + + yield* Effect.gen(function* () { + yield* sampleOnce(false); + while (true) { + const [currentPower, demand, intervals] = yield* Effect.all([ + Ref.get(powerState), + Ref.get(diagnosticsDemandSources).pipe(Effect.map((sources) => sources.size > 0)), + Ref.get(hostPowerIntervals), + ]); + const allowSuspendRecovery = yield* Effect.raceFirst( + Queue.take(sampleTriggers).pipe(Effect.as(false)), + Effect.sleep(sampleInterval(currentPower, demand, intervals)).pipe(Effect.as(true)), + ); + yield* sampleOnce(allowSuspendRecovery); + } + }).pipe(Effect.forkScoped); + + const handleControlForSource: DesktopTelemetryPublisher["Service"]["handleControlForSource"] = ( + sourceId, + message, + ) => { + switch (message.type) { + case "setDiagnosticsDemand": + return Ref.modify(diagnosticsDemandSources, (sources) => { + const previous = sources.size > 0; + const next = new Set(sources); + if (message.enabled) { + next.add(sourceId); + } else { + next.delete(sourceId); + } + return [[previous, next.size > 0] as const, next] as const; + }).pipe( + Effect.flatMap(([previous, enabled]) => + previous === enabled + ? Effect.void + : Queue.offer(sampleTriggers, undefined).pipe(Effect.asVoid), + ), + ); + case "setHostPowerIntervals": + return Ref.set(hostPowerIntervals, { + active: Duration.millis(message.activeIntervalMs), + idle: Duration.millis(message.idleIntervalMs), + }).pipe(Effect.andThen(Queue.offer(sampleTriggers, undefined)), Effect.asVoid); + } + }; + const removeControlSource: DesktopTelemetryPublisher["Service"]["removeControlSource"] = ( + sourceId, + ) => + Ref.modify(diagnosticsDemandSources, (sources) => { + const previous = sources.size > 0; + const next = new Set(sources); + next.delete(sourceId); + return [[previous, next.size > 0] as const, next] as const; + }).pipe( + Effect.flatMap(([previous, enabled]) => + previous === enabled + ? Effect.void + : Queue.offer(sampleTriggers, undefined).pipe(Effect.asVoid), + ), + ); + const handleControl: DesktopTelemetryPublisher["Service"]["handleControl"] = (message) => + handleControlForSource("legacy", message); + + const snapshots = Stream.unwrap( + Effect.gen(function* () { + const subscription = yield* PubSub.subscribe(changes); + const initial = yield* Ref.get(latest); + return Stream.concat( + Option.match(initial, { + onNone: () => Stream.empty, + onSome: Stream.make, + }), + Stream.fromSubscription(subscription), + ); + }), + ); + const encoded = Stream.concat( + Stream.make({ + version: 1, + type: "desktopTelemetryHello", + electronPid: process.pid, + } as const), + snapshots, + ).pipe(Stream.map((message) => textEncoder.encode(`${encodeMessage(message)}\n`))); + + return DesktopTelemetryPublisher.of({ + latest: Ref.get(latest), + changes: Stream.fromPubSub(changes), + encoded, + handleControl, + handleControlForSource, + removeControlSource, + }); +}); + +export const layer = Layer.effect(DesktopTelemetryPublisher, make()); ``` [Next page](https://git.cafe/pingdotgg/t3code/pull/2679/diffs?after=ImFwcHMvZGVza3RvcC9zcmMvdGVsZW1ldHJ5L0Rlc2t0b3BUZWxlbWV0cnlQdWJsaXNoZXIudHMi&expectedVersion=95&format=markdown)