diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 15dbd92e6e..6de6cbc7ba 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -1308,3 +1308,35 @@ Networking behavior for this smoke script: ### GitHub identity for shared agents See [execution GitHub identity](execution-github-identity.md) for the operation-time credential contract, continuation rules, runtime rollout, and acceptance-test requirements. + + +### Investigating polling load + +The company heartbeat-run and live-run lists load secret registries in one +company-scoped query per response. Registry reads project only +`paperclipSecretRedactions` from the run context. They do not load the full +prompt/context JSON. Decrypted values live only for that request and each run +uses its own registry. + +Hidden browser tabs suspend the company live-events connection and transcript +log reads. Returning to a visible tab refreshes active queries once and resumes +transcript reads from their retained offsets. A queued live-event invalidation +that flushes after the tab hides marks data stale without starting a refetch. +The developer-server health poll also stops in hidden tabs. + +Workspace detail responses share concurrent Git inspections and reuse their +results for up to five seconds after completion. The cache holds at most 256 +entries. Close-readiness checks, the terminal-workspace reaper, and the final +cleanup validation still inspect Git afresh. A display result never authorizes +worktree removal. + +The connection-health sweep selects only due IDs in SQL before applying its +limit. Legacy `paperclip_plugin` placeholder connections are excluded: their +tools run in plugin workers and do not have remote MCP endpoints. These rows +remain available; the sweep does not disable or delete plugin connections. + +When investigating an overloaded instance, distinguish request amplification +from stored configuration problems. Verify connection transport and endpoint +fields before disabling a connection. Verify workspace ownership, active runs, +Git state, and runtime-service readiness before closing a workspace. A missing +URL or old workspace timestamp alone does not prove that a row is disposable. diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs index c2246dc128..e02389e819 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs @@ -5812,7 +5812,9 @@ fn durable_descendant_lineage_survives_capacity_and_provider_restoration() { json!({"text": "Read test context."}), )) .unwrap(); - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + // Persisting 300 descendant notifications can exceed five seconds while + // the other provider tests contend for disk and CPU on a shared runner. + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); let mut completed = false; let mut children = std::collections::BTreeSet::new(); while std::time::Instant::now() < deadline && !completed { @@ -5830,7 +5832,11 @@ fn durable_descendant_lineage_survives_capacity_and_provider_restoration() { } std::thread::sleep(std::time::Duration::from_millis(1)); } - assert!(completed); + assert!( + completed, + "descendant run did not complete; observed {} of 300 children", + children.len() + ); assert_eq!(children.len(), 300); first.shutdown().unwrap(); drop(first); diff --git a/server/src/__tests__/agent-live-run-routes.test.ts b/server/src/__tests__/agent-live-run-routes.test.ts index ef1e4bbdae..55a74aa750 100644 --- a/server/src/__tests__/agent-live-run-routes.test.ts +++ b/server/src/__tests__/agent-live-run-routes.test.ts @@ -37,6 +37,7 @@ const mockInstanceSettingsService = vi.hoisted(() => ({ })); const mockRunSecretRedactionRegistry = vi.hoisted(() => ({ + redactForRuns: vi.fn(async (_companyId: string, values: unknown[]) => values), redactForRun: vi.fn( async (_companyId: string, _runId: string, value: unknown) => value, ), @@ -615,6 +616,8 @@ describe("agent live run routes", () => { expect(res.status, JSON.stringify(res.body)).toBe(200); expect(limit).toHaveBeenCalledWith(50); expect(res.body).toHaveLength(50); + expect(mockRunSecretRedactionRegistry.redactForRuns).toHaveBeenCalledTimes(1); + expect(mockRunSecretRedactionRegistry.redactForRun).not.toHaveBeenCalled(); expect(mockHeartbeatService.buildRunOutputSilence).toHaveBeenCalledTimes( 50, ); @@ -659,6 +662,8 @@ describe("agent live run routes", () => { expect(res.status, JSON.stringify(res.body)).toBe(200); expect(limit).toHaveBeenCalledWith(50); expect(res.body).toHaveLength(50); + expect(mockRunSecretRedactionRegistry.redactForRuns).toHaveBeenCalledTimes(1); + expect(mockRunSecretRedactionRegistry.redactForRun).not.toHaveBeenCalled(); }); it("does not pad with recent runs when no minCount is requested", async () => { diff --git a/server/src/__tests__/agent-secrets-routes.test.ts b/server/src/__tests__/agent-secrets-routes.test.ts index da0543c26c..60375c1400 100644 --- a/server/src/__tests__/agent-secrets-routes.test.ts +++ b/server/src/__tests__/agent-secrets-routes.test.ts @@ -19,6 +19,7 @@ import { secretAccessEvents, } from "@paperclipai/db"; import { LOW_TRUST_REVIEW_PRESET, type AgentApiKeyScope } from "@paperclipai/shared"; +import { REDACTED_EVENT_VALUE } from "../redaction.js"; import { errorHandler } from "../middleware/error-handler.js"; import { secretRoutes } from "../routes/secrets.js"; import { secretService } from "../services/secrets.js"; @@ -248,6 +249,24 @@ describeEmbeddedPostgres("agent secret routes", () => { expect((run.contextSnapshot as { paperclipSecretRedactions: unknown[] }).paperclipSecretRedactions).toHaveLength(1); }); + it("redacts batched runs from projected registries and enforces company scope", async () => { + const first = await seedAgentRun(); + const foreign = await seedAgentRun(); + const registry = createRunSecretRedactionRegistry(db); + await registry.register(first.companyId, first.heartbeatRunId, "first-secret-value"); + await registry.register(foreign.companyId, foreign.heartbeatRunId, "foreign-secret-value"); + const runs = [ + { id: first.heartbeatRunId, text: "first-secret-value foreign-secret-value", createdAt: new Date() }, + { id: foreign.heartbeatRunId, text: "foreign-secret-value", createdAt: new Date() }, + ]; + const redacted = await registry.redactForRuns(first.companyId, runs); + expect(redacted[0].text).toBe(`${REDACTED_EVENT_VALUE} foreign-secret-value`); + expect(redacted[0].createdAt).toEqual(runs[0].createdAt); + expect(redacted[1].text).toBe("foreign-secret-value"); + expect(await registry.redactForRun(first.companyId, first.heartbeatRunId, runs[0].text)) + .toBe(redacted[0].text); + }); + it("denies low-trust, task-bridge, and skill-test callers on both routes", async () => { const lowTrust = await seedAgentRun({ trustPreset: LOW_TRUST_REVIEW_PRESET, diff --git a/server/src/__tests__/run-secret-redaction.test.ts b/server/src/__tests__/run-secret-redaction.test.ts index fa2dd60a06..a4f73bf15c 100644 --- a/server/src/__tests__/run-secret-redaction.test.ts +++ b/server/src/__tests__/run-secret-redaction.test.ts @@ -1,6 +1,8 @@ -import { describe, expect, it } from "vitest"; +import { beforeEach, describe, expect, it, vi } from "vitest"; import { REDACTED_EVENT_VALUE } from "../redaction.js"; -import { redactRegisteredSecretValues } from "../services/run-secret-redaction.js"; +import type { Db } from "@paperclipai/db"; +import { PgDialect } from "drizzle-orm/pg-core"; +import { createRunSecretRedactionRegistry, redactRegisteredSecretValues } from "../services/run-secret-redaction.js"; const secret = "q2a-exact-secret-value"; @@ -76,3 +78,50 @@ describe("registered run secret redaction", () => { expect(result.createdAt.toISOString()).toBe("2026-08-06T12:00:00.000Z"); }); }); + +const { resolveVersion } = vi.hoisted(() => ({ resolveVersion: vi.fn(async ({ material }) => material.value as string) })); +vi.mock("../secrets/provider-registry.js", () => ({ getSecretProvider: () => ({ resolveVersion }) })); + +describe("batched run secret redaction", () => { + beforeEach(() => { resolveVersion.mockClear(); }); + + function fixture(rows: unknown[]) { + const where = vi.fn(async (_predicate: import("drizzle-orm").SQL | undefined) => rows); + const select = vi.fn((_columns: { contextSnapshot: import("drizzle-orm").SQL }) => ({ from: () => ({ where }) })); + return { registry: createRunSecretRedactionRegistry({ select } as unknown as Db), select, where }; + } + + it("reads only registry JSON once for 200 runs and resolves shared secrets once", async () => { + const contextSnapshot = { paperclipSecretRedactions: [{ fingerprintSha256: "shared", material: { value: secret } }] }; + const rows = Array.from({ length: 200 }, (_, i) => ({ id: `run-${i}`, contextSnapshot })); + const { registry, select, where } = fixture(rows); + const result = await registry.redactForRuns("company-1", rows.map(row => ({ ...row, stdoutExcerpt: secret }))); + expect(select).toHaveBeenCalledTimes(1); + expect(resolveVersion).toHaveBeenCalledTimes(1); + expect(result.every(run => run.stdoutExcerpt === REDACTED_EVENT_VALUE)).toBe(true); + expect(result[0].contextSnapshot).toEqual({}); + const dialect = new PgDialect(); + const predicate = dialect.sqlToQuery(where.mock.calls[0][0]); + expect(predicate.params).toContain("company-1"); + expect(predicate.sql).toContain('"company_id"'); + expect(dialect.sqlToQuery(select.mock.calls[0][0].contextSnapshot).sql).toContain("-> 'paperclipSecretRedactions'"); + }); + + it("keeps each run's registry separate and observes new registrations on the next request", async () => { + const rows = [{ id: "a", contextSnapshot: { paperclipSecretRedactions: [{ fingerprintSha256: "one", material: { value: secret } }] } }]; + const { registry } = fixture(rows); + expect(await registry.redactForRuns("company", [{ id: "a", text: secret }, { id: "b", text: secret }])) + .toEqual([{ id: "a", text: REDACTED_EVENT_VALUE }, { id: "b", text: secret }]); + rows[0].contextSnapshot.paperclipSecretRedactions.push({ fingerprintSha256: "two", material: { value: "new-secret" } }); + expect(await registry.redactForRuns("company", [{ id: "a", text: "new-secret" }])) + .toEqual([{ id: "a", text: REDACTED_EVENT_VALUE }]); + }); + + it("does not query for an empty list and fails closed on decryption failure", async () => { + const { registry, select } = fixture([{ id: "a", contextSnapshot: { paperclipSecretRedactions: [{ fingerprintSha256: "one", material: {} }] } }]); + expect(await registry.redactForRuns("company", [])).toEqual([]); + expect(select).not.toHaveBeenCalled(); + resolveVersion.mockRejectedValueOnce(new Error("unavailable")); + await expect(registry.redactForRuns("company", [{ id: "a", text: secret }])).rejects.toThrow("unavailable"); + }); +}); diff --git a/server/src/__tests__/tool-access-service.test.ts b/server/src/__tests__/tool-access-service.test.ts index 6d0f82998b..ab444a0878 100644 --- a/server/src/__tests__/tool-access-service.test.ts +++ b/server/src/__tests__/tool-access-service.test.ts @@ -17094,6 +17094,28 @@ describeEmbeddedPostgres("tool access service", () => { lastHealthAt: new Date(0), }) .returning(); + const [pluginApplication] = await db.insert(toolApplications).values({ + companyId: company.id, + applicationKey: `paperclip_plugin:fixture-${randomUUID()}`, + name: "Plugin placeholder", + type: "paperclip_plugin", + status: "active", + metadata: { source: "plugin_backfill" }, + }).returning(); + const [pluginConnection] = await db.insert(toolConnections).values({ + companyId: company.id, + applicationId: pluginApplication!.id, + name: "Plugin placeholder", + uid: `plugin-${randomUUID()}`, + connectionKind: "managed", + transport: "mcp_remote", + status: "active", + enabled: true, + config: { type: "paperclip_plugin" }, + transportConfig: { type: "paperclip_plugin" }, + healthStatus: "ok", + healthCheckedAt: null, + }).returning(); const connection = await service.createConnection(company.id, { name: "Swept remote", transport: "mcp_remote", @@ -17102,7 +17124,7 @@ describeEmbeddedPostgres("tool access service", () => { status: "active", }); - const sweep = await service.sweepConnectionHealth({ staleAfterMs: 0 }); + const sweep = await service.sweepConnectionHealth({ staleAfterMs: 0, limit: 1 }); const [updatedConnection] = await db .select() .from(toolConnections) @@ -17112,6 +17134,10 @@ describeEmbeddedPostgres("tool access service", () => { .from(toolConnections) .where(eq(toolConnections.id, chatConnection!.id)); + const [untouchedPlugin] = await db.select().from(toolConnections) + .where(eq(toolConnections.id, pluginConnection!.id)); + expect(untouchedPlugin).toMatchObject({ enabled: true, healthStatus: "ok", healthCheckedAt: null }); + expect(sweep).toMatchObject({ checked: 1, healthy: 0, diff --git a/server/src/__tests__/workspace-git-inspection-cache.test.ts b/server/src/__tests__/workspace-git-inspection-cache.test.ts new file mode 100644 index 0000000000..f4703cc225 --- /dev/null +++ b/server/src/__tests__/workspace-git-inspection-cache.test.ts @@ -0,0 +1,44 @@ +import type { ExecutionWorkspace } from "@paperclipai/shared"; +import { afterEach, expect, it, vi } from "vitest"; +import { createWorkspaceGitInspectionCache } from "../services/workspace-git-inspection-cache.js"; + +const workspace = { id: "workspace", companyId: "company", cwd: "/repo", baseRef: "master" } as ExecutionWorkspace; +afterEach(() => vi.useRealTimers()); + +it("coalesces concurrent display reads and expires after five seconds", async () => { + vi.useFakeTimers(); + const inspect = vi.fn(async () => ({ dirty: false })); + const read = createWorkspaceGitInspectionCache(inspect); + await Promise.all(Array.from({ length: 100 }, () => read(workspace))); + expect(inspect).toHaveBeenCalledTimes(1); + await read(workspace); + expect(inspect).toHaveBeenCalledTimes(1); + vi.advanceTimersByTime(5_000); + await read(workspace); + expect(inspect).toHaveBeenCalledTimes(2); + // Callers that authorize cleanup retain the uncached inspector. + await inspect(); + expect(inspect).toHaveBeenCalledTimes(3); +}); + +it("does not share results across companies, paths, base refs or workspace revisions", async () => { + const inspect = vi.fn(async () => null); + const read = createWorkspaceGitInspectionCache(inspect); + await read(workspace); + await read({ ...workspace, companyId: "other" }); + await read({ ...workspace, cwd: "/other" }); + await read({ ...workspace, baseRef: "other" }); + await read({ ...workspace, updatedAt: new Date() }); + expect(inspect).toHaveBeenCalledTimes(5); +}); + +it("retries failed inspections and bounds retained entries", async () => { + const inspect = vi.fn(async () => null).mockRejectedValueOnce(new Error("failed")); + const read = createWorkspaceGitInspectionCache(inspect); + await expect(read(workspace)).rejects.toThrow("failed"); + await read(workspace); + expect(inspect).toHaveBeenCalledTimes(2); + for (let i = 0; i < 256; i++) await read({ ...workspace, id: String(i) }); + await read(workspace); + expect(inspect).toHaveBeenCalledTimes(259); +}); diff --git a/server/src/routes/agents.ts b/server/src/routes/agents.ts index 133477eb13..6efa0888c9 100644 --- a/server/src/routes/agents.ts +++ b/server/src/routes/agents.ts @@ -6219,7 +6219,7 @@ export function agentRoutes( const limit = limitParam ? Math.max(1, Math.min(1000, parseInt(limitParam, 10) || 200)) : undefined; const summary = req.query.summary === "true" || req.query.summary === "1"; const runs = await heartbeat.list(companyId, agentId, limit, { summary }); - res.json(await Promise.all(runs.map((run) => runRedactions.redactForRun(companyId, run.id, run)))); + res.json(await runRedactions.redactForRuns(companyId, runs)); }); router.get("/companies/:companyId/provider-traces", async (req, res) => { @@ -6326,20 +6326,20 @@ export function agentRoutes( const rows = [...liveRuns, ...recentRuns]; const projections = await executionProjectionsForRuns(db, companyId, rows.map(run => run.id)); - res.json(await Promise.all(rows.map(async (run) => runRedactions.redactForRun(companyId, run.id, { + res.json(await runRedactions.redactForRuns(companyId, await Promise.all(rows.map(async (run) => ({ ...heartbeat.decorateActiveRunStatus(run), execution: projections.get(run.id) ?? null, outputSilence: await heartbeat.buildRunOutputSilence(run), - })))); + }))))); return; } const projections = await executionProjectionsForRuns(db, companyId, liveRuns.map(run => run.id)); - res.json(await Promise.all(liveRuns.map(async (run) => runRedactions.redactForRun(companyId, run.id, { + res.json(await runRedactions.redactForRuns(companyId, await Promise.all(liveRuns.map(async (run) => ({ ...heartbeat.decorateActiveRunStatus(run), execution: projections.get(run.id) ?? null, outputSilence: await heartbeat.buildRunOutputSilence(run), - })))); + }))))); }); router.get("/heartbeat-runs/:runId", async (req, res) => { diff --git a/server/src/services/execution-workspaces.ts b/server/src/services/execution-workspaces.ts index 8fa2aaf2c6..c1ec53ffd6 100644 --- a/server/src/services/execution-workspaces.ts +++ b/server/src/services/execution-workspaces.ts @@ -1,3 +1,4 @@ +import { createWorkspaceGitInspectionCache } from "./workspace-git-inspection-cache.js"; import { execFile } from "node:child_process"; import { createHash } from "node:crypto"; import fs from "node:fs/promises"; @@ -1268,7 +1269,12 @@ type WorkspaceOverviewIssueRow = WorkspaceOverviewLinkedIssue & { executionWorkspaceId: string; }; +const inspectGitForDisplay = createWorkspaceGitInspectionCache(inspectGitCloseReadiness); + export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServiceOptions = {}) { + const inspectDisplay = opts.inspectGitCloseReadiness + ? createWorkspaceGitInspectionCache(opts.inspectGitCloseReadiness) + : inspectGitForDisplay; const recoveryActionsSvc = issueRecoveryActionService(db); const resolvePullRequestDetails = opts.resolvePullRequestDetails ?? createPullRequestMergeDetailsResolver(db); const now = opts.now ?? (() => new Date()); @@ -1487,7 +1493,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic async function hydrateWorkspace(row: ExecutionWorkspaceRow, runtimeServices: WorkspaceRuntimeService[] = []) { const workspace = toExecutionWorkspace(row, runtimeServices); - const { git } = await (opts.inspectGitCloseReadiness ?? inspectGitCloseReadiness)(workspace); + const { git } = await inspectDisplay(workspace); const assessment = await assessDelivery(row, git); return toExecutionWorkspace(row, runtimeServices, assessment.deliveryState); } diff --git a/server/src/services/run-secret-redaction.ts b/server/src/services/run-secret-redaction.ts index 4ca7963e95..1dfbdbedc6 100644 --- a/server/src/services/run-secret-redaction.ts +++ b/server/src/services/run-secret-redaction.ts @@ -1,5 +1,5 @@ import { createHash } from "node:crypto"; -import { and, eq, or, sql } from "drizzle-orm"; +import { and, eq, inArray, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { heartbeatRuns } from "@paperclipai/db"; import { REDACTED_EVENT_VALUE } from "../redaction.js"; @@ -7,6 +7,8 @@ import { getSecretProvider } from "../secrets/provider-registry.js"; import type { StoredSecretVersionMaterial } from "../secrets/types.js"; const REGISTRY_KEY = "paperclipSecretRedactions"; +// Project only the registry: run contexts can contain megabytes of prompt data. +const registrySnapshot = sql`jsonb_build_object('paperclipSecretRedactions', ${heartbeatRuns.contextSnapshot} -> 'paperclipSecretRedactions')`; type RegistryEntry = { fingerprintSha256: string; @@ -70,14 +72,14 @@ export function createRunSecretRedactionRegistry(db: Db) { } async function valuesForRun(companyId: string, runId: string) { - const rows = await db.select({ contextSnapshot: heartbeatRuns.contextSnapshot }) + const rows = await db.select({ contextSnapshot: registrySnapshot }) .from(heartbeatRuns) .where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId))); return valuesForRuns(rows); } async function valuesForIssue(companyId: string, issueId: string) { - const rows = await db.select({ contextSnapshot: heartbeatRuns.contextSnapshot }) + const rows = await db.select({ contextSnapshot: registrySnapshot }) .from(heartbeatRuns) .where(and( eq(heartbeatRuns.companyId, companyId), @@ -116,6 +118,27 @@ export function createRunSecretRedactionRegistry(db: Db) { .where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId))); }); }, + redactForRuns: async (companyId: string, runs: T[]): Promise => { + if (runs.length === 0) return []; + const rows = await db.select({ id: heartbeatRuns.id, contextSnapshot: registrySnapshot }) + .from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, companyId), inArray(heartbeatRuns.id, runs.map((run) => run.id)))); + // Resolve each encrypted value once per request, but apply only each run's + // own registry. Do not retain plaintext secrets across requests. + const resolved = new Map>(); + const valuesByRun = new Map(await Promise.all(rows.map(async (row) => { + const values = await Promise.all(registryEntries(row.contextSnapshot).map((entry) => { + let value = resolved.get(entry.fingerprintSha256); + if (!value) { + value = provider.resolveVersion({ material: entry.material, externalRef: null }); + resolved.set(entry.fingerprintSha256, value); + } + return value; + })); + return [row.id, values.sort((a, b) => b.length - a.length)] as const; + }))); + return runs.map((run) => redactRegisteredSecretValues(run, valuesByRun.get(run.id) ?? [])); + }, redactForRun: async (companyId: string, runId: string, value: T): Promise => redactRegisteredSecretValues(value, await valuesForRun(companyId, runId)), redactForIssue: async (companyId: string, issueId: string, value: T): Promise => diff --git a/server/src/services/tool-access.ts b/server/src/services/tool-access.ts index f276559b3f..1b704acf04 100644 --- a/server/src/services/tool-access.ts +++ b/server/src/services/tool-access.ts @@ -15,8 +15,10 @@ import { isNotNull, isNull, lt, + lte, max, ne, + or, sql, } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -8015,26 +8017,32 @@ export function toolAccessService( const staleAfterMs = input.staleAfterMs ?? 15 * 60 * 1000; const limit = input.limit ?? 25; const cutoff = new Date(generatedAt.getTime() - staleAfterMs); - const connections = await db - .select() + // Legacy plugin backfills use a remote transport as a placeholder, but + // their tools run in the plugin worker and have no remote MCP endpoint. + // Select only due IDs in SQL so each scheduler tick does not decode every + // active connection's config and credential metadata. + const due = await db + .select({ id: toolConnections.id }) .from(toolConnections) + .innerJoin(toolApplications, and( + eq(toolApplications.id, toolConnections.applicationId), + eq(toolApplications.companyId, toolConnections.companyId), + )) .where( and( eq(toolConnections.enabled, true), eq(toolConnections.status, "active"), ne(toolConnections.transport, "chat_sdk"), + ne(toolApplications.type, "paperclip_plugin"), + or(isNull(toolConnections.healthCheckedAt), lte(toolConnections.healthCheckedAt, cutoff)), ), ) .orderBy( - asc(toolConnections.healthCheckedAt), + sql`${toolConnections.healthCheckedAt} asc nulls first`, asc(toolConnections.createdAt), - ); - const due = connections - .filter( - (connection) => - !connection.healthCheckedAt || connection.healthCheckedAt <= cutoff, + asc(toolConnections.id), ) - .slice(0, limit); + .limit(limit); let healthy = 0; let failed = 0; const failedConnectionIds: string[] = []; diff --git a/server/src/services/workspace-git-inspection-cache.ts b/server/src/services/workspace-git-inspection-cache.ts new file mode 100644 index 0000000000..89ed9403c7 --- /dev/null +++ b/server/src/services/workspace-git-inspection-cache.ts @@ -0,0 +1,30 @@ +import type { ExecutionWorkspace } from "@paperclipai/shared"; + +/** Short-lived display cache only. Destructive operations must inspect afresh. */ +export function createWorkspaceGitInspectionCache(inspect: (workspace: ExecutionWorkspace) => Promise) { + const entries = new Map }>(); + return (workspace: ExecutionWorkspace): Promise => { + const key = JSON.stringify([ + workspace.companyId, workspace.id, workspace.updatedAt, workspace.providerType, + workspace.providerRef, workspace.cwd, workspace.repoUrl, workspace.baseRef, + workspace.branchName, workspace.metadata, + ]); + const now = Date.now(); + const existing = entries.get(key); + if (existing && existing.expiresAt > now) return existing.promise; + for (const [candidate, entry] of entries) { + if (entry.expiresAt <= now) entries.delete(candidate); + } + if (entries.size >= 256) entries.delete(entries.keys().next().value!); + const entry = { expiresAt: Number.POSITIVE_INFINITY, promise: Promise.resolve().then(() => inspect(workspace)) }; + entries.set(key, entry); + entry.promise = entry.promise.then((result) => { + entry.expiresAt = Date.now() + 5_000; + return result; + }, (error) => { + if (entries.get(key) === entry) entries.delete(key); + throw error; + }); + return entry.promise; + }; +} diff --git a/ui/src/components/Layout.production.tsx b/ui/src/components/Layout.production.tsx index 62af6916b6..fc5e5f01a6 100644 --- a/ui/src/components/Layout.production.tsx +++ b/ui/src/components/Layout.production.tsx @@ -248,7 +248,7 @@ export function Layout() { { devServer?: { enabled?: boolean } } | undefined; return data?.devServer?.enabled ? 2000 : false; }, - refetchIntervalInBackground: true, + refetchIntervalInBackground: false, }); const keyboardShortcutsEnabled = useQuery({ diff --git a/ui/src/components/Layout.tsx b/ui/src/components/Layout.tsx index 704f3c8426..d305599f4c 100644 --- a/ui/src/components/Layout.tsx +++ b/ui/src/components/Layout.tsx @@ -235,7 +235,7 @@ export function Layout() { const data = query.state.data as { devServer?: { enabled?: boolean } } | undefined; return data?.devServer?.enabled ? 2000 : false; }, - refetchIntervalInBackground: true, + refetchIntervalInBackground: false, }); const keyboardShortcutsEnabled = useQuery({ queryKey: queryKeys.instance.generalSettings, diff --git a/ui/src/components/transcript/useLiveRunTranscripts.test.tsx b/ui/src/components/transcript/useLiveRunTranscripts.test.tsx index e78b8077d8..e82c282df4 100644 --- a/ui/src/components/transcript/useLiveRunTranscripts.test.tsx +++ b/ui/src/components/transcript/useLiveRunTranscripts.test.tsx @@ -78,6 +78,7 @@ describe("useLiveRunTranscripts", () => { const OriginalWebSocket = globalThis.WebSocket; beforeEach(() => { + vi.spyOn(document, "visibilityState", "get").mockReturnValue("visible"); FakeWebSocket.instances = []; useQueryMock.mockClear(); logMock.mockReset(); @@ -88,6 +89,45 @@ describe("useLiveRunTranscripts", () => { afterEach(() => { globalThis.WebSocket = OriginalWebSocket; + vi.restoreAllMocks(); + vi.useRealTimers(); + }); + + it("pauses hidden-tab reads and resumes at the retained log offset", async () => { + vi.useFakeTimers(); + const visibility = vi.spyOn(document, "visibilityState", "get").mockReturnValue("hidden"); + logMock.mockResolvedValue({ runId: "run-1", store: "memory", logRef: "log-1", content: "", nextOffset: 42 }); + const runs = [{ id: "run-1", status: "running", adapterType: "codex_local" }]; + function Harness() { + useLiveRunTranscripts({ companyId: "company-1", runs, enableRealtimeUpdates: false }); + return null; + } + const container = document.createElement("div"); + const root = createRoot(container); + try { + await act(async () => root.render()); + await act(async () => vi.advanceTimersByTimeAsync(10_000)); + expect(logMock).not.toHaveBeenCalled(); + expect(FakeWebSocket.instances).toHaveLength(0); + await act(async () => { + visibility.mockReturnValue("visible"); + document.dispatchEvent(new Event("visibilitychange")); + }); + expect(logMock).toHaveBeenCalledTimes(1); + await act(async () => { + visibility.mockReturnValue("hidden"); + document.dispatchEvent(new Event("visibilitychange")); + }); + await act(async () => vi.advanceTimersByTimeAsync(10_000)); + expect(logMock).toHaveBeenCalledTimes(1); + await act(async () => { + visibility.mockReturnValue("visible"); + document.dispatchEvent(new Event("visibilitychange")); + }); + expect(logMock).toHaveBeenLastCalledWith("run-1", 42, 256_000, expect.anything()); + } finally { + await act(async () => root.unmount()); + } }); it("waits for a connecting socket to open before closing it during cleanup", async () => { diff --git a/ui/src/components/transcript/useLiveRunTranscripts.ts b/ui/src/components/transcript/useLiveRunTranscripts.ts index a7fcd6bb8f..cce877edb7 100644 --- a/ui/src/components/transcript/useLiveRunTranscripts.ts +++ b/ui/src/components/transcript/useLiveRunTranscripts.ts @@ -1,3 +1,4 @@ +import { usePageVisibility } from "../../lib/page-visibility"; import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { readTranscriptRequest } from "./read-transcript-request"; import { useQuery } from "@tanstack/react-query"; @@ -102,6 +103,7 @@ export function useLiveRunTranscripts({ }: UseLiveRunTranscriptsOptions) { // Ticker consumers opt into the silent chunk-count cap; full task views use a // byte budget that collapses (not discards) the oldest output when exceeded. + const { visible } = usePageVisibility(); const retentionBudget: ChunkRetentionBudget = useMemo( () => typeof maxChunksPerRun === "number" @@ -293,6 +295,7 @@ export function useLiveRunTranscripts({ }, [normalizedRuns, pruneTick]); useEffect(() => { + if (!visible) return; const readableRuns = normalizedRuns.filter(canReadPersistedLog); if (readableRuns.length === 0) return; @@ -383,10 +386,10 @@ export function useLiveRunTranscripts({ controller.abort(); if (interval !== null) window.clearInterval(interval); }; - }, [enableRealtimeUpdates, logPollIntervalMs, logReadLimitBytes, normalizedRuns, runIdsKey, retryGeneration]); + }, [visible, enableRealtimeUpdates, logPollIntervalMs, logReadLimitBytes, normalizedRuns, runIdsKey, retryGeneration]); useEffect(() => { - if (!enableRealtimeUpdates) return; + if (!visible || !enableRealtimeUpdates) return; if (!companyId || activeRunIds.size === 0) return; let closed = false; @@ -515,7 +518,7 @@ export function useLiveRunTranscripts({ } } }; - }, [activeRunIds, companyId, enableRealtimeUpdates, runById]); + }, [visible, activeRunIds, companyId, enableRealtimeUpdates, runById]); const transcriptByRun = useMemo(() => { const next = new Map(); diff --git a/ui/src/components/useSummaryDraftStream.ts b/ui/src/components/useSummaryDraftStream.ts index 48fe9d09eb..4cd052dd63 100644 --- a/ui/src/components/useSummaryDraftStream.ts +++ b/ui/src/components/useSummaryDraftStream.ts @@ -1,3 +1,4 @@ +import { usePageVisibility } from "@/lib/page-visibility"; import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { useQuery } from "@tanstack/react-query"; import type { LiveEvent, SummarySlotIssueRef } from "@paperclipai/shared"; @@ -62,6 +63,7 @@ export function useSummaryDraftStream( companyId: string | null | undefined, generatingIssue: SummarySlotIssueRef | null, ): SummaryDraftStream { + const { visible } = usePageVisibility(); const issueId = generatingIssue?.id ?? null; const [runId, setRunId] = useState(null); const [chunks, setChunks] = useState([]); @@ -118,9 +120,14 @@ export function useSummaryDraftStream( if (fallbackRunId) setRunId((current) => current ?? fallbackRunId); }, [fallbackRunId]); + useEffect(() => { + logOffsetRef.current = 0; + pendingLogRowsRef.current = new Map(); + }, [runId]); + // Live token deltas over the shared company-events socket. useCompanyLiveEvent((event: LiveEvent) => { - if (!runId) return; + if (!visible || !runId) return; if (event.type !== "heartbeat.run.log") return; const payload = event.payload ?? {}; if (payload.runId !== runId) return; @@ -136,12 +143,12 @@ export function useSummaryDraftStream( // Hydrate already-emitted output and fill any gaps from the persisted run log. useEffect(() => { - if (!runId) return; - logOffsetRef.current = 0; - pendingLogRowsRef.current = new Map(); - + if (!visible || !runId) return; let cancelled = false; + let reading = false; const read = async () => { + if (reading || cancelled) return; + reading = true; try { const result = await heartbeatsApi.log(runId, logOffsetRef.current, LOG_READ_LIMIT_BYTES); if (cancelled) return; @@ -153,6 +160,8 @@ export function useSummaryDraftStream( } } catch { // Ignore transient/404 reads (log not yet flushed, run just started). + } finally { + reading = false; } }; @@ -162,7 +171,7 @@ export function useSummaryDraftStream( cancelled = true; window.clearInterval(interval); }; - }, [runId, appendChunks]); + }, [visible, runId, appendChunks]); const parse = useMemo(() => parseSummaryDraftStream(extractAssistantOutputText(chunks)), [chunks]); diff --git a/ui/src/context/LiveUpdatesProvider.hook.test.tsx b/ui/src/context/LiveUpdatesProvider.hook.test.tsx index 9d1957a3fb..41546fb571 100644 --- a/ui/src/context/LiveUpdatesProvider.hook.test.tsx +++ b/ui/src/context/LiveUpdatesProvider.hook.test.tsx @@ -181,6 +181,24 @@ describe("LiveUpdatesProvider socket run notification scope", () => { }))); } + it("disconnects while hidden and reconciles active queries once on return", async () => { + await receiveStatus({ runId: "child-run", agentId: "child-agent", status: "running" }); + const invalidate = vi.spyOn(queryClient, "invalidateQueries"); + const visibility = vi.spyOn(document, "visibilityState", "get"); + await reactAct(async () => { + visibility.mockReturnValue("hidden"); + document.dispatchEvent(new Event("visibilitychange")); + }); + expect(sockets[0].onmessage).toBeNull(); + invalidate.mockClear(); + await reactAct(async () => { + visibility.mockReturnValue("visible"); + document.dispatchEvent(new Event("visibilitychange")); + }); + await vi.waitFor(() => expect(sockets).toHaveLength(2)); + expect(invalidate).toHaveBeenCalledExactlyOnceWith({ type: "active" }, { cancelRefetch: false }); + }); + it.each(["parent-agent", "child-agent"])("shows an unrelated retryable failure without issueId for %s", async (agentId) => { // Match the retryable broadcast from execution-status-delivery.ts: it has // exact run identity but deliberately omits issueId and provider output. diff --git a/ui/src/context/LiveUpdatesProvider.tsx b/ui/src/context/LiveUpdatesProvider.tsx index 5ddec714f8..b2513f318f 100644 --- a/ui/src/context/LiveUpdatesProvider.tsx +++ b/ui/src/context/LiveUpdatesProvider.tsx @@ -1,3 +1,4 @@ +import { getPageVisibility, usePageVisibility } from "../lib/page-visibility"; import { createContext, useCallback, @@ -1787,6 +1788,8 @@ export const __liveUpdatesTestUtils = { }; export function LiveUpdatesProvider({ children }: { children: ReactNode }) { + const { visible } = usePageVisibility(); + const wasHidden = useRef(!visible); const { selectedCompanyId, selectedCompany } = useCompany(); const queryClient = useQueryClient(); const { pushToast } = useToastActions(); @@ -1853,7 +1856,17 @@ export function LiveUpdatesProvider({ children }: { children: ReactNode }) { }, [currentUserId]); useEffect(() => { + if (!visible) { + wasHidden.current = true; + invalidationBatcher.dispose(); + return; + } if (!canConnectSocket || !liveCompanyId) return; + if (wasHidden.current) { + wasHidden.current = false; + // Reconcile events missed while hidden, including completed runs/issues. + void queryClient.invalidateQueries({ type: "active" }, { cancelRefetch: false }); + } let closed = false; let reconnectAttempt = 0; @@ -1905,6 +1918,7 @@ export function LiveUpdatesProvider({ children }: { children: ReactNode }) { }; nextSocket.onmessage = (message) => { + if (!getPageVisibility().visible) return; const raw = typeof message.data === "string" ? message.data : ""; if (!raw) return; @@ -1961,6 +1975,9 @@ export function LiveUpdatesProvider({ children }: { children: ReactNode }) { closeSocketQuietly(activeSocket, "provider_unmount"); }; }, [ + visible, + invalidationBatcher, + queryClient, coalescingClient, liveCompanyId, pushToast, diff --git a/ui/src/lib/query-invalidation-batcher.test.ts b/ui/src/lib/query-invalidation-batcher.test.ts index 12ce094ac7..2fcb2976ed 100644 --- a/ui/src/lib/query-invalidation-batcher.test.ts +++ b/ui/src/lib/query-invalidation-batcher.test.ts @@ -18,7 +18,17 @@ function fakeClient() { describe("createInvalidationBatcher", () => { beforeEach(() => vi.useFakeTimers()); - afterEach(() => vi.useRealTimers()); + afterEach(() => { vi.useRealTimers(); vi.unstubAllGlobals(); }); + + it("marks queries stale without refetching when the tab hides before flush", async () => { + const { client } = fakeClient(); + const batcher = createInvalidationBatcher(client); + const pending = batcher.schedule({ queryKey: ["dashboard", "c1"] }); + vi.stubGlobal("document", { visibilityState: "hidden" }); + await batcher.flush(); + await pending; + expect(client.invalidateQueries).toHaveBeenCalledExactlyOnceWith({ queryKey: ["dashboard", "c1"], refetchType: "none" }); + }); it("coalesces repeated invalidations of the same key into one call per window", () => { const { client } = fakeClient(); @@ -111,7 +121,7 @@ describe("createInvalidationBatcher", () => { describe("createCoalescingQueryClient", () => { beforeEach(() => vi.useFakeTimers()); - afterEach(() => vi.useRealTimers()); + afterEach(() => { vi.useRealTimers(); vi.unstubAllGlobals(); }); it("batches invalidateQueries but passes other methods straight through", () => { const setQueryData = vi.fn(); diff --git a/ui/src/lib/query-invalidation-batcher.ts b/ui/src/lib/query-invalidation-batcher.ts index d5967aff5c..9d8807ebec 100644 --- a/ui/src/lib/query-invalidation-batcher.ts +++ b/ui/src/lib/query-invalidation-batcher.ts @@ -1,3 +1,4 @@ +import { getPageVisibility } from "./page-visibility"; import type { InvalidateQueryFilters, QueryClient } from "@tanstack/react-query"; /** @@ -62,7 +63,9 @@ export function createInvalidationBatcher( const filtersList = [...pending.values()]; pending.clear(); try { - await Promise.all(filtersList.map((filters) => queryClient.invalidateQueries(filters))); + await Promise.all(filtersList.map((filters) => queryClient.invalidateQueries( + getPageVisibility().visible ? filters : { ...filters, refetchType: "none" }, + ))); } finally { deferred?.resolve(); } diff --git a/ui/src/pages/AgentDetail.log-visibility.test.tsx b/ui/src/pages/AgentDetail.log-visibility.test.tsx new file mode 100644 index 0000000000..b2dca3a615 --- /dev/null +++ b/ui/src/pages/AgentDetail.log-visibility.test.tsx @@ -0,0 +1,63 @@ +// @vitest-environment jsdom +import { act } from "react"; +import { createRoot } from "react-dom/client"; +import type { HeartbeatRun } from "@paperclipai/shared"; +import { afterEach, expect, it, vi } from "vitest"; +import { LogViewer } from "./AgentDetail"; +import { LogViewer as ProductionLogViewer } from "./AgentDetail.production"; + +const { log, empty } = vi.hoisted(() => ({ log: vi.fn(), empty: [] })); +vi.mock("../api/heartbeats", () => ({ heartbeatsApi: { log } })); +vi.mock("@tanstack/react-query", async (original) => ({ + ...await original(), + useQuery: () => ({ data: empty }), +})); +vi.mock("../adapters", () => ({ + getUIAdapter: () => null, + onAdapterChange: () => () => {}, + buildTranscript: (lines: unknown[]) => lines, +})); +vi.mock("../components/transcript/RunTranscriptView", () => ({ + RunTranscriptView: ({ entries }: { entries: Array<{ chunk: string }> }) =>
{entries.map(line => line.chunk).join(" ")}
, +})); +(globalThis as typeof globalThis & { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true; +afterEach(() => { vi.restoreAllMocks(); log.mockReset(); }); + +it.each([LogViewer, ProductionLogViewer])("retains legacy history and reads only the next offset on visibility recovery (%#)", async (Viewer) => { + const visibility = vi.spyOn(document, "visibilityState", "get").mockReturnValue("visible"); + const row = (seq: number, chunk: string) => JSON.stringify({ seq, ts: `2026-09-10T12:00:0${seq}Z`, stream: "stdout", chunk }) + "\n"; + const first = row(1, "retained history"); + const second = row(2, "new output"); + log.mockResolvedValueOnce({ content: first, nextOffset: first.length }); + const run = { id: "run-1", companyId: "company-1", agentId: "agent-1", status: "succeeded", logRef: "log" } as HeartbeatRun; + const container = document.createElement("div"); + document.body.append(container); + const root = createRoot(container); + try { + await act(async () => root.render()); + expect(container.textContent).toContain("retained history"); + await act(async () => { + visibility.mockReturnValue("hidden"); + document.dispatchEvent(new Event("visibilitychange")); + }); + expect(log).toHaveBeenCalledTimes(1); + expect(container.textContent).toContain("retained history"); + let complete!: (value: { content: string; nextOffset: number }) => void; + log.mockImplementationOnce(() => new Promise(resolve => { complete = resolve; })); + await act(async () => { + visibility.mockReturnValue("visible"); + document.dispatchEvent(new Event("visibilitychange")); + }); + expect(container.textContent).toContain("retained history"); + expect(log).toHaveBeenLastCalledWith("run-1", first.length, expect.any(Number)); + await act(async () => complete({ content: second, nextOffset: first.length + second.length })); + expect(container.textContent).toContain("retained history new output"); + log.mockResolvedValueOnce({ content: first, nextOffset: first.length }); + await act(async () => root.render()); + expect(log).toHaveBeenLastCalledWith("run-2", 0, expect.any(Number)); + expect(container.textContent).not.toContain("new output"); + } finally { + await act(async () => root.unmount()); + container.remove(); + } +}); diff --git a/ui/src/pages/AgentDetail.production.tsx b/ui/src/pages/AgentDetail.production.tsx index 49a42258cc..519e6c5254 100644 --- a/ui/src/pages/AgentDetail.production.tsx +++ b/ui/src/pages/AgentDetail.production.tsx @@ -1,3 +1,5 @@ +import { mergeRunLogChunks, readChunkSeq } from "../lib/run-log-chunks"; +import { getPageVisibility, usePageVisibility } from "../lib/page-visibility"; import { useCallback, useEffect, useMemo, useState, useRef } from "react"; import { useParams, useNavigate, Link, Navigate, useBeforeUnload, type NavigateFunction } from "@/lib/router"; import { useQuery, useMutation, useQueryClient, type QueryClient } from "@tanstack/react-query"; @@ -383,6 +385,7 @@ function runMetrics(run: HeartbeatRun) { } export type RunLogChunk = { + seq?: number; ts: string; stream: "stdout" | "stderr" | "system"; chunk: string; @@ -3720,13 +3723,20 @@ function RunDetail({ run: initialRun, agentRouteId, adapterType, adapterConfig } /* ---- Log Viewer ---- */ -function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: string }) { +export function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: string }) { + const { visible } = usePageVisibility(); const [events, setEvents] = useState([]); - const [logLines, setLogLines] = useState>([]); + const [logLines, setLogLines] = useState([]); const [loading, setLoading] = useState(true); const [logLoading, setLogLoading] = useState(!!run.logRef); const [logError, setLogError] = useState(null); - const [logOffset, setLogOffset] = useState(0); + const [logOffset, setLogOffsetState] = useState(0); + const logOffsetRef = useRef(0); + const setLogOffset = useCallback((next: number | ((previous: number) => number)) => { + logOffsetRef.current = typeof next === "function" ? next(logOffsetRef.current) : next; + setLogOffsetState(logOffsetRef.current); + }, []); + const logMergeRefs = useRef({ seenChunkKeys: new Set(), trimmedSeqFloorByRun: new Map() }); const [hasMoreLog, setHasMoreLog] = useState(false); const [loadingMoreLog, setLoadingMoreLog] = useState(false); const [isFollowing, setIsFollowing] = useState(false); @@ -3753,6 +3763,12 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin return err instanceof ApiError && err.status === 404; } + function appendLogLines(incoming: RunLogChunk[]) { + setLogLines((previous) => mergeRunLogChunks(run.id, previous, incoming.map((line) => ({ + ...line, dedupeKey: `log:${run.id}:${line.ts}:${line.stream}:${line.chunk}`, + })), logMergeRefs.current, isLive ? MAX_LIVE_LOG_LINES : Number.POSITIVE_INFINITY).chunks); + } + function appendLogContent(content: string, finalize = false) { if (!content && !finalize) return; const combined = `${pendingLogLineRef.current}${content}`; @@ -3763,18 +3779,18 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin pendingLogLineRef.current = ""; } - const parsed: Array<{ ts: string; stream: "stdout" | "stderr" | "system"; chunk: string }> = []; + const parsed: RunLogChunk[] = []; for (const line of split) { const trimmed = line.trim(); if (!trimmed) continue; try { - const raw = JSON.parse(trimmed) as { ts?: unknown; stream?: unknown; chunk?: unknown }; + const raw = JSON.parse(trimmed) as { ts?: unknown; stream?: unknown; chunk?: unknown; seq?: unknown }; const stream = raw.stream === "stderr" || raw.stream === "system" ? raw.stream : "stdout"; const chunk = typeof raw.chunk === "string" ? raw.chunk : ""; const ts = typeof raw.ts === "string" ? raw.ts : new Date().toISOString(); if (!chunk) continue; - parsed.push({ ts, stream, chunk }); + parsed.push({ ts, stream, chunk, seq: readChunkSeq(raw.seq) }); } catch { // ignore malformed lines } @@ -3783,9 +3799,7 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin if (parsed.length > 0) { // Live runs stream forever, so cap the retained tail. Terminated runs are // paginated by the user via "Load more log" and keep their full history. - setLogLines((prev) => - isLive ? appendCapped(prev, parsed, MAX_LIVE_LOG_LINES) : [...prev, ...parsed], - ); + appendLogLines(parsed); } } @@ -3883,17 +3897,23 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin setIsFollowing((prev) => (prev ? prev : true)); }, [events.length, logLines.length, isLive, getScrollContainer]); - // Fetch persisted shell log + // Reset only when the log source changes, never when visibility changes. useEffect(() => { - let cancelled = false; pendingLogLineRef.current = ""; + logMergeRefs.current = { seenChunkKeys: new Set(), trimmedSeqFloorByRun: new Map() }; seenProgressLogLineKeysRef.current = new Set(); setLogLines([]); setLogOffset(0); setHasMoreLog(false); setLoadingMoreLog(false); setLogError(null); + }, [run.id, run.logRef, setLogOffset]); + // Fetch persisted shell log, retaining partial rows and offsets across hides. + useEffect(() => { + if (!visible) return; + let cancelled = false; + const offset = logOffsetRef.current; if (!run.logRef && !shouldPollShellLog) { setLogLoading(false); return () => { @@ -3904,10 +3924,10 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin setLogLoading(true); const load = async () => { try { - const result = await heartbeatsApi.log(run.id, 0, RUN_LOG_PAGE_BYTES); + const result = await heartbeatsApi.log(run.id, offset, RUN_LOG_PAGE_BYTES); if (cancelled) return; appendLogContent(result.content, result.nextOffset === undefined); - const next = result.nextOffset ?? result.content.length; + const next = result.nextOffset ?? offset + result.content.length; setLogOffset(next); setHasMoreLog(!shouldPollShellLog && result.nextOffset !== undefined); } catch (err) { @@ -3927,7 +3947,7 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin return () => { cancelled = true; }; - }, [run.id, run.logRef, run.logBytes, shouldPollShellLog]); + }, [visible, run.id, run.logRef, run.logBytes, shouldPollShellLog]); async function loadMorePersistedLog() { if (loadingMoreLog || !hasMoreLog) return; @@ -3948,27 +3968,42 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin // Poll for live updates useEffect(() => { - if (!isLive || isStreamingConnected) return; + if (!visible || !isLive || isStreamingConnected) return; + let pending = false; + let cancelled = false; const interval = setInterval(async () => { + if (pending || cancelled || !getPageVisibility().visible) return; + pending = true; const maxSeq = events.length > 0 ? Math.max(...events.map((e) => e.seq)) : 0; try { const newEvents = await heartbeatsApi.events(run.id, maxSeq, 100); + if (cancelled) return; if (newEvents.length > 0) { setEvents((prev) => appendCapped(prev, newEvents, MAX_LIVE_EVENTS)); } } catch { // ignore polling errors + } finally { + pending = false; } }, 2000); - return () => clearInterval(interval); - }, [run.id, isLive, isStreamingConnected, events]); + return () => { + cancelled = true; + clearInterval(interval); + }; + }, [visible, run.id, isLive, isStreamingConnected, events]); // Poll shell log for running runs useEffect(() => { - if (!shouldPollShellLog || isStreamingConnected) return; + if (!visible || !shouldPollShellLog || isStreamingConnected) return; + let pending = false; + let cancelled = false; const interval = setInterval(async () => { + if (pending || cancelled || !getPageVisibility().visible) return; + pending = true; try { const result = await heartbeatsApi.log(run.id, logOffset, 256_000); + if (cancelled) return; if (result.content) { appendLogContent(result.content, result.nextOffset === undefined); } @@ -3980,14 +4015,19 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin } catch (err) { if (isRunLogUnavailable(err)) return; // ignore polling errors + } finally { + pending = false; } }, 2000); - return () => clearInterval(interval); - }, [run.id, shouldPollShellLog, isStreamingConnected, logOffset]); + return () => { + cancelled = true; + clearInterval(interval); + }; + }, [visible, run.id, shouldPollShellLog, isStreamingConnected, logOffset]); // Stream live updates from websocket (primary path for running runs). useEffect(() => { - if (!isLive) return; + if (!visible || !isLive) return; let closed = false; let reconnectTimer: number | null = null; @@ -4031,7 +4071,7 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin const streamRaw = asNonEmptyString(payload.stream); const stream = streamRaw === "stderr" || streamRaw === "system" ? streamRaw : "stdout"; const ts = asNonEmptyString((payload as Record).ts) ?? event.createdAt; - setLogLines((prev) => appendCapped(prev, [{ ts, stream, chunk }], MAX_LIVE_LOG_LINES)); + appendLogLines([{ ts, stream, chunk, seq: readChunkSeq(payload.seq) }]); return; } @@ -4041,7 +4081,7 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin const key = heartbeatProgressLogLineKey(line); if (seenProgressLogLineKeysRef.current.has(key)) return; seenProgressLogLineKeysRef.current.add(key); - setLogLines((prev) => appendCapped(prev, [line], MAX_LIVE_LOG_LINES)); + appendLogLines([line]); return; } @@ -4106,7 +4146,7 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin socket.close(1000, "run_detail_unmount"); } }; - }, [isLive, run.companyId, run.id, run.agentId]); + }, [visible, isLive, run.companyId, run.id, run.agentId]); const censorUsernameInLogs = useQuery({ queryKey: queryKeys.instance.generalSettings, diff --git a/ui/src/pages/AgentDetail.tsx b/ui/src/pages/AgentDetail.tsx index e659c337d4..1ae0704863 100644 --- a/ui/src/pages/AgentDetail.tsx +++ b/ui/src/pages/AgentDetail.tsx @@ -1,3 +1,5 @@ +import { mergeRunLogChunks, readChunkSeq } from "../lib/run-log-chunks"; +import { getPageVisibility, usePageVisibility } from "../lib/page-visibility"; import { useCallback, useEffect, useMemo, useState, useRef } from "react"; import { useParams, useNavigate, Link, Navigate, useBeforeUnload, type NavigateFunction } from "@/lib/router"; import { useQuery, useMutation, useQueryClient, type QueryClient } from "@tanstack/react-query"; @@ -384,6 +386,7 @@ function runMetrics(run: HeartbeatRun) { } export type RunLogChunk = { + seq?: number; ts: string; stream: "stdout" | "stderr" | "system"; chunk: string; @@ -3796,13 +3799,20 @@ function RunDetail({ run: initialRun, agentRouteId, adapterType, adapterConfig } /* ---- Log Viewer ---- */ -function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: string }) { +export function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: string }) { + const { visible } = usePageVisibility(); const [events, setEvents] = useState([]); - const [logLines, setLogLines] = useState>([]); + const [logLines, setLogLines] = useState([]); const [loading, setLoading] = useState(true); const [logLoading, setLogLoading] = useState(!!run.logRef); const [logError, setLogError] = useState(null); - const [logOffset, setLogOffset] = useState(0); + const [logOffset, setLogOffsetState] = useState(0); + const logOffsetRef = useRef(0); + const setLogOffset = useCallback((next: number | ((previous: number) => number)) => { + logOffsetRef.current = typeof next === "function" ? next(logOffsetRef.current) : next; + setLogOffsetState(logOffsetRef.current); + }, []); + const logMergeRefs = useRef({ seenChunkKeys: new Set(), trimmedSeqFloorByRun: new Map() }); const [hasMoreLog, setHasMoreLog] = useState(false); const [loadingMoreLog, setLoadingMoreLog] = useState(false); const [isFollowing, setIsFollowing] = useState(false); @@ -3829,6 +3839,12 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin return err instanceof ApiError && err.status === 404; } + function appendLogLines(incoming: RunLogChunk[]) { + setLogLines((previous) => mergeRunLogChunks(run.id, previous, incoming.map((line) => ({ + ...line, dedupeKey: `log:${run.id}:${line.ts}:${line.stream}:${line.chunk}`, + })), logMergeRefs.current, isLive ? MAX_LIVE_LOG_LINES : Number.POSITIVE_INFINITY).chunks); + } + function appendLogContent(content: string, finalize = false) { if (!content && !finalize) return; const combined = `${pendingLogLineRef.current}${content}`; @@ -3839,18 +3855,18 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin pendingLogLineRef.current = ""; } - const parsed: Array<{ ts: string; stream: "stdout" | "stderr" | "system"; chunk: string }> = []; + const parsed: RunLogChunk[] = []; for (const line of split) { const trimmed = line.trim(); if (!trimmed) continue; try { - const raw = JSON.parse(trimmed) as { ts?: unknown; stream?: unknown; chunk?: unknown }; + const raw = JSON.parse(trimmed) as { ts?: unknown; stream?: unknown; chunk?: unknown; seq?: unknown }; const stream = raw.stream === "stderr" || raw.stream === "system" ? raw.stream : "stdout"; const chunk = typeof raw.chunk === "string" ? raw.chunk : ""; const ts = typeof raw.ts === "string" ? raw.ts : new Date().toISOString(); if (!chunk) continue; - parsed.push({ ts, stream, chunk }); + parsed.push({ ts, stream, chunk, seq: readChunkSeq(raw.seq) }); } catch { // ignore malformed lines } @@ -3859,9 +3875,7 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin if (parsed.length > 0) { // Live runs stream forever, so cap the retained tail. Terminated runs are // paginated by the user via "Load more log" and keep their full history. - setLogLines((prev) => - isLive ? appendCapped(prev, parsed, MAX_LIVE_LOG_LINES) : [...prev, ...parsed], - ); + appendLogLines(parsed); } } @@ -3959,17 +3973,23 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin setIsFollowing((prev) => (prev ? prev : true)); }, [events.length, logLines.length, isLive, getScrollContainer]); - // Fetch persisted shell log + // Reset only when the log source changes, never when visibility changes. useEffect(() => { - let cancelled = false; pendingLogLineRef.current = ""; + logMergeRefs.current = { seenChunkKeys: new Set(), trimmedSeqFloorByRun: new Map() }; seenProgressLogLineKeysRef.current = new Set(); setLogLines([]); setLogOffset(0); setHasMoreLog(false); setLoadingMoreLog(false); setLogError(null); + }, [run.id, run.logRef, setLogOffset]); + // Fetch persisted shell log, retaining partial rows and offsets across hides. + useEffect(() => { + if (!visible) return; + let cancelled = false; + const offset = logOffsetRef.current; if (!run.logRef && !shouldPollShellLog) { setLogLoading(false); return () => { @@ -3980,10 +4000,10 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin setLogLoading(true); const load = async () => { try { - const result = await heartbeatsApi.log(run.id, 0, RUN_LOG_PAGE_BYTES); + const result = await heartbeatsApi.log(run.id, offset, RUN_LOG_PAGE_BYTES); if (cancelled) return; appendLogContent(result.content, result.nextOffset === undefined); - const next = result.nextOffset ?? result.content.length; + const next = result.nextOffset ?? offset + result.content.length; setLogOffset(next); setHasMoreLog(!shouldPollShellLog && result.nextOffset !== undefined); } catch (err) { @@ -4003,7 +4023,7 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin return () => { cancelled = true; }; - }, [run.id, run.logRef, run.logBytes, shouldPollShellLog]); + }, [visible, run.id, run.logRef, run.logBytes, shouldPollShellLog]); async function loadMorePersistedLog() { if (loadingMoreLog || !hasMoreLog) return; @@ -4024,27 +4044,42 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin // Poll for live updates useEffect(() => { - if (!isLive || isStreamingConnected) return; + if (!visible || !isLive || isStreamingConnected) return; + let pending = false; + let cancelled = false; const interval = setInterval(async () => { + if (pending || cancelled || !getPageVisibility().visible) return; + pending = true; const maxSeq = events.length > 0 ? Math.max(...events.map((e) => e.seq)) : 0; try { const newEvents = await heartbeatsApi.events(run.id, maxSeq, 100); + if (cancelled) return; if (newEvents.length > 0) { setEvents((prev) => appendCapped(prev, newEvents, MAX_LIVE_EVENTS)); } } catch { // ignore polling errors + } finally { + pending = false; } }, 2000); - return () => clearInterval(interval); - }, [run.id, isLive, isStreamingConnected, events]); + return () => { + cancelled = true; + clearInterval(interval); + }; + }, [visible, run.id, isLive, isStreamingConnected, events]); // Poll shell log for running runs useEffect(() => { - if (!shouldPollShellLog || isStreamingConnected) return; + if (!visible || !shouldPollShellLog || isStreamingConnected) return; + let pending = false; + let cancelled = false; const interval = setInterval(async () => { + if (pending || cancelled || !getPageVisibility().visible) return; + pending = true; try { const result = await heartbeatsApi.log(run.id, logOffset, 256_000); + if (cancelled) return; if (result.content) { appendLogContent(result.content, result.nextOffset === undefined); } @@ -4056,14 +4091,19 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin } catch (err) { if (isRunLogUnavailable(err)) return; // ignore polling errors + } finally { + pending = false; } }, 2000); - return () => clearInterval(interval); - }, [run.id, shouldPollShellLog, isStreamingConnected, logOffset]); + return () => { + cancelled = true; + clearInterval(interval); + }; + }, [visible, run.id, shouldPollShellLog, isStreamingConnected, logOffset]); // Stream live updates from websocket (primary path for running runs). useEffect(() => { - if (!isLive) return; + if (!visible || !isLive) return; let closed = false; let reconnectTimer: number | null = null; @@ -4107,7 +4147,7 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin const streamRaw = asNonEmptyString(payload.stream); const stream = streamRaw === "stderr" || streamRaw === "system" ? streamRaw : "stdout"; const ts = asNonEmptyString((payload as Record).ts) ?? event.createdAt; - setLogLines((prev) => appendCapped(prev, [{ ts, stream, chunk }], MAX_LIVE_LOG_LINES)); + appendLogLines([{ ts, stream, chunk, seq: readChunkSeq(payload.seq) }]); return; } @@ -4117,7 +4157,7 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin const key = heartbeatProgressLogLineKey(line); if (seenProgressLogLineKeysRef.current.has(key)) return; seenProgressLogLineKeysRef.current.add(key); - setLogLines((prev) => appendCapped(prev, [line], MAX_LIVE_LOG_LINES)); + appendLogLines([line]); return; } @@ -4182,7 +4222,7 @@ function LogViewer({ run, adapterType }: { run: HeartbeatRun; adapterType: strin socket.close(1000, "run_detail_unmount"); } }; - }, [isLive, run.companyId, run.id, run.agentId]); + }, [visible, isLive, run.companyId, run.id, run.agentId]); const censorUsernameInLogs = useQuery({ queryKey: queryKeys.instance.generalSettings,