Remove api trigger kind and mark webhook as coming soon
Drop "api" from the trigger kind dropdown and disable the "webhook" option with a "COMING SOON" label until it's ready. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
246
server/src/__tests__/agent-permissions-routes.test.ts
Normal file
246
server/src/__tests__/agent-permissions-routes.test.ts
Normal file
@@ -0,0 +1,246 @@
|
||||
import express from "express";
|
||||
import request from "supertest";
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { agentRoutes } from "../routes/agents.js";
|
||||
import { errorHandler } from "../middleware/index.js";
|
||||
|
||||
const agentId = "11111111-1111-4111-8111-111111111111";
|
||||
const companyId = "22222222-2222-4222-8222-222222222222";
|
||||
|
||||
const baseAgent = {
|
||||
id: agentId,
|
||||
companyId,
|
||||
name: "Builder",
|
||||
urlKey: "builder",
|
||||
role: "engineer",
|
||||
title: "Builder",
|
||||
icon: null,
|
||||
status: "idle",
|
||||
reportsTo: null,
|
||||
capabilities: null,
|
||||
adapterType: "process",
|
||||
adapterConfig: {},
|
||||
runtimeConfig: {},
|
||||
budgetMonthlyCents: 0,
|
||||
spentMonthlyCents: 0,
|
||||
pauseReason: null,
|
||||
pausedAt: null,
|
||||
permissions: { canCreateAgents: false },
|
||||
lastHeartbeatAt: null,
|
||||
metadata: null,
|
||||
createdAt: new Date("2026-03-19T00:00:00.000Z"),
|
||||
updatedAt: new Date("2026-03-19T00:00:00.000Z"),
|
||||
};
|
||||
|
||||
const mockAgentService = vi.hoisted(() => ({
|
||||
getById: vi.fn(),
|
||||
create: vi.fn(),
|
||||
updatePermissions: vi.fn(),
|
||||
getChainOfCommand: vi.fn(),
|
||||
resolveByReference: vi.fn(),
|
||||
}));
|
||||
|
||||
const mockAccessService = vi.hoisted(() => ({
|
||||
canUser: vi.fn(),
|
||||
hasPermission: vi.fn(),
|
||||
getMembership: vi.fn(),
|
||||
ensureMembership: vi.fn(),
|
||||
listPrincipalGrants: vi.fn(),
|
||||
setPrincipalPermission: vi.fn(),
|
||||
}));
|
||||
|
||||
const mockApprovalService = vi.hoisted(() => ({
|
||||
create: vi.fn(),
|
||||
getById: vi.fn(),
|
||||
}));
|
||||
|
||||
const mockBudgetService = vi.hoisted(() => ({
|
||||
upsertPolicy: vi.fn(),
|
||||
}));
|
||||
|
||||
const mockHeartbeatService = vi.hoisted(() => ({
|
||||
listTaskSessions: vi.fn(),
|
||||
resetRuntimeSession: vi.fn(),
|
||||
}));
|
||||
|
||||
const mockIssueApprovalService = vi.hoisted(() => ({
|
||||
linkManyForApproval: vi.fn(),
|
||||
}));
|
||||
|
||||
const mockIssueService = vi.hoisted(() => ({
|
||||
list: vi.fn(),
|
||||
}));
|
||||
|
||||
const mockSecretService = vi.hoisted(() => ({
|
||||
normalizeAdapterConfigForPersistence: vi.fn(),
|
||||
resolveAdapterConfigForRuntime: vi.fn(),
|
||||
}));
|
||||
|
||||
const mockWorkspaceOperationService = vi.hoisted(() => ({}));
|
||||
const mockLogActivity = vi.hoisted(() => vi.fn());
|
||||
|
||||
vi.mock("../services/index.js", () => ({
|
||||
agentService: () => mockAgentService,
|
||||
accessService: () => mockAccessService,
|
||||
approvalService: () => mockApprovalService,
|
||||
budgetService: () => mockBudgetService,
|
||||
heartbeatService: () => mockHeartbeatService,
|
||||
issueApprovalService: () => mockIssueApprovalService,
|
||||
issueService: () => mockIssueService,
|
||||
logActivity: mockLogActivity,
|
||||
secretService: () => mockSecretService,
|
||||
workspaceOperationService: () => mockWorkspaceOperationService,
|
||||
}));
|
||||
|
||||
function createDbStub() {
|
||||
return {
|
||||
select: vi.fn().mockReturnValue({
|
||||
from: vi.fn().mockReturnValue({
|
||||
where: vi.fn().mockReturnValue({
|
||||
then: vi.fn().mockResolvedValue([{
|
||||
id: companyId,
|
||||
name: "Paperclip",
|
||||
requireBoardApprovalForNewAgents: false,
|
||||
}]),
|
||||
}),
|
||||
}),
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
function createApp(actor: Record<string, unknown>) {
|
||||
const app = express();
|
||||
app.use(express.json());
|
||||
app.use((req, _res, next) => {
|
||||
(req as any).actor = actor;
|
||||
next();
|
||||
});
|
||||
app.use("/api", agentRoutes(createDbStub() as any));
|
||||
app.use(errorHandler);
|
||||
return app;
|
||||
}
|
||||
|
||||
describe("agent permission routes", () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
mockAgentService.getById.mockResolvedValue(baseAgent);
|
||||
mockAgentService.getChainOfCommand.mockResolvedValue([]);
|
||||
mockAgentService.resolveByReference.mockResolvedValue({ ambiguous: false, agent: baseAgent });
|
||||
mockAgentService.create.mockResolvedValue(baseAgent);
|
||||
mockAgentService.updatePermissions.mockResolvedValue(baseAgent);
|
||||
mockAccessService.getMembership.mockResolvedValue({
|
||||
id: "membership-1",
|
||||
companyId,
|
||||
principalType: "agent",
|
||||
principalId: agentId,
|
||||
status: "active",
|
||||
membershipRole: "member",
|
||||
createdAt: new Date("2026-03-19T00:00:00.000Z"),
|
||||
updatedAt: new Date("2026-03-19T00:00:00.000Z"),
|
||||
});
|
||||
mockAccessService.listPrincipalGrants.mockResolvedValue([]);
|
||||
mockAccessService.ensureMembership.mockResolvedValue(undefined);
|
||||
mockAccessService.setPrincipalPermission.mockResolvedValue(undefined);
|
||||
mockBudgetService.upsertPolicy.mockResolvedValue(undefined);
|
||||
mockSecretService.normalizeAdapterConfigForPersistence.mockImplementation(async (_companyId, config) => config);
|
||||
mockSecretService.resolveAdapterConfigForRuntime.mockImplementation(async (_companyId, config) => ({ config }));
|
||||
mockLogActivity.mockResolvedValue(undefined);
|
||||
});
|
||||
|
||||
it("grants tasks:assign by default when board creates a new agent", async () => {
|
||||
const app = createApp({
|
||||
type: "board",
|
||||
userId: "board-user",
|
||||
source: "local_implicit",
|
||||
isInstanceAdmin: true,
|
||||
companyIds: [companyId],
|
||||
});
|
||||
|
||||
const res = await request(app)
|
||||
.post(`/api/companies/${companyId}/agents`)
|
||||
.send({
|
||||
name: "Builder",
|
||||
role: "engineer",
|
||||
adapterType: "process",
|
||||
adapterConfig: {},
|
||||
});
|
||||
|
||||
expect(res.status).toBe(201);
|
||||
expect(mockAccessService.ensureMembership).toHaveBeenCalledWith(
|
||||
companyId,
|
||||
"agent",
|
||||
agentId,
|
||||
"member",
|
||||
"active",
|
||||
);
|
||||
expect(mockAccessService.setPrincipalPermission).toHaveBeenCalledWith(
|
||||
companyId,
|
||||
"agent",
|
||||
agentId,
|
||||
"tasks:assign",
|
||||
true,
|
||||
"board-user",
|
||||
);
|
||||
});
|
||||
|
||||
it("exposes explicit task assignment access on agent detail", async () => {
|
||||
mockAccessService.listPrincipalGrants.mockResolvedValue([
|
||||
{
|
||||
id: "grant-1",
|
||||
companyId,
|
||||
principalType: "agent",
|
||||
principalId: agentId,
|
||||
permissionKey: "tasks:assign",
|
||||
scope: null,
|
||||
grantedByUserId: "board-user",
|
||||
createdAt: new Date("2026-03-19T00:00:00.000Z"),
|
||||
updatedAt: new Date("2026-03-19T00:00:00.000Z"),
|
||||
},
|
||||
]);
|
||||
|
||||
const app = createApp({
|
||||
type: "board",
|
||||
userId: "board-user",
|
||||
source: "local_implicit",
|
||||
isInstanceAdmin: true,
|
||||
companyIds: [companyId],
|
||||
});
|
||||
|
||||
const res = await request(app).get(`/api/agents/${agentId}`);
|
||||
|
||||
expect(res.status).toBe(200);
|
||||
expect(res.body.access.canAssignTasks).toBe(true);
|
||||
expect(res.body.access.taskAssignSource).toBe("explicit_grant");
|
||||
});
|
||||
|
||||
it("keeps task assignment enabled when agent creation privilege is enabled", async () => {
|
||||
mockAgentService.updatePermissions.mockResolvedValue({
|
||||
...baseAgent,
|
||||
permissions: { canCreateAgents: true },
|
||||
});
|
||||
|
||||
const app = createApp({
|
||||
type: "board",
|
||||
userId: "board-user",
|
||||
source: "local_implicit",
|
||||
isInstanceAdmin: true,
|
||||
companyIds: [companyId],
|
||||
});
|
||||
|
||||
const res = await request(app)
|
||||
.patch(`/api/agents/${agentId}/permissions`)
|
||||
.send({ canCreateAgents: true, canAssignTasks: false });
|
||||
|
||||
expect(res.status).toBe(200);
|
||||
expect(mockAccessService.setPrincipalPermission).toHaveBeenCalledWith(
|
||||
companyId,
|
||||
"agent",
|
||||
agentId,
|
||||
"tasks:assign",
|
||||
true,
|
||||
"board-user",
|
||||
);
|
||||
expect(res.body.access.canAssignTasks).toBe(true);
|
||||
expect(res.body.access.taskAssignSource).toBe("agent_creator");
|
||||
});
|
||||
});
|
||||
321
server/src/__tests__/heartbeat-process-recovery.test.ts
Normal file
321
server/src/__tests__/heartbeat-process-recovery.test.ts
Normal file
@@ -0,0 +1,321 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import fs from "node:fs";
|
||||
import net from "node:net";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { spawn, type ChildProcess } from "node:child_process";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
|
||||
import {
|
||||
applyPendingMigrations,
|
||||
createDb,
|
||||
ensurePostgresDatabase,
|
||||
agents,
|
||||
agentWakeupRequests,
|
||||
companies,
|
||||
heartbeatRunEvents,
|
||||
heartbeatRuns,
|
||||
issues,
|
||||
} from "@paperclipai/db";
|
||||
import { runningProcesses } from "../adapters/index.ts";
|
||||
import { heartbeatService } from "../services/heartbeat.ts";
|
||||
|
||||
type EmbeddedPostgresInstance = {
|
||||
initialise(): Promise<void>;
|
||||
start(): Promise<void>;
|
||||
stop(): Promise<void>;
|
||||
};
|
||||
|
||||
type EmbeddedPostgresCtor = new (opts: {
|
||||
databaseDir: string;
|
||||
user: string;
|
||||
password: string;
|
||||
port: number;
|
||||
persistent: boolean;
|
||||
initdbFlags?: string[];
|
||||
onLog?: (message: unknown) => void;
|
||||
onError?: (message: unknown) => void;
|
||||
}) => EmbeddedPostgresInstance;
|
||||
|
||||
async function getEmbeddedPostgresCtor(): Promise<EmbeddedPostgresCtor> {
|
||||
const mod = await import("embedded-postgres");
|
||||
return mod.default as EmbeddedPostgresCtor;
|
||||
}
|
||||
|
||||
async function getAvailablePort(): Promise<number> {
|
||||
return await new Promise((resolve, reject) => {
|
||||
const server = net.createServer();
|
||||
server.unref();
|
||||
server.on("error", reject);
|
||||
server.listen(0, "127.0.0.1", () => {
|
||||
const address = server.address();
|
||||
if (!address || typeof address === "string") {
|
||||
server.close(() => reject(new Error("Failed to allocate test port")));
|
||||
return;
|
||||
}
|
||||
const { port } = address;
|
||||
server.close((error) => {
|
||||
if (error) reject(error);
|
||||
else resolve(port);
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
async function startTempDatabase() {
|
||||
const dataDir = fs.mkdtempSync(path.join(os.tmpdir(), "paperclip-heartbeat-recovery-"));
|
||||
const port = await getAvailablePort();
|
||||
const EmbeddedPostgres = await getEmbeddedPostgresCtor();
|
||||
const instance = new EmbeddedPostgres({
|
||||
databaseDir: dataDir,
|
||||
user: "paperclip",
|
||||
password: "paperclip",
|
||||
port,
|
||||
persistent: true,
|
||||
initdbFlags: ["--encoding=UTF8", "--locale=C"],
|
||||
onLog: () => {},
|
||||
onError: () => {},
|
||||
});
|
||||
await instance.initialise();
|
||||
await instance.start();
|
||||
|
||||
const adminConnectionString = `postgres://paperclip:paperclip@127.0.0.1:${port}/postgres`;
|
||||
await ensurePostgresDatabase(adminConnectionString, "paperclip");
|
||||
const connectionString = `postgres://paperclip:paperclip@127.0.0.1:${port}/paperclip`;
|
||||
await applyPendingMigrations(connectionString);
|
||||
return { connectionString, instance, dataDir };
|
||||
}
|
||||
|
||||
function spawnAliveProcess() {
|
||||
return spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], {
|
||||
stdio: "ignore",
|
||||
});
|
||||
}
|
||||
|
||||
describe("heartbeat orphaned process recovery", () => {
|
||||
let db!: ReturnType<typeof createDb>;
|
||||
let instance: EmbeddedPostgresInstance | null = null;
|
||||
let dataDir = "";
|
||||
const childProcesses = new Set<ChildProcess>();
|
||||
|
||||
beforeAll(async () => {
|
||||
const started = await startTempDatabase();
|
||||
db = createDb(started.connectionString);
|
||||
instance = started.instance;
|
||||
dataDir = started.dataDir;
|
||||
}, 20_000);
|
||||
|
||||
afterEach(async () => {
|
||||
runningProcesses.clear();
|
||||
for (const child of childProcesses) {
|
||||
child.kill("SIGKILL");
|
||||
}
|
||||
childProcesses.clear();
|
||||
await db.delete(issues);
|
||||
await db.delete(heartbeatRunEvents);
|
||||
await db.delete(heartbeatRuns);
|
||||
await db.delete(agentWakeupRequests);
|
||||
await db.delete(agents);
|
||||
await db.delete(companies);
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
for (const child of childProcesses) {
|
||||
child.kill("SIGKILL");
|
||||
}
|
||||
childProcesses.clear();
|
||||
runningProcesses.clear();
|
||||
await instance?.stop();
|
||||
if (dataDir) {
|
||||
fs.rmSync(dataDir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
async function seedRunFixture(input?: {
|
||||
adapterType?: string;
|
||||
runStatus?: "running" | "queued" | "failed";
|
||||
processPid?: number | null;
|
||||
processLossRetryCount?: number;
|
||||
includeIssue?: boolean;
|
||||
runErrorCode?: string | null;
|
||||
runError?: string | null;
|
||||
}) {
|
||||
const companyId = randomUUID();
|
||||
const agentId = randomUUID();
|
||||
const runId = randomUUID();
|
||||
const wakeupRequestId = randomUUID();
|
||||
const issueId = randomUUID();
|
||||
const now = new Date("2026-03-19T00:00:00.000Z");
|
||||
const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`;
|
||||
|
||||
await db.insert(companies).values({
|
||||
id: companyId,
|
||||
name: "Paperclip",
|
||||
issuePrefix,
|
||||
requireBoardApprovalForNewAgents: false,
|
||||
});
|
||||
|
||||
await db.insert(agents).values({
|
||||
id: agentId,
|
||||
companyId,
|
||||
name: "CodexCoder",
|
||||
role: "engineer",
|
||||
status: "paused",
|
||||
adapterType: input?.adapterType ?? "codex_local",
|
||||
adapterConfig: {},
|
||||
runtimeConfig: {},
|
||||
permissions: {},
|
||||
});
|
||||
|
||||
await db.insert(agentWakeupRequests).values({
|
||||
id: wakeupRequestId,
|
||||
companyId,
|
||||
agentId,
|
||||
source: "assignment",
|
||||
triggerDetail: "system",
|
||||
reason: "issue_assigned",
|
||||
payload: input?.includeIssue === false ? {} : { issueId },
|
||||
status: "claimed",
|
||||
runId,
|
||||
claimedAt: now,
|
||||
});
|
||||
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: runId,
|
||||
companyId,
|
||||
agentId,
|
||||
invocationSource: "assignment",
|
||||
triggerDetail: "system",
|
||||
status: input?.runStatus ?? "running",
|
||||
wakeupRequestId,
|
||||
contextSnapshot: input?.includeIssue === false ? {} : { issueId },
|
||||
processPid: input?.processPid ?? null,
|
||||
processLossRetryCount: input?.processLossRetryCount ?? 0,
|
||||
errorCode: input?.runErrorCode ?? null,
|
||||
error: input?.runError ?? null,
|
||||
startedAt: now,
|
||||
updatedAt: new Date("2026-03-19T00:00:00.000Z"),
|
||||
});
|
||||
|
||||
if (input?.includeIssue !== false) {
|
||||
await db.insert(issues).values({
|
||||
id: issueId,
|
||||
companyId,
|
||||
title: "Recover local adapter after lost process",
|
||||
status: "in_progress",
|
||||
priority: "medium",
|
||||
assigneeAgentId: agentId,
|
||||
checkoutRunId: runId,
|
||||
executionRunId: runId,
|
||||
issueNumber: 1,
|
||||
identifier: `${issuePrefix}-1`,
|
||||
});
|
||||
}
|
||||
|
||||
return { companyId, agentId, runId, wakeupRequestId, issueId };
|
||||
}
|
||||
|
||||
it("keeps a local run active when the recorded pid is still alive", async () => {
|
||||
const child = spawnAliveProcess();
|
||||
childProcesses.add(child);
|
||||
expect(child.pid).toBeTypeOf("number");
|
||||
|
||||
const { runId, wakeupRequestId } = await seedRunFixture({
|
||||
processPid: child.pid ?? null,
|
||||
includeIssue: false,
|
||||
});
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
const result = await heartbeat.reapOrphanedRuns();
|
||||
expect(result.reaped).toBe(0);
|
||||
|
||||
const run = await heartbeat.getRun(runId);
|
||||
expect(run?.status).toBe("running");
|
||||
expect(run?.errorCode).toBe("process_detached");
|
||||
expect(run?.error).toContain(String(child.pid));
|
||||
|
||||
const wakeup = await db
|
||||
.select()
|
||||
.from(agentWakeupRequests)
|
||||
.where(eq(agentWakeupRequests.id, wakeupRequestId))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
expect(wakeup?.status).toBe("claimed");
|
||||
});
|
||||
|
||||
it("queues exactly one retry when the recorded local pid is dead", async () => {
|
||||
const { agentId, runId, issueId } = await seedRunFixture({
|
||||
processPid: 999_999_999,
|
||||
});
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
const result = await heartbeat.reapOrphanedRuns();
|
||||
expect(result.reaped).toBe(1);
|
||||
expect(result.runIds).toEqual([runId]);
|
||||
|
||||
const runs = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.agentId, agentId));
|
||||
expect(runs).toHaveLength(2);
|
||||
|
||||
const failedRun = runs.find((row) => row.id === runId);
|
||||
const retryRun = runs.find((row) => row.id !== runId);
|
||||
expect(failedRun?.status).toBe("failed");
|
||||
expect(failedRun?.errorCode).toBe("process_lost");
|
||||
expect(retryRun?.status).toBe("queued");
|
||||
expect(retryRun?.retryOfRunId).toBe(runId);
|
||||
expect(retryRun?.processLossRetryCount).toBe(1);
|
||||
|
||||
const issue = await db
|
||||
.select()
|
||||
.from(issues)
|
||||
.where(eq(issues.id, issueId))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
expect(issue?.executionRunId).toBe(retryRun?.id ?? null);
|
||||
expect(issue?.checkoutRunId).toBe(runId);
|
||||
});
|
||||
|
||||
it("does not queue a second retry after the first process-loss retry was already used", async () => {
|
||||
const { agentId, runId, issueId } = await seedRunFixture({
|
||||
processPid: 999_999_999,
|
||||
processLossRetryCount: 1,
|
||||
});
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
const result = await heartbeat.reapOrphanedRuns();
|
||||
expect(result.reaped).toBe(1);
|
||||
expect(result.runIds).toEqual([runId]);
|
||||
|
||||
const runs = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.agentId, agentId));
|
||||
expect(runs).toHaveLength(1);
|
||||
expect(runs[0]?.status).toBe("failed");
|
||||
|
||||
const issue = await db
|
||||
.select()
|
||||
.from(issues)
|
||||
.where(eq(issues.id, issueId))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
expect(issue?.executionRunId).toBeNull();
|
||||
expect(issue?.checkoutRunId).toBe(runId);
|
||||
});
|
||||
|
||||
it("clears the detached warning when the run reports activity again", async () => {
|
||||
const { runId } = await seedRunFixture({
|
||||
includeIssue: false,
|
||||
runErrorCode: "process_detached",
|
||||
runError: "Lost in-memory process handle, but child pid 123 is still alive",
|
||||
});
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
const updated = await heartbeat.reportRunActivity(runId);
|
||||
expect(updated?.errorCode).toBeNull();
|
||||
expect(updated?.error).toBeNull();
|
||||
|
||||
const run = await heartbeat.getRun(runId);
|
||||
expect(run?.errorCode).toBeNull();
|
||||
expect(run?.error).toBeNull();
|
||||
});
|
||||
});
|
||||
@@ -2450,6 +2450,14 @@ export function accessRoutes(
|
||||
"member",
|
||||
"active"
|
||||
);
|
||||
await access.setPrincipalPermission(
|
||||
companyId,
|
||||
"agent",
|
||||
created.id,
|
||||
"tasks:assign",
|
||||
true,
|
||||
req.actor.userId ?? null
|
||||
);
|
||||
const grants = grantsFromDefaults(
|
||||
invite.defaultsPayload as Record<string, unknown> | null,
|
||||
"agent"
|
||||
|
||||
@@ -84,6 +84,80 @@ export function agentRoutes(db: Db) {
|
||||
return Boolean((agent.permissions as Record<string, unknown>).canCreateAgents);
|
||||
}
|
||||
|
||||
async function buildAgentAccessState(agent: NonNullable<Awaited<ReturnType<typeof svc.getById>>>) {
|
||||
const membership = await access.getMembership(agent.companyId, "agent", agent.id);
|
||||
const grants = membership
|
||||
? await access.listPrincipalGrants(agent.companyId, "agent", agent.id)
|
||||
: [];
|
||||
const hasExplicitTaskAssignGrant = grants.some((grant) => grant.permissionKey === "tasks:assign");
|
||||
|
||||
if (agent.role === "ceo") {
|
||||
return {
|
||||
canAssignTasks: true,
|
||||
taskAssignSource: "ceo_role" as const,
|
||||
membership,
|
||||
grants,
|
||||
};
|
||||
}
|
||||
|
||||
if (canCreateAgents(agent)) {
|
||||
return {
|
||||
canAssignTasks: true,
|
||||
taskAssignSource: "agent_creator" as const,
|
||||
membership,
|
||||
grants,
|
||||
};
|
||||
}
|
||||
|
||||
if (hasExplicitTaskAssignGrant) {
|
||||
return {
|
||||
canAssignTasks: true,
|
||||
taskAssignSource: "explicit_grant" as const,
|
||||
membership,
|
||||
grants,
|
||||
};
|
||||
}
|
||||
|
||||
return {
|
||||
canAssignTasks: false,
|
||||
taskAssignSource: "none" as const,
|
||||
membership,
|
||||
grants,
|
||||
};
|
||||
}
|
||||
|
||||
async function buildAgentDetail(
|
||||
agent: NonNullable<Awaited<ReturnType<typeof svc.getById>>>,
|
||||
options?: { restricted?: boolean },
|
||||
) {
|
||||
const [chainOfCommand, accessState] = await Promise.all([
|
||||
svc.getChainOfCommand(agent.id),
|
||||
buildAgentAccessState(agent),
|
||||
]);
|
||||
|
||||
return {
|
||||
...(options?.restricted ? redactForRestrictedAgentView(agent) : agent),
|
||||
chainOfCommand,
|
||||
access: accessState,
|
||||
};
|
||||
}
|
||||
|
||||
async function applyDefaultAgentTaskAssignGrant(
|
||||
companyId: string,
|
||||
agentId: string,
|
||||
grantedByUserId: string | null,
|
||||
) {
|
||||
await access.ensureMembership(companyId, "agent", agentId, "member", "active");
|
||||
await access.setPrincipalPermission(
|
||||
companyId,
|
||||
"agent",
|
||||
agentId,
|
||||
"tasks:assign",
|
||||
true,
|
||||
grantedByUserId,
|
||||
);
|
||||
}
|
||||
|
||||
async function assertCanCreateAgentsForCompany(req: Request, companyId: string) {
|
||||
assertCompanyAccess(req, companyId);
|
||||
if (req.actor.type === "board") {
|
||||
@@ -799,8 +873,7 @@ export function agentRoutes(db: Db) {
|
||||
res.status(404).json({ error: "Agent not found" });
|
||||
return;
|
||||
}
|
||||
const chainOfCommand = await svc.getChainOfCommand(agent.id);
|
||||
res.json({ ...agent, chainOfCommand });
|
||||
res.json(await buildAgentDetail(agent));
|
||||
});
|
||||
|
||||
router.get("/agents/me/inbox-lite", async (req, res) => {
|
||||
@@ -842,13 +915,11 @@ export function agentRoutes(db: Db) {
|
||||
if (req.actor.type === "agent" && req.actor.agentId !== id) {
|
||||
const canRead = await actorCanReadConfigurationsForCompany(req, agent.companyId);
|
||||
if (!canRead) {
|
||||
const chainOfCommand = await svc.getChainOfCommand(agent.id);
|
||||
res.json({ ...redactForRestrictedAgentView(agent), chainOfCommand });
|
||||
res.json(await buildAgentDetail(agent, { restricted: true }));
|
||||
return;
|
||||
}
|
||||
}
|
||||
const chainOfCommand = await svc.getChainOfCommand(agent.id);
|
||||
res.json({ ...agent, chainOfCommand });
|
||||
res.json(await buildAgentDetail(agent));
|
||||
});
|
||||
|
||||
router.get("/agents/:id/configuration", async (req, res) => {
|
||||
@@ -1122,6 +1193,12 @@ export function agentRoutes(db: Db) {
|
||||
},
|
||||
});
|
||||
|
||||
await applyDefaultAgentTaskAssignGrant(
|
||||
companyId,
|
||||
agent.id,
|
||||
actor.actorType === "user" ? actor.actorId : null,
|
||||
);
|
||||
|
||||
if (approval) {
|
||||
await logActivity(db, {
|
||||
companyId,
|
||||
@@ -1197,6 +1274,12 @@ export function agentRoutes(db: Db) {
|
||||
},
|
||||
});
|
||||
|
||||
await applyDefaultAgentTaskAssignGrant(
|
||||
companyId,
|
||||
agent.id,
|
||||
req.actor.type === "board" ? (req.actor.userId ?? null) : null,
|
||||
);
|
||||
|
||||
if (agent.budgetMonthlyCents > 0) {
|
||||
await budgets.upsertPolicy(
|
||||
companyId,
|
||||
@@ -1240,6 +1323,18 @@ export function agentRoutes(db: Db) {
|
||||
return;
|
||||
}
|
||||
|
||||
const effectiveCanAssignTasks =
|
||||
agent.role === "ceo" || Boolean(agent.permissions?.canCreateAgents) || req.body.canAssignTasks;
|
||||
await access.ensureMembership(agent.companyId, "agent", agent.id, "member", "active");
|
||||
await access.setPrincipalPermission(
|
||||
agent.companyId,
|
||||
"agent",
|
||||
agent.id,
|
||||
"tasks:assign",
|
||||
effectiveCanAssignTasks,
|
||||
req.actor.type === "board" ? (req.actor.userId ?? null) : null,
|
||||
);
|
||||
|
||||
const actor = getActorInfo(req);
|
||||
await logActivity(db, {
|
||||
companyId: agent.companyId,
|
||||
@@ -1250,10 +1345,13 @@ export function agentRoutes(db: Db) {
|
||||
action: "agent.permissions_updated",
|
||||
entityType: "agent",
|
||||
entityId: agent.id,
|
||||
details: req.body,
|
||||
details: {
|
||||
canCreateAgents: agent.permissions?.canCreateAgents ?? false,
|
||||
canAssignTasks: effectiveCanAssignTasks,
|
||||
},
|
||||
});
|
||||
|
||||
res.json(agent);
|
||||
res.json(await buildAgentDetail(agent));
|
||||
});
|
||||
|
||||
router.patch("/agents/:id/instructions-path", validate(updateAgentInstructionsPathSchema), async (req, res) => {
|
||||
|
||||
@@ -7,6 +7,7 @@ import {
|
||||
createCompanySchema,
|
||||
updateCompanyBrandingSchema,
|
||||
updateCompanySchema,
|
||||
updateCompanyBrandingSchema,
|
||||
} from "@paperclipai/shared";
|
||||
import { forbidden } from "../errors.js";
|
||||
import { validate } from "../middleware/validate.js";
|
||||
@@ -90,9 +91,12 @@ export function companyRoutes(db: Db, storage?: StorageService) {
|
||||
});
|
||||
|
||||
router.get("/:companyId", async (req, res) => {
|
||||
assertBoard(req);
|
||||
const companyId = req.params.companyId as string;
|
||||
assertCompanyAccess(req, companyId);
|
||||
// Allow agents (CEO) to read their own company; board always allowed
|
||||
if (req.actor.type !== "agent") {
|
||||
assertBoard(req);
|
||||
}
|
||||
const company = await svc.getById(companyId);
|
||||
if (!company) {
|
||||
res.status(404).json({ error: "Company not found" });
|
||||
@@ -238,23 +242,44 @@ export function companyRoutes(db: Db, storage?: StorageService) {
|
||||
res.status(201).json(company);
|
||||
});
|
||||
|
||||
router.patch("/:companyId", validate(updateCompanySchema), async (req, res) => {
|
||||
assertBoard(req);
|
||||
router.patch("/:companyId", async (req, res) => {
|
||||
const companyId = req.params.companyId as string;
|
||||
assertCompanyAccess(req, companyId);
|
||||
const company = await svc.update(companyId, req.body);
|
||||
|
||||
const actor = getActorInfo(req);
|
||||
let body: Record<string, unknown>;
|
||||
|
||||
if (req.actor.type === "agent") {
|
||||
// Only CEO agents may update company branding fields
|
||||
const agentSvc = agentService(db);
|
||||
const actorAgent = req.actor.agentId ? await agentSvc.getById(req.actor.agentId) : null;
|
||||
if (!actorAgent || actorAgent.role !== "ceo") {
|
||||
throw forbidden("Only CEO agents or board users may update company settings");
|
||||
}
|
||||
if (actorAgent.companyId !== companyId) {
|
||||
throw forbidden("Agent key cannot access another company");
|
||||
}
|
||||
body = updateCompanyBrandingSchema.parse(req.body);
|
||||
} else {
|
||||
assertBoard(req);
|
||||
body = updateCompanySchema.parse(req.body);
|
||||
}
|
||||
|
||||
const company = await svc.update(companyId, body);
|
||||
if (!company) {
|
||||
res.status(404).json({ error: "Company not found" });
|
||||
return;
|
||||
}
|
||||
await logActivity(db, {
|
||||
companyId,
|
||||
actorType: "user",
|
||||
actorId: req.actor.userId ?? "board",
|
||||
actorType: actor.actorType,
|
||||
actorId: actor.actorId,
|
||||
agentId: actor.agentId,
|
||||
runId: actor.runId,
|
||||
action: "company.updated",
|
||||
entityType: "company",
|
||||
entityId: companyId,
|
||||
details: req.body,
|
||||
details: body,
|
||||
});
|
||||
res.json(company);
|
||||
});
|
||||
|
||||
@@ -826,6 +826,7 @@ export function issueRoutes(db: Db, storage: StorageService) {
|
||||
}
|
||||
if (!(await assertAgentRunCheckoutOwnership(req, res, existing))) return;
|
||||
|
||||
const actor = getActorInfo(req);
|
||||
const { comment: commentBody, hiddenAt: hiddenAtRaw, ...updateFields } = req.body;
|
||||
if (hiddenAtRaw !== undefined) {
|
||||
updateFields.hiddenAt = hiddenAtRaw ? new Date(hiddenAtRaw) : null;
|
||||
@@ -863,6 +864,11 @@ export function issueRoutes(db: Db, storage: StorageService) {
|
||||
}
|
||||
await routinesSvc.syncRunStatusForIssue(issue.id);
|
||||
|
||||
if (actor.runId) {
|
||||
await heartbeat.reportRunActivity(actor.runId).catch((err) =>
|
||||
logger.warn({ err, runId: actor.runId }, "failed to clear detached run warning after issue activity"));
|
||||
}
|
||||
|
||||
// Build activity details with previous values for changed fields
|
||||
const previous: Record<string, unknown> = {};
|
||||
for (const key of Object.keys(updateFields)) {
|
||||
@@ -871,7 +877,6 @@ export function issueRoutes(db: Db, storage: StorageService) {
|
||||
}
|
||||
}
|
||||
|
||||
const actor = getActorInfo(req);
|
||||
const hasFieldChanges = Object.keys(previous).length > 0;
|
||||
await logActivity(db, {
|
||||
companyId: issue.companyId,
|
||||
@@ -1285,6 +1290,11 @@ export function issueRoutes(db: Db, storage: StorageService) {
|
||||
userId: actor.actorType === "user" ? actor.actorId : undefined,
|
||||
});
|
||||
|
||||
if (actor.runId) {
|
||||
await heartbeat.reportRunActivity(actor.runId).catch((err) =>
|
||||
logger.warn({ err, runId: actor.runId }, "failed to clear detached run warning after issue comment"));
|
||||
}
|
||||
|
||||
await logActivity(db, {
|
||||
companyId: currentIssue.companyId,
|
||||
actorType: actor.actorType,
|
||||
|
||||
@@ -279,6 +279,86 @@ export function accessService(db: Db) {
|
||||
return sourceMemberships;
|
||||
}
|
||||
|
||||
async function listPrincipalGrants(
|
||||
companyId: string,
|
||||
principalType: PrincipalType,
|
||||
principalId: string,
|
||||
) {
|
||||
return db
|
||||
.select()
|
||||
.from(principalPermissionGrants)
|
||||
.where(
|
||||
and(
|
||||
eq(principalPermissionGrants.companyId, companyId),
|
||||
eq(principalPermissionGrants.principalType, principalType),
|
||||
eq(principalPermissionGrants.principalId, principalId),
|
||||
),
|
||||
)
|
||||
.orderBy(principalPermissionGrants.permissionKey);
|
||||
}
|
||||
|
||||
async function setPrincipalPermission(
|
||||
companyId: string,
|
||||
principalType: PrincipalType,
|
||||
principalId: string,
|
||||
permissionKey: PermissionKey,
|
||||
enabled: boolean,
|
||||
grantedByUserId: string | null,
|
||||
scope: Record<string, unknown> | null = null,
|
||||
) {
|
||||
if (!enabled) {
|
||||
await db
|
||||
.delete(principalPermissionGrants)
|
||||
.where(
|
||||
and(
|
||||
eq(principalPermissionGrants.companyId, companyId),
|
||||
eq(principalPermissionGrants.principalType, principalType),
|
||||
eq(principalPermissionGrants.principalId, principalId),
|
||||
eq(principalPermissionGrants.permissionKey, permissionKey),
|
||||
),
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
await ensureMembership(companyId, principalType, principalId, "member", "active");
|
||||
|
||||
const existing = await db
|
||||
.select()
|
||||
.from(principalPermissionGrants)
|
||||
.where(
|
||||
and(
|
||||
eq(principalPermissionGrants.companyId, companyId),
|
||||
eq(principalPermissionGrants.principalType, principalType),
|
||||
eq(principalPermissionGrants.principalId, principalId),
|
||||
eq(principalPermissionGrants.permissionKey, permissionKey),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
|
||||
if (existing) {
|
||||
await db
|
||||
.update(principalPermissionGrants)
|
||||
.set({
|
||||
scope,
|
||||
grantedByUserId,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(principalPermissionGrants.id, existing.id));
|
||||
return;
|
||||
}
|
||||
|
||||
await db.insert(principalPermissionGrants).values({
|
||||
companyId,
|
||||
principalType,
|
||||
principalId,
|
||||
permissionKey,
|
||||
scope,
|
||||
grantedByUserId,
|
||||
createdAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
});
|
||||
}
|
||||
|
||||
return {
|
||||
isInstanceAdmin,
|
||||
canUser,
|
||||
@@ -294,5 +374,7 @@ export function accessService(db: Db) {
|
||||
listUserCompanyAccess,
|
||||
setUserCompanyAccess,
|
||||
setPrincipalGrants,
|
||||
listPrincipalGrants,
|
||||
setPrincipalPermission,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -3076,6 +3076,15 @@ export function companyPortabilityService(db: Db, storage?: StorageService) {
|
||||
}
|
||||
|
||||
let created = await agents.create(targetCompany.id, patch);
|
||||
await access.ensureMembership(targetCompany.id, "agent", created.id, "member", "active");
|
||||
await access.setPrincipalPermission(
|
||||
targetCompany.id,
|
||||
"agent",
|
||||
created.id,
|
||||
"tasks:assign",
|
||||
true,
|
||||
actorUserId ?? null,
|
||||
);
|
||||
try {
|
||||
const materialized = await instructions.materializeManagedBundle(created, bundleFiles, {
|
||||
clearLegacyPromptTemplate: true,
|
||||
|
||||
@@ -61,6 +61,7 @@ const MAX_LIVE_LOG_CHUNK_BYTES = 8 * 1024;
|
||||
const HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT = 1;
|
||||
const HEARTBEAT_MAX_CONCURRENT_RUNS_MAX = 10;
|
||||
const DEFERRED_WAKE_CONTEXT_KEY = "_paperclipWakeContext";
|
||||
const DETACHED_PROCESS_ERROR_CODE = "process_detached";
|
||||
const startLocksByAgent = new Map<string, Promise<void>>();
|
||||
const REPO_ONLY_CWD_SENTINEL = "/__paperclip_repo_only__";
|
||||
const MANAGED_WORKSPACE_GIT_CLONE_TIMEOUT_MS = 10 * 60 * 1000;
|
||||
@@ -164,6 +165,10 @@ const heartbeatRunListColumns = {
|
||||
stderrExcerpt: sql<string | null>`NULL`.as("stderrExcerpt"),
|
||||
errorCode: heartbeatRuns.errorCode,
|
||||
externalRunId: heartbeatRuns.externalRunId,
|
||||
processPid: heartbeatRuns.processPid,
|
||||
processStartedAt: heartbeatRuns.processStartedAt,
|
||||
retryOfRunId: heartbeatRuns.retryOfRunId,
|
||||
processLossRetryCount: heartbeatRuns.processLossRetryCount,
|
||||
contextSnapshot: heartbeatRuns.contextSnapshot,
|
||||
createdAt: heartbeatRuns.createdAt,
|
||||
updatedAt: heartbeatRuns.updatedAt,
|
||||
@@ -599,6 +604,26 @@ function isSameTaskScope(left: string | null, right: string | null) {
|
||||
return (left ?? null) === (right ?? null);
|
||||
}
|
||||
|
||||
function isTrackedLocalChildProcessAdapter(adapterType: string) {
|
||||
return SESSIONED_LOCAL_ADAPTERS.has(adapterType);
|
||||
}
|
||||
|
||||
// A positive liveness check means some process currently owns the PID.
|
||||
// On Linux, PIDs can be recycled, so this is a best-effort signal rather
|
||||
// than proof that the original child is still alive.
|
||||
function isProcessAlive(pid: number | null | undefined) {
|
||||
if (typeof pid !== "number" || !Number.isInteger(pid) || pid <= 0) return false;
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
return true;
|
||||
} catch (error) {
|
||||
const code = (error as NodeJS.ErrnoException | undefined)?.code;
|
||||
if (code === "EPERM") return true;
|
||||
if (code === "ESRCH") return false;
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
function truncateDisplayId(value: string | null | undefined, max = 128) {
|
||||
if (!value) return null;
|
||||
return value.length > max ? value.slice(0, max) : value;
|
||||
@@ -1328,6 +1353,156 @@ export function heartbeatService(db: Db) {
|
||||
});
|
||||
}
|
||||
|
||||
async function nextRunEventSeq(runId: string) {
|
||||
const [row] = await db
|
||||
.select({ maxSeq: sql<number | null>`max(${heartbeatRunEvents.seq})` })
|
||||
.from(heartbeatRunEvents)
|
||||
.where(eq(heartbeatRunEvents.runId, runId));
|
||||
return Number(row?.maxSeq ?? 0) + 1;
|
||||
}
|
||||
|
||||
async function persistRunProcessMetadata(
|
||||
runId: string,
|
||||
meta: { pid: number; startedAt: string },
|
||||
) {
|
||||
const startedAt = new Date(meta.startedAt);
|
||||
return db
|
||||
.update(heartbeatRuns)
|
||||
.set({
|
||||
processPid: meta.pid,
|
||||
processStartedAt: Number.isNaN(startedAt.getTime()) ? new Date() : startedAt,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(heartbeatRuns.id, runId))
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null);
|
||||
}
|
||||
|
||||
async function clearDetachedRunWarning(runId: string) {
|
||||
const updated = await db
|
||||
.update(heartbeatRuns)
|
||||
.set({
|
||||
error: null,
|
||||
errorCode: null,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.status, "running"), eq(heartbeatRuns.errorCode, DETACHED_PROCESS_ERROR_CODE)))
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!updated) return null;
|
||||
|
||||
await appendRunEvent(updated, await nextRunEventSeq(updated.id), {
|
||||
eventType: "lifecycle",
|
||||
stream: "system",
|
||||
level: "info",
|
||||
message: "Detached child process reported activity; cleared detached warning",
|
||||
});
|
||||
return updated;
|
||||
}
|
||||
|
||||
async function enqueueProcessLossRetry(
|
||||
run: typeof heartbeatRuns.$inferSelect,
|
||||
agent: typeof agents.$inferSelect,
|
||||
now: Date,
|
||||
) {
|
||||
const contextSnapshot = parseObject(run.contextSnapshot);
|
||||
const issueId = readNonEmptyString(contextSnapshot.issueId);
|
||||
const taskKey = deriveTaskKey(contextSnapshot, null);
|
||||
const sessionBefore = await resolveSessionBeforeForWakeup(agent, taskKey);
|
||||
const retryContextSnapshot = {
|
||||
...contextSnapshot,
|
||||
retryOfRunId: run.id,
|
||||
wakeReason: "process_lost_retry",
|
||||
retryReason: "process_lost",
|
||||
};
|
||||
|
||||
const queued = await db.transaction(async (tx) => {
|
||||
const wakeupRequest = await tx
|
||||
.insert(agentWakeupRequests)
|
||||
.values({
|
||||
companyId: run.companyId,
|
||||
agentId: run.agentId,
|
||||
source: "automation",
|
||||
triggerDetail: "system",
|
||||
reason: "process_lost_retry",
|
||||
payload: {
|
||||
...(issueId ? { issueId } : {}),
|
||||
retryOfRunId: run.id,
|
||||
},
|
||||
status: "queued",
|
||||
requestedByActorType: "system",
|
||||
requestedByActorId: null,
|
||||
updatedAt: now,
|
||||
})
|
||||
.returning()
|
||||
.then((rows) => rows[0]);
|
||||
|
||||
const retryRun = await tx
|
||||
.insert(heartbeatRuns)
|
||||
.values({
|
||||
companyId: run.companyId,
|
||||
agentId: run.agentId,
|
||||
invocationSource: "automation",
|
||||
triggerDetail: "system",
|
||||
status: "queued",
|
||||
wakeupRequestId: wakeupRequest.id,
|
||||
contextSnapshot: retryContextSnapshot,
|
||||
sessionIdBefore: sessionBefore,
|
||||
retryOfRunId: run.id,
|
||||
processLossRetryCount: (run.processLossRetryCount ?? 0) + 1,
|
||||
updatedAt: now,
|
||||
})
|
||||
.returning()
|
||||
.then((rows) => rows[0]);
|
||||
|
||||
await tx
|
||||
.update(agentWakeupRequests)
|
||||
.set({
|
||||
runId: retryRun.id,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(eq(agentWakeupRequests.id, wakeupRequest.id));
|
||||
|
||||
if (issueId) {
|
||||
await tx
|
||||
.update(issues)
|
||||
.set({
|
||||
executionRunId: retryRun.id,
|
||||
executionAgentNameKey: normalizeAgentNameKey(agent.name),
|
||||
executionLockedAt: now,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(and(eq(issues.id, issueId), eq(issues.companyId, run.companyId), eq(issues.executionRunId, run.id)));
|
||||
}
|
||||
|
||||
return retryRun;
|
||||
});
|
||||
|
||||
publishLiveEvent({
|
||||
companyId: queued.companyId,
|
||||
type: "heartbeat.run.queued",
|
||||
payload: {
|
||||
runId: queued.id,
|
||||
agentId: queued.agentId,
|
||||
invocationSource: queued.invocationSource,
|
||||
triggerDetail: queued.triggerDetail,
|
||||
wakeupRequestId: queued.wakeupRequestId,
|
||||
},
|
||||
});
|
||||
|
||||
await appendRunEvent(queued, 1, {
|
||||
eventType: "lifecycle",
|
||||
stream: "system",
|
||||
level: "warn",
|
||||
message: "Queued automatic retry after orphaned child process was confirmed dead",
|
||||
payload: {
|
||||
retryOfRunId: run.id,
|
||||
},
|
||||
});
|
||||
|
||||
return queued;
|
||||
}
|
||||
|
||||
function parseHeartbeatPolicy(agent: typeof agents.$inferSelect) {
|
||||
const runtimeConfig = parseObject(agent.runtimeConfig);
|
||||
const heartbeat = parseObject(runtimeConfig.heartbeat);
|
||||
@@ -1455,13 +1630,17 @@ export function heartbeatService(db: Db) {
|
||||
|
||||
// Find all runs stuck in "running" state (queued runs are legitimately waiting; resumeQueuedRuns handles them)
|
||||
const activeRuns = await db
|
||||
.select()
|
||||
.select({
|
||||
run: heartbeatRuns,
|
||||
adapterType: agents.adapterType,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.innerJoin(agents, eq(heartbeatRuns.agentId, agents.id))
|
||||
.where(eq(heartbeatRuns.status, "running"));
|
||||
|
||||
const reaped: string[] = [];
|
||||
|
||||
for (const run of activeRuns) {
|
||||
for (const { run, adapterType } of activeRuns) {
|
||||
if (runningProcesses.has(run.id) || activeRunExecutions.has(run.id)) continue;
|
||||
|
||||
// Apply staleness threshold to avoid false positives
|
||||
@@ -1470,25 +1649,69 @@ export function heartbeatService(db: Db) {
|
||||
if (now.getTime() - refTime < staleThresholdMs) continue;
|
||||
}
|
||||
|
||||
await setRunStatus(run.id, "failed", {
|
||||
error: "Process lost -- server may have restarted",
|
||||
const tracksLocalChild = isTrackedLocalChildProcessAdapter(adapterType);
|
||||
if (tracksLocalChild && run.processPid && isProcessAlive(run.processPid)) {
|
||||
if (run.errorCode !== DETACHED_PROCESS_ERROR_CODE) {
|
||||
const detachedMessage = `Lost in-memory process handle, but child pid ${run.processPid} is still alive`;
|
||||
const detachedRun = await setRunStatus(run.id, "running", {
|
||||
error: detachedMessage,
|
||||
errorCode: DETACHED_PROCESS_ERROR_CODE,
|
||||
});
|
||||
if (detachedRun) {
|
||||
await appendRunEvent(detachedRun, await nextRunEventSeq(detachedRun.id), {
|
||||
eventType: "lifecycle",
|
||||
stream: "system",
|
||||
level: "warn",
|
||||
message: detachedMessage,
|
||||
payload: {
|
||||
processPid: run.processPid,
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
const shouldRetry = tracksLocalChild && !!run.processPid && (run.processLossRetryCount ?? 0) < 1;
|
||||
const baseMessage = run.processPid
|
||||
? `Process lost -- child pid ${run.processPid} is no longer running`
|
||||
: "Process lost -- server may have restarted";
|
||||
|
||||
let finalizedRun = await setRunStatus(run.id, "failed", {
|
||||
error: shouldRetry ? `${baseMessage}; retrying once` : baseMessage,
|
||||
errorCode: "process_lost",
|
||||
finishedAt: now,
|
||||
});
|
||||
await setWakeupStatus(run.wakeupRequestId, "failed", {
|
||||
finishedAt: now,
|
||||
error: "Process lost -- server may have restarted",
|
||||
error: shouldRetry ? `${baseMessage}; retrying once` : baseMessage,
|
||||
});
|
||||
const updatedRun = await getRun(run.id);
|
||||
if (updatedRun) {
|
||||
await appendRunEvent(updatedRun, 1, {
|
||||
eventType: "lifecycle",
|
||||
stream: "system",
|
||||
level: "error",
|
||||
message: "Process lost -- server may have restarted",
|
||||
});
|
||||
await releaseIssueExecutionAndPromote(updatedRun);
|
||||
if (!finalizedRun) finalizedRun = await getRun(run.id);
|
||||
if (!finalizedRun) continue;
|
||||
|
||||
let retriedRun: typeof heartbeatRuns.$inferSelect | null = null;
|
||||
if (shouldRetry) {
|
||||
const agent = await getAgent(run.agentId);
|
||||
if (agent) {
|
||||
retriedRun = await enqueueProcessLossRetry(finalizedRun, agent, now);
|
||||
}
|
||||
} else {
|
||||
await releaseIssueExecutionAndPromote(finalizedRun);
|
||||
}
|
||||
|
||||
await appendRunEvent(finalizedRun, await nextRunEventSeq(finalizedRun.id), {
|
||||
eventType: "lifecycle",
|
||||
stream: "system",
|
||||
level: "error",
|
||||
message: shouldRetry
|
||||
? `${baseMessage}; queued retry ${retriedRun?.id ?? ""}`.trim()
|
||||
: baseMessage,
|
||||
payload: {
|
||||
...(run.processPid ? { processPid: run.processPid } : {}),
|
||||
...(retriedRun ? { retryRunId: retriedRun.id } : {}),
|
||||
},
|
||||
});
|
||||
|
||||
await finalizeAgentStatus(run.agentId, "failed");
|
||||
await startNextQueuedRunForAgent(run.agentId);
|
||||
runningProcesses.delete(run.id);
|
||||
@@ -2159,6 +2382,9 @@ export function heartbeatService(db: Db) {
|
||||
context,
|
||||
onLog,
|
||||
onMeta: onAdapterMeta,
|
||||
onSpawn: async (meta) => {
|
||||
await persistRunProcessMetadata(run.id, meta);
|
||||
},
|
||||
authToken: authToken ?? undefined,
|
||||
});
|
||||
const adapterManagedRuntimeServices = adapterResult.runtimeServices
|
||||
@@ -3410,6 +3636,8 @@ export function heartbeatService(db: Db) {
|
||||
|
||||
wakeup: enqueueWakeup,
|
||||
|
||||
reportRunActivity: clearDetachedRunWarning,
|
||||
|
||||
reapOrphanedRuns,
|
||||
|
||||
resumeQueuedRuns,
|
||||
|
||||
Reference in New Issue
Block a user