Merge pull request #790 from paperclipai/paperclip-token-optimization
Optimize heartbeat token usage
This commit is contained in:
Dotta
208
server/src/__tests__/codex-local-execute.test.ts
Normal file
208
server/src/__tests__/codex-local-execute.test.ts
Normal file
@@ -0,0 +1,208 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { execute } from "@paperclipai/adapter-codex-local/server";
|
||||
|
||||
async function writeFakeCodexCommand(commandPath: string): Promise<void> {
|
||||
const script = `#!/usr/bin/env node
|
||||
const fs = require("node:fs");
|
||||
|
||||
const capturePath = process.env.PAPERCLIP_TEST_CAPTURE_PATH;
|
||||
const payload = {
|
||||
argv: process.argv.slice(2),
|
||||
prompt: fs.readFileSync(0, "utf8"),
|
||||
codexHome: process.env.CODEX_HOME || null,
|
||||
paperclipEnvKeys: Object.keys(process.env)
|
||||
.filter((key) => key.startsWith("PAPERCLIP_"))
|
||||
.sort(),
|
||||
};
|
||||
if (capturePath) {
|
||||
fs.writeFileSync(capturePath, JSON.stringify(payload), "utf8");
|
||||
}
|
||||
console.log(JSON.stringify({ type: "thread.started", thread_id: "codex-session-1" }));
|
||||
console.log(JSON.stringify({ type: "item.completed", item: { type: "agent_message", text: "hello" } }));
|
||||
console.log(JSON.stringify({ type: "turn.completed", usage: { input_tokens: 1, cached_input_tokens: 0, output_tokens: 1 } }));
|
||||
`;
|
||||
await fs.writeFile(commandPath, script, "utf8");
|
||||
await fs.chmod(commandPath, 0o755);
|
||||
}
|
||||
|
||||
type CapturePayload = {
|
||||
argv: string[];
|
||||
prompt: string;
|
||||
codexHome: string | null;
|
||||
paperclipEnvKeys: string[];
|
||||
};
|
||||
|
||||
describe("codex execute", () => {
|
||||
it("uses a worktree-isolated CODEX_HOME while preserving shared auth and config", async () => {
|
||||
const root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-codex-execute-"));
|
||||
const workspace = path.join(root, "workspace");
|
||||
const commandPath = path.join(root, "codex");
|
||||
const capturePath = path.join(root, "capture.json");
|
||||
const sharedCodexHome = path.join(root, "shared-codex-home");
|
||||
const paperclipHome = path.join(root, "paperclip-home");
|
||||
const isolatedCodexHome = path.join(paperclipHome, "instances", "worktree-1", "codex-home");
|
||||
await fs.mkdir(workspace, { recursive: true });
|
||||
await fs.mkdir(sharedCodexHome, { recursive: true });
|
||||
await fs.writeFile(path.join(sharedCodexHome, "auth.json"), '{"token":"shared"}\n', "utf8");
|
||||
await fs.writeFile(path.join(sharedCodexHome, "config.toml"), 'model = "codex-mini-latest"\n', "utf8");
|
||||
await writeFakeCodexCommand(commandPath);
|
||||
|
||||
const previousHome = process.env.HOME;
|
||||
const previousPaperclipHome = process.env.PAPERCLIP_HOME;
|
||||
const previousPaperclipInstanceId = process.env.PAPERCLIP_INSTANCE_ID;
|
||||
const previousPaperclipInWorktree = process.env.PAPERCLIP_IN_WORKTREE;
|
||||
const previousCodexHome = process.env.CODEX_HOME;
|
||||
process.env.HOME = root;
|
||||
process.env.PAPERCLIP_HOME = paperclipHome;
|
||||
process.env.PAPERCLIP_INSTANCE_ID = "worktree-1";
|
||||
process.env.PAPERCLIP_IN_WORKTREE = "true";
|
||||
process.env.CODEX_HOME = sharedCodexHome;
|
||||
|
||||
try {
|
||||
const result = await execute({
|
||||
runId: "run-1",
|
||||
agent: {
|
||||
id: "agent-1",
|
||||
companyId: "company-1",
|
||||
name: "Codex Coder",
|
||||
adapterType: "codex_local",
|
||||
adapterConfig: {},
|
||||
},
|
||||
runtime: {
|
||||
sessionId: null,
|
||||
sessionParams: null,
|
||||
sessionDisplayId: null,
|
||||
taskKey: null,
|
||||
},
|
||||
config: {
|
||||
command: commandPath,
|
||||
cwd: workspace,
|
||||
env: {
|
||||
PAPERCLIP_TEST_CAPTURE_PATH: capturePath,
|
||||
},
|
||||
promptTemplate: "Follow the paperclip heartbeat.",
|
||||
},
|
||||
context: {},
|
||||
authToken: "run-jwt-token",
|
||||
onLog: async () => {},
|
||||
});
|
||||
|
||||
expect(result.exitCode).toBe(0);
|
||||
expect(result.errorMessage).toBeNull();
|
||||
|
||||
const capture = JSON.parse(await fs.readFile(capturePath, "utf8")) as CapturePayload;
|
||||
expect(capture.codexHome).toBe(isolatedCodexHome);
|
||||
expect(capture.argv).toEqual(expect.arrayContaining(["exec", "--json", "-"]));
|
||||
expect(capture.prompt).toContain("Follow the paperclip heartbeat.");
|
||||
expect(capture.paperclipEnvKeys).toEqual(
|
||||
expect.arrayContaining([
|
||||
"PAPERCLIP_AGENT_ID",
|
||||
"PAPERCLIP_API_KEY",
|
||||
"PAPERCLIP_API_URL",
|
||||
"PAPERCLIP_COMPANY_ID",
|
||||
"PAPERCLIP_RUN_ID",
|
||||
]),
|
||||
);
|
||||
|
||||
const isolatedAuth = path.join(isolatedCodexHome, "auth.json");
|
||||
const isolatedConfig = path.join(isolatedCodexHome, "config.toml");
|
||||
const isolatedSkill = path.join(isolatedCodexHome, "skills", "paperclip");
|
||||
|
||||
expect((await fs.lstat(isolatedAuth)).isSymbolicLink()).toBe(true);
|
||||
expect(await fs.realpath(isolatedAuth)).toBe(await fs.realpath(path.join(sharedCodexHome, "auth.json")));
|
||||
expect((await fs.lstat(isolatedConfig)).isFile()).toBe(true);
|
||||
expect(await fs.readFile(isolatedConfig, "utf8")).toBe('model = "codex-mini-latest"\n');
|
||||
expect((await fs.lstat(isolatedSkill)).isSymbolicLink()).toBe(true);
|
||||
} finally {
|
||||
if (previousHome === undefined) delete process.env.HOME;
|
||||
else process.env.HOME = previousHome;
|
||||
if (previousPaperclipHome === undefined) delete process.env.PAPERCLIP_HOME;
|
||||
else process.env.PAPERCLIP_HOME = previousPaperclipHome;
|
||||
if (previousPaperclipInstanceId === undefined) delete process.env.PAPERCLIP_INSTANCE_ID;
|
||||
else process.env.PAPERCLIP_INSTANCE_ID = previousPaperclipInstanceId;
|
||||
if (previousPaperclipInWorktree === undefined) delete process.env.PAPERCLIP_IN_WORKTREE;
|
||||
else process.env.PAPERCLIP_IN_WORKTREE = previousPaperclipInWorktree;
|
||||
if (previousCodexHome === undefined) delete process.env.CODEX_HOME;
|
||||
else process.env.CODEX_HOME = previousCodexHome;
|
||||
await fs.rm(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("respects an explicit CODEX_HOME config override even in worktree mode", async () => {
|
||||
const root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-codex-execute-explicit-"));
|
||||
const workspace = path.join(root, "workspace");
|
||||
const commandPath = path.join(root, "codex");
|
||||
const capturePath = path.join(root, "capture.json");
|
||||
const sharedCodexHome = path.join(root, "shared-codex-home");
|
||||
const explicitCodexHome = path.join(root, "explicit-codex-home");
|
||||
const paperclipHome = path.join(root, "paperclip-home");
|
||||
await fs.mkdir(workspace, { recursive: true });
|
||||
await fs.mkdir(sharedCodexHome, { recursive: true });
|
||||
await fs.writeFile(path.join(sharedCodexHome, "auth.json"), '{"token":"shared"}\n', "utf8");
|
||||
await writeFakeCodexCommand(commandPath);
|
||||
|
||||
const previousHome = process.env.HOME;
|
||||
const previousPaperclipHome = process.env.PAPERCLIP_HOME;
|
||||
const previousPaperclipInstanceId = process.env.PAPERCLIP_INSTANCE_ID;
|
||||
const previousPaperclipInWorktree = process.env.PAPERCLIP_IN_WORKTREE;
|
||||
const previousCodexHome = process.env.CODEX_HOME;
|
||||
process.env.HOME = root;
|
||||
process.env.PAPERCLIP_HOME = paperclipHome;
|
||||
process.env.PAPERCLIP_INSTANCE_ID = "worktree-1";
|
||||
process.env.PAPERCLIP_IN_WORKTREE = "true";
|
||||
process.env.CODEX_HOME = sharedCodexHome;
|
||||
|
||||
try {
|
||||
const result = await execute({
|
||||
runId: "run-2",
|
||||
agent: {
|
||||
id: "agent-1",
|
||||
companyId: "company-1",
|
||||
name: "Codex Coder",
|
||||
adapterType: "codex_local",
|
||||
adapterConfig: {},
|
||||
},
|
||||
runtime: {
|
||||
sessionId: null,
|
||||
sessionParams: null,
|
||||
sessionDisplayId: null,
|
||||
taskKey: null,
|
||||
},
|
||||
config: {
|
||||
command: commandPath,
|
||||
cwd: workspace,
|
||||
env: {
|
||||
PAPERCLIP_TEST_CAPTURE_PATH: capturePath,
|
||||
CODEX_HOME: explicitCodexHome,
|
||||
},
|
||||
promptTemplate: "Follow the paperclip heartbeat.",
|
||||
},
|
||||
context: {},
|
||||
authToken: "run-jwt-token",
|
||||
onLog: async () => {},
|
||||
});
|
||||
|
||||
expect(result.exitCode).toBe(0);
|
||||
expect(result.errorMessage).toBeNull();
|
||||
|
||||
const capture = JSON.parse(await fs.readFile(capturePath, "utf8")) as CapturePayload;
|
||||
expect(capture.codexHome).toBe(explicitCodexHome);
|
||||
await expect(fs.lstat(path.join(paperclipHome, "instances", "worktree-1", "codex-home"))).rejects.toThrow();
|
||||
} finally {
|
||||
if (previousHome === undefined) delete process.env.HOME;
|
||||
else process.env.HOME = previousHome;
|
||||
if (previousPaperclipHome === undefined) delete process.env.PAPERCLIP_HOME;
|
||||
else process.env.PAPERCLIP_HOME = previousPaperclipHome;
|
||||
if (previousPaperclipInstanceId === undefined) delete process.env.PAPERCLIP_INSTANCE_ID;
|
||||
else process.env.PAPERCLIP_INSTANCE_ID = previousPaperclipInstanceId;
|
||||
if (previousPaperclipInWorktree === undefined) delete process.env.PAPERCLIP_IN_WORKTREE;
|
||||
else process.env.PAPERCLIP_IN_WORKTREE = previousPaperclipInWorktree;
|
||||
if (previousCodexHome === undefined) delete process.env.CODEX_HOME;
|
||||
else process.env.CODEX_HOME = previousCodexHome;
|
||||
await fs.rm(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
});
|
||||
91
server/src/__tests__/codex-local-skill-injection.test.ts
Normal file
91
server/src/__tests__/codex-local-skill-injection.test.ts
Normal file
@@ -0,0 +1,91 @@
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import { ensureCodexSkillsInjected } from "@paperclipai/adapter-codex-local/server";
|
||||
|
||||
async function makeTempDir(prefix: string): Promise<string> {
|
||||
return fs.mkdtemp(path.join(os.tmpdir(), prefix));
|
||||
}
|
||||
|
||||
async function createPaperclipRepoSkill(root: string, skillName: string) {
|
||||
await fs.mkdir(path.join(root, "server"), { recursive: true });
|
||||
await fs.mkdir(path.join(root, "packages", "adapter-utils"), { recursive: true });
|
||||
await fs.mkdir(path.join(root, "skills", skillName), { recursive: true });
|
||||
await fs.writeFile(path.join(root, "pnpm-workspace.yaml"), "packages:\n - packages/*\n", "utf8");
|
||||
await fs.writeFile(path.join(root, "package.json"), '{"name":"paperclip"}\n', "utf8");
|
||||
await fs.writeFile(
|
||||
path.join(root, "skills", skillName, "SKILL.md"),
|
||||
`---\nname: ${skillName}\n---\n`,
|
||||
"utf8",
|
||||
);
|
||||
}
|
||||
|
||||
async function createCustomSkill(root: string, skillName: string) {
|
||||
await fs.mkdir(path.join(root, "custom", skillName), { recursive: true });
|
||||
await fs.writeFile(
|
||||
path.join(root, "custom", skillName, "SKILL.md"),
|
||||
`---\nname: ${skillName}\n---\n`,
|
||||
"utf8",
|
||||
);
|
||||
}
|
||||
|
||||
describe("codex local adapter skill injection", () => {
|
||||
const cleanupDirs = new Set<string>();
|
||||
|
||||
afterEach(async () => {
|
||||
await Promise.all(Array.from(cleanupDirs).map((dir) => fs.rm(dir, { recursive: true, force: true })));
|
||||
cleanupDirs.clear();
|
||||
});
|
||||
|
||||
it("repairs a Codex Paperclip skill symlink that still points at another live checkout", async () => {
|
||||
const currentRepo = await makeTempDir("paperclip-codex-current-");
|
||||
const oldRepo = await makeTempDir("paperclip-codex-old-");
|
||||
const skillsHome = await makeTempDir("paperclip-codex-home-");
|
||||
cleanupDirs.add(currentRepo);
|
||||
cleanupDirs.add(oldRepo);
|
||||
cleanupDirs.add(skillsHome);
|
||||
|
||||
await createPaperclipRepoSkill(currentRepo, "paperclip");
|
||||
await createPaperclipRepoSkill(oldRepo, "paperclip");
|
||||
await fs.symlink(path.join(oldRepo, "skills", "paperclip"), path.join(skillsHome, "paperclip"));
|
||||
|
||||
const logs: string[] = [];
|
||||
await ensureCodexSkillsInjected(
|
||||
async (_stream, chunk) => {
|
||||
logs.push(chunk);
|
||||
},
|
||||
{
|
||||
skillsHome,
|
||||
skillsEntries: [{ name: "paperclip", source: path.join(currentRepo, "skills", "paperclip") }],
|
||||
},
|
||||
);
|
||||
|
||||
expect(await fs.realpath(path.join(skillsHome, "paperclip"))).toBe(
|
||||
await fs.realpath(path.join(currentRepo, "skills", "paperclip")),
|
||||
);
|
||||
expect(logs.some((line) => line.includes('Repaired Codex skill "paperclip"'))).toBe(true);
|
||||
});
|
||||
|
||||
it("preserves a custom Codex skill symlink outside Paperclip repo checkouts", async () => {
|
||||
const currentRepo = await makeTempDir("paperclip-codex-current-");
|
||||
const customRoot = await makeTempDir("paperclip-codex-custom-");
|
||||
const skillsHome = await makeTempDir("paperclip-codex-home-");
|
||||
cleanupDirs.add(currentRepo);
|
||||
cleanupDirs.add(customRoot);
|
||||
cleanupDirs.add(skillsHome);
|
||||
|
||||
await createPaperclipRepoSkill(currentRepo, "paperclip");
|
||||
await createCustomSkill(customRoot, "paperclip");
|
||||
await fs.symlink(path.join(customRoot, "custom", "paperclip"), path.join(skillsHome, "paperclip"));
|
||||
|
||||
await ensureCodexSkillsInjected(async () => {}, {
|
||||
skillsHome,
|
||||
skillsEntries: [{ name: "paperclip", source: path.join(currentRepo, "skills", "paperclip") }],
|
||||
});
|
||||
|
||||
expect(await fs.realpath(path.join(skillsHome, "paperclip"))).toBe(
|
||||
await fs.realpath(path.join(customRoot, "custom", "paperclip")),
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -93,16 +93,26 @@ describe("shouldResetTaskSessionForWake", () => {
|
||||
expect(shouldResetTaskSessionForWake({ wakeReason: "issue_assigned" })).toBe(true);
|
||||
});
|
||||
|
||||
it("resets session context on timer heartbeats", () => {
|
||||
expect(shouldResetTaskSessionForWake({ wakeSource: "timer" })).toBe(true);
|
||||
it("preserves session context on timer heartbeats", () => {
|
||||
expect(shouldResetTaskSessionForWake({ wakeSource: "timer" })).toBe(false);
|
||||
});
|
||||
|
||||
it("resets session context on manual on-demand invokes", () => {
|
||||
it("preserves session context on manual on-demand invokes by default", () => {
|
||||
expect(
|
||||
shouldResetTaskSessionForWake({
|
||||
wakeSource: "on_demand",
|
||||
wakeTriggerDetail: "manual",
|
||||
}),
|
||||
).toBe(false);
|
||||
});
|
||||
|
||||
it("resets session context when a fresh session is explicitly requested", () => {
|
||||
expect(
|
||||
shouldResetTaskSessionForWake({
|
||||
wakeSource: "on_demand",
|
||||
wakeTriggerDetail: "manual",
|
||||
forceFreshSession: true,
|
||||
}),
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
|
||||
@@ -97,7 +97,11 @@ function requestBaseUrl(req: Request) {
|
||||
|
||||
function readSkillMarkdown(skillName: string): string | null {
|
||||
const normalized = skillName.trim().toLowerCase();
|
||||
if (normalized !== "paperclip" && normalized !== "paperclip-create-agent")
|
||||
if (
|
||||
normalized !== "paperclip" &&
|
||||
normalized !== "paperclip-create-agent" &&
|
||||
normalized !== "para-memory-files"
|
||||
)
|
||||
return null;
|
||||
const moduleDir = path.dirname(fileURLToPath(import.meta.url));
|
||||
const candidates = [
|
||||
@@ -1610,6 +1614,10 @@ export function accessRoutes(
|
||||
res.json({
|
||||
skills: [
|
||||
{ name: "paperclip", path: "/api/skills/paperclip" },
|
||||
{
|
||||
name: "para-memory-files",
|
||||
path: "/api/skills/para-memory-files"
|
||||
},
|
||||
{
|
||||
name: "paperclip-create-agent",
|
||||
path: "/api/skills/paperclip-create-agent"
|
||||
|
||||
@@ -575,6 +575,34 @@ export function agentRoutes(db: Db) {
|
||||
res.json({ ...agent, chainOfCommand });
|
||||
});
|
||||
|
||||
router.get("/agents/me/inbox-lite", async (req, res) => {
|
||||
if (req.actor.type !== "agent" || !req.actor.agentId || !req.actor.companyId) {
|
||||
res.status(401).json({ error: "Agent authentication required" });
|
||||
return;
|
||||
}
|
||||
|
||||
const issuesSvc = issueService(db);
|
||||
const rows = await issuesSvc.list(req.actor.companyId, {
|
||||
assigneeAgentId: req.actor.agentId,
|
||||
status: "todo,in_progress,blocked",
|
||||
});
|
||||
|
||||
res.json(
|
||||
rows.map((issue) => ({
|
||||
id: issue.id,
|
||||
identifier: issue.identifier,
|
||||
title: issue.title,
|
||||
status: issue.status,
|
||||
priority: issue.priority,
|
||||
projectId: issue.projectId,
|
||||
goalId: issue.goalId,
|
||||
parentId: issue.parentId,
|
||||
updatedAt: issue.updatedAt,
|
||||
activeRun: issue.activeRun,
|
||||
})),
|
||||
);
|
||||
});
|
||||
|
||||
router.get("/agents/:id", async (req, res) => {
|
||||
const id = req.params.id as string;
|
||||
const agent = await svc.getById(id);
|
||||
@@ -1275,6 +1303,7 @@ export function agentRoutes(db: Db) {
|
||||
contextSnapshot: {
|
||||
triggeredBy: req.actor.type,
|
||||
actorId: req.actor.type === "agent" ? req.actor.agentId : req.actor.userId,
|
||||
forceFreshSession: req.body.forceFreshSession === true,
|
||||
},
|
||||
});
|
||||
|
||||
|
||||
@@ -28,6 +28,8 @@ import { assertCompanyAccess, getActorInfo } from "./authz.js";
|
||||
import { shouldWakeAssigneeOnCheckout } from "./issues-checkout-wakeup.js";
|
||||
import { isAllowedContentType, MAX_ATTACHMENT_BYTES } from "../attachment-types.js";
|
||||
|
||||
const MAX_ISSUE_COMMENT_LIMIT = 500;
|
||||
|
||||
export function issueRoutes(db: Db, storage: StorageService) {
|
||||
const router = Router();
|
||||
const svc = issueService(db);
|
||||
@@ -314,6 +316,79 @@ export function issueRoutes(db: Db, storage: StorageService) {
|
||||
});
|
||||
});
|
||||
|
||||
router.get("/issues/:id/heartbeat-context", async (req, res) => {
|
||||
const id = req.params.id as string;
|
||||
const issue = await svc.getById(id);
|
||||
if (!issue) {
|
||||
res.status(404).json({ error: "Issue not found" });
|
||||
return;
|
||||
}
|
||||
assertCompanyAccess(req, issue.companyId);
|
||||
|
||||
const wakeCommentId =
|
||||
typeof req.query.wakeCommentId === "string" && req.query.wakeCommentId.trim().length > 0
|
||||
? req.query.wakeCommentId.trim()
|
||||
: null;
|
||||
|
||||
const [ancestors, project, goal, commentCursor, wakeComment] = await Promise.all([
|
||||
svc.getAncestors(issue.id),
|
||||
issue.projectId ? projectsSvc.getById(issue.projectId) : null,
|
||||
issue.goalId
|
||||
? goalsSvc.getById(issue.goalId)
|
||||
: !issue.projectId
|
||||
? goalsSvc.getDefaultCompanyGoal(issue.companyId)
|
||||
: null,
|
||||
svc.getCommentCursor(issue.id),
|
||||
wakeCommentId ? svc.getComment(wakeCommentId) : null,
|
||||
]);
|
||||
|
||||
res.json({
|
||||
issue: {
|
||||
id: issue.id,
|
||||
identifier: issue.identifier,
|
||||
title: issue.title,
|
||||
description: issue.description,
|
||||
status: issue.status,
|
||||
priority: issue.priority,
|
||||
projectId: issue.projectId,
|
||||
goalId: goal?.id ?? issue.goalId,
|
||||
parentId: issue.parentId,
|
||||
assigneeAgentId: issue.assigneeAgentId,
|
||||
assigneeUserId: issue.assigneeUserId,
|
||||
updatedAt: issue.updatedAt,
|
||||
},
|
||||
ancestors: ancestors.map((ancestor) => ({
|
||||
id: ancestor.id,
|
||||
identifier: ancestor.identifier,
|
||||
title: ancestor.title,
|
||||
status: ancestor.status,
|
||||
priority: ancestor.priority,
|
||||
})),
|
||||
project: project
|
||||
? {
|
||||
id: project.id,
|
||||
name: project.name,
|
||||
status: project.status,
|
||||
targetDate: project.targetDate,
|
||||
}
|
||||
: null,
|
||||
goal: goal
|
||||
? {
|
||||
id: goal.id,
|
||||
title: goal.title,
|
||||
status: goal.status,
|
||||
level: goal.level,
|
||||
parentId: goal.parentId,
|
||||
}
|
||||
: null,
|
||||
commentCursor,
|
||||
wakeComment:
|
||||
wakeComment && wakeComment.issueId === issue.id
|
||||
? wakeComment
|
||||
: null,
|
||||
});
|
||||
});
|
||||
|
||||
router.post("/issues/:id/read", async (req, res) => {
|
||||
const id = req.params.id as string;
|
||||
const issue = await svc.getById(id);
|
||||
@@ -791,7 +866,29 @@ export function issueRoutes(db: Db, storage: StorageService) {
|
||||
return;
|
||||
}
|
||||
assertCompanyAccess(req, issue.companyId);
|
||||
const comments = await svc.listComments(id);
|
||||
const afterCommentId =
|
||||
typeof req.query.after === "string" && req.query.after.trim().length > 0
|
||||
? req.query.after.trim()
|
||||
: typeof req.query.afterCommentId === "string" && req.query.afterCommentId.trim().length > 0
|
||||
? req.query.afterCommentId.trim()
|
||||
: null;
|
||||
const order =
|
||||
typeof req.query.order === "string" && req.query.order.trim().toLowerCase() === "asc"
|
||||
? "asc"
|
||||
: "desc";
|
||||
const limitRaw =
|
||||
typeof req.query.limit === "string" && req.query.limit.trim().length > 0
|
||||
? Number(req.query.limit)
|
||||
: null;
|
||||
const limit =
|
||||
limitRaw && Number.isFinite(limitRaw) && limitRaw > 0
|
||||
? Math.min(Math.floor(limitRaw), MAX_ISSUE_COMMENT_LIMIT)
|
||||
: null;
|
||||
const comments = await svc.listComments(id, {
|
||||
afterCommentId,
|
||||
order,
|
||||
limit,
|
||||
});
|
||||
res.json(comments);
|
||||
});
|
||||
|
||||
|
||||
@@ -18,7 +18,7 @@ import { logger } from "../middleware/logger.js";
|
||||
import { publishLiveEvent } from "./live-events.js";
|
||||
import { getRunLogStore, type RunLogHandle } from "./run-log-store.js";
|
||||
import { getServerAdapter, runningProcesses } from "../adapters/index.js";
|
||||
import type { AdapterExecutionResult, AdapterInvocationMeta, AdapterSessionCodec } from "../adapters/index.js";
|
||||
import type { AdapterExecutionResult, AdapterInvocationMeta, AdapterSessionCodec, UsageSummary } from "../adapters/index.js";
|
||||
import { createLocalAgentJwt } from "../agent-auth-jwt.js";
|
||||
import { parseObject, asBoolean, asNumber, appendWithCap, MAX_EXCERPT_BYTES } from "../adapters/utils.js";
|
||||
import { costService } from "./costs.js";
|
||||
@@ -47,6 +47,14 @@ const HEARTBEAT_MAX_CONCURRENT_RUNS_MAX = 10;
|
||||
const DEFERRED_WAKE_CONTEXT_KEY = "_paperclipWakeContext";
|
||||
const startLocksByAgent = new Map<string, Promise<void>>();
|
||||
const REPO_ONLY_CWD_SENTINEL = "/__paperclip_repo_only__";
|
||||
const SESSIONED_LOCAL_ADAPTERS = new Set([
|
||||
"claude_local",
|
||||
"codex_local",
|
||||
"cursor",
|
||||
"gemini_local",
|
||||
"opencode_local",
|
||||
"pi_local",
|
||||
]);
|
||||
|
||||
const heartbeatRunListColumns = {
|
||||
id: heartbeatRuns.id,
|
||||
@@ -117,6 +125,26 @@ interface WakeupOptions {
|
||||
contextSnapshot?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
type UsageTotals = {
|
||||
inputTokens: number;
|
||||
cachedInputTokens: number;
|
||||
outputTokens: number;
|
||||
};
|
||||
|
||||
type SessionCompactionPolicy = {
|
||||
enabled: boolean;
|
||||
maxSessionRuns: number;
|
||||
maxRawInputTokens: number;
|
||||
maxSessionAgeHours: number;
|
||||
};
|
||||
|
||||
type SessionCompactionDecision = {
|
||||
rotate: boolean;
|
||||
reason: string | null;
|
||||
handoffMarkdown: string | null;
|
||||
previousRunId: string | null;
|
||||
};
|
||||
|
||||
interface ParsedIssueAssigneeAdapterOverrides {
|
||||
adapterConfig: Record<string, unknown> | null;
|
||||
useProjectWorkspace: boolean | null;
|
||||
@@ -142,6 +170,88 @@ function readNonEmptyString(value: unknown): string | null {
|
||||
return typeof value === "string" && value.trim().length > 0 ? value : null;
|
||||
}
|
||||
|
||||
function normalizeUsageTotals(usage: UsageSummary | null | undefined): UsageTotals | null {
|
||||
if (!usage) return null;
|
||||
return {
|
||||
inputTokens: Math.max(0, Math.floor(asNumber(usage.inputTokens, 0))),
|
||||
cachedInputTokens: Math.max(0, Math.floor(asNumber(usage.cachedInputTokens, 0))),
|
||||
outputTokens: Math.max(0, Math.floor(asNumber(usage.outputTokens, 0))),
|
||||
};
|
||||
}
|
||||
|
||||
function readRawUsageTotals(usageJson: unknown): UsageTotals | null {
|
||||
const parsed = parseObject(usageJson);
|
||||
if (Object.keys(parsed).length === 0) return null;
|
||||
|
||||
const inputTokens = Math.max(
|
||||
0,
|
||||
Math.floor(asNumber(parsed.rawInputTokens, asNumber(parsed.inputTokens, 0))),
|
||||
);
|
||||
const cachedInputTokens = Math.max(
|
||||
0,
|
||||
Math.floor(asNumber(parsed.rawCachedInputTokens, asNumber(parsed.cachedInputTokens, 0))),
|
||||
);
|
||||
const outputTokens = Math.max(
|
||||
0,
|
||||
Math.floor(asNumber(parsed.rawOutputTokens, asNumber(parsed.outputTokens, 0))),
|
||||
);
|
||||
|
||||
if (inputTokens <= 0 && cachedInputTokens <= 0 && outputTokens <= 0) {
|
||||
return null;
|
||||
}
|
||||
|
||||
return {
|
||||
inputTokens,
|
||||
cachedInputTokens,
|
||||
outputTokens,
|
||||
};
|
||||
}
|
||||
|
||||
function deriveNormalizedUsageDelta(current: UsageTotals | null, previous: UsageTotals | null): UsageTotals | null {
|
||||
if (!current) return null;
|
||||
if (!previous) return { ...current };
|
||||
|
||||
const inputTokens = current.inputTokens >= previous.inputTokens
|
||||
? current.inputTokens - previous.inputTokens
|
||||
: current.inputTokens;
|
||||
const cachedInputTokens = current.cachedInputTokens >= previous.cachedInputTokens
|
||||
? current.cachedInputTokens - previous.cachedInputTokens
|
||||
: current.cachedInputTokens;
|
||||
const outputTokens = current.outputTokens >= previous.outputTokens
|
||||
? current.outputTokens - previous.outputTokens
|
||||
: current.outputTokens;
|
||||
|
||||
return {
|
||||
inputTokens: Math.max(0, inputTokens),
|
||||
cachedInputTokens: Math.max(0, cachedInputTokens),
|
||||
outputTokens: Math.max(0, outputTokens),
|
||||
};
|
||||
}
|
||||
|
||||
function formatCount(value: number | null | undefined) {
|
||||
if (typeof value !== "number" || !Number.isFinite(value)) return "0";
|
||||
return value.toLocaleString("en-US");
|
||||
}
|
||||
|
||||
function parseSessionCompactionPolicy(agent: typeof agents.$inferSelect): SessionCompactionPolicy {
|
||||
const runtimeConfig = parseObject(agent.runtimeConfig);
|
||||
const heartbeat = parseObject(runtimeConfig.heartbeat);
|
||||
const compaction = parseObject(
|
||||
heartbeat.sessionCompaction ?? heartbeat.sessionRotation ?? runtimeConfig.sessionCompaction,
|
||||
);
|
||||
const supportsSessions = SESSIONED_LOCAL_ADAPTERS.has(agent.adapterType);
|
||||
const enabled = compaction.enabled === undefined
|
||||
? supportsSessions
|
||||
: asBoolean(compaction.enabled, supportsSessions);
|
||||
|
||||
return {
|
||||
enabled,
|
||||
maxSessionRuns: Math.max(0, Math.floor(asNumber(compaction.maxSessionRuns, 200))),
|
||||
maxRawInputTokens: Math.max(0, Math.floor(asNumber(compaction.maxRawInputTokens, 2_000_000))),
|
||||
maxSessionAgeHours: Math.max(0, Math.floor(asNumber(compaction.maxSessionAgeHours, 72))),
|
||||
};
|
||||
}
|
||||
|
||||
export function resolveRuntimeSessionParamsForWorkspace(input: {
|
||||
agentId: string;
|
||||
previousSessionParams: Record<string, unknown> | null;
|
||||
@@ -246,29 +356,20 @@ function deriveTaskKey(
|
||||
export function shouldResetTaskSessionForWake(
|
||||
contextSnapshot: Record<string, unknown> | null | undefined,
|
||||
) {
|
||||
if (contextSnapshot?.forceFreshSession === true) return true;
|
||||
|
||||
const wakeReason = readNonEmptyString(contextSnapshot?.wakeReason);
|
||||
if (wakeReason === "issue_assigned") return true;
|
||||
|
||||
const wakeSource = readNonEmptyString(contextSnapshot?.wakeSource);
|
||||
if (wakeSource === "timer") return true;
|
||||
|
||||
const wakeTriggerDetail = readNonEmptyString(contextSnapshot?.wakeTriggerDetail);
|
||||
return wakeSource === "on_demand" && wakeTriggerDetail === "manual";
|
||||
return false;
|
||||
}
|
||||
|
||||
function describeSessionResetReason(
|
||||
contextSnapshot: Record<string, unknown> | null | undefined,
|
||||
) {
|
||||
if (contextSnapshot?.forceFreshSession === true) return "forceFreshSession was requested";
|
||||
|
||||
const wakeReason = readNonEmptyString(contextSnapshot?.wakeReason);
|
||||
if (wakeReason === "issue_assigned") return "wake reason is issue_assigned";
|
||||
|
||||
const wakeSource = readNonEmptyString(contextSnapshot?.wakeSource);
|
||||
if (wakeSource === "timer") return "wake source is timer";
|
||||
|
||||
const wakeTriggerDetail = readNonEmptyString(contextSnapshot?.wakeTriggerDetail);
|
||||
if (wakeSource === "on_demand" && wakeTriggerDetail === "manual") {
|
||||
return "this is a manual invoke";
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -501,6 +602,176 @@ export function heartbeatService(db: Db) {
|
||||
.then((rows) => rows[0] ?? null);
|
||||
}
|
||||
|
||||
async function getLatestRunForSession(
|
||||
agentId: string,
|
||||
sessionId: string,
|
||||
opts?: { excludeRunId?: string | null },
|
||||
) {
|
||||
const conditions = [
|
||||
eq(heartbeatRuns.agentId, agentId),
|
||||
eq(heartbeatRuns.sessionIdAfter, sessionId),
|
||||
];
|
||||
if (opts?.excludeRunId) {
|
||||
conditions.push(sql`${heartbeatRuns.id} <> ${opts.excludeRunId}`);
|
||||
}
|
||||
return db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(and(...conditions))
|
||||
.orderBy(desc(heartbeatRuns.createdAt))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
}
|
||||
|
||||
async function getOldestRunForSession(agentId: string, sessionId: string) {
|
||||
return db
|
||||
.select({
|
||||
id: heartbeatRuns.id,
|
||||
createdAt: heartbeatRuns.createdAt,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.agentId, agentId), eq(heartbeatRuns.sessionIdAfter, sessionId)))
|
||||
.orderBy(asc(heartbeatRuns.createdAt), asc(heartbeatRuns.id))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
}
|
||||
|
||||
async function resolveNormalizedUsageForSession(input: {
|
||||
agentId: string;
|
||||
runId: string;
|
||||
sessionId: string | null;
|
||||
rawUsage: UsageTotals | null;
|
||||
}) {
|
||||
const { agentId, runId, sessionId, rawUsage } = input;
|
||||
if (!sessionId || !rawUsage) {
|
||||
return {
|
||||
normalizedUsage: rawUsage,
|
||||
previousRawUsage: null as UsageTotals | null,
|
||||
derivedFromSessionTotals: false,
|
||||
};
|
||||
}
|
||||
|
||||
const previousRun = await getLatestRunForSession(agentId, sessionId, { excludeRunId: runId });
|
||||
const previousRawUsage = readRawUsageTotals(previousRun?.usageJson);
|
||||
return {
|
||||
normalizedUsage: deriveNormalizedUsageDelta(rawUsage, previousRawUsage),
|
||||
previousRawUsage,
|
||||
derivedFromSessionTotals: previousRawUsage !== null,
|
||||
};
|
||||
}
|
||||
|
||||
async function evaluateSessionCompaction(input: {
|
||||
agent: typeof agents.$inferSelect;
|
||||
sessionId: string | null;
|
||||
issueId: string | null;
|
||||
}): Promise<SessionCompactionDecision> {
|
||||
const { agent, sessionId, issueId } = input;
|
||||
if (!sessionId) {
|
||||
return {
|
||||
rotate: false,
|
||||
reason: null,
|
||||
handoffMarkdown: null,
|
||||
previousRunId: null,
|
||||
};
|
||||
}
|
||||
|
||||
const policy = parseSessionCompactionPolicy(agent);
|
||||
if (!policy.enabled) {
|
||||
return {
|
||||
rotate: false,
|
||||
reason: null,
|
||||
handoffMarkdown: null,
|
||||
previousRunId: null,
|
||||
};
|
||||
}
|
||||
|
||||
const fetchLimit = Math.max(policy.maxSessionRuns > 0 ? policy.maxSessionRuns + 1 : 0, 4);
|
||||
const runs = await db
|
||||
.select({
|
||||
id: heartbeatRuns.id,
|
||||
createdAt: heartbeatRuns.createdAt,
|
||||
usageJson: heartbeatRuns.usageJson,
|
||||
resultJson: heartbeatRuns.resultJson,
|
||||
error: heartbeatRuns.error,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.agentId, agent.id), eq(heartbeatRuns.sessionIdAfter, sessionId)))
|
||||
.orderBy(desc(heartbeatRuns.createdAt))
|
||||
.limit(fetchLimit);
|
||||
|
||||
if (runs.length === 0) {
|
||||
return {
|
||||
rotate: false,
|
||||
reason: null,
|
||||
handoffMarkdown: null,
|
||||
previousRunId: null,
|
||||
};
|
||||
}
|
||||
|
||||
const latestRun = runs[0] ?? null;
|
||||
const oldestRun =
|
||||
policy.maxSessionAgeHours > 0
|
||||
? await getOldestRunForSession(agent.id, sessionId)
|
||||
: runs[runs.length - 1] ?? latestRun;
|
||||
const latestRawUsage = readRawUsageTotals(latestRun?.usageJson);
|
||||
const sessionAgeHours =
|
||||
latestRun && oldestRun
|
||||
? Math.max(
|
||||
0,
|
||||
(new Date(latestRun.createdAt).getTime() - new Date(oldestRun.createdAt).getTime()) / (1000 * 60 * 60),
|
||||
)
|
||||
: 0;
|
||||
|
||||
let reason: string | null = null;
|
||||
if (policy.maxSessionRuns > 0 && runs.length > policy.maxSessionRuns) {
|
||||
reason = `session exceeded ${policy.maxSessionRuns} runs`;
|
||||
} else if (
|
||||
policy.maxRawInputTokens > 0 &&
|
||||
latestRawUsage &&
|
||||
latestRawUsage.inputTokens >= policy.maxRawInputTokens
|
||||
) {
|
||||
reason =
|
||||
`session raw input reached ${formatCount(latestRawUsage.inputTokens)} tokens ` +
|
||||
`(threshold ${formatCount(policy.maxRawInputTokens)})`;
|
||||
} else if (policy.maxSessionAgeHours > 0 && sessionAgeHours >= policy.maxSessionAgeHours) {
|
||||
reason = `session age reached ${Math.floor(sessionAgeHours)} hours`;
|
||||
}
|
||||
|
||||
if (!reason || !latestRun) {
|
||||
return {
|
||||
rotate: false,
|
||||
reason: null,
|
||||
handoffMarkdown: null,
|
||||
previousRunId: latestRun?.id ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
const latestSummary = summarizeHeartbeatRunResultJson(latestRun.resultJson);
|
||||
const latestTextSummary =
|
||||
readNonEmptyString(latestSummary?.summary) ??
|
||||
readNonEmptyString(latestSummary?.result) ??
|
||||
readNonEmptyString(latestSummary?.message) ??
|
||||
readNonEmptyString(latestRun.error);
|
||||
|
||||
const handoffMarkdown = [
|
||||
"Paperclip session handoff:",
|
||||
`- Previous session: ${sessionId}`,
|
||||
issueId ? `- Issue: ${issueId}` : "",
|
||||
`- Rotation reason: ${reason}`,
|
||||
latestTextSummary ? `- Last run summary: ${latestTextSummary}` : "",
|
||||
"Continue from the current task state. Rebuild only the minimum context you need.",
|
||||
]
|
||||
.filter(Boolean)
|
||||
.join("\n");
|
||||
|
||||
return {
|
||||
rotate: true,
|
||||
reason,
|
||||
handoffMarkdown,
|
||||
previousRunId: latestRun.id,
|
||||
};
|
||||
}
|
||||
|
||||
async function resolveSessionBeforeForWakeup(
|
||||
agent: typeof agents.$inferSelect,
|
||||
taskKey: string | null,
|
||||
@@ -1016,9 +1287,10 @@ export function heartbeatService(db: Db) {
|
||||
run: typeof heartbeatRuns.$inferSelect,
|
||||
result: AdapterExecutionResult,
|
||||
session: { legacySessionId: string | null },
|
||||
normalizedUsage?: UsageTotals | null,
|
||||
) {
|
||||
await ensureRuntimeState(agent);
|
||||
const usage = result.usage;
|
||||
const usage = normalizedUsage ?? normalizeUsageTotals(result.usage);
|
||||
const inputTokens = usage?.inputTokens ?? 0;
|
||||
const outputTokens = usage?.outputTokens ?? 0;
|
||||
const cachedInputTokens = usage?.cachedInputTokens ?? 0;
|
||||
@@ -1270,15 +1542,42 @@ export function heartbeatService(db: Db) {
|
||||
context.projectId = executionWorkspace.projectId;
|
||||
}
|
||||
const runtimeSessionFallback = taskKey || resetTaskSession ? null : runtime.sessionId;
|
||||
const previousSessionDisplayId = truncateDisplayId(
|
||||
let previousSessionDisplayId = truncateDisplayId(
|
||||
taskSessionForRun?.sessionDisplayId ??
|
||||
(sessionCodec.getDisplayId ? sessionCodec.getDisplayId(runtimeSessionParams) : null) ??
|
||||
readNonEmptyString(runtimeSessionParams?.sessionId) ??
|
||||
runtimeSessionFallback,
|
||||
);
|
||||
let runtimeSessionIdForAdapter =
|
||||
readNonEmptyString(runtimeSessionParams?.sessionId) ?? runtimeSessionFallback;
|
||||
let runtimeSessionParamsForAdapter = runtimeSessionParams;
|
||||
|
||||
const sessionCompaction = await evaluateSessionCompaction({
|
||||
agent,
|
||||
sessionId: previousSessionDisplayId ?? runtimeSessionIdForAdapter,
|
||||
issueId,
|
||||
});
|
||||
if (sessionCompaction.rotate) {
|
||||
context.paperclipSessionHandoffMarkdown = sessionCompaction.handoffMarkdown;
|
||||
context.paperclipSessionRotationReason = sessionCompaction.reason;
|
||||
context.paperclipPreviousSessionId = previousSessionDisplayId ?? runtimeSessionIdForAdapter;
|
||||
runtimeSessionIdForAdapter = null;
|
||||
runtimeSessionParamsForAdapter = null;
|
||||
previousSessionDisplayId = null;
|
||||
if (sessionCompaction.reason) {
|
||||
runtimeWorkspaceWarnings.push(
|
||||
`Starting a fresh session because ${sessionCompaction.reason}.`,
|
||||
);
|
||||
}
|
||||
} else {
|
||||
delete context.paperclipSessionHandoffMarkdown;
|
||||
delete context.paperclipSessionRotationReason;
|
||||
delete context.paperclipPreviousSessionId;
|
||||
}
|
||||
|
||||
const runtimeForAdapter = {
|
||||
sessionId: readNonEmptyString(runtimeSessionParams?.sessionId) ?? runtimeSessionFallback,
|
||||
sessionParams: runtimeSessionParams,
|
||||
sessionId: runtimeSessionIdForAdapter,
|
||||
sessionParams: runtimeSessionParamsForAdapter,
|
||||
sessionDisplayId: previousSessionDisplayId,
|
||||
taskKey,
|
||||
};
|
||||
@@ -1522,6 +1821,14 @@ export function heartbeatService(db: Db) {
|
||||
previousDisplayId: runtimeForAdapter.sessionDisplayId,
|
||||
previousLegacySessionId: runtimeForAdapter.sessionId,
|
||||
});
|
||||
const rawUsage = normalizeUsageTotals(adapterResult.usage);
|
||||
const sessionUsageResolution = await resolveNormalizedUsageForSession({
|
||||
agentId: agent.id,
|
||||
runId: run.id,
|
||||
sessionId: nextSessionState.displayId ?? nextSessionState.legacySessionId,
|
||||
rawUsage,
|
||||
});
|
||||
const normalizedUsage = sessionUsageResolution.normalizedUsage;
|
||||
|
||||
let outcome: "succeeded" | "failed" | "cancelled" | "timed_out";
|
||||
const latestRun = await getRun(run.id);
|
||||
@@ -1550,9 +1857,23 @@ export function heartbeatService(db: Db) {
|
||||
: "failed";
|
||||
|
||||
const usageJson =
|
||||
adapterResult.usage || adapterResult.costUsd != null
|
||||
normalizedUsage || adapterResult.costUsd != null
|
||||
? ({
|
||||
...(adapterResult.usage ?? {}),
|
||||
...(normalizedUsage ?? {}),
|
||||
...(rawUsage ? {
|
||||
rawInputTokens: rawUsage.inputTokens,
|
||||
rawCachedInputTokens: rawUsage.cachedInputTokens,
|
||||
rawOutputTokens: rawUsage.outputTokens,
|
||||
} : {}),
|
||||
...(sessionUsageResolution.derivedFromSessionTotals ? { usageSource: "session_delta" } : {}),
|
||||
...((nextSessionState.displayId ?? nextSessionState.legacySessionId)
|
||||
? { persistedSessionId: nextSessionState.displayId ?? nextSessionState.legacySessionId }
|
||||
: {}),
|
||||
sessionReused: runtimeForAdapter.sessionId != null || runtimeForAdapter.sessionDisplayId != null,
|
||||
taskSessionReused: taskSessionForRun != null,
|
||||
freshSession: runtimeForAdapter.sessionId == null && runtimeForAdapter.sessionDisplayId == null,
|
||||
sessionRotated: sessionCompaction.rotate,
|
||||
sessionRotationReason: sessionCompaction.reason,
|
||||
...(adapterResult.costUsd != null ? { costUsd: adapterResult.costUsd } : {}),
|
||||
...(adapterResult.billingType ? { billingType: adapterResult.billingType } : {}),
|
||||
} as Record<string, unknown>)
|
||||
@@ -1609,7 +1930,7 @@ export function heartbeatService(db: Db) {
|
||||
if (finalizedRun) {
|
||||
await updateRuntimeState(agent, finalizedRun, adapterResult, {
|
||||
legacySessionId: nextSessionState.legacySessionId,
|
||||
});
|
||||
}, normalizedUsage);
|
||||
if (taskKey) {
|
||||
if (adapterResult.clearSession || (!nextSessionState.params && !nextSessionState.displayId)) {
|
||||
await clearTaskSessions(agent.companyId, agent.id, {
|
||||
|
||||
@@ -27,6 +27,7 @@ import { resolveIssueGoalId, resolveNextIssueGoalId } from "./issue-goal-fallbac
|
||||
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;
|
||||
|
||||
function assertTransition(from: string, to: string) {
|
||||
if (from === to) return;
|
||||
@@ -1060,13 +1061,86 @@ export function issueService(db: Db) {
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null),
|
||||
|
||||
listComments: (issueId: string) =>
|
||||
db
|
||||
listComments: async (
|
||||
issueId: string,
|
||||
opts?: {
|
||||
afterCommentId?: string | null;
|
||||
order?: "asc" | "desc";
|
||||
limit?: number | null;
|
||||
},
|
||||
) => {
|
||||
const order = opts?.order === "asc" ? "asc" : "desc";
|
||||
const afterCommentId = opts?.afterCommentId?.trim() || null;
|
||||
const limit =
|
||||
opts?.limit && opts.limit > 0
|
||||
? Math.min(Math.floor(opts.limit), MAX_ISSUE_COMMENT_PAGE_LIMIT)
|
||||
: null;
|
||||
|
||||
const conditions = [eq(issueComments.issueId, issueId)];
|
||||
if (afterCommentId) {
|
||||
const anchor = await db
|
||||
.select({
|
||||
id: issueComments.id,
|
||||
createdAt: issueComments.createdAt,
|
||||
})
|
||||
.from(issueComments)
|
||||
.where(and(eq(issueComments.issueId, issueId), eq(issueComments.id, afterCommentId)))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
|
||||
if (!anchor) return [];
|
||||
conditions.push(
|
||||
order === "asc"
|
||||
? sql<boolean>`(
|
||||
${issueComments.createdAt} > ${anchor.createdAt}
|
||||
OR (${issueComments.createdAt} = ${anchor.createdAt} AND ${issueComments.id} > ${anchor.id})
|
||||
)`
|
||||
: sql<boolean>`(
|
||||
${issueComments.createdAt} < ${anchor.createdAt}
|
||||
OR (${issueComments.createdAt} = ${anchor.createdAt} AND ${issueComments.id} < ${anchor.id})
|
||||
)`,
|
||||
);
|
||||
}
|
||||
|
||||
const query = db
|
||||
.select()
|
||||
.from(issueComments)
|
||||
.where(eq(issueComments.issueId, issueId))
|
||||
.orderBy(desc(issueComments.createdAt))
|
||||
.then((comments) => comments.map(redactIssueComment)),
|
||||
.where(and(...conditions))
|
||||
.orderBy(
|
||||
order === "asc" ? asc(issueComments.createdAt) : desc(issueComments.createdAt),
|
||||
order === "asc" ? asc(issueComments.id) : desc(issueComments.id),
|
||||
);
|
||||
|
||||
const comments = limit ? await query.limit(limit) : await query;
|
||||
return comments.map(redactIssueComment);
|
||||
},
|
||||
|
||||
getCommentCursor: async (issueId: string) => {
|
||||
const [latest, countRow] = await Promise.all([
|
||||
db
|
||||
.select({
|
||||
latestCommentId: issueComments.id,
|
||||
latestCommentAt: issueComments.createdAt,
|
||||
})
|
||||
.from(issueComments)
|
||||
.where(eq(issueComments.issueId, issueId))
|
||||
.orderBy(desc(issueComments.createdAt), desc(issueComments.id))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null),
|
||||
db
|
||||
.select({
|
||||
totalComments: sql<number>`count(*)::int`,
|
||||
})
|
||||
.from(issueComments)
|
||||
.where(eq(issueComments.issueId, issueId))
|
||||
.then((rows) => rows[0] ?? null),
|
||||
]);
|
||||
|
||||
return {
|
||||
totalComments: Number(countRow?.totalComments ?? 0),
|
||||
latestCommentId: latest?.latestCommentId ?? null,
|
||||
latestCommentAt: latest?.latestCommentAt ?? null,
|
||||
};
|
||||
},
|
||||
|
||||
getComment: (commentId: string) =>
|
||||
db
|
||||
|
||||
Reference in New Issue
Block a user