diff --git a/packages/adapter-utils/src/server-utils.ts b/packages/adapter-utils/src/server-utils.ts index ba44249fbd0..189a16ac890 100644 --- a/packages/adapter-utils/src/server-utils.ts +++ b/packages/adapter-utils/src/server-utils.ts @@ -180,6 +180,11 @@ export function appendWithCap(prev: string, chunk: string, cap = MAX_CAPTURE_BYT return combined.length > cap ? combined.slice(combined.length - cap) : combined; } +function resumeReadable(readable: { resume: () => unknown; destroyed?: boolean } | null | undefined) { + if (!readable || readable.destroyed) return; + readable.resume(); +} + export function resolvePathValue(obj: Record, dottedPath: string) { const parts = dottedPath.split("."); let cursor: unknown = obj; @@ -1134,19 +1139,25 @@ export async function runChildProcess( : null; child.stdout?.on("data", (chunk: unknown) => { + const readable = child.stdout; + readable?.pause(); const text = String(chunk); stdout = appendWithCap(stdout, text); logChain = logChain .then(() => opts.onLog("stdout", text)) - .catch((err) => onLogError(err, runId, "failed to append stdout log chunk")); + .catch((err) => onLogError(err, runId, "failed to append stdout log chunk")) + .finally(() => resumeReadable(readable)); }); child.stderr?.on("data", (chunk: unknown) => { + const readable = child.stderr; + readable?.pause(); const text = String(chunk); stderr = appendWithCap(stderr, text); logChain = logChain .then(() => opts.onLog("stderr", text)) - .catch((err) => onLogError(err, runId, "failed to append stderr log chunk")); + .catch((err) => onLogError(err, runId, "failed to append stderr log chunk")) + .finally(() => resumeReadable(readable)); }); const stdin = child.stdin; diff --git a/server/src/__tests__/agent-permissions-routes.test.ts b/server/src/__tests__/agent-permissions-routes.test.ts index eed083b573f..284b21b34b9 100644 --- a/server/src/__tests__/agent-permissions-routes.test.ts +++ b/server/src/__tests__/agent-permissions-routes.test.ts @@ -109,6 +109,7 @@ function registerModuleMocks() { companySkillService: () => mockCompanySkillService, budgetService: () => mockBudgetService, heartbeatService: () => mockHeartbeatService, + ISSUE_LIST_DEFAULT_LIMIT: 500, issueApprovalService: () => mockIssueApprovalService, issueService: () => mockIssueService, logActivity: mockLogActivity, @@ -674,6 +675,12 @@ describe("agent permission routes", () => { status: "todo", }, ]); + expect(mockIssueService.list).toHaveBeenCalledWith(companyId, { + touchedByUserId: "board-user", + inboxArchivedByUserId: "board-user", + status: "backlog,todo,in_progress,in_review,blocked,done", + limit: 500, + }); }); it("rejects heartbeat cancellation outside the caller company scope", async () => { diff --git a/server/src/__tests__/heartbeat-list.test.ts b/server/src/__tests__/heartbeat-list.test.ts index a3e6620f726..34ad0fd33c5 100644 --- a/server/src/__tests__/heartbeat-list.test.ts +++ b/server/src/__tests__/heartbeat-list.test.ts @@ -5,7 +5,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; -import { heartbeatService } from "../services/heartbeat.ts"; +import { boundHeartbeatRunEventPayloadForStorage, heartbeatService } from "../services/heartbeat.ts"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; @@ -190,3 +190,25 @@ describeEmbeddedPostgres("heartbeat list", () => { expect(result).not.toHaveProperty("nestedHuge"); }); }); + +describe("heartbeat run event payload bounding", () => { + it("truncates oversized adapter metadata before storage", () => { + const payload = boundHeartbeatRunEventPayloadForStorage({ + adapterType: "codex_local", + prompt: "x".repeat(40_000), + context: { + issueId: "issue-1", + memory: "y".repeat(40_000), + }, + }); + + expect(payload.adapterType).toBe("codex_local"); + expect(typeof payload.prompt).toBe("string"); + expect((payload.prompt as string).length).toBeLessThan(20_000); + expect(payload.prompt).toContain("[truncated"); + expect(payload.context).toMatchObject({ + issueId: "issue-1", + }); + expect(JSON.stringify(payload).length).toBeLessThan(45_000); + }); +}); diff --git a/server/src/__tests__/issues-service.test.ts b/server/src/__tests__/issues-service.test.ts index 3e233634dfe..95279ad4705 100644 --- a/server/src/__tests__/issues-service.test.ts +++ b/server/src/__tests__/issues-service.test.ts @@ -21,12 +21,20 @@ import { startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; import { instanceSettingsService } from "../services/instance-settings.ts"; -import { issueService } from "../services/issues.ts"; +import { clampIssueListLimit, ISSUE_LIST_MAX_LIMIT, issueService } from "../services/issues.ts"; import { buildProjectMentionHref } from "@paperclipai/shared"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; +describe("issue list limit helpers", () => { + it("clamps untrusted issue-list limits to the server maximum", () => { + expect(clampIssueListLimit(0)).toBe(1); + expect(clampIssueListLimit(25.9)).toBe(25); + expect(clampIssueListLimit(ISSUE_LIST_MAX_LIMIT + 10)).toBe(ISSUE_LIST_MAX_LIMIT); + }); +}); + async function ensureIssueRelationsTable(db: ReturnType) { await db.execute(sql.raw(` CREATE TABLE IF NOT EXISTS "issue_relations" ( diff --git a/server/src/routes/agents.ts b/server/src/routes/agents.ts index 2ef48962e22..fda62a3c213 100644 --- a/server/src/routes/agents.ts +++ b/server/src/routes/agents.ts @@ -37,6 +37,7 @@ import { companySkillService, budgetService, heartbeatService, + ISSUE_LIST_DEFAULT_LIMIT, issueApprovalService, issueService, logActivity, @@ -75,6 +76,15 @@ import { } from "../services/default-agent-instructions.js"; import { getTelemetryClient } from "../telemetry.js"; +const RUN_LOG_DEFAULT_LIMIT_BYTES = 256_000; +const RUN_LOG_MAX_LIMIT_BYTES = 1024 * 1024; + +function readRunLogLimitBytes(value: unknown) { + const parsed = Number(value ?? RUN_LOG_DEFAULT_LIMIT_BYTES); + if (!Number.isFinite(parsed)) return RUN_LOG_DEFAULT_LIMIT_BYTES; + return Math.max(1, Math.min(RUN_LOG_MAX_LIMIT_BYTES, Math.trunc(parsed))); +} + export function agentRoutes(db: Db) { // Legacy hardcoded maps — used as fallback when adapter module does not // declare capability flags explicitly. @@ -1168,6 +1178,7 @@ export function agentRoutes(db: Db) { assigneeAgentId: req.actor.agentId, status: "todo,in_progress,blocked", includeRoutineExecutions: true, + limit: ISSUE_LIST_DEFAULT_LIMIT, }); res.json( @@ -1198,6 +1209,7 @@ export function agentRoutes(db: Db) { touchedByUserId: query.userId, inboxArchivedByUserId: query.userId, status: query.status, + limit: ISSUE_LIST_DEFAULT_LIMIT, }); res.json(rows); @@ -2518,10 +2530,10 @@ export function agentRoutes(db: Db) { assertCompanyAccess(req, run.companyId); const offset = Number(req.query.offset ?? 0); - const limitBytes = Number(req.query.limitBytes ?? 256000); + const limitBytes = readRunLogLimitBytes(req.query.limitBytes); const result = await heartbeat.readLog(run, { offset: Number.isFinite(offset) ? offset : 0, - limitBytes: Number.isFinite(limitBytes) ? limitBytes : 256000, + limitBytes, }); res.set("Cache-Control", "no-cache, no-store"); @@ -2553,10 +2565,10 @@ export function agentRoutes(db: Db) { assertCompanyAccess(req, operation.companyId); const offset = Number(req.query.offset ?? 0); - const limitBytes = Number(req.query.limitBytes ?? 256000); + const limitBytes = readRunLogLimitBytes(req.query.limitBytes); const result = await workspaceOperations.readLog(operationId, { offset: Number.isFinite(offset) ? offset : 0, - limitBytes: Number.isFinite(limitBytes) ? limitBytes : 256000, + limitBytes, }); res.set("Cache-Control", "no-cache, no-store"); diff --git a/server/src/routes/execution-workspaces.ts b/server/src/routes/execution-workspaces.ts index 20e80c53e43..07895b92ed5 100644 --- a/server/src/routes/execution-workspaces.ts +++ b/server/src/routes/execution-workspaces.ts @@ -28,6 +28,9 @@ import { collectExecutionWorkspaceCommandPaths, } from "./workspace-command-authz.js"; import { assertCanManageExecutionWorkspaceRuntimeServices } from "./workspace-runtime-service-authz.js"; +import { appendWithCap } from "../adapters/utils.js"; + +const WORKSPACE_CONTROL_OUTPUT_MAX_CHARS = 256 * 1024; export function executionWorkspaceRoutes(db: Db) { const router = Router(); @@ -209,8 +212,8 @@ export function executionWorkspaceRoutes(db: Db) { executionWorkspaceId: existing.id, }); let runtimeServiceCount = existing.runtimeServices?.length ?? 0; - const stdout: string[] = []; - const stderr: string[] = []; + let stdout = ""; + let stderr = ""; const operation = await recorder.recordOperation({ phase: action === "stop" ? "workspace_teardown" : "workspace_provision", @@ -310,8 +313,8 @@ export function executionWorkspaceRoutes(db: Db) { } const onLog = async (stream: "stdout" | "stderr", chunk: string) => { - if (stream === "stdout") stdout.push(chunk); - else stderr.push(chunk); + if (stream === "stdout") stdout = appendWithCap(stdout, chunk, WORKSPACE_CONTROL_OUTPUT_MAX_CHARS); + else stderr = appendWithCap(stderr, chunk, WORKSPACE_CONTROL_OUTPUT_MAX_CHARS); }; if (action === "stop" || action === "restart") { @@ -382,8 +385,8 @@ export function executionWorkspaceRoutes(db: Db) { return { status: "succeeded", - stdout: stdout.join(""), - stderr: stderr.join(""), + stdout, + stderr, system: action === "stop" ? "Stopped execution workspace runtime services.\n" diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 9d8220a023a..b1e4ebebd04 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -38,7 +38,10 @@ import { heartbeatService, instanceSettingsService, issueApprovalService, + ISSUE_LIST_DEFAULT_LIMIT, + ISSUE_LIST_MAX_LIMIT, issueService, + clampIssueListLimit, documentService, logActivity, projectService, @@ -616,8 +619,10 @@ export function issueRoutes( ? req.actor.userId : unreadForUserFilterRaw; const rawLimit = req.query.limit as string | undefined; - const parsedLimit = rawLimit ? Number.parseInt(rawLimit, 10) : null; - const limit = parsedLimit ?? undefined; + const parsedLimit = rawLimit !== undefined && /^\d+$/.test(rawLimit) + ? Number.parseInt(rawLimit, 10) + : null; + const limit = parsedLimit === null ? ISSUE_LIST_DEFAULT_LIMIT : clampIssueListLimit(parsedLimit); if (assigneeUserFilterRaw === "me" && (!assigneeUserId || req.actor.type !== "board")) { res.status(403).json({ error: "assigneeUserId=me requires board authentication" }); @@ -636,7 +641,7 @@ export function issueRoutes( return; } if (rawLimit !== undefined && (parsedLimit === null || !Number.isInteger(parsedLimit) || parsedLimit <= 0)) { - res.status(400).json({ error: "limit must be a positive integer" }); + res.status(400).json({ error: `limit must be a positive integer up to ${ISSUE_LIST_MAX_LIMIT}` }); return; } diff --git a/server/src/routes/projects.ts b/server/src/routes/projects.ts index f4d3e84863b..036b0de60ff 100644 --- a/server/src/routes/projects.ts +++ b/server/src/routes/projects.ts @@ -29,6 +29,9 @@ import { } from "./workspace-command-authz.js"; import { assertCanManageProjectWorkspaceRuntimeServices } from "./workspace-runtime-service-authz.js"; import { getTelemetryClient } from "../telemetry.js"; +import { appendWithCap } from "../adapters/utils.js"; + +const WORKSPACE_CONTROL_OUTPUT_MAX_CHARS = 256 * 1024; export function projectRoutes(db: Db) { const router = Router(); @@ -377,8 +380,8 @@ export function projectRoutes(db: Db) { const actor = getActorInfo(req); const recorder = workspaceOperations.createRecorder({ companyId: project.companyId }); let runtimeServiceCount = workspace.runtimeServices?.length ?? 0; - const stdout: string[] = []; - const stderr: string[] = []; + let stdout = ""; + let stderr = ""; const operation = await recorder.recordOperation({ phase: action === "stop" ? "workspace_teardown" : "workspace_provision", @@ -440,8 +443,8 @@ export function projectRoutes(db: Db) { } const onLog = async (stream: "stdout" | "stderr", chunk: string) => { - if (stream === "stdout") stdout.push(chunk); - else stderr.push(chunk); + if (stream === "stdout") stdout = appendWithCap(stdout, chunk, WORKSPACE_CONTROL_OUTPUT_MAX_CHARS); + else stderr = appendWithCap(stderr, chunk, WORKSPACE_CONTROL_OUTPUT_MAX_CHARS); }; if (action === "stop" || action === "restart") { @@ -514,8 +517,8 @@ export function projectRoutes(db: Db) { return { status: "succeeded", - stdout: stdout.join(""), - stderr: stderr.join(""), + stdout, + stderr, system: action === "stop" ? "Stopped project workspace runtime services.\n" diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index af9c568697d..1dfdf7bb108 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -79,6 +79,10 @@ import { extractSkillMentionIds } from "@paperclipai/shared"; const MAX_LIVE_LOG_CHUNK_BYTES = 8 * 1024; const MAX_PERSISTED_LOG_CHUNK_CHARS = 64 * 1024; +const MAX_RUN_EVENT_PAYLOAD_STRING_CHARS = 16 * 1024; +const MAX_RUN_EVENT_PAYLOAD_ARRAY_ITEMS = 50; +const MAX_RUN_EVENT_PAYLOAD_OBJECT_KEYS = 100; +const MAX_RUN_EVENT_PAYLOAD_DEPTH = 6; const HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT = 1; const HEARTBEAT_MAX_CONCURRENT_RUNS_MAX = 10; const DEFERRED_WAKE_CONTEXT_KEY = "_paperclipWakeContext"; @@ -498,6 +502,80 @@ function appendExcerpt(prev: string, chunk: string) { return appendWithCap(prev, chunk, MAX_EXCERPT_BYTES); } +function truncateRunEventString(value: string) { + if (value.length <= MAX_RUN_EVENT_PAYLOAD_STRING_CHARS) return value; + const omittedChars = value.length - MAX_RUN_EVENT_PAYLOAD_STRING_CHARS; + return `${value.slice(0, MAX_RUN_EVENT_PAYLOAD_STRING_CHARS)}\n[truncated ${omittedChars} chars]`; +} + +function boundRunEventValue(value: unknown, depth: number, seen: WeakSet): unknown { + if (typeof value === "string") { + return truncateRunEventString(value); + } + if ( + value === null + || typeof value === "number" + || typeof value === "boolean" + ) { + return value; + } + if (value instanceof Date) { + return value.toISOString(); + } + if (Array.isArray(value)) { + if (depth >= MAX_RUN_EVENT_PAYLOAD_DEPTH) { + return { + _truncated: true, + type: "array", + originalLength: value.length, + }; + } + const bounded = value + .slice(0, MAX_RUN_EVENT_PAYLOAD_ARRAY_ITEMS) + .map((entry) => boundRunEventValue(entry, depth + 1, seen)); + if (value.length > MAX_RUN_EVENT_PAYLOAD_ARRAY_ITEMS) { + bounded.push({ + _truncated: true, + omittedItems: value.length - MAX_RUN_EVENT_PAYLOAD_ARRAY_ITEMS, + }); + } + return bounded; + } + if (typeof value !== "object" || value === undefined) { + return null; + } + if (seen.has(value)) { + return "[Circular]"; + } + seen.add(value); + const entries = Object.entries(value as Record); + if (depth >= MAX_RUN_EVENT_PAYLOAD_DEPTH) { + const bounded = { + _truncated: true, + type: "object", + keys: entries.map(([key]) => key).slice(0, 20), + }; + seen.delete(value); + return bounded; + } + + const out: Record = {}; + for (const [key, entryValue] of entries.slice(0, MAX_RUN_EVENT_PAYLOAD_OBJECT_KEYS)) { + out[key] = boundRunEventValue(entryValue, depth + 1, seen); + } + if (entries.length > MAX_RUN_EVENT_PAYLOAD_OBJECT_KEYS) { + out._truncated = true; + out._omittedKeys = entries.length - MAX_RUN_EVENT_PAYLOAD_OBJECT_KEYS; + } + seen.delete(value); + return out; +} + +export function boundHeartbeatRunEventPayloadForStorage(payload: Record): Record { + const bounded = boundRunEventValue(payload, 0, new WeakSet()); + return parseObject(bounded) ?? { _truncated: true }; +} + 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}`, @@ -2203,9 +2281,12 @@ export function heartbeatService(db: Db) { const sanitizedMessage = event.message ? redactCurrentUserText(event.message, currentUserRedactionOptions) : event.message; - const sanitizedPayload = event.payload - ? redactCurrentUserValue(event.payload, currentUserRedactionOptions) + const boundedPayload = event.payload + ? boundHeartbeatRunEventPayloadForStorage(event.payload) : event.payload; + const sanitizedPayload = boundedPayload + ? redactCurrentUserValue(boundedPayload, currentUserRedactionOptions) + : boundedPayload; await db.insert(heartbeatRunEvents).values({ companyId: run.companyId, @@ -5310,7 +5391,9 @@ export function heartbeatService(db: Db) { store: run.logStore, logRef: run.logRef, ...result, - content: redactCurrentUserText(result.content, await getCurrentUserRedactionOptions()), + // Run-log chunks are already redacted before they are appended to the store. + // Rewriting the full chunk again on every poll creates avoidable string copies. + content: result.content, }; }, diff --git a/server/src/services/index.ts b/server/src/services/index.ts index c279dd13be5..945e0afc506 100644 --- a/server/src/services/index.ts +++ b/server/src/services/index.ts @@ -6,7 +6,13 @@ export { agentInstructionsService, syncInstructionsBundleConfigFromFilePath } fr export { assetService } from "./assets.js"; export { documentService, extractLegacyPlanBody } from "./documents.js"; export { projectService } from "./projects.js"; -export { issueService, type IssueFilters } from "./issues.js"; +export { + clampIssueListLimit, + ISSUE_LIST_DEFAULT_LIMIT, + ISSUE_LIST_MAX_LIMIT, + issueService, + type IssueFilters, +} from "./issues.js"; export { issueApprovalService } from "./issue-approvals.js"; export { goalService } from "./goals.js"; export { activityService, type ActivityFilters } from "./activity.js"; diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 5d045e47ae0..e3472f13a57 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -38,6 +38,9 @@ import { getDefaultCompanyGoal } from "./goals.js"; const ALL_ISSUE_STATUSES = ["backlog", "todo", "in_progress", "in_review", "blocked", "done", "cancelled"]; const MAX_ISSUE_COMMENT_PAGE_LIMIT = 500; +export const ISSUE_LIST_DEFAULT_LIMIT = 500; +export const ISSUE_LIST_MAX_LIMIT = 1000; +const ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE = 500; function assertTransition(from: string, to: string) { if (from === to) return; @@ -103,6 +106,10 @@ type IssueUserCommentStats = { myLastCommentAt: Date | null; lastExternalCommentAt: Date | null; }; +type IssueReadStat = { + issueId: string; + myLastReadAt: Date | null; +}; type IssueLastActivityStat = { issueId: string; latestCommentAt: Date | null; @@ -138,6 +145,18 @@ function escapeLikePattern(value: string): string { return value.replace(/[\\%_]/g, "\\$&"); } +export function clampIssueListLimit(limit: number): number { + return Math.min(ISSUE_LIST_MAX_LIMIT, Math.max(1, Math.floor(limit))); +} + +function chunkList(values: T[], size: number): T[][] { + const chunks: T[][] = []; + for (let index = 0; index < values.length; index += size) { + chunks.push(values.slice(index, index + size)); + } + return chunks; +} + async function getProjectDefaultGoalId( db: ProjectGoalReader, companyId: string, @@ -460,20 +479,22 @@ function latestIssueActivityAt(...values: Array> { const map = new Map(); if (issueIds.length === 0) return map; - const rows = await dbOrTx - .select({ - issueId: issueLabels.issueId, - label: labels, - }) - .from(issueLabels) - .innerJoin(labels, eq(issueLabels.labelId, labels.id)) - .where(inArray(issueLabels.issueId, issueIds)) - .orderBy(asc(labels.name), asc(labels.id)); + for (const issueIdChunk of chunkList(issueIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) { + const rows = await dbOrTx + .select({ + issueId: issueLabels.issueId, + label: labels, + }) + .from(issueLabels) + .innerJoin(labels, eq(issueLabels.labelId, labels.id)) + .where(inArray(issueLabels.issueId, issueIdChunk)) + .orderBy(asc(labels.name), asc(labels.id)); - for (const row of rows) { - const existing = map.get(row.issueId); - if (existing) existing.push(row.label); - else map.set(row.issueId, [row.label]); + for (const row of rows) { + const existing = map.get(row.issueId); + if (existing) existing.push(row.label); + else map.set(row.issueId, [row.label]); + } } return map; } @@ -503,27 +524,29 @@ async function activeRunMapForIssues( .filter((id): id is string => id != null); if (runIds.length === 0) return map; - const rows = await dbOrTx - .select({ - id: heartbeatRuns.id, - status: heartbeatRuns.status, - agentId: heartbeatRuns.agentId, - invocationSource: heartbeatRuns.invocationSource, - triggerDetail: heartbeatRuns.triggerDetail, - startedAt: heartbeatRuns.startedAt, - finishedAt: heartbeatRuns.finishedAt, - createdAt: heartbeatRuns.createdAt, - }) - .from(heartbeatRuns) - .where( - and( - inArray(heartbeatRuns.id, runIds), - inArray(heartbeatRuns.status, ACTIVE_RUN_STATUSES), - ), - ); + for (const runIdChunk of chunkList([...new Set(runIds)], ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) { + const rows = await dbOrTx + .select({ + id: heartbeatRuns.id, + status: heartbeatRuns.status, + agentId: heartbeatRuns.agentId, + invocationSource: heartbeatRuns.invocationSource, + triggerDetail: heartbeatRuns.triggerDetail, + startedAt: heartbeatRuns.startedAt, + finishedAt: heartbeatRuns.finishedAt, + createdAt: heartbeatRuns.createdAt, + }) + .from(heartbeatRuns) + .where( + and( + inArray(heartbeatRuns.id, runIdChunk), + inArray(heartbeatRuns.status, ACTIVE_RUN_STATUSES), + ), + ); - for (const row of rows) { - map.set(row.id, row); + for (const row of rows) { + map.set(row.id, row); + } } return map; } @@ -583,6 +606,131 @@ function withActiveRuns( })); } +async function userCommentStatsForIssues( + dbOrTx: any, + companyId: string, + userId: string, + issueIds: string[], +): Promise { + const stats: IssueUserCommentStats[] = []; + for (const issueIdChunk of chunkList(issueIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) { + const rows = await dbOrTx + .select({ + issueId: issueComments.issueId, + myLastCommentAt: sql` + MAX(CASE WHEN ${issueComments.authorUserId} = ${userId} THEN ${issueComments.createdAt} END) + `, + lastExternalCommentAt: sql` + MAX( + CASE + WHEN ${issueComments.authorUserId} IS NULL OR ${issueComments.authorUserId} <> ${userId} + THEN ${issueComments.createdAt} + END + ) + `, + }) + .from(issueComments) + .where( + and( + eq(issueComments.companyId, companyId), + inArray(issueComments.issueId, issueIdChunk), + ), + ) + .groupBy(issueComments.issueId); + stats.push(...rows); + } + return stats; +} + +async function userReadStatsForIssues( + dbOrTx: any, + companyId: string, + userId: string, + issueIds: string[], +): Promise { + const stats: IssueReadStat[] = []; + for (const issueIdChunk of chunkList(issueIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) { + const rows = await dbOrTx + .select({ + issueId: issueReadStates.issueId, + myLastReadAt: issueReadStates.lastReadAt, + }) + .from(issueReadStates) + .where( + and( + eq(issueReadStates.companyId, companyId), + eq(issueReadStates.userId, userId), + inArray(issueReadStates.issueId, issueIdChunk), + ), + ); + stats.push(...rows); + } + return stats; +} + +async function lastActivityStatsForIssues( + dbOrTx: any, + companyId: string, + issueIds: string[], +): Promise { + const byIssueId = new Map(); + for (const issueIdChunk of chunkList(issueIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) { + const [commentRows, logRows] = await Promise.all([ + dbOrTx + .select({ + issueId: issueComments.issueId, + latestCommentAt: sql`MAX(${issueComments.createdAt})`, + }) + .from(issueComments) + .where( + and( + eq(issueComments.companyId, companyId), + inArray(issueComments.issueId, issueIdChunk), + ), + ) + .groupBy(issueComments.issueId), + dbOrTx + .select({ + issueId: activityLog.entityId, + latestLogAt: sql`MAX(${activityLog.createdAt})`, + }) + .from(activityLog) + .where( + and( + eq(activityLog.companyId, companyId), + eq(activityLog.entityType, "issue"), + inArray(activityLog.entityId, issueIdChunk), + sql`${activityLog.action} NOT IN (${sql.join( + ISSUE_LOCAL_INBOX_ACTIVITY_ACTIONS.map((action) => sql`${action}`), + sql`, `, + )})`, + ), + ) + .groupBy(activityLog.entityId), + ]); + + for (const row of commentRows) { + byIssueId.set(row.issueId, { + issueId: row.issueId, + latestCommentAt: row.latestCommentAt, + latestLogAt: null, + }); + } + for (const row of logRows) { + const existing = byIssueId.get(row.issueId); + if (existing) existing.latestLogAt = row.latestLogAt; + else { + byIssueId.set(row.issueId, { + issueId: row.issueId, + latestCommentAt: null, + latestLogAt: row.latestLogAt, + }); + } + } + } + return [...byIssueId.values()]; +} + export function issueService(db: Db) { const instanceSettings = instanceSettingsService(db); @@ -1071,99 +1219,12 @@ export function issueService(db: Db) { const issueIds = withRuns.map((row) => row.id); const [statsRows, readRows, lastActivityRows] = await Promise.all([ contextUserId - ? db - .select({ - issueId: issueComments.issueId, - myLastCommentAt: sql` - MAX(CASE WHEN ${issueComments.authorUserId} = ${contextUserId} THEN ${issueComments.createdAt} END) - `, - lastExternalCommentAt: sql` - MAX( - CASE - WHEN ${issueComments.authorUserId} IS NULL OR ${issueComments.authorUserId} <> ${contextUserId} - THEN ${issueComments.createdAt} - END - ) - `, - }) - .from(issueComments) - .where( - and( - eq(issueComments.companyId, companyId), - inArray(issueComments.issueId, issueIds), - ), - ) - .groupBy(issueComments.issueId) + ? userCommentStatsForIssues(db, companyId, contextUserId, issueIds) : Promise.resolve([]), contextUserId - ? db - .select({ - issueId: issueReadStates.issueId, - myLastReadAt: issueReadStates.lastReadAt, - }) - .from(issueReadStates) - .where( - and( - eq(issueReadStates.companyId, companyId), - eq(issueReadStates.userId, contextUserId), - inArray(issueReadStates.issueId, issueIds), - ), - ) + ? userReadStatsForIssues(db, companyId, contextUserId, issueIds) : Promise.resolve([]), - Promise.all([ - db - .select({ - issueId: issueComments.issueId, - latestCommentAt: sql`MAX(${issueComments.createdAt})`, - }) - .from(issueComments) - .where( - and( - eq(issueComments.companyId, companyId), - inArray(issueComments.issueId, issueIds), - ), - ) - .groupBy(issueComments.issueId), - db - .select({ - issueId: activityLog.entityId, - latestLogAt: sql`MAX(${activityLog.createdAt})`, - }) - .from(activityLog) - .where( - and( - eq(activityLog.companyId, companyId), - eq(activityLog.entityType, "issue"), - inArray(activityLog.entityId, issueIds), - sql`${activityLog.action} NOT IN (${sql.join( - ISSUE_LOCAL_INBOX_ACTIVITY_ACTIONS.map((action) => sql`${action}`), - sql`, `, - )})`, - ), - ) - .groupBy(activityLog.entityId), - ]).then(([commentRows, logRows]) => { - const byIssueId = new Map(); - for (const row of commentRows) { - byIssueId.set(row.issueId, { - issueId: row.issueId, - latestCommentAt: row.latestCommentAt, - latestLogAt: null, - }); - } - for (const row of logRows) { - const existing = byIssueId.get(row.issueId); - if (existing) existing.latestLogAt = row.latestLogAt; - else { - byIssueId.set(row.issueId, { - issueId: row.issueId, - latestCommentAt: null, - latestLogAt: row.latestLogAt, - }); - } - } - return [...byIssueId.values()]; - }), + lastActivityStatsForIssues(db, companyId, issueIds), ]); const statsByIssueId = new Map(statsRows.map((row) => [row.issueId, row])); const lastActivityByIssueId = new Map(lastActivityRows.map((row) => [row.issueId, row])); diff --git a/server/src/services/workspace-operations.ts b/server/src/services/workspace-operations.ts index b20a9ed74c2..9509e9824bc 100644 --- a/server/src/services/workspace-operations.ts +++ b/server/src/services/workspace-operations.ts @@ -250,9 +250,9 @@ export function workspaceOperationService(db: Db) { store: operation.logStore, logRef: operation.logRef, ...result, - content: redactCurrentUserText(result.content, { - enabled: (await instanceSettings.getGeneral()).censorUsernameInLogs, - }), + // Workspace-operation log chunks are sanitized before append-time storage. + // Returning the stored chunk avoids another whole-string rewrite per poll. + content: result.content, }; }, }; diff --git a/server/src/services/workspace-runtime.ts b/server/src/services/workspace-runtime.ts index 87a093a50a4..f3a556c0cf4 100644 --- a/server/src/services/workspace-runtime.ts +++ b/server/src/services/workspace-runtime.ts @@ -413,32 +413,33 @@ function formatCommandForDisplay(command: string, args: string[]) { .join(" "); } +function trimToLastBytes(value: string, limit: number) { + const byteLength = Buffer.byteLength(value, "utf8"); + if (byteLength <= limit) return value; + return Buffer.from(value, "utf8").subarray(byteLength - limit).toString("utf8"); +} + function createProcessOutputCapture(maxBytes: number): ProcessOutputAccumulator { const limit = Math.max(1, Math.trunc(maxBytes)); - let chunks: string[] = []; + let text = ""; let truncated = false; let totalBytes = 0; return { append(chunk: string) { if (!chunk) return; - chunks.push(chunk); totalBytes += Buffer.byteLength(chunk, "utf8"); - let currentBytes = chunks.reduce((sum, value) => sum + Buffer.byteLength(value, "utf8"), 0); - if (currentBytes <= limit) return; - - const combined = Buffer.from(chunks.join(""), "utf8"); - const tail = combined.subarray(Math.max(0, combined.length - limit)).toString("utf8"); - chunks = [tail]; - truncated = true; - currentBytes = Buffer.byteLength(tail, "utf8"); - if (currentBytes > limit) { - chunks = [Buffer.from(tail, "utf8").subarray(Math.max(0, currentBytes - limit)).toString("utf8")]; + const combined = text + chunk; + if (Buffer.byteLength(combined, "utf8") <= limit) { + text = combined; + return; } + + text = trimToLastBytes(combined, limit); + truncated = true; }, finish(): ProcessOutputCapture { - const text = chunks.join(""); if (!truncated) { return { text, diff --git a/ui/src/hooks/useInboxBadge.ts b/ui/src/hooks/useInboxBadge.ts index 6c4083b8ead..1e401eda790 100644 --- a/ui/src/hooks/useInboxBadge.ts +++ b/ui/src/hooks/useInboxBadge.ts @@ -21,6 +21,7 @@ import { } from "../lib/inbox"; const INBOX_ISSUE_STATUSES = "backlog,todo,in_progress,in_review,blocked,done"; +const INBOX_BADGE_ISSUE_LIMIT = 500; const INBOX_BADGE_HEARTBEAT_RUN_LIMIT = 200; export function useDismissedInboxAlerts() { @@ -180,6 +181,7 @@ export function useInboxBadge(companyId: string | null | undefined) { touchedByUserId: "me", inboxArchivedByUserId: "me", status: INBOX_ISSUE_STATUSES, + limit: INBOX_BADGE_ISSUE_LIMIT, }), enabled: !!companyId, }); diff --git a/ui/src/pages/Inbox.tsx b/ui/src/pages/Inbox.tsx index f58d9688a2e..33827634290 100644 --- a/ui/src/pages/Inbox.tsx +++ b/ui/src/pages/Inbox.tsx @@ -93,6 +93,7 @@ import { } from "lucide-react"; const INBOX_HEARTBEAT_RUN_LIMIT = 200; +const INBOX_ISSUE_LIST_LIMIT = 500; import { Input } from "@/components/ui/input"; import { PageTabBar } from "../components/PageTabBar"; import type { Approval, HeartbeatRun, Issue, JoinRequest } from "@paperclipai/shared"; @@ -714,9 +715,9 @@ export function Inbox() { const isolatedWorkspacesEnabled = experimentalSettings?.enableIsolatedWorkspaces === true; const { data: executionWorkspaces = [] } = useQuery({ queryKey: selectedCompanyId - ? queryKeys.executionWorkspaces.list(selectedCompanyId) + ? queryKeys.executionWorkspaces.summaryList(selectedCompanyId) : ["execution-workspaces", "__disabled__"], - queryFn: () => executionWorkspacesApi.list(selectedCompanyId!), + queryFn: () => executionWorkspacesApi.listSummaries(selectedCompanyId!), enabled: !!selectedCompanyId && isolatedWorkspacesEnabled, }); @@ -776,7 +777,11 @@ export function Inbox() { const { data: issues, isLoading: isIssuesLoading } = useQuery({ queryKey: [...queryKeys.issues.list(selectedCompanyId!), "with-routine-executions"], - queryFn: () => issuesApi.list(selectedCompanyId!, { includeRoutineExecutions: true }), + queryFn: () => + issuesApi.list(selectedCompanyId!, { + includeRoutineExecutions: true, + limit: INBOX_ISSUE_LIST_LIMIT, + }), enabled: !!selectedCompanyId, }); const { @@ -790,6 +795,7 @@ export function Inbox() { inboxArchivedByUserId: "me", status: INBOX_MINE_ISSUE_STATUS_FILTER, includeRoutineExecutions: true, + limit: INBOX_ISSUE_LIST_LIMIT, }), enabled: !!selectedCompanyId, }); @@ -803,6 +809,7 @@ export function Inbox() { touchedByUserId: "me", status: INBOX_MINE_ISSUE_STATUS_FILTER, includeRoutineExecutions: true, + limit: INBOX_ISSUE_LIST_LIMIT, }), enabled: !!selectedCompanyId, }); @@ -1937,7 +1944,7 @@ export function Inbox() { enableRoutineVisibilityFilter buttonVariant="outline" iconOnly - workspaces={isolatedWorkspacesEnabled ? executionWorkspaces.filter((w) => w.mode === "isolated_workspace" && w.status === "active").map((w) => ({ id: w.id, name: w.name })) : undefined} + workspaces={isolatedWorkspacesEnabled ? executionWorkspaces.filter((w) => w.mode === "isolated_workspace").map((w) => ({ id: w.id, name: w.name })) : undefined} />