fix(server): bound high-volume issue and log reads

This commit is contained in:
Dotta committed 2026-04-19 21:45:59 -05:00
1 parent 4dd09091e3
commit 302ee69a51
15 files changed
+401 -170

No files matched your search

+13 -2
View File
@@ -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<string, unknown>, 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;
@@ -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 () => {
+23 -1
View File
@@ -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);
});
});
+9 -1
View File
@@ -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<typeof createDb>) {
await db.execute(sql.raw(`
CREATE TABLE IF NOT EXISTS "issue_relations" (
+16 -4
View File
@@ -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");
+9 -6
View File
@@ -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"
+8 -3
View File
@@ -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;
}
+9 -6
View File
@@ -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"
+86 -3
View File
@@ -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<object>): 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<string, unknown>);
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<string, unknown> = {};
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<string, unknown>): Record<string, unknown> {
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,
};
},
+7 -1
View File
@@ -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";
+184 -123
View File
@@ -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<T>(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<Date | string | null | undefined
async function labelMapForIssues(dbOrTx: any, issueIds: string[]): Promise<Map<string, IssueLabelRow[]>> {
const map = new Map<string, IssueLabelRow[]>();
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<IssueUserCommentStats[]> {
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<Date | null>`
MAX(CASE WHEN ${issueComments.authorUserId} = ${userId} THEN ${issueComments.createdAt} END)
`,
lastExternalCommentAt: sql<Date | null>`
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<IssueReadStat[]> {
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<IssueLastActivityStat[]> {
const byIssueId = new Map<string, IssueLastActivityStat>();
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<Date | null>`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<Date | null>`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<Date | null>`
MAX(CASE WHEN ${issueComments.authorUserId} = ${contextUserId} THEN ${issueComments.createdAt} END)
`,
lastExternalCommentAt: sql<Date | null>`
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<Date | null>`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<Date | null>`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<string, IssueLastActivityStat>();
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]));
+3 -3
View File
@@ -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,
};
},
};
+14 -13
View File
@@ -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,
+2
View File
@@ -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,
});
+11 -4
View File
@@ -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}
/>
<Popover>
<PopoverTrigger asChild>