From 0ad46672efc310c303a2e4c7ba236338dd7b2e89 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 19 Apr 2026 07:55:15 -0500 Subject: [PATCH] Fix heartbeat runs on legacy SQL_ASCII databases --- .../adapter-utils/src/server-utils.test.ts | 12 ++- packages/adapter-utils/src/server-utils.ts | 11 +++ server/src/adapters/utils.ts | 1 + server/src/services/heartbeat.ts | 91 +++++++++++++++---- server/src/services/instance-settings.ts | 11 ++- vitest.config.ts | 1 + 6 files changed, 107 insertions(+), 20 deletions(-) diff --git a/packages/adapter-utils/src/server-utils.test.ts b/packages/adapter-utils/src/server-utils.test.ts index bcdc4ba09e0..ff910757122 100644 --- a/packages/adapter-utils/src/server-utils.test.ts +++ b/packages/adapter-utils/src/server-utils.test.ts @@ -1,6 +1,6 @@ import { randomUUID } from "node:crypto"; import { describe, expect, it } from "vitest"; -import { runChildProcess } from "./server-utils.js"; +import { appendWithByteCap, runChildProcess } from "./server-utils.js"; function isPidAlive(pid: number) { try { @@ -86,3 +86,13 @@ describe("runChildProcess", () => { expect(await waitForPidExit(descendantPid!, 2_000)).toBe(true); }); }); + +describe("appendWithByteCap", () => { + it("keeps valid UTF-8 when trimming through multibyte text", () => { + const output = appendWithByteCap("prefix ", "hello — world", 7); + + expect(output).not.toContain("\uFFFD"); + expect(Buffer.from(output, "utf8").toString("utf8")).toBe(output); + expect(Buffer.byteLength(output, "utf8")).toBeLessThanOrEqual(7); + }); +}); diff --git a/packages/adapter-utils/src/server-utils.ts b/packages/adapter-utils/src/server-utils.ts index 189a16ac890..24e5ede5a41 100644 --- a/packages/adapter-utils/src/server-utils.ts +++ b/packages/adapter-utils/src/server-utils.ts @@ -180,6 +180,17 @@ export function appendWithCap(prev: string, chunk: string, cap = MAX_CAPTURE_BYT return combined.length > cap ? combined.slice(combined.length - cap) : combined; } +export function appendWithByteCap(prev: string, chunk: string, cap = MAX_CAPTURE_BYTES) { + const combined = prev + chunk; + const bytes = Buffer.byteLength(combined, "utf8"); + if (bytes <= cap) return combined; + + const buffer = Buffer.from(combined, "utf8"); + let start = Math.max(0, bytes - cap); + while (start < buffer.length && (buffer[start]! & 0xc0) === 0x80) start += 1; + return buffer.subarray(start).toString("utf8"); +} + function resumeReadable(readable: { resume: () => unknown; destroyed?: boolean } | null | undefined) { if (!readable || readable.destroyed) return; readable.resume(); diff --git a/server/src/adapters/utils.ts b/server/src/adapters/utils.ts index 96931b0f166..da3767dd270 100644 --- a/server/src/adapters/utils.ts +++ b/server/src/adapters/utils.ts @@ -24,6 +24,7 @@ export const asBoolean = serverUtils.asBoolean; export const asStringArray = serverUtils.asStringArray; export const parseJson = serverUtils.parseJson; export const appendWithCap = serverUtils.appendWithCap; +export const appendWithByteCap = serverUtils.appendWithByteCap; export const resolvePathValue = serverUtils.resolvePathValue; export const renderTemplate = serverUtils.renderTemplate; export const redactEnvForLogs = serverUtils.redactEnvForLogs; diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index d3c68c4be52..e333cefcade 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -25,7 +25,7 @@ import { getRunLogStore, type RunLogHandle } from "./run-log-store.js"; import { getServerAdapter, runningProcesses } from "../adapters/index.js"; import type { AdapterExecutionResult, AdapterInvocationMeta, AdapterSessionCodec, UsageSummary } from "../adapters/index.js"; import { createLocalAgentJwt } from "../agent-auth-jwt.js"; -import { parseObject, asBoolean, asNumber, appendWithCap, MAX_EXCERPT_BYTES } from "../adapters/utils.js"; +import { parseObject, asBoolean, asNumber, appendWithByteCap, MAX_EXCERPT_BYTES } from "../adapters/utils.js"; import { costService } from "./costs.js"; import { trackAgentFirstHeartbeat } from "@paperclipai/shared/telemetry"; import { getTelemetryClient } from "../telemetry.js"; @@ -480,6 +480,15 @@ const heartbeatRunSafeColumns = { resultJson: heartbeatRunSafeResultJsonColumn, } as const; +const heartbeatRunSqlAsciiSafeColumns = { + ...getTableColumns(heartbeatRuns), + processGroupId: heartbeatRunProcessGroupIdColumn, + error: sql`NULL`.as("error"), + resultJson: sql | null>`NULL`.as("resultJson"), + stdoutExcerpt: sql`NULL`.as("stdoutExcerpt"), + stderrExcerpt: sql`NULL`.as("stderrExcerpt"), +} as const; + const heartbeatRunLogAccessColumns = { id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, @@ -500,7 +509,7 @@ const heartbeatRunIssueSummaryColumns = { } as const; function appendExcerpt(prev: string, chunk: string) { - return appendWithCap(prev, chunk, MAX_EXCERPT_BYTES); + return appendWithByteCap(prev, chunk, MAX_EXCERPT_BYTES); } function truncateRunEventString(value: string) { @@ -1609,6 +1618,26 @@ export function heartbeatService(db: Db) { cancelWorkForScope: cancelBudgetScopeWork, }; const budgets = budgetService(db, budgetHooks); + let unsafeTextProjectionPromise: Promise | null = null; + + async function hasUnsafeTextProjectionDatabase() { + if (!unsafeTextProjectionPromise) { + unsafeTextProjectionPromise = db + .execute(sql`select current_setting('server_encoding') as server_encoding`) + .then((rows) => { + const first = Array.isArray(rows) ? rows[0] : null; + const serverEncoding = typeof first === "object" && first !== null + ? (first as Record).server_encoding + : null; + return typeof serverEncoding === "string" && serverEncoding.toUpperCase() === "SQL_ASCII"; + }) + .catch((err) => { + logger.warn({ err }, "failed to inspect database server encoding; using conservative heartbeat result projection"); + return true; + }); + } + return unsafeTextProjectionPromise; + } async function getAgent(agentId: string) { return db @@ -1619,8 +1648,15 @@ export function heartbeatService(db: Db) { } async function getRun(runId: string, opts?: { unsafeFullResultJson?: boolean }) { + const safeForLegacyEncoding = !opts?.unsafeFullResultJson && await hasUnsafeTextProjectionDatabase(); return db - .select(opts?.unsafeFullResultJson ? getTableColumns(heartbeatRuns) : heartbeatRunSafeColumns) + .select( + opts?.unsafeFullResultJson + ? getTableColumns(heartbeatRuns) + : safeForLegacyEncoding + ? heartbeatRunSqlAsciiSafeColumns + : heartbeatRunSafeColumns, + ) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, runId)) .then((rows) => rows[0] ?? null); @@ -5244,12 +5280,21 @@ export function heartbeatService(db: Db) { return { list: async (companyId: string, agentId?: string, limit?: number) => { + const safeForLegacyEncoding = await hasUnsafeTextProjectionDatabase(); const query = db - .select({ - ...heartbeatRunListColumns, - ...heartbeatRunListContextColumns, - ...heartbeatRunListResultColumns, - }) + .select( + safeForLegacyEncoding + ? { + ...heartbeatRunListColumns, + error: sql`NULL`.as("error"), + ...heartbeatRunListContextColumns, + } + : { + ...heartbeatRunListColumns, + ...heartbeatRunListContextColumns, + ...heartbeatRunListResultColumns, + }, + ) .from(heartbeatRuns) .where( agentId @@ -5277,7 +5322,15 @@ export function heartbeatService(db: Db) { resultCostUsd, resultCostUsdCamel, ...rest - } = row; + } = row as typeof row & { + resultSummary?: string | null; + resultResult?: string | null; + resultMessage?: string | null; + resultError?: string | null; + resultTotalCostUsd?: string | null; + resultCostUsd?: string | null; + resultCostUsdCamel?: string | null; + }; return { ...rest, @@ -5291,15 +5344,17 @@ export function heartbeatService(db: Db) { wakeSource: contextWakeSource, wakeTriggerDetail: contextWakeTriggerDetail, }), - resultJson: summarizeHeartbeatRunListResultJson({ - summary: resultSummary, - result: resultResult, - message: resultMessage, - error: resultError, - totalCostUsd: resultTotalCostUsd, - costUsd: resultCostUsd, - costUsdCamel: resultCostUsdCamel, - }), + resultJson: safeForLegacyEncoding + ? null + : summarizeHeartbeatRunListResultJson({ + summary: resultSummary, + result: resultResult, + message: resultMessage, + error: resultError, + totalCostUsd: resultTotalCostUsd, + costUsd: resultCostUsd, + costUsdCamel: resultCostUsdCamel, + }), }; }); }, diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 65a1263226d..06e6f9df577 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -85,7 +85,16 @@ export function instanceSettingsService(db: Db) { }) .returning(); - return created; + if (created) return created; + + const raced = await db + .select() + .from(instanceSettings) + .where(eq(instanceSettings.singletonKey, DEFAULT_SINGLETON_KEY)) + .then((rows) => rows[0] ?? null); + if (raced) return raced; + + throw new Error("Failed to initialize instance settings row"); } return { diff --git a/vitest.config.ts b/vitest.config.ts index 3ec05081bad..e2378ea8e92 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -4,6 +4,7 @@ export default defineConfig({ test: { projects: [ "packages/db", + "packages/adapter-utils", "packages/adapters/codex-local", "packages/adapters/opencode-local", "server",