/* * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information * regarding copyright ownership. The ASF licenses this file * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, * software distributed under the License is distributed on an * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * KIND, either express or implied. See the License for the * specific language governing permissions and limitations * under the License. */ import { spawn } from 'node:child_process'; import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { dirname, join } from 'node:path'; import { fileURLToPath } from 'node:url'; import { requireComputerUseLabRoot } from './lab-root.mjs'; const here = dirname(fileURLToPath(import.meta.url)); const repoRoot = join(here, '../..'); const harnessPath = join(here, 'process-restart-harness.mjs'); const monitorPath = join(here, 'real-e2e-monitor.swift'); const labRoot = requireComputerUseLabRoot(); const statePath = join(labRoot, 'test-app', 'runtime', 'state.json'); const SOAK_ROUNDS = 5; const FIXTURE_BUNDLE_ID = 'com.openai.codex.cualab'; const delay = (milliseconds) => new Promise((resolve) => setTimeout(resolve, milliseconds)); function runChild(file, args, options = {}) { return new Promise((resolve, reject) => { const child = spawn(file, args, { cwd: options.cwd, env: options.env, stdio: options.stdio ?? ['ignore', 'inherit', 'inherit'], }); child.once('error', reject); child.once('exit', (code, signal) => { if (code === 0) resolve(); else reject(new Error(`${file} failed (${signal ?? code})`)); }); }); } async function runFixtureScript(name, options = {}) { await runChild(join(labRoot, 'test-app', name), [], options); } async function runBuilds() { for (const workspace of ['@maka/core', '@maka/runtime', '@maka/computer-use']) { await runChild('npm', ['--workspace', workspace, 'run', 'build'], { cwd: repoRoot, }); } await runChild(process.execPath, ['scripts/computer-use.mjs', 'prepare'], { cwd: repoRoot }); } async function frontmostApplication() { const script = [ 'tell application "System Events"', 'set frontProcess to first application process whose frontmost is true', 'return (unix id of frontProcess as text) & tab & (bundle identifier of frontProcess as text)', 'end tell', ].join('\n'); let output = ''; await new Promise((resolve, reject) => { const child = spawn('/usr/bin/osascript', ['-e', script], { stdio: ['ignore', 'pipe', 'inherit'], }); child.stdout.setEncoding('utf8'); child.stdout.on('data', (chunk) => { output += chunk; }); child.once('error', reject); child.once('exit', (code) => { if (code === 0) resolve(); else reject(new Error(`failed to capture frontmost application (${code})`)); }); }); const [pid, bundleIdentifier] = output.trim().split('\t'); if (!Number.isInteger(Number(pid)) || !bundleIdentifier) { throw new Error(`invalid frontmost application identity: ${output.trim()}`); } return { pid: Number(pid), bundleIdentifier }; } async function restoreFrontmost(application) { const escaped = application.bundleIdentifier.replaceAll('\\', '\\\\').replaceAll('"', '\\"'); const script = [ 'tell application "System Events"', `if exists (first application process whose unix id is ${application.pid}) then`, `set frontmost of first application process whose unix id is ${application.pid} to true`, 'else', `tell application id "${escaped}" to activate`, 'end if', 'end tell', ].join('\n'); await runChild('/usr/bin/osascript', ['-e', script], { stdio: ['ignore', 'ignore', 'ignore'], }); } async function pointerLocation() { let output = ''; await new Promise((resolve, reject) => { const child = spawn('swift', [monitorPath, '--snapshot'], { stdio: ['ignore', 'pipe', 'inherit'], }); child.stdout.setEncoding('utf8'); child.stdout.on('data', (chunk) => { output += chunk; }); child.once('error', reject); child.once('exit', (code) => { if (code === 0) resolve(); else reject(new Error(`failed to capture desktop snapshot (${code})`)); }); }); const [kind, , , xText, yText] = output.trim().split('\t'); const x = Number(xText); const y = Number(yText); if (kind !== 'READY' || !Number.isFinite(x) || !Number.isFinite(y)) { throw new Error(`invalid desktop snapshot: ${output.trim()}`); } return { x, y }; } async function terminateChild(child, label, timeoutMs = 3_000) { if (!child || child.exitCode !== null || child.signalCode !== null) return; const exited = new Promise((resolve) => child.once('exit', resolve)); child.kill('SIGTERM'); if (await Promise.race([exited.then(() => true), delay(timeoutMs).then(() => false)])) { return; } child.kill('SIGKILL'); if (!(await Promise.race([exited.then(() => true), delay(timeoutMs).then(() => false)]))) { throw new Error(`${label} did not exit after SIGKILL`); } } async function waitForJson(path, label, timeoutMs = 15_000) { const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { try { return JSON.parse(await readFile(path, 'utf8')); } catch (error) { if (error?.code !== 'ENOENT' && !(error instanceof SyntaxError)) throw error; } await delay(50); } throw new Error(`${label} timeout`); } async function waitForRestartedState(oldPID, oldWebContentPID, timeoutMs = 15_000) { const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { try { const state = JSON.parse(await readFile(statePath, 'utf8')); const newPID = state?.oop?.hostPID; const newWebContentPID = state?.oop?.webContentPID; if ( Number.isInteger(newPID) && newPID > 0 && newPID !== oldPID && Number.isInteger(newWebContentPID) && newWebContentPID > 0 && newWebContentPID !== oldWebContentPID ) { try { process.kill(newPID, 0); return state; } catch { // The state raced process exit; wait for the next launched instance. } } } catch (error) { if (error?.code !== 'ENOENT' && !(error instanceof SyntaxError)) throw error; } await delay(50); } throw new Error( `fixture restart state did not advance from host/WebContent ` + `${oldPID}/${oldWebContentPID}`, ); } async function waitForInitialState(timeoutMs = 15_000) { const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { try { const state = JSON.parse(await readFile(statePath, 'utf8')); const hostPID = state?.oop?.hostPID; const webContentPID = state?.oop?.webContentPID; if ( Number.isInteger(hostPID) && hostPID > 0 && Number.isInteger(webContentPID) && webContentPID > 0 ) { process.kill(hostPID, 0); return state; } } catch (error) { if (error?.code !== 'ENOENT' && error?.code !== 'ESRCH' && !(error instanceof SyntaxError)) throw error; } await delay(50); } throw new Error('initial fixture state did not publish live host/WebContent PIDs'); } function startFocusMonitor() { const child = spawn( 'swift', [monitorPath, '--concurrent-user', '0', '--deny-frontmost-bundle', FIXTURE_BUNDLE_ID], { stdio: ['ignore', 'pipe', 'pipe'], }, ); let buffer = ''; let stderr = ''; let readyResolve; let readyReject; let failureResolve; let readySettled = false; let failureSettled = false; const ready = new Promise((resolve, reject) => { readyResolve = resolve; readyReject = reject; }); const failure = new Promise((resolve) => { failureResolve = resolve; }); const fail = (error) => { const normalized = error instanceof Error ? error : new Error(String(error)); if (!readySettled) { readySettled = true; readyReject(normalized); } if (!failureSettled) { failureSettled = true; failureResolve(normalized); } }; child.stdout.setEncoding('utf8'); child.stdout.on('data', (chunk) => { buffer += chunk; const lines = buffer.split('\n'); buffer = lines.pop() ?? ''; for (const line of lines.map((value) => value.trim()).filter(Boolean)) { const [kind, ...fields] = line.split('\t'); if (kind === 'READY') { readySettled = true; readyResolve({ frontmostPID: Number(fields[1]), pointer: { x: Number(fields[2]), y: Number(fields[3]) }, physicalInputAgeSeconds: Number(fields[4]), bundleIdentifier: fields[5], }); } else if (kind === 'CHANGE' || kind === 'ERROR') { fail(new Error(fields.join('\t') || line)); } } }); child.stderr.setEncoding('utf8'); child.stderr.on('data', (chunk) => { stderr += chunk; }); child.on('error', fail); child.on('exit', (code, signal) => { if (!failureSettled && code !== 0 && signal !== 'SIGTERM') { fail( new Error( `focus monitor exited (${signal ?? code})` + `${stderr.trim() ? `: ${stderr.trim()}` : ''}`, ), ); } }); return { child, ready, failure, stop: () => terminateChild(child, 'restart focus monitor'), }; } async function run() { const originalFrontmost = await frontmostApplication(); const temporaryDirectory = await mkdtemp(join(tmpdir(), 'maka-cu-restart-e2e-')); let fixtureTouched = false; let caffeinate; let harness; let monitor; try { caffeinate = spawn('/usr/bin/caffeinate', ['-dimsu'], { stdio: ['ignore', 'ignore', 'inherit'], }); await runBuilds(); fixtureTouched = true; await runFixtureScript('stop.sh'); await runFixtureScript('reset.sh'); const pointerBefore = await pointerLocation(); monitor = startFocusMonitor(); await Promise.race([ monitor.ready, delay(10_000).then(() => { throw new Error('restart focus monitor startup timeout'); }), ]); await runFixtureScript('launch.sh', { env: { ...process.env, CUA_LAB_BACKGROUND: '1' }, }); const oldState = await waitForInitialState(); const oldPID = oldState?.oop?.hostPID; if (!Number.isInteger(oldPID) || oldPID <= 0) { throw new Error('old synthetic fixture did not publish a valid PID'); } harness = spawn(process.execPath, [harnessPath], { cwd: repoRoot, env: { ...process.env, MAKA_CU_RESTART_OLD_PID: String(oldPID), MAKA_CU_RESTART_SOAK_ROUNDS: String(SOAK_ROUNDS), MAKA_CU_RESTART_TEMP_DIR: temporaryDirectory, }, stdio: ['ignore', 'inherit', 'inherit'], }); const exit = new Promise((resolve, reject) => { harness.once('error', reject); harness.once('exit', (code, signal) => resolve({ code, signal })); }); let currentPID = oldPID; for (let round = 1; round <= SOAK_ROUNDS; round += 1) { const requestPath = join(temporaryDirectory, `restart-request-${round}.json`); const completePath = join(temporaryDirectory, `restart-complete-${round}.json`); const request = await Promise.race([ waitForJson(requestPath, `restart request ${round}`), exit.then((result) => { throw new Error( `restart harness exited before round ${round} ` + `(${result.signal ?? result.code})`, ); }), monitor.failure.then((error) => { throw error; }), ]); if (request.round !== round || request.oldPID !== currentPID) { throw new Error(`restart request ${round} identity mismatch`); } if (!Number.isInteger(request.oldWebContentPID) || request.oldWebContentPID <= 0) { throw new Error(`restart request ${round} has no old WebContent PID`); } await runFixtureScript('stop.sh'); await runFixtureScript('launch.sh', { env: { ...process.env, CUA_LAB_BACKGROUND: '1' }, }); const newState = await waitForRestartedState(currentPID, request.oldWebContentPID); const newPID = newState?.oop?.hostPID; if (!Number.isInteger(newPID) || newPID <= 0 || currentPID === newPID) { throw new Error( `fixture restart ${round} did not create a new process: ` + `${currentPID} -> ${newPID}`, ); } await writeFile( completePath, `${JSON.stringify({ round, oldPID: currentPID, newPID, newWebContentPID: newState.oop.webContentPID, })}\n`, { flag: 'wx', mode: 0o600 }, ); currentPID = newPID; } const result = await Promise.race([ exit, monitor.failure.then((error) => { throw error; }), ]); if (result.code !== 0) { throw new Error(`restart soak E2E failed (${result.signal ?? result.code})`); } const pointerAfter = await pointerLocation(); const displacement = Math.hypot( pointerAfter.x - pointerBefore.x, pointerAfter.y - pointerBefore.y, ); process.stdout.write(`Synthetic fixture never became frontmost across ${SOAK_ROUNDS} rounds\n`); process.stdout.write(`Concurrent user pointer displacement observed: ${displacement}\n`); } finally { await terminateChild(harness, 'restart E2E harness').catch(() => {}); await monitor?.stop().catch(() => {}); if (fixtureTouched) await runFixtureScript('stop.sh').catch(() => {}); const finalFrontmost = await frontmostApplication().catch(() => undefined); if (finalFrontmost?.bundleIdentifier === FIXTURE_BUNDLE_ID) { await restoreFrontmost(originalFrontmost).catch(() => {}); } await terminateChild(caffeinate, 'restart caffeinate').catch(() => {}); await rm(temporaryDirectory, { recursive: true, force: true }); } } run().catch((error) => { console.error('Computer Use process restart E2E failed:', error); process.exitCode = 1; });