mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-24 18:19:05 +02:00
fix(heartbeat): compact run logs and trim polling noise
This commit is contained in:
1 parent
42544053c4
commit
0c3120d372
11 files changed
+146
-14
No files matched your search
@@ -0,0 +1,24 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { compactRunLogChunk } from "../services/heartbeat.js";
|
||||
|
||||
describe("compactRunLogChunk", () => {
|
||||
it("redacts inline base64 image data from structured log chunks", () => {
|
||||
const base64 = "A".repeat(4096);
|
||||
const chunk = `{"type":"user","message":{"content":[{"type":"image","source":{"type":"base64","data":"${base64}"}}]}}\n`;
|
||||
|
||||
const compacted = compactRunLogChunk(chunk);
|
||||
|
||||
expect(compacted).not.toContain(base64);
|
||||
expect(compacted).toContain("[omitted base64 image data: 4096 chars]");
|
||||
});
|
||||
|
||||
it("truncates oversized chunks after sanitizing them", () => {
|
||||
const chunk = `${"x".repeat(90_000)}tail`;
|
||||
|
||||
const compacted = compactRunLogChunk(chunk, 16_384);
|
||||
|
||||
expect(compacted.length).toBeLessThan(chunk.length);
|
||||
expect(compacted).toContain("[paperclip truncated run log chunk:");
|
||||
expect(compacted.endsWith("tail")).toBe(true);
|
||||
});
|
||||
});
|
||||
@@ -8,6 +8,20 @@ describe("shouldSilenceHttpSuccessLog", () => {
|
||||
|
||||
it("silences successful polling endpoints", () => {
|
||||
expect(shouldSilenceHttpSuccessLog("GET", "/api/health", 200)).toBe(true);
|
||||
expect(
|
||||
shouldSilenceHttpSuccessLog(
|
||||
"GET",
|
||||
"/api/companies/5cbe79ee-acb3-4597-896e-7662742593cd/heartbeat-runs",
|
||||
200,
|
||||
),
|
||||
).toBe(true);
|
||||
expect(
|
||||
shouldSilenceHttpSuccessLog(
|
||||
"GET",
|
||||
"/api/heartbeat-runs/b7044268-19b6-4b3a-a9f3-9c57dce70253/log?offset=1103894&limitBytes=256000",
|
||||
200,
|
||||
),
|
||||
).toBe(true);
|
||||
expect(
|
||||
shouldSilenceHttpSuccessLog(
|
||||
"GET",
|
||||
@@ -22,6 +36,20 @@ describe("shouldSilenceHttpSuccessLog", () => {
|
||||
200,
|
||||
),
|
||||
).toBe(true);
|
||||
expect(
|
||||
shouldSilenceHttpSuccessLog(
|
||||
"GET",
|
||||
"/api/companies/5cbe79ee-acb3-4597-896e-7662742593cd/issues?includeRoutineExecutions=true",
|
||||
200,
|
||||
),
|
||||
).toBe(true);
|
||||
expect(
|
||||
shouldSilenceHttpSuccessLog(
|
||||
"GET",
|
||||
"/api/companies/5cbe79ee-acb3-4597-896e-7662742593cd/activity",
|
||||
200,
|
||||
),
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
it("silences successful static asset requests", () => {
|
||||
|
||||
@@ -2,8 +2,13 @@ const SILENCED_SUCCESS_METHODS = new Set(["GET", "HEAD"]);
|
||||
|
||||
const SILENCED_SUCCESS_API_PATHS = [
|
||||
/^\/api\/health(?:\/|$)/,
|
||||
/^\/api\/companies\/[^/]+\/activity(?:\/|$)/,
|
||||
/^\/api\/companies\/[^/]+\/dashboard(?:\/|$)/,
|
||||
/^\/api\/companies\/[^/]+\/heartbeat-runs(?:\/|$)/,
|
||||
/^\/api\/companies\/[^/]+\/issues(?:\/|$)/,
|
||||
/^\/api\/companies\/[^/]+\/live-runs(?:\/|$)/,
|
||||
/^\/api\/companies\/[^/]+\/sidebar-badges(?:\/|$)/,
|
||||
/^\/api\/heartbeat-runs\/[^/]+\/log(?:\/|$)/,
|
||||
];
|
||||
|
||||
const SILENCED_SUCCESS_STATIC_PREFIXES = [
|
||||
|
||||
@@ -1433,7 +1433,7 @@ export function agentRoutes(db: Db) {
|
||||
});
|
||||
const telemetryClient = getTelemetryClient();
|
||||
if (telemetryClient) {
|
||||
trackAgentCreated(telemetryClient, { agentRole: agent.role });
|
||||
trackAgentCreated(telemetryClient, { agentRole: agent.role, agentId: agent.id });
|
||||
}
|
||||
|
||||
await applyDefaultAgentTaskAssignGrant(
|
||||
@@ -1521,7 +1521,7 @@ export function agentRoutes(db: Db) {
|
||||
});
|
||||
const telemetryClient = getTelemetryClient();
|
||||
if (telemetryClient) {
|
||||
trackAgentCreated(telemetryClient, { agentRole: agent.role });
|
||||
trackAgentCreated(telemetryClient, { agentRole: agent.role, agentId: agent.id });
|
||||
}
|
||||
|
||||
await applyDefaultAgentTaskAssignGrant(
|
||||
|
||||
@@ -75,6 +75,7 @@ import {
|
||||
import { extractSkillMentionIds } from "@paperclipai/shared";
|
||||
|
||||
const MAX_LIVE_LOG_CHUNK_BYTES = 8 * 1024;
|
||||
const MAX_PERSISTED_LOG_CHUNK_CHARS = 64 * 1024;
|
||||
const HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT = 1;
|
||||
const HEARTBEAT_MAX_CONCURRENT_RUNS_MAX = 10;
|
||||
const DEFERRED_WAKE_CONTEXT_KEY = "_paperclipWakeContext";
|
||||
@@ -98,6 +99,7 @@ const SESSIONED_LOCAL_ADAPTERS = new Set([
|
||||
"opencode_local",
|
||||
"pi_local",
|
||||
]);
|
||||
const INLINE_BASE64_IMAGE_DATA_RE = /("type":"image","source":\{"type":"base64","data":")([A-Za-z0-9+/=]{1024,})(")/g;
|
||||
|
||||
type RuntimeConfigSecretResolver = Pick<
|
||||
ReturnType<typeof secretService>,
|
||||
@@ -413,6 +415,23 @@ function appendExcerpt(prev: string, chunk: string) {
|
||||
return appendWithCap(prev, chunk, MAX_EXCERPT_BYTES);
|
||||
}
|
||||
|
||||
function redactInlineBase64ImageData(chunk: string) {
|
||||
return chunk.replace(INLINE_BASE64_IMAGE_DATA_RE, (_match, prefix: string, data: string, suffix: string) =>
|
||||
`${prefix}[omitted base64 image data: ${data.length} chars]${suffix}`,
|
||||
);
|
||||
}
|
||||
|
||||
export function compactRunLogChunk(chunk: string, maxChars = MAX_PERSISTED_LOG_CHUNK_CHARS) {
|
||||
const normalized = redactInlineBase64ImageData(chunk);
|
||||
if (normalized.length <= maxChars) return normalized;
|
||||
|
||||
const headChars = Math.max(0, Math.floor(maxChars * 0.6));
|
||||
const tailChars = Math.max(0, Math.floor(maxChars * 0.25));
|
||||
const omittedChars = Math.max(0, normalized.length - headChars - tailChars);
|
||||
const marker = `\n[paperclip truncated run log chunk: omitted ${omittedChars} chars]\n`;
|
||||
return `${normalized.slice(0, headChars)}${marker}${normalized.slice(normalized.length - tailChars)}`;
|
||||
}
|
||||
|
||||
function normalizeMaxConcurrentRuns(value: unknown) {
|
||||
const parsed = Math.floor(asNumber(value, HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT));
|
||||
if (!Number.isFinite(parsed)) return HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT;
|
||||
@@ -2517,7 +2536,7 @@ export function heartbeatService(db: Db) {
|
||||
|
||||
if (isFirstHeartbeat && updated) {
|
||||
const tc = getTelemetryClient();
|
||||
if (tc) trackAgentFirstHeartbeat(tc, { agentRole: updated.role });
|
||||
if (tc) trackAgentFirstHeartbeat(tc, { agentRole: updated.role, agentId: updated.id });
|
||||
}
|
||||
|
||||
if (updated) {
|
||||
@@ -3532,7 +3551,9 @@ export function heartbeatService(db: Db) {
|
||||
|
||||
const currentUserRedactionOptions = await getCurrentUserRedactionOptions();
|
||||
const onLog = async (stream: "stdout" | "stderr", chunk: string) => {
|
||||
const sanitizedChunk = redactCurrentUserText(chunk, currentUserRedactionOptions);
|
||||
const sanitizedChunk = compactRunLogChunk(
|
||||
redactCurrentUserText(chunk, currentUserRedactionOptions),
|
||||
);
|
||||
if (stream === "stdout") stdoutExcerpt = appendExcerpt(stdoutExcerpt, sanitizedChunk);
|
||||
if (stream === "stderr") stderrExcerpt = appendExcerpt(stderrExcerpt, sanitizedChunk);
|
||||
const ts = new Date().toISOString();
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
import { act } from "react";
|
||||
import { createRoot } from "react-dom/client";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { ApiError } from "../../api/client";
|
||||
import { useLiveRunTranscripts } from "./useLiveRunTranscripts";
|
||||
|
||||
const { useQueryMock, logMock } = vi.hoisted(() => ({
|
||||
@@ -188,4 +189,40 @@ describe("useLiveRunTranscripts", () => {
|
||||
});
|
||||
container.remove();
|
||||
});
|
||||
|
||||
it("stops retrying terminal runs whose persisted log never existed", async () => {
|
||||
logMock.mockReset();
|
||||
logMock.mockRejectedValue(new ApiError("Run log not found", 404, { error: "Run log not found" }));
|
||||
|
||||
function Harness() {
|
||||
useLiveRunTranscripts({
|
||||
companyId: "company-1",
|
||||
runs: [{ id: "run-404", status: "failed", adapterType: "codex_local" }],
|
||||
});
|
||||
return null;
|
||||
}
|
||||
|
||||
const container = document.createElement("div");
|
||||
document.body.appendChild(container);
|
||||
const root = createRoot(container);
|
||||
|
||||
await act(async () => {
|
||||
root.render(<Harness />);
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
expect(logMock).toHaveBeenCalledTimes(1);
|
||||
|
||||
await act(async () => {
|
||||
root.render(<Harness />);
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
expect(logMock).toHaveBeenCalledTimes(1);
|
||||
|
||||
act(() => {
|
||||
root.unmount();
|
||||
});
|
||||
container.remove();
|
||||
});
|
||||
});
|
||||
@@ -1,6 +1,7 @@
|
||||
import { useEffect, useMemo, useRef, useState } from "react";
|
||||
import { useQuery } from "@tanstack/react-query";
|
||||
import type { LiveEvent } from "@paperclipai/shared";
|
||||
import { ApiError } from "../../api/client";
|
||||
import { instanceSettingsApi } from "../../api/instanceSettings";
|
||||
import { heartbeatsApi } from "../../api/heartbeats";
|
||||
import { buildTranscript, getUIAdapter, onAdapterChange, type RunLogChunk, type TranscriptEntry } from "../../adapters";
|
||||
@@ -85,6 +86,7 @@ export function useLiveRunTranscripts({
|
||||
const seenChunkKeysRef = useRef(new Set<string>());
|
||||
const pendingLogRowsByRunRef = useRef(new Map<string, string>());
|
||||
const logOffsetByRunRef = useRef(new Map<string, number>());
|
||||
const missingTerminalLogRunIdsRef = useRef(new Set<string>());
|
||||
// Tick counter to force transcript recomputation when dynamic parser loads
|
||||
const [parserTick, setParserTick] = useState(0);
|
||||
useEffect(() => {
|
||||
@@ -160,6 +162,11 @@ export function useLiveRunTranscripts({
|
||||
logOffsetByRunRef.current.delete(runId);
|
||||
}
|
||||
}
|
||||
for (const runId of missingTerminalLogRunIdsRef.current.keys()) {
|
||||
if (!knownRunIds.has(runId)) {
|
||||
missingTerminalLogRunIdsRef.current.delete(runId);
|
||||
}
|
||||
}
|
||||
}, [normalizedRuns]);
|
||||
|
||||
useEffect(() => {
|
||||
@@ -168,6 +175,9 @@ export function useLiveRunTranscripts({
|
||||
let cancelled = false;
|
||||
|
||||
const readRunLog = async (run: RunTranscriptSource) => {
|
||||
if (missingTerminalLogRunIdsRef.current.has(run.id)) {
|
||||
return;
|
||||
}
|
||||
const offset = logOffsetByRunRef.current.get(run.id) ?? 0;
|
||||
try {
|
||||
const result = await heartbeatsApi.log(run.id, offset, LOG_READ_LIMIT_BYTES);
|
||||
@@ -182,8 +192,10 @@ export function useLiveRunTranscripts({
|
||||
if (result.content.length > 0) {
|
||||
logOffsetByRunRef.current.set(run.id, offset + result.content.length);
|
||||
}
|
||||
} catch {
|
||||
// Ignore log read errors while output is initializing.
|
||||
} catch (error) {
|
||||
if (error instanceof ApiError && error.status === 404 && isTerminalStatus(run.status)) {
|
||||
missingTerminalLogRunIdsRef.current.add(run.id);
|
||||
}
|
||||
} finally {
|
||||
if (!cancelled) {
|
||||
setHydratedRunIds((prev) => {
|
||||
|
||||
@@ -20,6 +20,7 @@ import {
|
||||
} from "../lib/inbox";
|
||||
|
||||
const INBOX_ISSUE_STATUSES = "backlog,todo,in_progress,in_review,blocked,done";
|
||||
const INBOX_BADGE_HEARTBEAT_RUN_LIMIT = 200;
|
||||
|
||||
export function useDismissedInboxAlerts() {
|
||||
const [dismissed, setDismissed] = useState<Set<string>>(loadDismissedInboxAlerts);
|
||||
@@ -181,8 +182,8 @@ export function useInboxBadge(companyId: string | null | undefined) {
|
||||
const mineIssues = useMemo(() => getRecentTouchedIssues(mineIssuesRaw), [mineIssuesRaw]);
|
||||
|
||||
const { data: heartbeatRuns = [] } = useQuery({
|
||||
queryKey: queryKeys.heartbeats(companyId!),
|
||||
queryFn: () => heartbeatsApi.list(companyId!),
|
||||
queryKey: [...queryKeys.heartbeats(companyId!), "limit", INBOX_BADGE_HEARTBEAT_RUN_LIMIT],
|
||||
queryFn: () => heartbeatsApi.list(companyId!, undefined, INBOX_BADGE_HEARTBEAT_RUN_LIMIT),
|
||||
enabled: !!companyId,
|
||||
});
|
||||
|
||||
|
||||
@@ -81,8 +81,8 @@ export function Agents() {
|
||||
});
|
||||
|
||||
const { data: runs } = useQuery({
|
||||
queryKey: queryKeys.heartbeats(selectedCompanyId!),
|
||||
queryFn: () => heartbeatsApi.list(selectedCompanyId!),
|
||||
queryKey: [...queryKeys.liveRuns(selectedCompanyId!), "agents-page"],
|
||||
queryFn: () => heartbeatsApi.liveRunsForCompany(selectedCompanyId!),
|
||||
enabled: !!selectedCompanyId,
|
||||
refetchInterval: 15_000,
|
||||
});
|
||||
|
||||
@@ -26,6 +26,8 @@ import { PageSkeleton } from "../components/PageSkeleton";
|
||||
import type { Agent, Issue } from "@paperclipai/shared";
|
||||
import { PluginSlotOutlet } from "@/plugins/slots";
|
||||
|
||||
const DASHBOARD_HEARTBEAT_RUN_LIMIT = 100;
|
||||
|
||||
function getRecentIssues(issues: Issue[]): Issue[] {
|
||||
return [...issues]
|
||||
.sort((a, b) => new Date(b.updatedAt).getTime() - new Date(a.updatedAt).getTime());
|
||||
@@ -75,8 +77,8 @@ export function Dashboard() {
|
||||
});
|
||||
|
||||
const { data: runs } = useQuery({
|
||||
queryKey: queryKeys.heartbeats(selectedCompanyId!),
|
||||
queryFn: () => heartbeatsApi.list(selectedCompanyId!),
|
||||
queryKey: [...queryKeys.heartbeats(selectedCompanyId!), "limit", DASHBOARD_HEARTBEAT_RUN_LIMIT],
|
||||
queryFn: () => heartbeatsApi.list(selectedCompanyId!, undefined, DASHBOARD_HEARTBEAT_RUN_LIMIT),
|
||||
enabled: !!selectedCompanyId,
|
||||
});
|
||||
|
||||
|
||||
@@ -87,6 +87,8 @@ import {
|
||||
Search,
|
||||
ListTree,
|
||||
} from "lucide-react";
|
||||
|
||||
const INBOX_HEARTBEAT_RUN_LIMIT = 200;
|
||||
import { Input } from "@/components/ui/input";
|
||||
import { PageTabBar } from "../components/PageTabBar";
|
||||
import type { Approval, HeartbeatRun, Issue, JoinRequest } from "@paperclipai/shared";
|
||||
@@ -788,8 +790,8 @@ export function Inbox() {
|
||||
});
|
||||
|
||||
const { data: heartbeatRuns, isLoading: isRunsLoading } = useQuery({
|
||||
queryKey: queryKeys.heartbeats(selectedCompanyId!),
|
||||
queryFn: () => heartbeatsApi.list(selectedCompanyId!),
|
||||
queryKey: [...queryKeys.heartbeats(selectedCompanyId!), "limit", INBOX_HEARTBEAT_RUN_LIMIT],
|
||||
queryFn: () => heartbeatsApi.list(selectedCompanyId!, undefined, INBOX_HEARTBEAT_RUN_LIMIT),
|
||||
enabled: !!selectedCompanyId,
|
||||
});
|
||||
|
||||
|
||||
Reference in new issue
Block a user