diff --git a/.env.example b/.env.example index 76040e1..e9c808b 100644 --- a/.env.example +++ b/.env.example @@ -14,6 +14,11 @@ TELEGRAM_ALLOWED_USER_IDS=123456789 #PUBLIC_URL=https://codex.example.com TELEX_TUNNEL=auto +# Release 0.0.35 migrates to Wirebot in Docker on startup. Set off BEFORE the +# first migration to keep running native Telex (also useful for source development). +# After migration, a scheduled shell check replaces the resident telex service. +TELEX_MIGRATION=auto + # Optional runtime paths and listener settings. TELEX_DATA_DIR=.telex CODEX_WORKSPACE=.telex/workspace diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6de9357..9dad306 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -20,6 +20,8 @@ jobs: cache: npm - run: npm ci - run: npm run check + - name: Verify migration, service handoff, and published Wirebot image + run: sudo env "PATH=$PATH" TMPDIR=/root TELEX_TEST_DOCKER=1 TELEX_TEST_SYSTEMD=1 npm run test:migration - run: npm run build - name: Load E2E entry point run: npm test -- --help diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 5541480..a5657fe 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -23,6 +23,8 @@ jobs: test "$GITHUB_REF_NAME" = "v$package_version" - run: npm ci - run: npm run check + - name: Verify migration and published Wirebot image + run: sudo env "PATH=$PATH" TMPDIR=/root TELEX_TEST_DOCKER=1 TELEX_TEST_SYSTEMD=1 npm run test:migration - run: npm run build - name: Load E2E entry point run: npm test -- --help diff --git a/CHANGELOG.md b/CHANGELOG.md index 1496133..b3b7670 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,18 @@ All notable changes to Telex are documented in this file. ## Unreleased +## [0.0.35] - 2026-09-06 + +### Changed + +- Migrate 0.0.34 instances to Wirebot's Docker image on startup, preserving the original data + and copying authentication, conversations, memory, schedules, settings, and workspace files. +- Retire the telex service after migration and use a scheduled shell check for Docker image + updates, with no resident updater process. Run update transactions only when needed, retaining + startup checks, data snapshots, rollback, and recovery after interrupted updates. +- Notify Telegram users after the migrated instance is ready; retain native Telex if initial + preparation or startup fails. Set `TELEX_MIGRATION=off` before migration to defer it. + ## [0.0.34] - 2026-08-02 ### Fixed diff --git a/README.md b/README.md index a461efc..99f9c6b 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,105 @@ # Telex +Development continues in [Wirebot](https://github.com/sadfun/wirebot). **Telex 0.0.35 is the +migration release for existing 0.0.34 instances.** New installations should use Wirebot directly. + +## Automatic migration to Wirebot + +The normal Telex release updater installs this release and restarts the existing service. Before +starting a bot, it prepares Docker, pulls `ghcr.io/sadfun/wirebot:latest`, and copies the configured +data and workspace into `~/.local/share/telex-wirebot//` (or the corresponding +`XDG_DATA_HOME` directory). The original Telex files and configuration remain intact. + +The copy includes Codex authentication, sessions and SQLite databases (including WAL files), +memory, skills, configuration, conversation-to-thread mappings and history, schedules, delivery +records, settings, attachments, and workspace files. The copy is mounted inside the container at +the original absolute paths as well as `/data`, preserving stored rollout paths and thread working +directories. The workspace's `.wirebot` is linked to `.telex`, so existing schedule memories and +attachment paths still resolve. No host Docker socket is exposed to the agent. + +The default `workspace-write` Codex sandbox setting becomes `danger-full-access` **inside the +container**, matching Wirebot's machine model: ordinary Docker cannot run Codex's nested +bubblewrap sandbox. Approval policy and other Codex settings stay intact. The original host +configuration remains unchanged. Wirebot logs at least at `info` level so the updater can verify +its startup event. + +Once Wirebot reports that its connectors and scheduler are ready, and its health endpoint confirms +the existing Codex account, each allowlisted Telegram user receives: + +> Telex just got a big update! Your instance is now [Wirebot](https://github.com/sadfun/wirebot), +> the next evolution of telex. +> +> Your login, conversations, memory, and schedules are preserved. Slack and Discord are now +> supported, and updates are automatic. Just keep chatting. + +Successful deliveries are recorded per user; failed deliveries retry. A crash between Telegram +accepting a message and saving its receipt can cause that message to repeat. + +### Docker and the host service + +- Linux: reuse a local Docker daemon, including rootless Docker. If Docker is missing, install + it using a pinned, checksum-verified official Docker installer and enable its system service. + Installation needs root or passwordless sudo; the updater never waits for a password. +- macOS: reuse Docker Desktop or Colima. If necessary, install Docker and Colima through an + existing Homebrew installation and enable Colima's login service. Homebrew or Docker Desktop + must already be available. Linux and macOS on x86-64 and ARM64 are supported. +- Migration installs a **systemd timer and oneshot service** on Linux, or a **launchd job with + `StartInterval`** on macOS, then disables and stops the old telex service. There is no updater + process between checks. A small shell script checks every `TELEX_UPDATE_INTERVAL_HOURS` + (six hours by default), pulls before stopping anything, and exits if the image is unchanged. + Node runs only for the first notification, an image replacement, or recovery; it exits after + that work. Each replacement + retains a stopped-data snapshot and restores both image and data if startup fails. The latest + snapshot is kept at the migrated data directory's `-backup` sibling and rotated on the next + update. Interrupted replacements recover at the next scheduled check or after reboot. +- Docker's `unless-stopped` restart policy keeps Wirebot running after Telex exits. Stopping + the updater disables checks, not the bot: use `docker stop ` to stop Wirebot, + after stopping its updater. Keep the migration release and Node installed for occasional + update transactions. The standard installer services are retired automatically; if using a + custom process manager, remove its old `telex start` job. Source development can use + `TELEX_MIGRATION=off npm run dev` to run native Telex. + +The private instance directory contains `migration.json` (container name, Docker endpoint, data +path and notification credentials), `wirebot.env`, and `update.sh`. Extra environment variables from the launcher's env file are +retained, excluding host-only paths and replaced runtime settings. Wirebot's HTTP port is +published at the original Telex host/port, so an existing `PUBLIC_URL` reverse proxy keeps working. +Slack and Discord are available but require their own connector credentials; migration does not +enable them automatically. + +Allow disk space for the migrated data/workspace and another copy for update rollback, in addition +to the Docker image. External symlink targets, host-installed programs, host keychain credentials, +and services bound to the host's localhost need container-compatible configuration. Remote Docker +daemons and paths that conflict with the image's system directories are rejected. + +### Deferring migration and recovery + +Set `TELEX_MIGRATION=off` in `telex.env` **before migration** to defer it. If preparation or initial +startup fails, the original Telex instance resumes; inspect the telex service logs, fix the +reported prerequisite, and restart that service to retry. Failed copies are retained for inspection. +No success announcement is sent on failure. + +After migration, `TELEX_MIGRATION=off` cannot switch back to the stale original state, and +`telex update` / `telex update --rollback` refuse to replace the migration release. For recovery, +stop the updater first, inspect `migration.json`, and back up the **migrated** data before +changing the container or image. The original Telex directory is a pre-migration backup and does +not contain conversations or token refreshes made since migration. Do not run it alongside Wirebot. + +For a container named `telex-wirebot-`, the Linux timer and oneshot service are +both named `telex-wirebot--update`. Use `systemctl --user disable --now +telex-wirebot--update.timer` and `systemctl --user stop +telex-wirebot--update.service` before maintenance (omit `--user` for a root system +installation). On macOS, disable and boot out `gui/$(id -u)/com.sadfun.telex-wirebot--update` +with `launchctl`. Linux logs go to the journal; macOS retains the last check's `update.log` in +the instance directory. Running `/bin/sh /update.sh` checks immediately. + +Run `npm run test:migration` for the isolated migration/rollback regression check. On a machine +with Docker, `TELEX_TEST_DOCKER=1 npm run test:migration` also boots the published image against a +simulated Telegram API and verifies its mounted data, readiness, and snapshot restoration. CI +and release builds also set `TELEX_TEST_SYSTEMD=1` to verify real service retirement and that the +oneshot updater has no remaining process. No real Telegram or ChatGPT credentials are used. + +The rest of this README documents native Telex and its pre-migration installation. + Telex is a self-hosted Telegram bridge for OpenAI Codex. Telegram is only the transport: a dedicated [Codex app-server](https://github.com/openai/codex/blob/main/codex-rs/app-server/README.md) owns threads, turns, tools, approvals, authentication, and configuration. Telex supports private conversations, scheduled runs, automatic Telegram voice-message transcription, photos and files in both directions, forwarded and replied-to context, polls and other structured messages, streamed replies and thinking, interactive approvals, guest mentions, persistent Codex threads, and an authenticated settings Mini App. It installs a pinned Codex CLI into isolated application storage, so it never depends on a global Codex installation. diff --git a/package-lock.json b/package-lock.json index ad8520a..ad830cd 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "telex", - "version": "0.0.34", + "version": "0.0.35", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "telex", - "version": "0.0.34", + "version": "0.0.35", "license": "FSL-1.1-MIT", "dependencies": { "@grammyjs/runner": "^2.0.3", diff --git a/package.json b/package.json index 7935dcc..c076e5f 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "telex", - "version": "0.0.34", + "version": "0.0.35", "private": true, "type": "module", "description": "Telex — a Telegram bridge for the OpenAI Codex app-server.", @@ -25,6 +25,7 @@ "dev": "npm run build && node --env-file-if-exists=.env --watch dist/cli/main.js start", "start": "node --env-file-if-exists=.env dist/cli/main.js start", "test": "npm run test:e2e --", + "test:migration": "node --import tsx --test src/update/wirebot.test.ts", "test:e2e": "node --import tsx src/core/e2e/cli.ts core", "test:telegram": "node --import tsx src/core/e2e/cli.ts telegram", "protocol:generate-bindings": "node --import tsx src/cli/main.ts codex check --apply --version $(cat codex.version)", diff --git a/src/automations/store.ts b/src/automations/store.ts index e420aa5..6767d06 100644 --- a/src/automations/store.ts +++ b/src/automations/store.ts @@ -13,7 +13,7 @@ import { type ProviderReference, } from "./types.js"; -const storedStateSchema = z.preprocess( +export const storedStateSchema = z.preprocess( migrateStoredState, z.strictObject({ version: z.literal(1), diff --git a/src/cli/main.ts b/src/cli/main.ts index e0f728a..77a7a66 100644 --- a/src/cli/main.ts +++ b/src/cli/main.ts @@ -39,15 +39,25 @@ async function main(args: readonly string[]): Promise { switch (args[0]) { case "start": { if (args.length !== 1) throw new Error(usage); - const { runTelex } = await import("../index.js"); - const result = await runTelex(); - return result.reason === "updated" ? 75 : 0; + const { runWithWirebot } = await import("../update/wirebot.js"); + const result = await runWithWirebot(async () => { + const { runTelex } = await import("../index.js"); + return await runTelex(); + }); + return result?.reason === "updated" ? 75 : 0; } case "version": { if (args.length !== 1) throw new Error(usage); console.log(await readTelexVersion(projectRootFrom(import.meta.url))); return 0; } + case "wirebot-update": { + const directory = args[1]; + if (args.length !== 2 || directory === undefined) throw new Error(usage); + const { maintainWirebot } = await import("../update/wirebot.js"); + await maintainWirebot(directory); + return 0; + } case "update": return await updateTelex(parseUpdateArguments(args)); case "codex": @@ -58,6 +68,8 @@ async function main(args: readonly string[]): Promise { } async function updateTelex(args: UpdateArguments): Promise { + const { assertTelexReleaseUpdatesAllowed } = await import("../update/wirebot.js"); + await assertTelexReleaseUpdatesAllowed(); const projectRoot = projectRootFrom(import.meta.url); const config = loadUpdateConfig(); const currentVersion = await readTelexVersion(projectRoot); diff --git a/src/config/env.ts b/src/config/env.ts index 75348f4..eac8f26 100644 --- a/src/config/env.ts +++ b/src/config/env.ts @@ -24,6 +24,7 @@ const envSchema = z.object({ .refine((value) => new URL(value).protocol === "https:", "PUBLIC_URL must use HTTPS") .optional(), TELEX_TUNNEL: z.enum(["auto", "off"]).default("auto"), + TELEX_MIGRATION: z.enum(["auto", "off"]).default("auto"), TELEX_DATA_DIR: z.string().min(1).default(".telex"), CODEX_WORKSPACE: z.string().min(1).default(".telex/workspace"), HOST: z.string().min(1).default("127.0.0.1"), @@ -38,6 +39,7 @@ interface AppConfig { readonly telegramPollTimeout: number; readonly publicUrl: string | undefined; readonly tunnelMode: "auto" | "off"; + readonly migrationMode: "auto" | "off"; readonly dataDirectory: string; readonly workspace: string; readonly updateMode: UpdateMode; @@ -72,6 +74,7 @@ export function loadAppConfig(environment: NodeJS.ProcessEnv = process.env): App telegramPollTimeout: parsed.TELEGRAM_POLL_TIMEOUT, publicUrl: parsed.PUBLIC_URL?.replace(/\/$/, ""), tunnelMode: parsed.TELEX_TUNNEL, + migrationMode: parsed.TELEX_MIGRATION, dataDirectory: resolve(parsed.TELEX_DATA_DIR), workspace: resolve(parsed.CODEX_WORKSPACE), host: parsed.HOST, diff --git a/src/core/conversation-store.ts b/src/core/conversation-store.ts index badde05..176fef8 100644 --- a/src/core/conversation-store.ts +++ b/src/core/conversation-store.ts @@ -7,7 +7,7 @@ const conversationStateSchema = z.object({ previousThreadIds: z.array(z.string().min(1)).max(10), }); -const storedStateSchema = z.object({ +export const storedStateSchema = z.object({ version: z.literal(2), conversations: z.record(z.string(), conversationStateSchema), }); diff --git a/src/core/settings-store.ts b/src/core/settings-store.ts index 7111ae6..db9802f 100644 --- a/src/core/settings-store.ts +++ b/src/core/settings-store.ts @@ -6,7 +6,7 @@ const settingsSchema = z.strictObject({ remoteClientContext: z.boolean(), }); -const storedSettingsSchema = settingsSchema.extend({ +export const storedSettingsSchema = settingsSchema.extend({ version: z.literal(1), }); diff --git a/src/shared/process.ts b/src/shared/process.ts index f35e06d..3ea8041 100644 --- a/src/shared/process.ts +++ b/src/shared/process.ts @@ -12,6 +12,8 @@ export async function runCommand( options: Readonly<{ cwd: string; env?: NodeJS.ProcessEnv; + signal?: AbortSignal; + timeout?: number; }>, ): Promise { return await new Promise((resolve, reject) => { @@ -20,13 +22,15 @@ export async function runCommand( env: options.env ?? process.env, stdio: ["ignore", "pipe", "pipe"], shell: false, + ...(options.signal === undefined ? {} : { signal: options.signal }), + ...(options.timeout === undefined ? {} : { timeout: options.timeout }), }); const stdout: Buffer[] = []; const stderr: Buffer[] = []; child.stdout.on("data", (chunk: Buffer) => stdout.push(chunk)); child.stderr.on("data", (chunk: Buffer) => stderr.push(chunk)); child.once("error", reject); - child.once("exit", (code, signal) => { + child.once("close", (code, signal) => { const result = { stdout: Buffer.concat(stdout).toString("utf8"), stderr: Buffer.concat(stderr).toString("utf8"), diff --git a/src/update/docker.ts b/src/update/docker.ts new file mode 100644 index 0000000..b71404f --- /dev/null +++ b/src/update/docker.ts @@ -0,0 +1,354 @@ +import { createHash } from "node:crypto"; +import { constants } from "node:fs"; +import { access, mkdir } from "node:fs/promises"; +import { basename, delimiter, dirname, join } from "node:path"; +import { setTimeout as delay } from "node:timers/promises"; +import { z } from "zod"; +import { externalProcessEnvironment } from "../shared/environment.js"; +import { atomicWriteFile } from "../shared/fs.js"; +import type { Logger } from "../shared/logger.js"; +import { runCommand } from "../shared/process.js"; + +export const wirebotImage = "ghcr.io/sadfun/wirebot:latest"; +export function bindMount(source: string, target: string, readonly = false): string { + return ["type=bind", `source=${source}`, `target=${target}`, ...(readonly ? ["readonly"] : [])] + .map((field) => `"${field.replaceAll('"', '""')}"`) + .join(","); +} +const installerRevision = "4e16be805b3af595aa448d1a4def857841f1c62c"; +const installerSha256 = "f51e472f1ffb1cf2516a9fd55ab7d7d1ed8d07288d31dce93de3c66524b92997"; +const containerSchema = z.object({ + Image: z.string(), + Config: z.object({ Labels: z.record(z.string(), z.string()).nullable() }), + State: z.object({ Running: z.boolean(), Restarting: z.boolean(), StartedAt: z.string() }), +}); + +function commandEnvironment(): NodeJS.ProcessEnv { + return externalProcessEnvironment({ + PATH: [ + process.env.PATH, + "/opt/homebrew/bin", + "/usr/local/bin", + "/usr/bin", + "/bin", + "/usr/sbin", + "/sbin", + ] + .filter(Boolean) + .join(delimiter), + NONINTERACTIVE: "1", + HOMEBREW_NO_AUTO_UPDATE: "1", + }); +} + +async function executable(name: string): Promise { + for (const directory of (commandEnvironment().PATH ?? "").split(delimiter)) { + if (!directory) continue; + const candidate = join(directory, name); + try { + await access(candidate, constants.X_OK); + return candidate; + } catch { + // Try the next PATH entry, including Homebrew's paths in launchd sessions. + } + } + return undefined; +} + +async function command(args: readonly string[], signal?: AbortSignal): Promise { + const [program, ...rest] = args; + if (program === undefined) throw new Error("Missing command"); + const environment = commandEnvironment(); + if (args.includes("--host")) { + delete environment.DOCKER_HOST; + delete environment.DOCKER_CONTEXT; + delete environment.DOCKER_TLS_VERIFY; + } + const result = await runCommand(program, rest, { + cwd: "/", + env: environment, + timeout: 20 * 60_000, + ...(signal === undefined ? {} : { signal }), + }); + return result.stdout.trim(); +} + +/** The endpoint is pinned so a later `docker context use` cannot move this bot. */ +export class Docker { + public readonly command: readonly string[]; + + public constructor(command: readonly string[]) { + this.command = command; + } + + public async run(args: readonly string[], signal?: AbortSignal): Promise { + return await command([...this.command, ...args], signal); + } + + public async ensureRunning(signal: AbortSignal): Promise { + try { + await this.run(["info"], signal); + return; + } catch { + signal.throwIfAborted(); + } + const host = this.command[this.command.indexOf("--host") + 1] ?? ""; + if (process.platform === "darwin") { + if (host.includes("/.colima/")) { + const colima = await executable("colima"); + if (colima === undefined) throw new Error("Start the Colima daemon used by Wirebot"); + await command( + [colima, "start", "--profile", basename(dirname(new URL(host).pathname))], + signal, + ); + } else { + await command(["/usr/bin/open", "-a", "Docker"], signal); + } + } else if (host.includes("/run/user/")) { + await command(["systemctl", "--user", "start", "docker"], signal); + } else { + await command( + [...(process.getuid?.() === 0 ? [] : ["sudo", "-n"]), "systemctl", "start", "docker"], + signal, + ); + } + for (let attempt = 0; attempt < 60; attempt += 1) { + try { + await this.run(["info"], signal); + return; + } catch { + await delay(2_000, undefined, { signal }); + } + } + throw new Error("Wirebot's Docker daemon is unavailable"); + } + + public async container(name: string) { + const id = await this.run(["container", "ls", "-aq", "--filter", `name=^/${name}$`]); + if (!id) return undefined; + return containerSchema.parse(JSON.parse(await this.run(["inspect", name]))[0]); + } + + /** Copy with container ownership intact; the host user need not own files under /data. */ + public async copyData( + source: string, + destination: string, + image: string, + signal?: AbortSignal, + ): Promise { + await mkdir(destination, { recursive: true, mode: 0o700 }); + const name = `telex-wirebot-copy-${createHash("sha256").update([source, destination].sort().join("\0")).digest("hex").slice(0, 12)}`; + const existing = await this.container(name); + if (existing !== undefined) { + if (existing.Config.Labels?.["io.github.sadfun.telex-copy"] !== name) + throw new Error(`Container ${name} belongs to another application`); + await this.run(["rm", "--force", name]); + } + await this.run( + [ + "run", + "--rm", + "--name", + name, + "--label", + `io.github.sadfun.telex-copy=${name}`, + "--network", + "none", + "--user", + "0", + "--entrypoint", + "sh", + "--mount", + bindMount(source, "/source", true), + "--mount", + bindMount(destination, "/destination"), + image, + "-c", + "find /destination -mindepth 1 -delete && cp -a /source/. /destination/", + ], + signal, + ); + } + + public async waitUntilReady( + name: string, + expectAuthentication: boolean, + signal: AbortSignal, + ): Promise { + const deadline = Date.now() + 180_000; + while (Date.now() < deadline) { + signal.throwIfAborted(); + const container = await this.container(name); + if (container === undefined || !container.State.Running || container.State.Restarting) { + throw new Error("Wirebot exited before it was ready; inspect its Docker logs"); + } + // /healthz starts before the connectors. The ready log confirms polling and schedules too. + const logs = await this.run(["logs", "--since", container.State.StartedAt, name], signal); + const ready = logs.split("\n").some((line) => { + try { + return JSON.parse(line).message === "Wirebot is ready"; + } catch { + return false; + } + }); + if (ready) { + const health = JSON.parse( + await this.run( + [ + "exec", + name, + "curl", + "--fail", + "--silent", + "--max-time", + "10", + "http://127.0.0.1:8787/healthz", + ], + signal, + ), + ); + if ( + health.ok === true && + (health.codex === "authenticated" || + (!expectAuthentication && ["needs_login", "not_required"].includes(health.codex))) + ) + return; + } + await delay(2_000, undefined, { signal }); + } + throw new Error( + "Wirebot did not become ready with the migrated Codex account within 3 minutes", + ); + } +} + +/** Installs only when missing. Unattended services never wait for a sudo password. */ +export async function ensureDocker( + directory: string, + logger: Logger, + signal: AbortSignal, +): Promise { + if (!["linux", "darwin"].includes(process.platform) || !["x64", "arm64"].includes(process.arch)) { + throw new Error("Wirebot requires Linux or macOS on x86-64 or ARM64"); + } + let binary = await executable("docker"); + if (binary !== undefined) { + const host = + process.env.DOCKER_HOST ?? + (await command( + [binary, "context", "inspect", "--format", "{{.Endpoints.docker.Host}}"], + signal, + )); + if (!host.startsWith("unix://")) { + throw new Error( + "Migration requires a local Docker Unix socket; remote daemons cannot mount this user's data", + ); + } + const docker = new Docker([binary, "--host", host]); + try { + await docker.run(["info"], signal); + return docker; + } catch { + signal.throwIfAborted(); + if (process.env.DOCKER_HOST !== undefined) { + throw new Error( + "The configured DOCKER_HOST is unavailable; start that Docker daemon and restart telex", + ); + } + } + } + + logger.info("Preparing Docker for Wirebot"); + if (process.platform === "darwin") { + try { + await access("/Applications/Docker.app"); + await command(["/usr/bin/open", "-a", "Docker"], signal); + } catch { + const brew = await executable("brew"); + if (brew === undefined) { + throw new Error( + "Install Homebrew or start Docker Desktop, then restart telex to finish migrating", + ); + } + await command([brew, "install", "docker", "colima"], signal); + const colima = await executable("colima"); + if (colima === undefined) throw new Error("Homebrew did not install Colima"); + await command([colima, "start", "--mount", `${directory}:w`], signal); + await command([brew, "services", "start", "colima"], signal); + } + binary = await executable("docker"); + if (binary === undefined) throw new Error("Docker CLI is missing after installation"); + for (let attempt = 0; attempt < 60; attempt += 1) { + try { + const host = await command( + [binary, "context", "inspect", "--format", "{{.Endpoints.docker.Host}}"], + signal, + ); + if (!host.startsWith("unix://")) throw new Error("Docker must use a local Unix socket"); + const docker = new Docker([binary, "--host", host]); + await docker.run(["info"], signal); + return docker; + } catch { + await delay(2_000, undefined, { signal }); + } + } + throw new Error("Docker did not start; start Docker Desktop or Colima and restart telex"); + } + + // Rootless Docker may simply need its existing user service restarted. + if (binary !== undefined) { + try { + await command(["systemctl", "--user", "start", "docker"], signal); + const host = await command( + [binary, "context", "inspect", "--format", "{{.Endpoints.docker.Host}}"], + signal, + ); + if (host.startsWith("unix://")) { + const docker = new Docker([binary, "--host", host]); + await docker.run(["info"], signal); + return docker; + } + } catch { + signal.throwIfAborted(); + } + } + const privilege = process.getuid?.() === 0 ? [] : ["sudo", "-n"]; + if (binary !== undefined) { + try { + const docker = new Docker([...privilege, binary, "--host", "unix:///var/run/docker.sock"]); + await docker.run(["info"], signal); + return docker; + } catch { + signal.throwIfAborted(); + } + } + try { + await command([...privilege, "true"], signal); + } catch { + throw new Error( + "Docker setup needs root or passwordless sudo. Install/start Docker for this user, then restart telex", + ); + } + if (binary === undefined) { + const response = await fetch( + `https://raw.githubusercontent.com/docker/docker-install/${installerRevision}/install.sh`, + { + signal: AbortSignal.any([signal, AbortSignal.timeout(30_000)]), + }, + ); + if (!response.ok) throw new Error("Could not download the Docker installer"); + const script = await response.text(); + if (createHash("sha256").update(script).digest("hex") !== installerSha256) { + throw new Error("Docker installer checksum mismatch"); + } + const installer = join(directory, "install-docker.sh"); + await atomicWriteFile(installer, script); + await command([...privilege, "sh", installer], signal); + binary = await executable("docker"); + } + if (binary === undefined) throw new Error("Docker CLI is missing after installation"); + await command([...privilege, "systemctl", "enable", "--now", "docker"], signal); + const docker = new Docker([...privilege, binary, "--host", "unix:///var/run/docker.sock"]); + await docker.run(["info"], signal); + return docker; +} diff --git a/src/update/wirebot-service.ts b/src/update/wirebot-service.ts new file mode 100644 index 0000000..4731ec5 --- /dev/null +++ b/src/update/wirebot-service.ts @@ -0,0 +1,225 @@ +import { homedir } from "node:os"; +import { join, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; +import { externalProcessEnvironment } from "../shared/environment.js"; +import { atomicWriteFile } from "../shared/fs.js"; +import { runCommand } from "../shared/process.js"; +import { wirebotImage } from "./docker.js"; + +function quote(value: string): string { + return `'${value.replaceAll("'", "'\\''")}'`; +} + +function shellCommand(args: readonly string[]): string { + return args.map(quote).join(" "); +} + +async function hostCommand(args: readonly string[]): Promise { + const [command, ...rest] = args; + if (command === undefined) throw new Error("Missing service command"); + const result = await runCommand(command, rest, { + cwd: "/", + env: externalProcessEnvironment(), + timeout: 60_000, + }); + return result.stdout.trim(); +} + +/** No resident process: only pull/inspect on the common path; Node handles rare transactions. */ +export function updaterScript( + directory: string, + name: string, + dockerCommand: readonly string[], + maintenanceCommand: readonly string[], + retireService: string, +): string { + return `#!/bin/sh +set -eu +umask 077 +cd ${quote(directory)} +export PATH=/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin +unset DOCKER_HOST DOCKER_CONTEXT DOCKER_TLS_VERIFY +docker() { ${shellCommand(dockerCommand)} "$@"; } +maintain() { + : > maintenance-needed + exec ${shellCommand(maintenanceCommand)} +} + +# Only the replacement job can stop Telex without killing its own migration midway. +if [ ! -f service-retired ]; then +${retireService} + : > service-retired +fi + +# launchd bootout can return before the old process has released its PID lock. +attempt=0 +while [ -f runtime.lock ]; do + pid=$(cat runtime.lock) || break + case "$pid" in ''|*[!0-9]*|0) echo "Invalid migration lock" >&2; exit 1 ;; esac + kill -0 "$pid" 2>/dev/null || break + attempt=$((attempt + 1)) + [ "$attempt" -lt 60 ] || exit 1 + sleep 1 +done + +[ -f current-image ] && [ ! -f maintenance-needed ] || maintain +current=$(cat current-image) +case "$current" in sha256:*) ;; *) maintain ;; esac +status=$(docker inspect --format '{{.State.Running}} {{.State.Restarting}} {{.Image}} {{index .Config.Labels "io.github.sadfun.telex-migration"}}' ${quote(name)}) || maintain +[ "$status" = "true false $current ${name}" ] || maintain +docker pull ${quote(wirebotImage)} +latest=$(docker image inspect --format '{{.Id}}' ${quote(wirebotImage)}) +[ "$latest" = "$current" ] || maintain +`; +} + +export function systemdUnits(script: string, seconds: number): { service: string; timer: string } { + if (/[\r\n\0]/u.test(script)) throw new Error("Unsupported control character in service path"); + const argument = `"${script + .replaceAll("\\", "\\\\") + .replaceAll('"', '\\"') + .replaceAll("%", "%%") + .replaceAll("$", () => "$$")}"`; + return { + service: `[Unit] +Description=Wirebot image update +After=network-online.target +Wants=network-online.target + +[Service] +Type=oneshot +ExecStart=/bin/sh ${argument} +TimeoutStartSec=90min +Restart=no +`, + timer: `[Unit] +Description=Check for Wirebot image updates + +[Timer] +OnBootSec=1min +OnUnitInactiveSec=${seconds}s +AccuracySec=1min + +[Install] +WantedBy=timers.target +`, + }; +} + +export function launchAgent(label: string, script: string, seconds: number): string { + const xml = (value: string) => + value.replaceAll("&", "&").replaceAll("<", "<").replaceAll(">", ">"); + return ` + + +Label${xml(label)} +ProgramArguments/bin/sh${xml(script)} +RunAtLoad +StartInterval${seconds} + +`; +} + +export const wirebotService = { + async prepare( + directory: string, + state: { name: string; dockerCommand: readonly string[] }, + intervalMs: number, + ): Promise<() => Promise> { + const seconds = Math.max(60, Math.ceil(intervalMs / 1_000)); + const script = join(directory, "update.sh"); + // The installer's launcher keeps the stable Node path; Homebrew may remove this + // process's versioned Cellar executable during its next upgrade. + const maintenance = [ + ...(process.env.TELEX_INSTALL_DIR + ? [join(resolve(process.env.TELEX_INSTALL_DIR), "bin", "telex")] + : [process.execPath, fileURLToPath(new URL("../cli/main.js", import.meta.url))]), + "wirebot-update", + directory, + ]; + const name = `${state.name}-update`; + if (process.platform === "linux") { + let manager = ["systemctl", "--user"]; + // Root installations may use a system unit instead of the installer's user unit. + if (process.getuid?.() === 0) { + const pid = await hostCommand([ + "systemctl", + "show", + "telex.service", + "-p", + "MainPID", + "--value", + ]); + if (pid === String(process.pid)) manager = ["systemctl"]; + } + try { + await hostCommand([...manager, "show-environment"]); + } catch (error) { + if (process.getuid?.() !== 0) throw error; + manager = ["systemctl"]; + await hostCommand([...manager, "show-environment"]); + } + const unitDirectory = manager.includes("--user") + ? join(process.env.XDG_CONFIG_HOME ?? join(homedir(), ".config"), "systemd", "user") + : "/etc/systemd/system"; + const units = systemdUnits(script, seconds); + await atomicWriteFile(join(unitDirectory, `${name}.service`), units.service); + await atomicWriteFile(join(unitDirectory, `${name}.timer`), units.timer); + const ctl = shellCommand(manager); + await atomicWriteFile( + script, + updaterScript( + directory, + state.name, + state.dockerCommand, + maintenance, + ` if [ "$(${ctl} show telex.service -p LoadState --value)" = loaded ]; then + ${ctl} disable --now telex.service + fi`, + ), + 0o700, + ); + await hostCommand([...manager, "daemon-reload"]); + return async () => { + await hostCommand([...manager, "enable", "--now", `${name}.timer`]); + await hostCommand([...manager, "start", "--no-block", `${name}.service`]); + }; + } + if (process.platform === "darwin") { + const domain = `gui/${process.getuid?.()}`; + await hostCommand(["/bin/launchctl", "print", domain]); + const label = `com.sadfun.${name}`; + const plist = join(homedir(), "Library", "LaunchAgents", `${label}.plist`); + await atomicWriteFile(plist, launchAgent(label, script, seconds)); + await atomicWriteFile( + script, + updaterScript( + directory, + state.name, + state.dockerCommand, + maintenance, + ` if /bin/launchctl print ${quote(`${domain}/com.sadfun.telex`)} >/dev/null 2>&1; then + /bin/launchctl disable ${quote(`${domain}/com.sadfun.telex`)} + /bin/launchctl bootout ${quote(`${domain}/com.sadfun.telex`)} + fi`, + ) + // Keep only the latest invocation's log, without another daemon for log rotation. + .replace( + "umask 077\n", + `umask 077\nexec >${quote(join(directory, "update.log"))} 2>&1\n`, + ), + 0o700, + ); + return async () => { + // An already-loaded job must not be booted out while it might be replacing the image. + try { + await hostCommand(["/bin/launchctl", "print", `${domain}/${label}`]); + } catch { + await hostCommand(["/bin/launchctl", "enable", `${domain}/${label}`]); + await hostCommand(["/bin/launchctl", "bootstrap", domain, plist]); + } + }; + } + throw new Error("Wirebot auto-updates require systemd or launchd"); + }, +}; diff --git a/src/update/wirebot.test.ts b/src/update/wirebot.test.ts new file mode 100644 index 0000000..5be6bb2 --- /dev/null +++ b/src/update/wirebot.test.ts @@ -0,0 +1,701 @@ +import assert from "node:assert/strict"; +import { + cp, + mkdir, + mkdtemp, + readdir, + readFile, + readlink, + rm, + stat, + writeFile, +} from "node:fs/promises"; +import { createServer } from "node:http"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { test } from "node:test"; +import { loadAppConfig } from "../config/env.js"; +import { atomicWriteJson } from "../shared/fs.js"; +import { Logger } from "../shared/logger.js"; +import { runCommand } from "../shared/process.js"; +import { bindMount, Docker, wirebotImage } from "./docker.js"; +import { + acquireRuntimeLock, + maintainWirebot, + migrationDirectory, + migrationMessage, + prepareMigration, + runWithWirebot, + updateWirebot, +} from "./wirebot.js"; +import { launchAgent, systemdUnits, updaterScript, wirebotService } from "./wirebot-service.js"; + +test("migration preserves 0.0.34 state and survives failed/interrupted image updates", { + timeout: 30_000, +}, async (t) => { + const root = await mkdtemp(join(tmpdir(), "telex-migration-test-")); + const savedEnv = { ...process.env }; + const logger = new Logger("error"); + const oldImage = `sha256:${"a".repeat(64)}`; + const newImage = `sha256:${"b".repeat(64)}`; + let latestImage = oldImage; + let failPull = false; + let failReady = false; + let blockCheckpoint: string | undefined; + let liveData: string | undefined; + let checks = 0; + const commands: string[][] = []; + let activations = 0; + let failService = false; + let failActivation = false; + t.mock.method(wirebotService, "prepare", async (directory: string) => { + if (failService) throw new Error("simulated unavailable service manager"); + return async () => { + activations += 1; + await assert.rejects(stat(join(directory, "runtime.lock")), { code: "ENOENT" }); + if (failActivation) throw new Error("simulated scheduler activation failure"); + await maintainWirebot(directory); + }; + }); + const containers = new Map< + string, + { + Image: string; + Config: { Labels: Record }; + State: { Running: boolean; Restarting: boolean; StartedAt: string }; + } + >(); + t.mock.method(Docker.prototype, "container", async (name: string) => containers.get(name)); + t.mock.method(Docker.prototype, "copyData", async (source: string, destination: string) => { + await rm(destination, { recursive: true, force: true }); + await cp(source, destination, { recursive: true, verbatimSymlinks: true }); + }); + t.mock.method(Docker.prototype, "waitUntilReady", async (_name: string, expectAuth: boolean) => { + checks += 1; + if (checks === 1) assert.equal(expectAuth, true); + if (blockCheckpoint !== undefined) { + await rm(blockCheckpoint); + await mkdir(blockCheckpoint); + } + if (failReady) { + assert(liveData); + await writeFile(join(liveData, "conversations.json"), "incompatible candidate state"); + throw new Error("simulated unhealthy candidate"); + } + }); + t.mock.method(Docker.prototype, "run", async (args: string[]) => { + commands.push([...args]); + const last = args.at(-1) ?? ""; + switch (args[0]) { + case "info": + return "Docker is running"; + case "pull": + if (failPull) throw new Error("simulated registry outage"); + assert.equal(last, wirebotImage); + return "pulled"; + case "image": + return latestImage; + case "run": { + const name = args[args.indexOf("--name") + 1] ?? ""; + assert(!containers.has(name)); + assert( + ![...containers.values()].some((container) => container.State.Running), + "two bot instances must never poll concurrently", + ); + const label = args[args.indexOf("--label") + 1] ?? ""; + containers.set(name, { + Image: last, + Config: { Labels: { [label.split("=")[0] ?? ""]: name } }, + State: { Running: true, Restarting: false, StartedAt: new Date().toISOString() }, + }); + return name; + } + case "stop": { + const container = containers.get(last); + assert(container); + container.State.Running = false; + return last; + } + case "start": { + const container = containers.get(last); + assert(container); + container.State.Running = true; + return last; + } + case "rename": { + const name = args[1] ?? ""; + const container = containers.get(name); + assert(container); + assert(!containers.has(last)); + containers.delete(name); + containers.set(last, container); + return last; + } + case "rm": + containers.delete(last); + return last; + default: + throw new Error(`Unexpected Docker command: ${args.join(" ")}`); + } + }); + const messages: Array<{ chat_id: number; text: string }> = []; + let failNotice = false; + const telegram = createServer(async (request, response) => { + assert(request.url?.endsWith("/sendMessage")); + const chunks = []; + for await (const chunk of request) chunks.push(chunk); + messages.push(JSON.parse(Buffer.concat(chunks).toString())); + assert(checks > 0, "the announcement must follow readiness verification"); + response.setHeader("Content-Type", "application/json"); + if (failNotice) { + response.statusCode = 503; + response.end(JSON.stringify({ ok: false, error_code: 503, description: "Try later" })); + return; + } + response.end(JSON.stringify({ ok: true, result: { message_id: messages.length } })); + }); + await new Promise((resolve) => telegram.listen(0, "127.0.0.1", resolve)); + try { + const address = telegram.address(); + assert(address && typeof address !== "string"); + Object.assign(process.env, { + XDG_DATA_HOME: join(root, "host-state"), + TELEX_DATA_DIR: join(root, "telex data"), + CODEX_WORKSPACE: join(root, "custom workspace"), + TELEGRAM_BOT_TOKEN: "123456:fake-token-for-migration-test", + TELEGRAM_ALLOWED_USER_IDS: "123456", + TELEGRAM_API_BASE: `http://127.0.0.1:${address.port}`, + TELEX_TUNNEL: "off", + TELEX_MIGRATION: "auto", + PUBLIC_URL: "https://example.com", + PORT: "9876", + LOG_LEVEL: "error", + }); + const config = loadAppConfig(); + const directory = migrationDirectory(config); + const stateFile = join(directory, "migration.json"); + await mkdir(directory, { recursive: true, mode: 0o700 }); + const memory = join(config.workspace, ".telex", "automations", "schedule-1"); + await mkdir(memory, { recursive: true }); + await writeFile(join(memory, "memory.md"), "Remember the user's project."); + await mkdir(join(config.dataDirectory, "codex-home", "sessions"), { recursive: true }); + await mkdir(join(config.dataDirectory, "codex-home", "memories")); + const stateData = { + version: 2, + conversations: { + "telegram:123456:0": { activeThreadId: "thread-a", previousThreadIds: ["thread-b"] }, + "telegram:-123:42": { activeThreadId: "topic-thread", previousThreadIds: [] }, + "telegram:123:direct:7": { activeThreadId: "direct-thread", previousThreadIds: [] }, + }, + }; + const settings = { version: 1, remoteClientContext: false }; + const instant = "2026-09-06T12:00:00Z"; + const reference = { provider: "telegram", resource: "conversation", id: "telegram:123456:0" }; + const automation = { + id: "schedule-1", + owner: { ...reference, resource: "user", id: "123456" }, + conversation: reference, + deliveryTarget: { ...reference, resource: "destination", id: "opaque-target" }, + name: "Check project", + prompt: "Check my project", + status: "active", + schedule: { rrule: "FREQ=DAILY", startAt: instant, timeZone: "Europe/Berlin" }, + threadId: "thread-a", + notificationPolicy: "on-result", + model: null, + reasoningEffort: null, + nextRunAt: instant, + lastRunAt: null, + deferredUntil: null, + deferralReason: null, + createdAt: instant, + updatedAt: instant, + revision: 2, + }; + const automations = { + version: 1, + automations: { "schedule-1": automation }, + runs: {}, + notifications: {}, + }; + const authBytes = + '{"auth_mode":"chatgpt","tokens":{"access_token":"preserve-me","refresh_token":"refresh-me","account_id":"account-1"}}'; + await atomicWriteJson(join(config.dataDirectory, "conversations.json"), stateData); + await atomicWriteJson(join(config.dataDirectory, "settings.json"), settings); + await atomicWriteJson(join(config.dataDirectory, "automations.json"), automations); + await writeFile(join(config.dataDirectory, "codex-home", "auth.json"), authBytes, { + mode: 0o600, + }); + await writeFile( + join(config.dataDirectory, "codex-home", "config.toml"), + 'cli_auth_credentials_store = "file"\nmodel = "custom-model"\n', + ); + await writeFile( + join(config.dataDirectory, "codex-home", "sessions", "rollout.jsonl"), + '{"thread_id":"thread-a"}\n', + ); + const database = Buffer.from([0, 17, 255, 42]); + await writeFile(join(config.dataDirectory, "codex-home", "state_5.sqlite"), database); + await writeFile(join(config.dataDirectory, "codex-home", "state_5.sqlite-wal"), database); + await writeFile( + join(config.dataDirectory, "codex-home", "memories", "MEMORY.md"), + "Durable Codex memory", + ); + const docker = new Docker(["fake-docker"]); + let state = await prepareMigration(config, directory, docker, oldImage); + liveData = state.dataDirectory; + const copiedData = join( + directory, + (await readdir(directory)).find((name) => name.startsWith("data-")) ?? "missing", + ); + for (const [file, expected] of [ + ["conversations.json", stateData], + ["settings.json", settings], + ["automations.json", automations], + ] as const) { + assert.deepEqual(JSON.parse(await readFile(join(copiedData, file), "utf8")), expected); + assert.deepEqual( + JSON.parse(await readFile(join(config.dataDirectory, file), "utf8")), + expected, + ); + } + assert.equal(await readFile(join(copiedData, "codex-home", "auth.json"), "utf8"), authBytes); + assert.deepEqual( + await readFile(join(copiedData, "codex-home", "state_5.sqlite-wal")), + database, + ); + assert.equal( + await readFile(join(copiedData, "codex-home", "memories", "MEMORY.md"), "utf8"), + "Durable Codex memory", + ); + const copiedWorkspace = join(copiedData, ".telex-external-workspace"); + assert.equal(await readlink(join(copiedWorkspace, ".wirebot")), ".telex"); + assert.equal( + await readFile( + join(copiedWorkspace, ".wirebot", "automations", "schedule-1", "memory.md"), + "utf8", + ), + "Remember the user's project.", + ); + assert(state.runArguments.some((arg) => arg.includes(`target=${config.workspace}`))); + assert(state.runArguments.some((arg) => arg.includes(`target=${config.dataDirectory}`))); + const environment = await readFile(join(directory, "wirebot.env"), "utf8"); + assert(environment.includes(`WIREBOT_DATA_DIR=${config.dataDirectory}\n`)); + assert(environment.includes("WIREBOT_TUNNEL=off\n")); + assert(environment.includes("PUBLIC_URL=https://example.com\n")); + assert(environment.includes("HOST=0.0.0.0\nPORT=8787\n")); + assert.equal((await stat(join(directory, "wirebot.env"))).mode & 0o777, 0o600); + await atomicWriteJson(stateFile, state); + const noFallback = async () => { + assert.fail("an active migration must not run Telex"); + }; + await runWithWirebot(noFallback); + state = JSON.parse(await readFile(stateFile, "utf8")); + assert.equal(state.status, "active"); + assert.deepEqual(state.notifiedUserIds, [123456]); + assert.deepEqual(messages, [ + { + chat_id: 123456, + text: migrationMessage, + parse_mode: "HTML", + link_preview_options: { is_disabled: true }, + }, + ]); + assert.equal( + containers.get(state.name)?.State.Running, + true, + "the migration process exits while Docker keeps Wirebot running", + ); + assert.equal(activations, 1); + assert.equal(await readFile(join(directory, "current-image"), "utf8"), `${oldImage}\n`); + await assert.rejects(stat(join(directory, "maintenance-needed")), { code: "ENOENT" }); + // Even an opt-out cannot accidentally launch stale Telex alongside migrated Wirebot. + process.env.TELEX_MIGRATION = "off"; + await runWithWirebot(noFallback); + assert.equal(messages.length, 1, "successful announcements must not repeat on restart"); + await docker.run(["start", state.name]); + + const signal = new AbortController().signal; + const beforePull = commands.length; + failPull = true; + await assert.rejects( + updateWirebot(docker, stateFile, state, signal, logger), + /registry outage/u, + ); + assert(!commands.slice(beforePull).some((args) => args[0] === "stop")); + failPull = false; + latestImage = newImage; + failReady = true; + await assert.rejects(updateWirebot(docker, stateFile, state, signal, logger), /unhealthy/u); + assert.equal(containers.get(state.name)?.Image, oldImage); + assert.equal(containers.get(state.name)?.State.Running, true); + assert(!containers.has(`${state.name}-previous`)); + assert.equal(JSON.parse(await readFile(stateFile, "utf8")).image, oldImage); + assert.deepEqual( + JSON.parse(await readFile(join(liveData, "conversations.json"), "utf8")), + stateData, + "rollback must restore data modified by the failed candidate", + ); + failReady = false; + state = await updateWirebot(docker, stateFile, state, signal, logger); + assert.equal(state.image, newImage); + assert.equal(containers.get(state.name)?.Image, newImage); + const beforeNoop = commands.length; + await updateWirebot(docker, stateFile, state, signal, logger); + assert(!commands.slice(beforeNoop).some((args) => args[0] === "stop")); + + // Recover a crash after renaming the old container but before creating its replacement. + await docker.run(["stop", state.name]); + await docker.run(["rename", state.name, `${state.name}-previous`]); + await runWithWirebot(noFallback); + assert(containers.has(state.name)); + assert(!containers.has(`${state.name}-previous`)); + + const lock = join(directory, "runtime.lock"); + const unlock = await acquireRuntimeLock(lock); + await assert.rejects(acquireRuntimeLock(lock), /already running/u); + await unlock(); + await atomicWriteJson(join(config.dataDirectory, "conversations.json"), { + version: 2, + conversations: { broken: {} }, + }); + await assert.rejects(prepareMigration(config, directory, docker, oldImage)); + assert.equal( + await readFile(join(config.dataDirectory, "codex-home", "auth.json"), "utf8"), + authBytes, + ); + await atomicWriteJson(join(config.dataDirectory, "conversations.json"), stateData); + containers.clear(); + state = { ...state, status: "prepared", notifiedUserIds: [] }; + await atomicWriteJson(stateFile, state); + failService = true; + assert.equal(await runWithWirebot(async () => "telex-fallback"), "telex-fallback"); + assert.equal(containers.size, 0, "missing scheduler must fail before starting Wirebot"); + assert.equal(messages.length, 1); + failService = false; + await atomicWriteJson(stateFile, state); + failReady = true; + assert.equal(await runWithWirebot(async () => "telex-fallback"), "telex-fallback"); + assert.equal(containers.size, 0); + assert.equal(messages.length, 1, "failed migration must not announce success"); + assert.deepEqual( + JSON.parse(await readFile(join(config.dataDirectory, "conversations.json"), "utf8")), + stateData, + ); + failReady = false; + await atomicWriteJson(stateFile, state); + failActivation = true; + await assert.rejects(runWithWirebot(noFallback), /scheduler activation failure/u); + assert.equal(JSON.parse(await readFile(stateFile, "utf8")).status, "active"); + assert.equal(containers.get(state.name)?.State.Running, true); + assert.equal(messages.length, 1, "activation failure must not announce a completed handoff"); + failActivation = false; + // Readiness is the handoff boundary: checkpoint failure must never start stale Telex. + await atomicWriteJson(stateFile, state); + blockCheckpoint = stateFile; + await assert.rejects(runWithWirebot(noFallback)); + assert.equal(containers.get(state.name)?.State.Running, true); + assert.equal(messages.length, 1); + blockCheckpoint = undefined; + await rm(stateFile, { recursive: true }); + await atomicWriteJson(stateFile, { ...state, status: "active" }); + failNotice = true; + await maintainWirebot(directory); + assert.deepEqual(JSON.parse(await readFile(stateFile, "utf8")).notifiedUserIds, []); + await stat(join(directory, "maintenance-needed")); + failNotice = false; + await maintainWirebot(directory); + await maintainWirebot(directory); + assert.equal(messages.length, 3, "retry the failed notice once, then remember delivery"); + await assert.rejects(stat(join(directory, "maintenance-needed")), { code: "ENOENT" }); + } finally { + await new Promise((resolve) => telegram.close(() => resolve())); + for (const key of Object.keys(process.env)) if (!(key in savedEnv)) delete process.env[key]; + Object.assign(process.env, savedEnv); + await rm(root, { recursive: true, force: true }); + } +}); + +test("scheduled shell checks exit without Node when the image is unchanged", async () => { + const root = await mkdtemp(join(tmpdir(), "wirebot shell '$-")); + const name = "telex-wirebot-0123456789ab"; + const oldImage = `sha256:${"a".repeat(64)}`; + const newImage = `sha256:${"b".repeat(64)}`; + const script = join(root, "update.sh"); + const docker = join(root, "docker"); + const maintenance = join(root, "maintain"); + try { + await writeFile( + docker, + `#!/bin/sh +set -eu +printf '%s\\n' "$*" >> docker-calls +case "$1" in + inspect) cat status ;; + pull) [ ! -f fail-pull ] ;; + image) cat latest ;; + *) exit 2 ;; +esac +`, + { mode: 0o700 }, + ); + await writeFile( + maintenance, + `#!/bin/sh +set -eu +[ "$1" = 'literal $HOME $(touch injected)' ] +printf 'called\\n' >> maintenance-calls +[ ! -f fail-maintenance ] +`, + { mode: 0o700 }, + ); + await writeFile( + script, + updaterScript( + root, + name, + [docker], + [maintenance, "literal $HOME $(touch injected)"], + " printf 'retired\\n' >> retirement", + ), + ); + await writeFile(join(root, "current-image"), oldImage); + await writeFile(join(root, "latest"), oldImage); + await writeFile(join(root, "status"), `true false ${oldImage} ${name}`); + const check = () => runCommand("/bin/sh", [script], { cwd: root, timeout: 5_000 }); + await check(); + await check(); + assert.equal(await readFile(join(root, "retirement"), "utf8"), "retired\n"); + await assert.rejects(stat(join(root, "maintenance-calls")), { code: "ENOENT" }); + await assert.rejects(stat(join(root, "maintenance-needed")), { code: "ENOENT" }); + + await writeFile(join(root, "fail-pull"), ""); + await assert.rejects(check()); + await assert.rejects(stat(join(root, "maintenance-calls")), { code: "ENOENT" }); + await rm(join(root, "fail-pull")); + await writeFile(join(root, "latest"), newImage); + await check(); + assert.equal(await readFile(join(root, "maintenance-calls"), "utf8"), "called\n"); + await stat(join(root, "maintenance-needed")); + await assert.rejects(stat(join(root, "injected")), { code: "ENOENT" }); + // A failed/interrupted transaction must retry even if the registry tag changes back. + await writeFile(join(root, "latest"), oldImage); + await writeFile(join(root, "fail-maintenance"), ""); + await assert.rejects(check()); + await stat(join(root, "maintenance-needed")); + await rm(join(root, "fail-maintenance")); + await check(); + assert.equal((await readFile(join(root, "maintenance-calls"), "utf8")).split("\n").length, 4); + await rm(join(root, "maintenance-needed")); + await writeFile(join(root, "status"), `false false ${oldImage} ${name}`); + await check(); + await stat(join(root, "maintenance-needed")); + const units = systemdUnits(script, 21_600); + assert.match(units.service, /Type=oneshot/u); + assert.match(units.service, /Restart=no/u); + assert.match(units.service, /\$\$/u); + assert.match(units.timer, /OnUnitInactiveSec=21600s/u); + const plist = launchAgent(name, `${script}&`, 21_600); + assert(!plist.includes("KeepAlive")); + assert.match(plist, /StartInterval<\/key>21600<\/integer>/u); + assert(plist.includes("&")); + if (process.platform === "darwin") { + await writeFile(join(root, "test.plist"), plist); + await runCommand("plutil", ["-lint", join(root, "test.plist")], { cwd: root }); + } + if (process.env.TELEX_TEST_SYSTEMD === "1") { + assert.equal(process.platform, "linux"); + assert.equal(process.getuid?.(), 0); + const unit = `telex-test-${crypto.randomUUID()}`; + const legacy = `${unit}-legacy.service`; + const files = [legacy, `${unit}.service`, `${unit}.timer`]; + const ctl = (...args: string[]) => + runCommand("systemctl", args, { cwd: root, timeout: 20_000 }); + try { + await writeFile( + join("/run/systemd/system", legacy), + "[Service]\nExecStart=/bin/sleep infinity\nRestart=always\n[Install]\nWantedBy=multi-user.target\n", + ); + await writeFile(join("/run/systemd/system", `${unit}.service`), units.service); + await writeFile(join("/run/systemd/system", `${unit}.timer`), units.timer); + await writeFile( + script, + updaterScript( + root, + name, + [docker], + [maintenance, "literal $HOME $(touch injected)"], + ` systemctl disable --now ${legacy}`, + ), + ); + await rm(join(root, "service-retired")); + await rm(join(root, "maintenance-needed")); + await writeFile(join(root, "status"), `true false ${oldImage} ${name}`); + const before = await readFile(join(root, "maintenance-calls"), "utf8"); + await ctl("daemon-reload"); + await ctl("enable", "--now", legacy); + await ctl("enable", "--now", `${unit}.timer`); + await ctl("start", `${unit}.service`); + assert.equal( + (await ctl("show", legacy, "-p", "ActiveState", "--value")).stdout.trim(), + "inactive", + ); + assert.equal( + (await ctl("show", legacy, "-p", "UnitFileState", "--value")).stdout.trim(), + "disabled", + ); + assert.equal( + (await ctl("show", `${unit}.service`, "-p", "MainPID", "--value")).stdout.trim(), + "0", + ); + assert.equal( + (await ctl("show", `${unit}.timer`, "-p", "ActiveState", "--value")).stdout.trim(), + "active", + ); + assert.equal(await readFile(join(root, "maintenance-calls"), "utf8"), before); + } finally { + await ctl("disable", "--now", `${unit}.timer`, legacy); + await ctl("stop", `${unit}.service`); + for (const file of files) await rm(join("/run/systemd/system", file), { force: true }); + await ctl("daemon-reload"); + } + } + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + +test("the published Wirebot image boots with migrated data and a simulated Telegram API", { + skip: process.env.TELEX_TEST_DOCKER !== "1", + timeout: 600_000, +}, async () => { + const root = await mkdtemp(join(tmpdir(), "telex-container-test-")); + const docker = new Docker(["docker"]); + const signal = AbortSignal.timeout(500_000); + let name: string | undefined; + let image: string | undefined; + let polls = 0; + const telegram = createServer((request, response) => { + const method = request.url?.split("/").at(-1); + if (method === "getUpdates") polls += 1; + const result = + method === "getMe" + ? { id: 123456, is_bot: true, first_name: "Migration Test", username: "migration_test_bot" } + : method === "getUpdates" + ? [] + : true; + const reply = () => { + response.setHeader("Content-Type", "application/json"); + response.end(JSON.stringify({ ok: true, result })); + }; + if (method === "getUpdates") setTimeout(reply, 100); + else reply(); + }); + await new Promise((resolve) => telegram.listen(0, "0.0.0.0", resolve)); + try { + const address = telegram.address(); + assert(address && typeof address !== "string"); + const config = loadAppConfig({ + TELEGRAM_BOT_TOKEN: "123456:fake-token-for-container-test", + TELEGRAM_ALLOWED_USER_IDS: "123456", + TELEGRAM_API_BASE: `http://host.docker.internal:${address.port}`, + TELEX_DATA_DIR: join(root, "original-data"), + CODEX_WORKSPACE: join(root, "original-data", "workspace"), + TELEX_TUNNEL: "off", + }); + const directory = join(root, "migration"); + await mkdir(directory); + await mkdir(join(config.dataDirectory, "codex-home"), { recursive: true }); + await mkdir(join(config.workspace, ".telex", "automations", "schedule-1"), { recursive: true }); + await writeFile( + join(config.workspace, ".telex", "automations", "schedule-1", "memory.md"), + "preserved memory", + ); + const codexConfig = + 'approval_policy = "on-request"\nsandbox_mode = "workspace-write"\ncli_auth_credentials_store = "file"\n'; + await writeFile(join(config.dataDirectory, "codex-home", "config.toml"), codexConfig); + await atomicWriteJson(join(config.dataDirectory, "conversations.json"), { + version: 2, + conversations: { + "telegram:123456:0": { + activeThreadId: "preserved-thread", + previousThreadIds: ["previous-thread"], + }, + }, + }); + await docker.run(["pull", wirebotImage], signal); + image = await docker.run(["image", "inspect", "--format", "{{.Id}}", wirebotImage], signal); + const state = await prepareMigration(config, directory, docker, image); + name = state.name; + await docker.run( + [ + "run", + "--detach", + ...state.runArguments, + "--add-host", + "host.docker.internal:host-gateway", + image, + ], + signal, + ); + try { + await docker.waitUntilReady(name, false, signal); + } catch (error) { + console.error(await docker.run(["logs", name])); + throw error; + } + assert(polls > 0); + assert.equal( + await docker.run([ + "exec", + name, + "cat", + join(config.workspace, ".wirebot", "automations", "schedule-1", "memory.md"), + ]), + "preserved memory", + ); + const mapping = JSON.parse( + await docker.run(["exec", name, "cat", join(config.dataDirectory, "conversations.json")]), + ); + assert.equal(mapping.conversations["telegram:123456:0"].activeThreadId, "preserved-thread"); + const copiedConfig = await docker.run([ + "exec", + name, + "cat", + join(config.dataDirectory, "codex-home", "config.toml"), + ]); + assert(copiedConfig.includes('sandbox_mode = "danger-full-access"')); + assert(copiedConfig.includes('approval_policy = "on-request"')); + assert.equal( + await readFile(join(config.dataDirectory, "codex-home", "config.toml"), "utf8"), + codexConfig, + ); + await docker.run(["stop", "--time", "60", name]); + await docker.copyData(state.dataDirectory, `${state.dataDirectory}-backup`, image); + await docker.copyData(`${state.dataDirectory}-backup`, state.dataDirectory, image); + await docker.run(["start", name]); + await docker.waitUntilReady(name, false, signal); + } finally { + if (name !== undefined && (await docker.container(name)) !== undefined) + await docker.run(["rm", "--force", name]); + await new Promise((resolve) => telegram.close(() => resolve())); + if (image !== undefined) { + await docker.run([ + "run", + "--rm", + "--network", + "none", + "--user", + "0", + "--entrypoint", + "sh", + "--mount", + bindMount(root, "/cleanup"), + image, + "-c", + "find /cleanup -mindepth 1 -delete", + ]); + } + await rm(root, { recursive: true, force: true }); + } +}); diff --git a/src/update/wirebot.ts b/src/update/wirebot.ts new file mode 100644 index 0000000..736f072 --- /dev/null +++ b/src/update/wirebot.ts @@ -0,0 +1,593 @@ +import { createHash } from "node:crypto"; +import { lookup } from "node:dns/promises"; +import { cp, lstat, mkdir, open, readFile, realpath, rm, symlink } from "node:fs/promises"; +import { createServer } from "node:net"; +import { homedir } from "node:os"; +import { isAbsolute, join, relative, resolve, sep } from "node:path"; +import { parseEnv } from "node:util"; +import { Api } from "grammy"; +import { z } from "zod"; +import { storedStateSchema as automationSchema } from "../automations/store.js"; +import { loadAppConfig } from "../config/env.js"; +import { storedStateSchema as conversationSchema } from "../core/conversation-store.js"; +import { storedSettingsSchema as settingsSchema } from "../core/settings-store.js"; +import { atomicWriteFile, atomicWriteJson } from "../shared/fs.js"; +import { Logger } from "../shared/logger.js"; +import { bindMount, Docker, ensureDocker, wirebotImage } from "./docker.js"; +import { wirebotService } from "./wirebot-service.js"; + +type Config = ReturnType; +const ownerLabel = "io.github.sadfun.telex-migration"; +const stateSchema = z.object({ + version: z.literal(1), + status: z.enum(["prepared", "active"]), + name: z.string().regex(/^telex-wirebot-[a-f0-9]{12}$/), + image: z.string().regex(/^sha256:[a-f0-9]{64}$/), + dockerCommand: z.array(z.string()).min(1), + runArguments: z.array(z.string()), + dataDirectory: z.string().min(1), + updating: z.boolean(), + expectAuthentication: z.boolean(), + notifiedUserIds: z.array(z.number().int().positive()), + notification: z.object({ + token: z.string(), + apiBase: z.string(), + userIds: z.array(z.number().int().positive()), + }), +}); +type MigrationState = z.infer; + +export const migrationMessage = + 'Telex just got a big update! Your instance is now Wirebot, the next evolution of telex.\n\n' + + "Your login, conversations, memory, and schedules are preserved. Slack and Discord are now supported, and updates are automatic. Just keep chatting."; + +export function migrationDirectory(config: Pick): string { + const id = createHash("sha256").update(config.dataDirectory).digest("hex").slice(0, 12); + return join(process.env.XDG_DATA_HOME ?? join(homedir(), ".local", "share"), "telex-wirebot", id); +} + +export async function assertTelexReleaseUpdatesAllowed(): Promise { + const directory = migrationDirectory({ + dataDirectory: resolve(process.env.TELEX_DATA_DIR ?? ".telex"), + }); + if (await exists(join(directory, "migration.json"))) { + throw new Error( + "This instance has moved to Wirebot. A scheduled updater maintains its Docker image; do not roll back Telex against the obsolete source snapshot. See README.md for recovery instructions.", + ); + } +} + +/** Called before starting ANY Telex resources, after the old release has shut down. */ +export async function runWithWirebot(runTelex: () => Promise): Promise { + const config = loadAppConfig(); + const directory = migrationDirectory(config); + const stateFile = join(directory, "migration.json"); + let state = await readState(stateFile); + if (state === undefined && config.migrationMode === "off") return await runTelex(); + const logger = new Logger(config.logLevel, { component: "wirebot-migration" }); + await mkdir(directory, { recursive: true, mode: 0o700 }); + const unlock = await acquireRuntimeLock(join(directory, "runtime.lock")); + const abort = new AbortController(); + const stop = () => abort.abort(); + process.once("SIGINT", stop); + process.once("SIGTERM", stop); + let docker: Docker | undefined; + let activateUpdater: (() => Promise) | undefined; + try { + // Read again under the lock: another launcher may just have completed migration. + state = await readState(stateFile); + if (state === undefined) { + // Old releases have no PID lock. Reserve their HTTP port while taking the snapshot. + const reservation = createServer(); + await new Promise((resolve, reject) => { + reservation.once("error", reject); + reservation.listen(config.port, config.host, resolve); + }); + try { + docker = await ensureDocker(directory, logger, abort.signal); + logger.info("Pulling Wirebot before migrating; original Telex data will be retained"); + await docker.run(["pull", wirebotImage], abort.signal); + const image = await docker.run( + ["image", "inspect", "--format", "{{.Id}}", wirebotImage], + abort.signal, + ); + const prepared = await prepareMigration(config, directory, docker, image); + await atomicWriteJson(stateFile, prepared); + state = prepared; + } catch (error) { + logger.error( + "Wirebot preparation failed; continuing to run Telex. Fix the error and restart the service to retry", + error, + ); + } finally { + await new Promise((resolve) => reservation.close(() => resolve())); + } + } + + if (state !== undefined) { + docker ??= new Docker(state.dockerCommand); + await docker.ensureRunning(abort.signal); + // An active migration must never fall back to the obsolete source snapshot. + if (state.status === "active") { + activateUpdater = await wirebotService.prepare(directory, state, config.updateIntervalMs); + state = await reconcileContainers(docker, stateFile, state); + await startContainer(docker, state, abort.signal); + } else { + try { + // Verify and write the replacement service before crossing the migration boundary. + activateUpdater = await wirebotService.prepare(directory, state, config.updateIntervalMs); + await startContainer(docker, state, abort.signal); + } catch (error) { + // Cleanup must succeed before Telex can resume polling with its original state. + await removeOwnedContainer(docker, state.name); + await rm(stateFile); + state = undefined; + activateUpdater = undefined; + logger.error("Wirebot did not start; the original Telex instance is preserved", error); + } + if (state !== undefined) { + // Once ready, Wirebot may already accept messages. A failed checkpoint must retry + // against this copy, never start Telex with the now-obsolete original data. + const active = { ...state, status: "active" as const }; + await atomicWriteJson(stateFile, active); + state = active; + logger.info("Migrated to Wirebot; handing updates to the operating system's timer", { + directory, + }); + } + } + } + + if (state !== undefined) { + // The first scheduled run delivers the notice, after the replacement service is installed. + await atomicWriteFile(join(directory, "maintenance-needed"), ""); + } else if (!abort.signal.aborted) { + process.off("SIGINT", stop); + process.off("SIGTERM", stop); + return await runTelex(); + } + } finally { + process.off("SIGINT", stop); + process.off("SIGTERM", stop); + try { + if (abort.signal.aborted && state?.status === "prepared" && docker !== undefined) { + await docker.run(["stop", "--time", "60", state.name]); + } + } finally { + await unlock(); + } + } + // Release the migration lock before the new job stops the old, restarting Telex service. + await activateUpdater?.(); + return undefined; +} + +/** Runs only when the shell check detects a changed image or unfinished maintenance. */ +export async function maintainWirebot(directory: string): Promise { + const unlock = await acquireRuntimeLock(join(directory, "runtime.lock")); + const stateFile = join(directory, "migration.json"); + const logger = new Logger("info", { component: "wirebot-updater" }); + const abort = new AbortController(); + const stop = () => abort.abort(); + process.once("SIGINT", stop); + process.once("SIGTERM", stop); + try { + let state = await readState(stateFile); + if (state?.status !== "active") throw new Error("Wirebot migration is not active"); + await atomicWriteFile(join(directory, "maintenance-needed"), ""); + const docker = new Docker(state.dockerCommand); + await docker.ensureRunning(abort.signal); + state = await reconcileContainers(docker, stateFile, state); + await startContainer(docker, state, abort.signal); + state = await notifyMigratedUsers(stateFile, state, logger); + state = await updateWirebot(docker, stateFile, state, abort.signal, logger); + // Keep the marker on any failure, including a crash between the JSON and cache writes. + await atomicWriteFile(join(directory, "current-image"), `${state.image}\n`); + if (state.notification.userIds.every((id) => state.notifiedUserIds.includes(id))) + await rm(join(directory, "maintenance-needed")); + } finally { + process.off("SIGINT", stop); + process.off("SIGTERM", stop); + await unlock(); + } +} + +export async function prepareMigration( + config: Config, + directory: string, + docker: Docker, + image: string, +): Promise { + await mkdir(config.dataDirectory, { recursive: true }); + await mkdir(config.workspace, { recursive: true }); + const sourceData = await realpath(config.dataDirectory); + const sourceWorkspace = await realpath(config.workspace); + const root = await realpath(directory); + if (within(sourceData, root) || within(sourceWorkspace, root)) { + throw new Error( + "The migration directory must be outside Telex's data and workspace; set XDG_DATA_HOME to a separate directory", + ); + } + const data = join(root, `data-${crypto.randomUUID()}`); + await cp(sourceData, data, { recursive: true, preserveTimestamps: true, verbatimSymlinks: true }); + let workspace: string; + if (within(sourceData, sourceWorkspace)) { + workspace = join(data, relative(sourceData, sourceWorkspace)); + } else { + workspace = join(data, ".telex-external-workspace"); + if (await exists(workspace)) + throw new Error("Reserved migration workspace path already exists"); + await cp(sourceWorkspace, workspace, { + recursive: true, + preserveTimestamps: true, + verbatimSymlinks: true, + }); + } + // Both names resolve to the same files, including paths embedded in old scheduled threads. + if (await exists(join(workspace, ".telex"))) { + if (await exists(join(workspace, ".wirebot"))) { + throw new Error( + "Both .telex and .wirebot already exist in the workspace; reconcile them before migrating", + ); + } + await symlink(".telex", join(workspace, ".wirebot")); + } + for (const [name, schema] of [ + ["conversations.json", conversationSchema], + ["settings.json", settingsSchema], + ["automations.json", automationSchema], + ] as const) { + const file = join(data, name); + if (await exists(file)) + await atomicWriteJson(file, schema.parse(JSON.parse(await readFile(file, "utf8")))); + } + const authFile = join(data, "codex-home", "auth.json"); + const auth = (await exists(authFile)) ? JSON.parse(await readFile(authFile, "utf8")) : undefined; + const codexConfig = join(data, "codex-home", "config.toml"); + if (await exists(codexConfig)) { + const contents = await readFile(codexConfig, "utf8"); + const table = contents.search(/^\s*\[/mu); + const rootConfig = table < 0 ? contents : contents.slice(0, table); + // Wirebot uses the container as its sandbox. Nested bubblewrap cannot run in standard Docker. + const migrated = + rootConfig.replace( + /^(\s*sandbox_mode\s*=\s*)(["'])workspace-write\2/gmu, + '$1"danger-full-access"', + ) + (table < 0 ? "" : contents.slice(table)); + if (migrated !== contents) await atomicWriteFile(codexConfig, migrated); + } + if ( + !auth && + (await exists(codexConfig)) && + /^\s*cli_auth_credentials_store\s*=\s*["'](?:keyring|auto)["']/mu.test( + await readFile(codexConfig, "utf8"), + ) + ) { + throw new Error( + 'Codex credentials are in the host keychain. Set cli_auth_credentials_store = "file", sign in, and restart telex before migrating', + ); + } + const environment = await migrationEnvironment(config); + const envFile = join(root, "wirebot.env"); + await atomicWriteFile(envFile, environment); + const mounts = new Map([["/data", data]]); + if ( + [config.dataDirectory, sourceData, config.workspace, sourceWorkspace].some((path) => + within("/root", path), + ) + ) { + // The image's /root is mode 0700; root-owned Telex installs must remain reachable by UID 1000. + await mkdir(join(data, "home"), { recursive: true, mode: 0o700 }); + mounts.set("/root", join(data, "home")); + } + for (const destination of new Set([config.dataDirectory, sourceData])) + mounts.set(destination, data); + for (const destination of new Set([config.workspace, sourceWorkspace])) + mounts.set(destination, workspace); + // Original absolute paths preserve SQLite rollout paths, config references, and thread cwd. + const mountArgs = [...mounts].flatMap(([destination, source]) => { + if ( + [ + "/", + "/etc", + "/usr", + "/bin", + "/opt", + "/opt/wirebot", + "/proc", + "/sys", + "/dev", + "/data/home", + ].some( + (reserved) => + destination === reserved || (reserved !== "/" && within(reserved, destination)), + ) + ) { + throw new Error(`Cannot preserve the host path ${destination} inside the Wirebot image`); + } + return ["--mount", bindMount(source, destination)]; + }); + const host = (await lookup(config.host)).address; + const id = createHash("sha256").update(config.dataDirectory).digest("hex").slice(0, 12); + const name = `telex-wirebot-${id}`; + return stateSchema.parse({ + version: 1, + status: "prepared", + name, + image, + dockerCommand: docker.command, + dataDirectory: data, + updating: false, + runArguments: [ + "--name", + name, + "--label", + `${ownerLabel}=${name}`, + "--restart", + "unless-stopped", + "--log-opt", + "max-size=10m", + "--log-opt", + "max-file=3", + "--env-file", + envFile, + "--publish", + `${host.includes(":") ? `[${host}]` : host}:${config.port}:8787`, + ...mountArgs, + ], + expectAuthentication: Boolean(auth?.tokens?.access_token || auth?.OPENAI_API_KEY), + notifiedUserIds: [], + notification: { + token: config.telegramToken, + apiBase: config.telegramApiBase, + userIds: [...config.allowedUserIds], + }, + }); +} + +export async function migrationEnvironment(config: Config): Promise { + // Retain extra keys from the actual --env-file(s), including custom MCP/provider credentials. + const environment: Record = {}; + for (const argument of process.execArgv) { + const match = /^--env-file(?:-if-exists)?=(.+)$/.exec(argument); + if (match?.[1] !== undefined && (await exists(resolve(match[1])))) { + for (const [key, value] of Object.entries( + parseEnv(await readFile(resolve(match[1]), "utf8")), + )) { + if (value !== undefined) environment[key] = process.env[key] ?? value; + } + } + } + for (const key of Object.keys(environment)) { + if ( + key.startsWith("TELEX_") || + [ + "HOME", + "PATH", + "USER", + "LOGNAME", + "SHELL", + "CODEX_HOME", + "WIREBOT_TOOLCHAINS_DIR", + "WIREBOT_ASSETS_DIR", + ].includes(key) + ) + delete environment[key]; + } + Object.assign(environment, { + TELEGRAM_BOT_TOKEN: config.telegramToken, + TELEGRAM_ALLOWED_USER_IDS: [...config.allowedUserIds].join(","), + TELEGRAM_API_BASE: config.telegramApiBase, + TELEGRAM_POLL_TIMEOUT: String(config.telegramPollTimeout), + WIREBOT_TUNNEL: config.tunnelMode, + WIREBOT_CONTAINER: "1", + CODEX_CHECK_UPDATES: "false", + WIREBOT_DATA_DIR: config.dataDirectory, + CODEX_WORKSPACE: config.workspace, + HOST: "0.0.0.0", + PORT: "8787", + // Startup verification needs Wirebot's structured ready event. + LOG_LEVEL: config.logLevel === "debug" ? "debug" : "info", + ...(config.publicUrl === undefined ? {} : { PUBLIC_URL: config.publicUrl }), + }); + return Object.entries(environment) + .map(([key, value]) => { + if (/[\r\n\0]/u.test(value)) + throw new Error( + `Docker env files cannot preserve multiline ${key}; configure this value as a file before migrating`, + ); + return `${key}=${value}\n`; + }) + .join(""); +} + +export async function updateWirebot( + docker: Docker, + stateFile: string, + state: MigrationState, + signal: AbortSignal, + logger: Logger, +): Promise { + state = await reconcileContainers(docker, stateFile, state); + await docker.run(["pull", wirebotImage], signal); + const image = await docker.run(["image", "inspect", "--format", "{{.Id}}", wirebotImage], signal); + if (image === state.image) return state; + const previous = `${state.name}-previous`; + await ownedContainer(docker, state.name); + try { + await docker.run(["stop", "--time", "60", state.name], signal); + await docker.copyData( + state.dataDirectory, + `${state.dataDirectory}-backup`, + state.image, + signal, + ); + const pending = { ...state, updating: true }; + await atomicWriteJson(stateFile, pending); + state = pending; + await docker.run(["rename", state.name, previous], signal); + await startContainer(docker, { ...state, image }, signal); + } catch (error) { + state = await reconcileContainers(docker, stateFile, state); + await docker.run(["start", state.name]); + throw error; + } + // A ready candidate can already have accepted messages. Retry a failed commit on restart. + const candidate = { ...state, image, updating: false }; + await atomicWriteJson(stateFile, candidate); + state = candidate; + logger.info("Updated Wirebot image", { image }); + await removeOwnedContainer(docker, previous); + return state; +} + +async function reconcileContainers( + docker: Docker, + stateFile: string, + state: MigrationState, +): Promise { + const previous = `${state.name}-previous`; + const prior = await ownedContainer(docker, previous); + const current = await ownedContainer(docker, state.name); + if (state.updating) { + if (prior !== undefined && current?.State.Running && current.Image !== state.image) { + let ready = false; + try { + await docker.waitUntilReady(state.name, false, AbortSignal.timeout(180_000)); + ready = true; + } catch { + // The replacement never became usable; restore its pre-update snapshot below. + } + if (ready) { + state = { ...state, image: current.Image, updating: false }; + await atomicWriteJson(stateFile, state); + await removeOwnedContainer(docker, previous); + return state; + } + } + if (current?.State.Running) await docker.run(["stop", "--time", "60", state.name]); + if (prior !== undefined) { + await removeOwnedContainer(docker, state.name); + await docker.run(["rename", previous, state.name]); + } + await docker.copyData(`${state.dataDirectory}-backup`, state.dataDirectory, state.image); + state = { ...state, updating: false }; + await atomicWriteJson(stateFile, state); + } else if (prior !== undefined && current?.Image === state.image) { + await removeOwnedContainer(docker, previous); + } else if (prior !== undefined) { + await removeOwnedContainer(docker, state.name); + await docker.run(["rename", previous, state.name]); + } + return state; +} + +async function startContainer( + docker: Docker, + state: MigrationState, + signal: AbortSignal, +): Promise { + const container = await ownedContainer(docker, state.name); + if (container === undefined) + await docker.run(["run", "--detach", ...state.runArguments, state.image], signal); + else if (!container.State.Running) await docker.run(["start", state.name], signal); + await docker.waitUntilReady( + state.name, + state.status === "prepared" && state.expectAuthentication, + signal, + ); +} + +async function ownedContainer(docker: Docker, name: string) { + const container = await docker.container(name); + if ( + container !== undefined && + container.Config.Labels?.[ownerLabel] !== name.replace(/-previous$/u, "") + ) { + throw new Error(`Container ${name} belongs to another application; refusing to change it`); + } + return container; +} + +async function removeOwnedContainer(docker: Docker, name: string): Promise { + if ((await ownedContainer(docker, name)) !== undefined) await docker.run(["rm", "--force", name]); +} + +async function notifyMigratedUsers( + file: string, + state: MigrationState, + logger: Logger, +): Promise { + const api = new Api(state.notification.token, { + apiRoot: state.notification.apiBase, + timeoutSeconds: 15, + }); + for (const userId of state.notification.userIds) { + if (state.notifiedUserIds.includes(userId)) continue; + try { + await api.sendMessage(userId, migrationMessage, { + parse_mode: "HTML", + link_preview_options: { is_disabled: true }, + }); + const notified = { ...state, notifiedUserIds: [...state.notifiedUserIds, userId] }; + await atomicWriteJson(file, notified); + state = notified; + } catch { + logger.warn("Could not deliver the Wirebot migration notice; will retry", { userId }); + } + } + return state; +} + +async function readState(file: string): Promise { + return (await exists(file)) + ? stateSchema.parse(JSON.parse(await readFile(file, "utf8"))) + : undefined; +} + +function within(parent: string, child: string): boolean { + const path = relative(parent, child); + return path === "" || (path !== ".." && !path.startsWith(`..${sep}`) && !isAbsolute(path)); +} + +async function exists(file: string): Promise { + try { + await lstat(file); + return true; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return false; + throw error; + } +} + +export async function acquireRuntimeLock(file: string): Promise<() => Promise> { + for (let attempt = 0; attempt < 2; attempt += 1) { + try { + const handle = await open(file, "wx", 0o600); + try { + await handle.writeFile(String(process.pid)); + } finally { + await handle.close(); + } + return async () => { + await rm(file); + }; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; + const pid = Number(await readFile(file, "utf8")); + if (!Number.isSafeInteger(pid) || pid <= 0) + throw new Error(`Invalid migration lock at ${file}; remove it only after stopping telex`); + try { + process.kill(pid, 0); + } catch (probeError) { + if ((probeError as NodeJS.ErrnoException).code === "ESRCH") { + await rm(file); + continue; + } + throw probeError; + } + throw new Error(`Telex or its Wirebot updater is already running (PID ${pid})`); + } + } + throw new Error(`Could not acquire migration lock: ${file}`); +}