d671a59306
The $AGENT_HOME environment variable was referenced by skills (e.g. para-memory-files) but never actually set, causing runtime errors like "/HEARTBEAT.md: No such file or directory" when agents tried to resolve paths relative to their home directory. Add agentHome to the paperclipWorkspace context in the heartbeat service and propagate it as the AGENT_HOME env var in all local adapters. Co-Authored-By: Paperclip <noreply@paperclip.ing>
2921 lines
98 KiB
TypeScript
2921 lines
98 KiB
TypeScript
import fs from "node:fs/promises";
|
|
import path from "node:path";
|
|
import { and, asc, desc, eq, gt, inArray, sql } from "drizzle-orm";
|
|
import type { Db } from "@paperclipai/db";
|
|
import {
|
|
agents,
|
|
agentRuntimeState,
|
|
agentTaskSessions,
|
|
agentWakeupRequests,
|
|
heartbeatRunEvents,
|
|
heartbeatRuns,
|
|
issues,
|
|
projects,
|
|
projectWorkspaces,
|
|
} from "@paperclipai/db";
|
|
import { conflict, notFound } from "../errors.js";
|
|
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, 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";
|
|
import { secretService } from "./secrets.js";
|
|
import { resolveDefaultAgentWorkspaceDir } from "../home-paths.js";
|
|
import { summarizeHeartbeatRunResultJson } from "./heartbeat-run-summary.js";
|
|
import {
|
|
buildWorkspaceReadyComment,
|
|
ensureRuntimeServicesForRun,
|
|
persistAdapterManagedRuntimeServices,
|
|
realizeExecutionWorkspace,
|
|
releaseRuntimeServicesForRun,
|
|
} from "./workspace-runtime.js";
|
|
import { issueService } from "./issues.js";
|
|
import {
|
|
buildExecutionWorkspaceAdapterConfig,
|
|
parseIssueExecutionWorkspaceSettings,
|
|
parseProjectExecutionWorkspacePolicy,
|
|
resolveExecutionWorkspaceMode,
|
|
} from "./execution-workspace-policy.js";
|
|
import { redactCurrentUserText, redactCurrentUserValue } from "../log-redaction.js";
|
|
|
|
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 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,
|
|
companyId: heartbeatRuns.companyId,
|
|
agentId: heartbeatRuns.agentId,
|
|
invocationSource: heartbeatRuns.invocationSource,
|
|
triggerDetail: heartbeatRuns.triggerDetail,
|
|
status: heartbeatRuns.status,
|
|
startedAt: heartbeatRuns.startedAt,
|
|
finishedAt: heartbeatRuns.finishedAt,
|
|
error: heartbeatRuns.error,
|
|
wakeupRequestId: heartbeatRuns.wakeupRequestId,
|
|
exitCode: heartbeatRuns.exitCode,
|
|
signal: heartbeatRuns.signal,
|
|
usageJson: heartbeatRuns.usageJson,
|
|
resultJson: heartbeatRuns.resultJson,
|
|
sessionIdBefore: heartbeatRuns.sessionIdBefore,
|
|
sessionIdAfter: heartbeatRuns.sessionIdAfter,
|
|
logStore: heartbeatRuns.logStore,
|
|
logRef: heartbeatRuns.logRef,
|
|
logBytes: heartbeatRuns.logBytes,
|
|
logSha256: heartbeatRuns.logSha256,
|
|
logCompressed: heartbeatRuns.logCompressed,
|
|
stdoutExcerpt: sql<string | null>`NULL`.as("stdoutExcerpt"),
|
|
stderrExcerpt: sql<string | null>`NULL`.as("stderrExcerpt"),
|
|
errorCode: heartbeatRuns.errorCode,
|
|
externalRunId: heartbeatRuns.externalRunId,
|
|
contextSnapshot: heartbeatRuns.contextSnapshot,
|
|
createdAt: heartbeatRuns.createdAt,
|
|
updatedAt: heartbeatRuns.updatedAt,
|
|
} as const;
|
|
|
|
function appendExcerpt(prev: string, chunk: string) {
|
|
return appendWithCap(prev, chunk, MAX_EXCERPT_BYTES);
|
|
}
|
|
|
|
function normalizeMaxConcurrentRuns(value: unknown) {
|
|
const parsed = Math.floor(asNumber(value, HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT));
|
|
if (!Number.isFinite(parsed)) return HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT;
|
|
return Math.max(HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT, Math.min(HEARTBEAT_MAX_CONCURRENT_RUNS_MAX, parsed));
|
|
}
|
|
|
|
async function withAgentStartLock<T>(agentId: string, fn: () => Promise<T>) {
|
|
const previous = startLocksByAgent.get(agentId) ?? Promise.resolve();
|
|
const run = previous.then(fn);
|
|
const marker = run.then(
|
|
() => undefined,
|
|
() => undefined,
|
|
);
|
|
startLocksByAgent.set(agentId, marker);
|
|
try {
|
|
return await run;
|
|
} finally {
|
|
if (startLocksByAgent.get(agentId) === marker) {
|
|
startLocksByAgent.delete(agentId);
|
|
}
|
|
}
|
|
}
|
|
|
|
interface WakeupOptions {
|
|
source?: "timer" | "assignment" | "on_demand" | "automation";
|
|
triggerDetail?: "manual" | "ping" | "callback" | "system";
|
|
reason?: string | null;
|
|
payload?: Record<string, unknown> | null;
|
|
idempotencyKey?: string | null;
|
|
requestedByActorType?: "user" | "agent" | "system";
|
|
requestedByActorId?: string | null;
|
|
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;
|
|
}
|
|
|
|
export type ResolvedWorkspaceForRun = {
|
|
cwd: string;
|
|
source: "project_primary" | "task_session" | "agent_home";
|
|
projectId: string | null;
|
|
workspaceId: string | null;
|
|
repoUrl: string | null;
|
|
repoRef: string | null;
|
|
workspaceHints: Array<{
|
|
workspaceId: string;
|
|
cwd: string | null;
|
|
repoUrl: string | null;
|
|
repoRef: string | null;
|
|
}>;
|
|
warnings: string[];
|
|
};
|
|
|
|
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;
|
|
resolvedWorkspace: ResolvedWorkspaceForRun;
|
|
}) {
|
|
const { agentId, previousSessionParams, resolvedWorkspace } = input;
|
|
const previousSessionId = readNonEmptyString(previousSessionParams?.sessionId);
|
|
const previousCwd = readNonEmptyString(previousSessionParams?.cwd);
|
|
if (!previousSessionId || !previousCwd) {
|
|
return {
|
|
sessionParams: previousSessionParams,
|
|
warning: null as string | null,
|
|
};
|
|
}
|
|
if (resolvedWorkspace.source !== "project_primary") {
|
|
return {
|
|
sessionParams: previousSessionParams,
|
|
warning: null as string | null,
|
|
};
|
|
}
|
|
const projectCwd = readNonEmptyString(resolvedWorkspace.cwd);
|
|
if (!projectCwd) {
|
|
return {
|
|
sessionParams: previousSessionParams,
|
|
warning: null as string | null,
|
|
};
|
|
}
|
|
const fallbackAgentHomeCwd = resolveDefaultAgentWorkspaceDir(agentId);
|
|
if (path.resolve(previousCwd) !== path.resolve(fallbackAgentHomeCwd)) {
|
|
return {
|
|
sessionParams: previousSessionParams,
|
|
warning: null as string | null,
|
|
};
|
|
}
|
|
if (path.resolve(projectCwd) === path.resolve(previousCwd)) {
|
|
return {
|
|
sessionParams: previousSessionParams,
|
|
warning: null as string | null,
|
|
};
|
|
}
|
|
const previousWorkspaceId = readNonEmptyString(previousSessionParams?.workspaceId);
|
|
if (
|
|
previousWorkspaceId &&
|
|
resolvedWorkspace.workspaceId &&
|
|
previousWorkspaceId !== resolvedWorkspace.workspaceId
|
|
) {
|
|
return {
|
|
sessionParams: previousSessionParams,
|
|
warning: null as string | null,
|
|
};
|
|
}
|
|
|
|
const migratedSessionParams: Record<string, unknown> = {
|
|
...(previousSessionParams ?? {}),
|
|
cwd: projectCwd,
|
|
};
|
|
if (resolvedWorkspace.workspaceId) migratedSessionParams.workspaceId = resolvedWorkspace.workspaceId;
|
|
if (resolvedWorkspace.repoUrl) migratedSessionParams.repoUrl = resolvedWorkspace.repoUrl;
|
|
if (resolvedWorkspace.repoRef) migratedSessionParams.repoRef = resolvedWorkspace.repoRef;
|
|
|
|
return {
|
|
sessionParams: migratedSessionParams,
|
|
warning:
|
|
`Project workspace "${projectCwd}" is now available. ` +
|
|
`Attempting to resume session "${previousSessionId}" that was previously saved in fallback workspace "${previousCwd}".`,
|
|
};
|
|
}
|
|
|
|
function parseIssueAssigneeAdapterOverrides(
|
|
raw: unknown,
|
|
): ParsedIssueAssigneeAdapterOverrides | null {
|
|
const parsed = parseObject(raw);
|
|
const parsedAdapterConfig = parseObject(parsed.adapterConfig);
|
|
const adapterConfig =
|
|
Object.keys(parsedAdapterConfig).length > 0 ? parsedAdapterConfig : null;
|
|
const useProjectWorkspace =
|
|
typeof parsed.useProjectWorkspace === "boolean"
|
|
? parsed.useProjectWorkspace
|
|
: null;
|
|
if (!adapterConfig && useProjectWorkspace === null) return null;
|
|
return {
|
|
adapterConfig,
|
|
useProjectWorkspace,
|
|
};
|
|
}
|
|
|
|
function deriveTaskKey(
|
|
contextSnapshot: Record<string, unknown> | null | undefined,
|
|
payload: Record<string, unknown> | null | undefined,
|
|
) {
|
|
return (
|
|
readNonEmptyString(contextSnapshot?.taskKey) ??
|
|
readNonEmptyString(contextSnapshot?.taskId) ??
|
|
readNonEmptyString(contextSnapshot?.issueId) ??
|
|
readNonEmptyString(payload?.taskKey) ??
|
|
readNonEmptyString(payload?.taskId) ??
|
|
readNonEmptyString(payload?.issueId) ??
|
|
null
|
|
);
|
|
}
|
|
|
|
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;
|
|
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";
|
|
return null;
|
|
}
|
|
|
|
function deriveCommentId(
|
|
contextSnapshot: Record<string, unknown> | null | undefined,
|
|
payload: Record<string, unknown> | null | undefined,
|
|
) {
|
|
return (
|
|
readNonEmptyString(contextSnapshot?.wakeCommentId) ??
|
|
readNonEmptyString(contextSnapshot?.commentId) ??
|
|
readNonEmptyString(payload?.commentId) ??
|
|
null
|
|
);
|
|
}
|
|
|
|
function enrichWakeContextSnapshot(input: {
|
|
contextSnapshot: Record<string, unknown>;
|
|
reason: string | null;
|
|
source: WakeupOptions["source"];
|
|
triggerDetail: WakeupOptions["triggerDetail"] | null;
|
|
payload: Record<string, unknown> | null;
|
|
}) {
|
|
const { contextSnapshot, reason, source, triggerDetail, payload } = input;
|
|
const issueIdFromPayload = readNonEmptyString(payload?.["issueId"]);
|
|
const commentIdFromPayload = readNonEmptyString(payload?.["commentId"]);
|
|
const taskKey = deriveTaskKey(contextSnapshot, payload);
|
|
const wakeCommentId = deriveCommentId(contextSnapshot, payload);
|
|
|
|
if (!readNonEmptyString(contextSnapshot["wakeReason"]) && reason) {
|
|
contextSnapshot.wakeReason = reason;
|
|
}
|
|
if (!readNonEmptyString(contextSnapshot["issueId"]) && issueIdFromPayload) {
|
|
contextSnapshot.issueId = issueIdFromPayload;
|
|
}
|
|
if (!readNonEmptyString(contextSnapshot["taskId"]) && issueIdFromPayload) {
|
|
contextSnapshot.taskId = issueIdFromPayload;
|
|
}
|
|
if (!readNonEmptyString(contextSnapshot["taskKey"]) && taskKey) {
|
|
contextSnapshot.taskKey = taskKey;
|
|
}
|
|
if (!readNonEmptyString(contextSnapshot["commentId"]) && commentIdFromPayload) {
|
|
contextSnapshot.commentId = commentIdFromPayload;
|
|
}
|
|
if (!readNonEmptyString(contextSnapshot["wakeCommentId"]) && wakeCommentId) {
|
|
contextSnapshot.wakeCommentId = wakeCommentId;
|
|
}
|
|
if (!readNonEmptyString(contextSnapshot["wakeSource"]) && source) {
|
|
contextSnapshot.wakeSource = source;
|
|
}
|
|
if (!readNonEmptyString(contextSnapshot["wakeTriggerDetail"]) && triggerDetail) {
|
|
contextSnapshot.wakeTriggerDetail = triggerDetail;
|
|
}
|
|
|
|
return {
|
|
contextSnapshot,
|
|
issueIdFromPayload,
|
|
commentIdFromPayload,
|
|
taskKey,
|
|
wakeCommentId,
|
|
};
|
|
}
|
|
|
|
function mergeCoalescedContextSnapshot(
|
|
existingRaw: unknown,
|
|
incoming: Record<string, unknown>,
|
|
) {
|
|
const existing = parseObject(existingRaw);
|
|
const merged: Record<string, unknown> = {
|
|
...existing,
|
|
...incoming,
|
|
};
|
|
const commentId = deriveCommentId(incoming, null);
|
|
if (commentId) {
|
|
merged.commentId = commentId;
|
|
merged.wakeCommentId = commentId;
|
|
}
|
|
return merged;
|
|
}
|
|
|
|
function runTaskKey(run: typeof heartbeatRuns.$inferSelect) {
|
|
return deriveTaskKey(run.contextSnapshot as Record<string, unknown> | null, null);
|
|
}
|
|
|
|
function isSameTaskScope(left: string | null, right: string | null) {
|
|
return (left ?? null) === (right ?? null);
|
|
}
|
|
|
|
function truncateDisplayId(value: string | null | undefined, max = 128) {
|
|
if (!value) return null;
|
|
return value.length > max ? value.slice(0, max) : value;
|
|
}
|
|
|
|
function normalizeAgentNameKey(value: string | null | undefined) {
|
|
if (typeof value !== "string") return null;
|
|
const normalized = value.trim().toLowerCase();
|
|
return normalized.length > 0 ? normalized : null;
|
|
}
|
|
|
|
const defaultSessionCodec: AdapterSessionCodec = {
|
|
deserialize(raw: unknown) {
|
|
const asObj = parseObject(raw);
|
|
if (Object.keys(asObj).length > 0) return asObj;
|
|
const sessionId = readNonEmptyString((raw as Record<string, unknown> | null)?.sessionId);
|
|
if (sessionId) return { sessionId };
|
|
return null;
|
|
},
|
|
serialize(params: Record<string, unknown> | null) {
|
|
if (!params || Object.keys(params).length === 0) return null;
|
|
return params;
|
|
},
|
|
getDisplayId(params: Record<string, unknown> | null) {
|
|
return readNonEmptyString(params?.sessionId);
|
|
},
|
|
};
|
|
|
|
function getAdapterSessionCodec(adapterType: string) {
|
|
const adapter = getServerAdapter(adapterType);
|
|
return adapter.sessionCodec ?? defaultSessionCodec;
|
|
}
|
|
|
|
function normalizeSessionParams(params: Record<string, unknown> | null | undefined) {
|
|
if (!params) return null;
|
|
return Object.keys(params).length > 0 ? params : null;
|
|
}
|
|
|
|
function resolveNextSessionState(input: {
|
|
codec: AdapterSessionCodec;
|
|
adapterResult: AdapterExecutionResult;
|
|
previousParams: Record<string, unknown> | null;
|
|
previousDisplayId: string | null;
|
|
previousLegacySessionId: string | null;
|
|
}) {
|
|
const { codec, adapterResult, previousParams, previousDisplayId, previousLegacySessionId } = input;
|
|
|
|
if (adapterResult.clearSession) {
|
|
return {
|
|
params: null as Record<string, unknown> | null,
|
|
displayId: null as string | null,
|
|
legacySessionId: null as string | null,
|
|
};
|
|
}
|
|
|
|
const explicitParams = adapterResult.sessionParams;
|
|
const hasExplicitParams = adapterResult.sessionParams !== undefined;
|
|
const hasExplicitSessionId = adapterResult.sessionId !== undefined;
|
|
const explicitSessionId = readNonEmptyString(adapterResult.sessionId);
|
|
const hasExplicitDisplay = adapterResult.sessionDisplayId !== undefined;
|
|
const explicitDisplayId = readNonEmptyString(adapterResult.sessionDisplayId);
|
|
const shouldUsePrevious = !hasExplicitParams && !hasExplicitSessionId && !hasExplicitDisplay;
|
|
|
|
const candidateParams =
|
|
hasExplicitParams
|
|
? explicitParams
|
|
: hasExplicitSessionId
|
|
? (explicitSessionId ? { sessionId: explicitSessionId } : null)
|
|
: previousParams;
|
|
|
|
const serialized = normalizeSessionParams(codec.serialize(normalizeSessionParams(candidateParams) ?? null));
|
|
const deserialized = normalizeSessionParams(codec.deserialize(serialized));
|
|
|
|
const displayId = truncateDisplayId(
|
|
explicitDisplayId ??
|
|
(codec.getDisplayId ? codec.getDisplayId(deserialized) : null) ??
|
|
readNonEmptyString(deserialized?.sessionId) ??
|
|
(shouldUsePrevious ? previousDisplayId : null) ??
|
|
explicitSessionId ??
|
|
(shouldUsePrevious ? previousLegacySessionId : null),
|
|
);
|
|
|
|
const legacySessionId =
|
|
explicitSessionId ??
|
|
readNonEmptyString(deserialized?.sessionId) ??
|
|
displayId ??
|
|
(shouldUsePrevious ? previousLegacySessionId : null);
|
|
|
|
return {
|
|
params: serialized,
|
|
displayId,
|
|
legacySessionId,
|
|
};
|
|
}
|
|
|
|
export function heartbeatService(db: Db) {
|
|
const runLogStore = getRunLogStore();
|
|
const secretsSvc = secretService(db);
|
|
const issuesSvc = issueService(db);
|
|
const activeRunExecutions = new Set<string>();
|
|
|
|
async function getAgent(agentId: string) {
|
|
return db
|
|
.select()
|
|
.from(agents)
|
|
.where(eq(agents.id, agentId))
|
|
.then((rows) => rows[0] ?? null);
|
|
}
|
|
|
|
async function getRun(runId: string) {
|
|
return db
|
|
.select()
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, runId))
|
|
.then((rows) => rows[0] ?? null);
|
|
}
|
|
|
|
async function getRuntimeState(agentId: string) {
|
|
return db
|
|
.select()
|
|
.from(agentRuntimeState)
|
|
.where(eq(agentRuntimeState.agentId, agentId))
|
|
.then((rows) => rows[0] ?? null);
|
|
}
|
|
|
|
async function getTaskSession(
|
|
companyId: string,
|
|
agentId: string,
|
|
adapterType: string,
|
|
taskKey: string,
|
|
) {
|
|
return db
|
|
.select()
|
|
.from(agentTaskSessions)
|
|
.where(
|
|
and(
|
|
eq(agentTaskSessions.companyId, companyId),
|
|
eq(agentTaskSessions.agentId, agentId),
|
|
eq(agentTaskSessions.adapterType, adapterType),
|
|
eq(agentTaskSessions.taskKey, taskKey),
|
|
),
|
|
)
|
|
.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,
|
|
) {
|
|
if (taskKey) {
|
|
const codec = getAdapterSessionCodec(agent.adapterType);
|
|
const existingTaskSession = await getTaskSession(
|
|
agent.companyId,
|
|
agent.id,
|
|
agent.adapterType,
|
|
taskKey,
|
|
);
|
|
const parsedParams = normalizeSessionParams(
|
|
codec.deserialize(existingTaskSession?.sessionParamsJson ?? null),
|
|
);
|
|
return truncateDisplayId(
|
|
existingTaskSession?.sessionDisplayId ??
|
|
(codec.getDisplayId ? codec.getDisplayId(parsedParams) : null) ??
|
|
readNonEmptyString(parsedParams?.sessionId),
|
|
);
|
|
}
|
|
|
|
const runtimeForRun = await getRuntimeState(agent.id);
|
|
return runtimeForRun?.sessionId ?? null;
|
|
}
|
|
|
|
async function resolveWorkspaceForRun(
|
|
agent: typeof agents.$inferSelect,
|
|
context: Record<string, unknown>,
|
|
previousSessionParams: Record<string, unknown> | null,
|
|
opts?: { useProjectWorkspace?: boolean | null },
|
|
): Promise<ResolvedWorkspaceForRun> {
|
|
const issueId = readNonEmptyString(context.issueId);
|
|
const contextProjectId = readNonEmptyString(context.projectId);
|
|
const issueProjectId = issueId
|
|
? await db
|
|
.select({ projectId: issues.projectId })
|
|
.from(issues)
|
|
.where(and(eq(issues.id, issueId), eq(issues.companyId, agent.companyId)))
|
|
.then((rows) => rows[0]?.projectId ?? null)
|
|
: null;
|
|
const resolvedProjectId = issueProjectId ?? contextProjectId;
|
|
const useProjectWorkspace = opts?.useProjectWorkspace !== false;
|
|
const workspaceProjectId = useProjectWorkspace ? resolvedProjectId : null;
|
|
|
|
const projectWorkspaceRows = workspaceProjectId
|
|
? await db
|
|
.select()
|
|
.from(projectWorkspaces)
|
|
.where(
|
|
and(
|
|
eq(projectWorkspaces.companyId, agent.companyId),
|
|
eq(projectWorkspaces.projectId, workspaceProjectId),
|
|
),
|
|
)
|
|
.orderBy(asc(projectWorkspaces.createdAt), asc(projectWorkspaces.id))
|
|
: [];
|
|
|
|
const workspaceHints = projectWorkspaceRows.map((workspace) => ({
|
|
workspaceId: workspace.id,
|
|
cwd: readNonEmptyString(workspace.cwd),
|
|
repoUrl: readNonEmptyString(workspace.repoUrl),
|
|
repoRef: readNonEmptyString(workspace.repoRef),
|
|
}));
|
|
|
|
if (projectWorkspaceRows.length > 0) {
|
|
const missingProjectCwds: string[] = [];
|
|
let hasConfiguredProjectCwd = false;
|
|
for (const workspace of projectWorkspaceRows) {
|
|
const projectCwd = readNonEmptyString(workspace.cwd);
|
|
if (!projectCwd || projectCwd === REPO_ONLY_CWD_SENTINEL) {
|
|
continue;
|
|
}
|
|
hasConfiguredProjectCwd = true;
|
|
const projectCwdExists = await fs
|
|
.stat(projectCwd)
|
|
.then((stats) => stats.isDirectory())
|
|
.catch(() => false);
|
|
if (projectCwdExists) {
|
|
return {
|
|
cwd: projectCwd,
|
|
source: "project_primary" as const,
|
|
projectId: resolvedProjectId,
|
|
workspaceId: workspace.id,
|
|
repoUrl: workspace.repoUrl,
|
|
repoRef: workspace.repoRef,
|
|
workspaceHints,
|
|
warnings: [],
|
|
};
|
|
}
|
|
missingProjectCwds.push(projectCwd);
|
|
}
|
|
|
|
const fallbackCwd = resolveDefaultAgentWorkspaceDir(agent.id);
|
|
await fs.mkdir(fallbackCwd, { recursive: true });
|
|
const warnings: string[] = [];
|
|
if (missingProjectCwds.length > 0) {
|
|
const firstMissing = missingProjectCwds[0];
|
|
const extraMissingCount = Math.max(0, missingProjectCwds.length - 1);
|
|
warnings.push(
|
|
extraMissingCount > 0
|
|
? `Project workspace path "${firstMissing}" and ${extraMissingCount} other configured path(s) are not available yet. Using fallback workspace "${fallbackCwd}" for this run.`
|
|
: `Project workspace path "${firstMissing}" is not available yet. Using fallback workspace "${fallbackCwd}" for this run.`,
|
|
);
|
|
} else if (!hasConfiguredProjectCwd) {
|
|
warnings.push(
|
|
`Project workspace has no local cwd configured. Using fallback workspace "${fallbackCwd}" for this run.`,
|
|
);
|
|
}
|
|
return {
|
|
cwd: fallbackCwd,
|
|
source: "project_primary" as const,
|
|
projectId: resolvedProjectId,
|
|
workspaceId: projectWorkspaceRows[0]?.id ?? null,
|
|
repoUrl: projectWorkspaceRows[0]?.repoUrl ?? null,
|
|
repoRef: projectWorkspaceRows[0]?.repoRef ?? null,
|
|
workspaceHints,
|
|
warnings,
|
|
};
|
|
}
|
|
|
|
const sessionCwd = readNonEmptyString(previousSessionParams?.cwd);
|
|
if (sessionCwd) {
|
|
const sessionCwdExists = await fs
|
|
.stat(sessionCwd)
|
|
.then((stats) => stats.isDirectory())
|
|
.catch(() => false);
|
|
if (sessionCwdExists) {
|
|
return {
|
|
cwd: sessionCwd,
|
|
source: "task_session" as const,
|
|
projectId: resolvedProjectId,
|
|
workspaceId: readNonEmptyString(previousSessionParams?.workspaceId),
|
|
repoUrl: readNonEmptyString(previousSessionParams?.repoUrl),
|
|
repoRef: readNonEmptyString(previousSessionParams?.repoRef),
|
|
workspaceHints,
|
|
warnings: [],
|
|
};
|
|
}
|
|
}
|
|
|
|
const cwd = resolveDefaultAgentWorkspaceDir(agent.id);
|
|
await fs.mkdir(cwd, { recursive: true });
|
|
const warnings: string[] = [];
|
|
if (sessionCwd) {
|
|
warnings.push(
|
|
`Saved session workspace "${sessionCwd}" is not available. Using fallback workspace "${cwd}" for this run.`,
|
|
);
|
|
} else if (resolvedProjectId) {
|
|
warnings.push(
|
|
`No project workspace directory is currently available for this issue. Using fallback workspace "${cwd}" for this run.`,
|
|
);
|
|
} else {
|
|
warnings.push(
|
|
`No project or prior session workspace was available. Using fallback workspace "${cwd}" for this run.`,
|
|
);
|
|
}
|
|
return {
|
|
cwd,
|
|
source: "agent_home" as const,
|
|
projectId: resolvedProjectId,
|
|
workspaceId: null,
|
|
repoUrl: null,
|
|
repoRef: null,
|
|
workspaceHints,
|
|
warnings,
|
|
};
|
|
}
|
|
|
|
async function upsertTaskSession(input: {
|
|
companyId: string;
|
|
agentId: string;
|
|
adapterType: string;
|
|
taskKey: string;
|
|
sessionParamsJson: Record<string, unknown> | null;
|
|
sessionDisplayId: string | null;
|
|
lastRunId: string | null;
|
|
lastError: string | null;
|
|
}) {
|
|
const existing = await getTaskSession(
|
|
input.companyId,
|
|
input.agentId,
|
|
input.adapterType,
|
|
input.taskKey,
|
|
);
|
|
if (existing) {
|
|
return db
|
|
.update(agentTaskSessions)
|
|
.set({
|
|
sessionParamsJson: input.sessionParamsJson,
|
|
sessionDisplayId: input.sessionDisplayId,
|
|
lastRunId: input.lastRunId,
|
|
lastError: input.lastError,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(agentTaskSessions.id, existing.id))
|
|
.returning()
|
|
.then((rows) => rows[0] ?? null);
|
|
}
|
|
|
|
return db
|
|
.insert(agentTaskSessions)
|
|
.values({
|
|
companyId: input.companyId,
|
|
agentId: input.agentId,
|
|
adapterType: input.adapterType,
|
|
taskKey: input.taskKey,
|
|
sessionParamsJson: input.sessionParamsJson,
|
|
sessionDisplayId: input.sessionDisplayId,
|
|
lastRunId: input.lastRunId,
|
|
lastError: input.lastError,
|
|
})
|
|
.returning()
|
|
.then((rows) => rows[0] ?? null);
|
|
}
|
|
|
|
async function clearTaskSessions(
|
|
companyId: string,
|
|
agentId: string,
|
|
opts?: { taskKey?: string | null; adapterType?: string | null },
|
|
) {
|
|
const conditions = [
|
|
eq(agentTaskSessions.companyId, companyId),
|
|
eq(agentTaskSessions.agentId, agentId),
|
|
];
|
|
if (opts?.taskKey) {
|
|
conditions.push(eq(agentTaskSessions.taskKey, opts.taskKey));
|
|
}
|
|
if (opts?.adapterType) {
|
|
conditions.push(eq(agentTaskSessions.adapterType, opts.adapterType));
|
|
}
|
|
|
|
return db
|
|
.delete(agentTaskSessions)
|
|
.where(and(...conditions))
|
|
.returning()
|
|
.then((rows) => rows.length);
|
|
}
|
|
|
|
async function ensureRuntimeState(agent: typeof agents.$inferSelect) {
|
|
const existing = await getRuntimeState(agent.id);
|
|
if (existing) return existing;
|
|
|
|
return db
|
|
.insert(agentRuntimeState)
|
|
.values({
|
|
agentId: agent.id,
|
|
companyId: agent.companyId,
|
|
adapterType: agent.adapterType,
|
|
stateJson: {},
|
|
})
|
|
.returning()
|
|
.then((rows) => rows[0]);
|
|
}
|
|
|
|
async function setRunStatus(
|
|
runId: string,
|
|
status: string,
|
|
patch?: Partial<typeof heartbeatRuns.$inferInsert>,
|
|
) {
|
|
const updated = await db
|
|
.update(heartbeatRuns)
|
|
.set({ status, ...patch, updatedAt: new Date() })
|
|
.where(eq(heartbeatRuns.id, runId))
|
|
.returning()
|
|
.then((rows) => rows[0] ?? null);
|
|
|
|
if (updated) {
|
|
publishLiveEvent({
|
|
companyId: updated.companyId,
|
|
type: "heartbeat.run.status",
|
|
payload: {
|
|
runId: updated.id,
|
|
agentId: updated.agentId,
|
|
status: updated.status,
|
|
invocationSource: updated.invocationSource,
|
|
triggerDetail: updated.triggerDetail,
|
|
error: updated.error ?? null,
|
|
errorCode: updated.errorCode ?? null,
|
|
startedAt: updated.startedAt ? new Date(updated.startedAt).toISOString() : null,
|
|
finishedAt: updated.finishedAt ? new Date(updated.finishedAt).toISOString() : null,
|
|
},
|
|
});
|
|
}
|
|
|
|
return updated;
|
|
}
|
|
|
|
async function setWakeupStatus(
|
|
wakeupRequestId: string | null | undefined,
|
|
status: string,
|
|
patch?: Partial<typeof agentWakeupRequests.$inferInsert>,
|
|
) {
|
|
if (!wakeupRequestId) return;
|
|
await db
|
|
.update(agentWakeupRequests)
|
|
.set({ status, ...patch, updatedAt: new Date() })
|
|
.where(eq(agentWakeupRequests.id, wakeupRequestId));
|
|
}
|
|
|
|
async function appendRunEvent(
|
|
run: typeof heartbeatRuns.$inferSelect,
|
|
seq: number,
|
|
event: {
|
|
eventType: string;
|
|
stream?: "system" | "stdout" | "stderr";
|
|
level?: "info" | "warn" | "error";
|
|
color?: string;
|
|
message?: string;
|
|
payload?: Record<string, unknown>;
|
|
},
|
|
) {
|
|
const sanitizedMessage = event.message ? redactCurrentUserText(event.message) : event.message;
|
|
const sanitizedPayload = event.payload ? redactCurrentUserValue(event.payload) : event.payload;
|
|
|
|
await db.insert(heartbeatRunEvents).values({
|
|
companyId: run.companyId,
|
|
runId: run.id,
|
|
agentId: run.agentId,
|
|
seq,
|
|
eventType: event.eventType,
|
|
stream: event.stream,
|
|
level: event.level,
|
|
color: event.color,
|
|
message: sanitizedMessage,
|
|
payload: sanitizedPayload,
|
|
});
|
|
|
|
publishLiveEvent({
|
|
companyId: run.companyId,
|
|
type: "heartbeat.run.event",
|
|
payload: {
|
|
runId: run.id,
|
|
agentId: run.agentId,
|
|
seq,
|
|
eventType: event.eventType,
|
|
stream: event.stream ?? null,
|
|
level: event.level ?? null,
|
|
color: event.color ?? null,
|
|
message: sanitizedMessage ?? null,
|
|
payload: sanitizedPayload ?? null,
|
|
},
|
|
});
|
|
}
|
|
|
|
function parseHeartbeatPolicy(agent: typeof agents.$inferSelect) {
|
|
const runtimeConfig = parseObject(agent.runtimeConfig);
|
|
const heartbeat = parseObject(runtimeConfig.heartbeat);
|
|
|
|
return {
|
|
enabled: asBoolean(heartbeat.enabled, true),
|
|
intervalSec: Math.max(0, asNumber(heartbeat.intervalSec, 0)),
|
|
wakeOnDemand: asBoolean(heartbeat.wakeOnDemand ?? heartbeat.wakeOnAssignment ?? heartbeat.wakeOnOnDemand ?? heartbeat.wakeOnAutomation, true),
|
|
maxConcurrentRuns: normalizeMaxConcurrentRuns(heartbeat.maxConcurrentRuns),
|
|
};
|
|
}
|
|
|
|
async function countRunningRunsForAgent(agentId: string) {
|
|
const [{ count }] = await db
|
|
.select({ count: sql<number>`count(*)` })
|
|
.from(heartbeatRuns)
|
|
.where(and(eq(heartbeatRuns.agentId, agentId), eq(heartbeatRuns.status, "running")));
|
|
return Number(count ?? 0);
|
|
}
|
|
|
|
async function claimQueuedRun(run: typeof heartbeatRuns.$inferSelect) {
|
|
if (run.status !== "queued") return run;
|
|
const claimedAt = new Date();
|
|
const claimed = await db
|
|
.update(heartbeatRuns)
|
|
.set({
|
|
status: "running",
|
|
startedAt: run.startedAt ?? claimedAt,
|
|
updatedAt: claimedAt,
|
|
})
|
|
.where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.status, "queued")))
|
|
.returning()
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!claimed) return null;
|
|
|
|
publishLiveEvent({
|
|
companyId: claimed.companyId,
|
|
type: "heartbeat.run.status",
|
|
payload: {
|
|
runId: claimed.id,
|
|
agentId: claimed.agentId,
|
|
status: claimed.status,
|
|
invocationSource: claimed.invocationSource,
|
|
triggerDetail: claimed.triggerDetail,
|
|
error: claimed.error ?? null,
|
|
errorCode: claimed.errorCode ?? null,
|
|
startedAt: claimed.startedAt ? new Date(claimed.startedAt).toISOString() : null,
|
|
finishedAt: claimed.finishedAt ? new Date(claimed.finishedAt).toISOString() : null,
|
|
},
|
|
});
|
|
|
|
await setWakeupStatus(claimed.wakeupRequestId, "claimed", { claimedAt });
|
|
return claimed;
|
|
}
|
|
|
|
async function finalizeAgentStatus(
|
|
agentId: string,
|
|
outcome: "succeeded" | "failed" | "cancelled" | "timed_out",
|
|
) {
|
|
const existing = await getAgent(agentId);
|
|
if (!existing) return;
|
|
|
|
if (existing.status === "paused" || existing.status === "terminated") {
|
|
return;
|
|
}
|
|
|
|
const runningCount = await countRunningRunsForAgent(agentId);
|
|
const nextStatus =
|
|
runningCount > 0
|
|
? "running"
|
|
: outcome === "succeeded" || outcome === "cancelled"
|
|
? "idle"
|
|
: "error";
|
|
|
|
const updated = await db
|
|
.update(agents)
|
|
.set({
|
|
status: nextStatus,
|
|
lastHeartbeatAt: new Date(),
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(agents.id, agentId))
|
|
.returning()
|
|
.then((rows) => rows[0] ?? null);
|
|
|
|
if (updated) {
|
|
publishLiveEvent({
|
|
companyId: updated.companyId,
|
|
type: "agent.status",
|
|
payload: {
|
|
agentId: updated.id,
|
|
status: updated.status,
|
|
lastHeartbeatAt: updated.lastHeartbeatAt
|
|
? new Date(updated.lastHeartbeatAt).toISOString()
|
|
: null,
|
|
outcome,
|
|
},
|
|
});
|
|
}
|
|
}
|
|
|
|
async function reapOrphanedRuns(opts?: { staleThresholdMs?: number }) {
|
|
const staleThresholdMs = opts?.staleThresholdMs ?? 0;
|
|
const now = new Date();
|
|
|
|
// Find all runs in "queued" or "running" state
|
|
const activeRuns = await db
|
|
.select()
|
|
.from(heartbeatRuns)
|
|
.where(inArray(heartbeatRuns.status, ["queued", "running"]));
|
|
|
|
const reaped: string[] = [];
|
|
|
|
for (const run of activeRuns) {
|
|
if (runningProcesses.has(run.id) || activeRunExecutions.has(run.id)) continue;
|
|
|
|
// Apply staleness threshold to avoid false positives
|
|
if (staleThresholdMs > 0) {
|
|
const refTime = run.updatedAt ? new Date(run.updatedAt).getTime() : 0;
|
|
if (now.getTime() - refTime < staleThresholdMs) continue;
|
|
}
|
|
|
|
await setRunStatus(run.id, "failed", {
|
|
error: "Process lost -- server may have restarted",
|
|
errorCode: "process_lost",
|
|
finishedAt: now,
|
|
});
|
|
await setWakeupStatus(run.wakeupRequestId, "failed", {
|
|
finishedAt: now,
|
|
error: "Process lost -- server may have restarted",
|
|
});
|
|
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);
|
|
}
|
|
await finalizeAgentStatus(run.agentId, "failed");
|
|
await startNextQueuedRunForAgent(run.agentId);
|
|
runningProcesses.delete(run.id);
|
|
reaped.push(run.id);
|
|
}
|
|
|
|
if (reaped.length > 0) {
|
|
logger.warn({ reapedCount: reaped.length, runIds: reaped }, "reaped orphaned heartbeat runs");
|
|
}
|
|
return { reaped: reaped.length, runIds: reaped };
|
|
}
|
|
|
|
async function resumeQueuedRuns() {
|
|
const queuedRuns = await db
|
|
.select({ agentId: heartbeatRuns.agentId })
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.status, "queued"));
|
|
|
|
const agentIds = [...new Set(queuedRuns.map((r) => r.agentId))];
|
|
for (const agentId of agentIds) {
|
|
await startNextQueuedRunForAgent(agentId);
|
|
}
|
|
}
|
|
|
|
async function updateRuntimeState(
|
|
agent: typeof agents.$inferSelect,
|
|
run: typeof heartbeatRuns.$inferSelect,
|
|
result: AdapterExecutionResult,
|
|
session: { legacySessionId: string | null },
|
|
normalizedUsage?: UsageTotals | null,
|
|
) {
|
|
await ensureRuntimeState(agent);
|
|
const usage = normalizedUsage ?? normalizeUsageTotals(result.usage);
|
|
const inputTokens = usage?.inputTokens ?? 0;
|
|
const outputTokens = usage?.outputTokens ?? 0;
|
|
const cachedInputTokens = usage?.cachedInputTokens ?? 0;
|
|
const additionalCostCents = Math.max(0, Math.round((result.costUsd ?? 0) * 100));
|
|
const hasTokenUsage = inputTokens > 0 || outputTokens > 0 || cachedInputTokens > 0;
|
|
|
|
await db
|
|
.update(agentRuntimeState)
|
|
.set({
|
|
adapterType: agent.adapterType,
|
|
sessionId: session.legacySessionId,
|
|
lastRunId: run.id,
|
|
lastRunStatus: run.status,
|
|
lastError: result.errorMessage ?? null,
|
|
totalInputTokens: sql`${agentRuntimeState.totalInputTokens} + ${inputTokens}`,
|
|
totalOutputTokens: sql`${agentRuntimeState.totalOutputTokens} + ${outputTokens}`,
|
|
totalCachedInputTokens: sql`${agentRuntimeState.totalCachedInputTokens} + ${cachedInputTokens}`,
|
|
totalCostCents: sql`${agentRuntimeState.totalCostCents} + ${additionalCostCents}`,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(agentRuntimeState.agentId, agent.id));
|
|
|
|
if (additionalCostCents > 0 || hasTokenUsage) {
|
|
const costs = costService(db);
|
|
await costs.createEvent(agent.companyId, {
|
|
agentId: agent.id,
|
|
provider: result.provider ?? "unknown",
|
|
model: result.model ?? "unknown",
|
|
inputTokens,
|
|
outputTokens,
|
|
costCents: additionalCostCents,
|
|
occurredAt: new Date(),
|
|
});
|
|
}
|
|
}
|
|
|
|
async function startNextQueuedRunForAgent(agentId: string) {
|
|
return withAgentStartLock(agentId, async () => {
|
|
const agent = await getAgent(agentId);
|
|
if (!agent) return [];
|
|
const policy = parseHeartbeatPolicy(agent);
|
|
const runningCount = await countRunningRunsForAgent(agentId);
|
|
const availableSlots = Math.max(0, policy.maxConcurrentRuns - runningCount);
|
|
if (availableSlots <= 0) return [];
|
|
|
|
const queuedRuns = await db
|
|
.select()
|
|
.from(heartbeatRuns)
|
|
.where(and(eq(heartbeatRuns.agentId, agentId), eq(heartbeatRuns.status, "queued")))
|
|
.orderBy(asc(heartbeatRuns.createdAt))
|
|
.limit(availableSlots);
|
|
if (queuedRuns.length === 0) return [];
|
|
|
|
const claimedRuns: Array<typeof heartbeatRuns.$inferSelect> = [];
|
|
for (const queuedRun of queuedRuns) {
|
|
const claimed = await claimQueuedRun(queuedRun);
|
|
if (claimed) claimedRuns.push(claimed);
|
|
}
|
|
if (claimedRuns.length === 0) return [];
|
|
|
|
for (const claimedRun of claimedRuns) {
|
|
void executeRun(claimedRun.id).catch((err) => {
|
|
logger.error({ err, runId: claimedRun.id }, "queued heartbeat execution failed");
|
|
});
|
|
}
|
|
return claimedRuns;
|
|
});
|
|
}
|
|
|
|
async function executeRun(runId: string) {
|
|
let run = await getRun(runId);
|
|
if (!run) return;
|
|
if (run.status !== "queued" && run.status !== "running") return;
|
|
|
|
if (run.status === "queued") {
|
|
const claimed = await claimQueuedRun(run);
|
|
if (!claimed) {
|
|
// Another worker has already claimed or finalized this run.
|
|
return;
|
|
}
|
|
run = claimed;
|
|
}
|
|
|
|
activeRunExecutions.add(run.id);
|
|
|
|
try {
|
|
const agent = await getAgent(run.agentId);
|
|
if (!agent) {
|
|
await setRunStatus(runId, "failed", {
|
|
error: "Agent not found",
|
|
errorCode: "agent_not_found",
|
|
finishedAt: new Date(),
|
|
});
|
|
await setWakeupStatus(run.wakeupRequestId, "failed", {
|
|
finishedAt: new Date(),
|
|
error: "Agent not found",
|
|
});
|
|
const failedRun = await getRun(runId);
|
|
if (failedRun) await releaseIssueExecutionAndPromote(failedRun);
|
|
return;
|
|
}
|
|
|
|
const runtime = await ensureRuntimeState(agent);
|
|
const context = parseObject(run.contextSnapshot);
|
|
const taskKey = deriveTaskKey(context, null);
|
|
const sessionCodec = getAdapterSessionCodec(agent.adapterType);
|
|
const issueId = readNonEmptyString(context.issueId);
|
|
const issueAssigneeConfig = issueId
|
|
? await db
|
|
.select({
|
|
projectId: issues.projectId,
|
|
assigneeAgentId: issues.assigneeAgentId,
|
|
assigneeAdapterOverrides: issues.assigneeAdapterOverrides,
|
|
executionWorkspaceSettings: issues.executionWorkspaceSettings,
|
|
})
|
|
.from(issues)
|
|
.where(and(eq(issues.id, issueId), eq(issues.companyId, agent.companyId)))
|
|
.then((rows) => rows[0] ?? null)
|
|
: null;
|
|
const issueAssigneeOverrides =
|
|
issueAssigneeConfig && issueAssigneeConfig.assigneeAgentId === agent.id
|
|
? parseIssueAssigneeAdapterOverrides(
|
|
issueAssigneeConfig.assigneeAdapterOverrides,
|
|
)
|
|
: null;
|
|
const issueExecutionWorkspaceSettings = parseIssueExecutionWorkspaceSettings(
|
|
issueAssigneeConfig?.executionWorkspaceSettings,
|
|
);
|
|
const contextProjectId = readNonEmptyString(context.projectId);
|
|
const executionProjectId = issueAssigneeConfig?.projectId ?? contextProjectId;
|
|
const projectExecutionWorkspacePolicy = executionProjectId
|
|
? await db
|
|
.select({ executionWorkspacePolicy: projects.executionWorkspacePolicy })
|
|
.from(projects)
|
|
.where(and(eq(projects.id, executionProjectId), eq(projects.companyId, agent.companyId)))
|
|
.then((rows) => parseProjectExecutionWorkspacePolicy(rows[0]?.executionWorkspacePolicy))
|
|
: null;
|
|
const taskSession = taskKey
|
|
? await getTaskSession(agent.companyId, agent.id, agent.adapterType, taskKey)
|
|
: null;
|
|
const resetTaskSession = shouldResetTaskSessionForWake(context);
|
|
const sessionResetReason = describeSessionResetReason(context);
|
|
const taskSessionForRun = resetTaskSession ? null : taskSession;
|
|
const previousSessionParams = normalizeSessionParams(
|
|
sessionCodec.deserialize(taskSessionForRun?.sessionParamsJson ?? null),
|
|
);
|
|
const config = parseObject(agent.adapterConfig);
|
|
const executionWorkspaceMode = resolveExecutionWorkspaceMode({
|
|
projectPolicy: projectExecutionWorkspacePolicy,
|
|
issueSettings: issueExecutionWorkspaceSettings,
|
|
legacyUseProjectWorkspace: issueAssigneeOverrides?.useProjectWorkspace ?? null,
|
|
});
|
|
const resolvedWorkspace = await resolveWorkspaceForRun(
|
|
agent,
|
|
context,
|
|
previousSessionParams,
|
|
{ useProjectWorkspace: executionWorkspaceMode !== "agent_default" },
|
|
);
|
|
const workspaceManagedConfig = buildExecutionWorkspaceAdapterConfig({
|
|
agentConfig: config,
|
|
projectPolicy: projectExecutionWorkspacePolicy,
|
|
issueSettings: issueExecutionWorkspaceSettings,
|
|
mode: executionWorkspaceMode,
|
|
legacyUseProjectWorkspace: issueAssigneeOverrides?.useProjectWorkspace ?? null,
|
|
});
|
|
const mergedConfig = issueAssigneeOverrides?.adapterConfig
|
|
? { ...workspaceManagedConfig, ...issueAssigneeOverrides.adapterConfig }
|
|
: workspaceManagedConfig;
|
|
const { config: resolvedConfig, secretKeys } = await secretsSvc.resolveAdapterConfigForRuntime(
|
|
agent.companyId,
|
|
mergedConfig,
|
|
);
|
|
const issueRef = issueId
|
|
? await db
|
|
.select({
|
|
id: issues.id,
|
|
identifier: issues.identifier,
|
|
title: issues.title,
|
|
})
|
|
.from(issues)
|
|
.where(and(eq(issues.id, issueId), eq(issues.companyId, agent.companyId)))
|
|
.then((rows) => rows[0] ?? null)
|
|
: null;
|
|
const executionWorkspace = await realizeExecutionWorkspace({
|
|
base: {
|
|
baseCwd: resolvedWorkspace.cwd,
|
|
source: resolvedWorkspace.source,
|
|
projectId: resolvedWorkspace.projectId,
|
|
workspaceId: resolvedWorkspace.workspaceId,
|
|
repoUrl: resolvedWorkspace.repoUrl,
|
|
repoRef: resolvedWorkspace.repoRef,
|
|
},
|
|
config: resolvedConfig,
|
|
issue: issueRef,
|
|
agent: {
|
|
id: agent.id,
|
|
name: agent.name,
|
|
companyId: agent.companyId,
|
|
},
|
|
});
|
|
const runtimeSessionResolution = resolveRuntimeSessionParamsForWorkspace({
|
|
agentId: agent.id,
|
|
previousSessionParams,
|
|
resolvedWorkspace: {
|
|
...resolvedWorkspace,
|
|
cwd: executionWorkspace.cwd,
|
|
},
|
|
});
|
|
const runtimeSessionParams = runtimeSessionResolution.sessionParams;
|
|
const runtimeWorkspaceWarnings = [
|
|
...resolvedWorkspace.warnings,
|
|
...executionWorkspace.warnings,
|
|
...(runtimeSessionResolution.warning ? [runtimeSessionResolution.warning] : []),
|
|
...(resetTaskSession && sessionResetReason
|
|
? [
|
|
taskKey
|
|
? `Skipping saved session resume for task "${taskKey}" because ${sessionResetReason}.`
|
|
: `Skipping saved session resume because ${sessionResetReason}.`,
|
|
]
|
|
: []),
|
|
];
|
|
context.paperclipWorkspace = {
|
|
cwd: executionWorkspace.cwd,
|
|
source: executionWorkspace.source,
|
|
mode: executionWorkspaceMode,
|
|
strategy: executionWorkspace.strategy,
|
|
projectId: executionWorkspace.projectId,
|
|
workspaceId: executionWorkspace.workspaceId,
|
|
repoUrl: executionWorkspace.repoUrl,
|
|
repoRef: executionWorkspace.repoRef,
|
|
branchName: executionWorkspace.branchName,
|
|
worktreePath: executionWorkspace.worktreePath,
|
|
agentHome: resolveDefaultAgentWorkspaceDir(agent.id),
|
|
};
|
|
context.paperclipWorkspaces = resolvedWorkspace.workspaceHints;
|
|
const runtimeServiceIntents = (() => {
|
|
const runtimeConfig = parseObject(resolvedConfig.workspaceRuntime);
|
|
return Array.isArray(runtimeConfig.services)
|
|
? runtimeConfig.services.filter(
|
|
(value): value is Record<string, unknown> => typeof value === "object" && value !== null,
|
|
)
|
|
: [];
|
|
})();
|
|
if (runtimeServiceIntents.length > 0) {
|
|
context.paperclipRuntimeServiceIntents = runtimeServiceIntents;
|
|
} else {
|
|
delete context.paperclipRuntimeServiceIntents;
|
|
}
|
|
if (executionWorkspace.projectId && !readNonEmptyString(context.projectId)) {
|
|
context.projectId = executionWorkspace.projectId;
|
|
}
|
|
const runtimeSessionFallback = taskKey || resetTaskSession ? null : runtime.sessionId;
|
|
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: runtimeSessionIdForAdapter,
|
|
sessionParams: runtimeSessionParamsForAdapter,
|
|
sessionDisplayId: previousSessionDisplayId,
|
|
taskKey,
|
|
};
|
|
|
|
let seq = 1;
|
|
let handle: RunLogHandle | null = null;
|
|
let stdoutExcerpt = "";
|
|
let stderrExcerpt = "";
|
|
try {
|
|
const startedAt = run.startedAt ?? new Date();
|
|
const runningWithSession = await db
|
|
.update(heartbeatRuns)
|
|
.set({
|
|
startedAt,
|
|
sessionIdBefore: runtimeForAdapter.sessionDisplayId ?? runtimeForAdapter.sessionId,
|
|
contextSnapshot: context,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(heartbeatRuns.id, run.id))
|
|
.returning()
|
|
.then((rows) => rows[0] ?? null);
|
|
if (runningWithSession) run = runningWithSession;
|
|
|
|
const runningAgent = await db
|
|
.update(agents)
|
|
.set({ status: "running", updatedAt: new Date() })
|
|
.where(eq(agents.id, agent.id))
|
|
.returning()
|
|
.then((rows) => rows[0] ?? null);
|
|
|
|
if (runningAgent) {
|
|
publishLiveEvent({
|
|
companyId: runningAgent.companyId,
|
|
type: "agent.status",
|
|
payload: {
|
|
agentId: runningAgent.id,
|
|
status: runningAgent.status,
|
|
outcome: "running",
|
|
},
|
|
});
|
|
}
|
|
|
|
const currentRun = run;
|
|
await appendRunEvent(currentRun, seq++, {
|
|
eventType: "lifecycle",
|
|
stream: "system",
|
|
level: "info",
|
|
message: "run started",
|
|
});
|
|
|
|
handle = await runLogStore.begin({
|
|
companyId: run.companyId,
|
|
agentId: run.agentId,
|
|
runId,
|
|
});
|
|
|
|
await db
|
|
.update(heartbeatRuns)
|
|
.set({
|
|
logStore: handle.store,
|
|
logRef: handle.logRef,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(heartbeatRuns.id, runId));
|
|
|
|
const onLog = async (stream: "stdout" | "stderr", chunk: string) => {
|
|
const sanitizedChunk = redactCurrentUserText(chunk);
|
|
if (stream === "stdout") stdoutExcerpt = appendExcerpt(stdoutExcerpt, sanitizedChunk);
|
|
if (stream === "stderr") stderrExcerpt = appendExcerpt(stderrExcerpt, sanitizedChunk);
|
|
const ts = new Date().toISOString();
|
|
|
|
if (handle) {
|
|
await runLogStore.append(handle, {
|
|
stream,
|
|
chunk: sanitizedChunk,
|
|
ts,
|
|
});
|
|
}
|
|
|
|
const payloadChunk =
|
|
sanitizedChunk.length > MAX_LIVE_LOG_CHUNK_BYTES
|
|
? sanitizedChunk.slice(sanitizedChunk.length - MAX_LIVE_LOG_CHUNK_BYTES)
|
|
: sanitizedChunk;
|
|
|
|
publishLiveEvent({
|
|
companyId: run.companyId,
|
|
type: "heartbeat.run.log",
|
|
payload: {
|
|
runId: run.id,
|
|
agentId: run.agentId,
|
|
ts,
|
|
stream,
|
|
chunk: payloadChunk,
|
|
truncated: payloadChunk.length !== sanitizedChunk.length,
|
|
},
|
|
});
|
|
};
|
|
for (const warning of runtimeWorkspaceWarnings) {
|
|
await onLog("stderr", `[paperclip] ${warning}\n`);
|
|
}
|
|
const adapterEnv = Object.fromEntries(
|
|
Object.entries(parseObject(resolvedConfig.env)).filter(
|
|
(entry): entry is [string, string] => typeof entry[0] === "string" && typeof entry[1] === "string",
|
|
),
|
|
);
|
|
const runtimeServices = await ensureRuntimeServicesForRun({
|
|
db,
|
|
runId: run.id,
|
|
agent: {
|
|
id: agent.id,
|
|
name: agent.name,
|
|
companyId: agent.companyId,
|
|
},
|
|
issue: issueRef,
|
|
workspace: executionWorkspace,
|
|
config: resolvedConfig,
|
|
adapterEnv,
|
|
onLog,
|
|
});
|
|
if (runtimeServices.length > 0) {
|
|
context.paperclipRuntimeServices = runtimeServices;
|
|
context.paperclipRuntimePrimaryUrl =
|
|
runtimeServices.find((service) => readNonEmptyString(service.url))?.url ?? null;
|
|
await db
|
|
.update(heartbeatRuns)
|
|
.set({
|
|
contextSnapshot: context,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(heartbeatRuns.id, run.id));
|
|
}
|
|
if (issueId && (executionWorkspace.created || runtimeServices.some((service) => !service.reused))) {
|
|
try {
|
|
await issuesSvc.addComment(
|
|
issueId,
|
|
buildWorkspaceReadyComment({
|
|
workspace: executionWorkspace,
|
|
runtimeServices,
|
|
}),
|
|
{ agentId: agent.id },
|
|
);
|
|
} catch (err) {
|
|
await onLog(
|
|
"stderr",
|
|
`[paperclip] Failed to post workspace-ready comment: ${err instanceof Error ? err.message : String(err)}\n`,
|
|
);
|
|
}
|
|
}
|
|
const onAdapterMeta = async (meta: AdapterInvocationMeta) => {
|
|
if (meta.env && secretKeys.size > 0) {
|
|
for (const key of secretKeys) {
|
|
if (key in meta.env) meta.env[key] = "***REDACTED***";
|
|
}
|
|
}
|
|
await appendRunEvent(currentRun, seq++, {
|
|
eventType: "adapter.invoke",
|
|
stream: "system",
|
|
level: "info",
|
|
message: "adapter invocation",
|
|
payload: meta as unknown as Record<string, unknown>,
|
|
});
|
|
};
|
|
|
|
const adapter = getServerAdapter(agent.adapterType);
|
|
const authToken = adapter.supportsLocalAgentJwt
|
|
? createLocalAgentJwt(agent.id, agent.companyId, agent.adapterType, run.id)
|
|
: null;
|
|
if (adapter.supportsLocalAgentJwt && !authToken) {
|
|
logger.warn(
|
|
{
|
|
companyId: agent.companyId,
|
|
agentId: agent.id,
|
|
runId: run.id,
|
|
adapterType: agent.adapterType,
|
|
},
|
|
"local agent jwt secret missing or invalid; running without injected PAPERCLIP_API_KEY",
|
|
);
|
|
}
|
|
const adapterResult = await adapter.execute({
|
|
runId: run.id,
|
|
agent,
|
|
runtime: runtimeForAdapter,
|
|
config: resolvedConfig,
|
|
context,
|
|
onLog,
|
|
onMeta: onAdapterMeta,
|
|
authToken: authToken ?? undefined,
|
|
});
|
|
const adapterManagedRuntimeServices = adapterResult.runtimeServices
|
|
? await persistAdapterManagedRuntimeServices({
|
|
db,
|
|
adapterType: agent.adapterType,
|
|
runId: run.id,
|
|
agent: {
|
|
id: agent.id,
|
|
name: agent.name,
|
|
companyId: agent.companyId,
|
|
},
|
|
issue: issueRef,
|
|
workspace: executionWorkspace,
|
|
reports: adapterResult.runtimeServices,
|
|
})
|
|
: [];
|
|
if (adapterManagedRuntimeServices.length > 0) {
|
|
const combinedRuntimeServices = [
|
|
...runtimeServices,
|
|
...adapterManagedRuntimeServices,
|
|
];
|
|
context.paperclipRuntimeServices = combinedRuntimeServices;
|
|
context.paperclipRuntimePrimaryUrl =
|
|
combinedRuntimeServices.find((service) => readNonEmptyString(service.url))?.url ?? null;
|
|
await db
|
|
.update(heartbeatRuns)
|
|
.set({
|
|
contextSnapshot: context,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(heartbeatRuns.id, run.id));
|
|
if (issueId) {
|
|
try {
|
|
await issuesSvc.addComment(
|
|
issueId,
|
|
buildWorkspaceReadyComment({
|
|
workspace: executionWorkspace,
|
|
runtimeServices: adapterManagedRuntimeServices,
|
|
}),
|
|
{ agentId: agent.id },
|
|
);
|
|
} catch (err) {
|
|
await onLog(
|
|
"stderr",
|
|
`[paperclip] Failed to post adapter-managed runtime comment: ${err instanceof Error ? err.message : String(err)}\n`,
|
|
);
|
|
}
|
|
}
|
|
}
|
|
const nextSessionState = resolveNextSessionState({
|
|
codec: sessionCodec,
|
|
adapterResult,
|
|
previousParams: previousSessionParams,
|
|
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);
|
|
if (latestRun?.status === "cancelled") {
|
|
outcome = "cancelled";
|
|
} else if (adapterResult.timedOut) {
|
|
outcome = "timed_out";
|
|
} else if ((adapterResult.exitCode ?? 0) === 0 && !adapterResult.errorMessage) {
|
|
outcome = "succeeded";
|
|
} else {
|
|
outcome = "failed";
|
|
}
|
|
|
|
let logSummary: { bytes: number; sha256?: string; compressed: boolean } | null = null;
|
|
if (handle) {
|
|
logSummary = await runLogStore.finalize(handle);
|
|
}
|
|
|
|
const status =
|
|
outcome === "succeeded"
|
|
? "succeeded"
|
|
: outcome === "cancelled"
|
|
? "cancelled"
|
|
: outcome === "timed_out"
|
|
? "timed_out"
|
|
: "failed";
|
|
|
|
const usageJson =
|
|
normalizedUsage || adapterResult.costUsd != null
|
|
? ({
|
|
...(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>)
|
|
: null;
|
|
|
|
await setRunStatus(run.id, status, {
|
|
finishedAt: new Date(),
|
|
error:
|
|
outcome === "succeeded"
|
|
? null
|
|
: redactCurrentUserText(
|
|
adapterResult.errorMessage ?? (outcome === "timed_out" ? "Timed out" : "Adapter failed"),
|
|
),
|
|
errorCode:
|
|
outcome === "timed_out"
|
|
? "timeout"
|
|
: outcome === "cancelled"
|
|
? "cancelled"
|
|
: outcome === "failed"
|
|
? (adapterResult.errorCode ?? "adapter_failed")
|
|
: null,
|
|
exitCode: adapterResult.exitCode,
|
|
signal: adapterResult.signal,
|
|
usageJson,
|
|
resultJson: adapterResult.resultJson ?? null,
|
|
sessionIdAfter: nextSessionState.displayId ?? nextSessionState.legacySessionId,
|
|
stdoutExcerpt,
|
|
stderrExcerpt,
|
|
logBytes: logSummary?.bytes,
|
|
logSha256: logSummary?.sha256,
|
|
logCompressed: logSummary?.compressed ?? false,
|
|
});
|
|
|
|
await setWakeupStatus(run.wakeupRequestId, outcome === "succeeded" ? "completed" : status, {
|
|
finishedAt: new Date(),
|
|
error: adapterResult.errorMessage ?? null,
|
|
});
|
|
|
|
const finalizedRun = await getRun(run.id);
|
|
if (finalizedRun) {
|
|
await appendRunEvent(finalizedRun, seq++, {
|
|
eventType: "lifecycle",
|
|
stream: "system",
|
|
level: outcome === "succeeded" ? "info" : "error",
|
|
message: `run ${outcome}`,
|
|
payload: {
|
|
status,
|
|
exitCode: adapterResult.exitCode,
|
|
},
|
|
});
|
|
await releaseIssueExecutionAndPromote(finalizedRun);
|
|
}
|
|
|
|
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, {
|
|
taskKey,
|
|
adapterType: agent.adapterType,
|
|
});
|
|
} else {
|
|
await upsertTaskSession({
|
|
companyId: agent.companyId,
|
|
agentId: agent.id,
|
|
adapterType: agent.adapterType,
|
|
taskKey,
|
|
sessionParamsJson: nextSessionState.params,
|
|
sessionDisplayId: nextSessionState.displayId,
|
|
lastRunId: finalizedRun.id,
|
|
lastError: outcome === "succeeded" ? null : (adapterResult.errorMessage ?? "run_failed"),
|
|
});
|
|
}
|
|
}
|
|
}
|
|
await finalizeAgentStatus(agent.id, outcome);
|
|
} catch (err) {
|
|
const message = redactCurrentUserText(err instanceof Error ? err.message : "Unknown adapter failure");
|
|
logger.error({ err, runId }, "heartbeat execution failed");
|
|
|
|
let logSummary: { bytes: number; sha256?: string; compressed: boolean } | null = null;
|
|
if (handle) {
|
|
try {
|
|
logSummary = await runLogStore.finalize(handle);
|
|
} catch (finalizeErr) {
|
|
logger.warn({ err: finalizeErr, runId }, "failed to finalize run log after error");
|
|
}
|
|
}
|
|
|
|
const failedRun = await setRunStatus(run.id, "failed", {
|
|
error: message,
|
|
errorCode: "adapter_failed",
|
|
finishedAt: new Date(),
|
|
stdoutExcerpt,
|
|
stderrExcerpt,
|
|
logBytes: logSummary?.bytes,
|
|
logSha256: logSummary?.sha256,
|
|
logCompressed: logSummary?.compressed ?? false,
|
|
});
|
|
await setWakeupStatus(run.wakeupRequestId, "failed", {
|
|
finishedAt: new Date(),
|
|
error: message,
|
|
});
|
|
|
|
if (failedRun) {
|
|
await appendRunEvent(failedRun, seq++, {
|
|
eventType: "error",
|
|
stream: "system",
|
|
level: "error",
|
|
message,
|
|
});
|
|
await releaseIssueExecutionAndPromote(failedRun);
|
|
|
|
await updateRuntimeState(agent, failedRun, {
|
|
exitCode: null,
|
|
signal: null,
|
|
timedOut: false,
|
|
errorMessage: message,
|
|
}, {
|
|
legacySessionId: runtimeForAdapter.sessionId,
|
|
});
|
|
|
|
if (taskKey && (previousSessionParams || previousSessionDisplayId || taskSession)) {
|
|
await upsertTaskSession({
|
|
companyId: agent.companyId,
|
|
agentId: agent.id,
|
|
adapterType: agent.adapterType,
|
|
taskKey,
|
|
sessionParamsJson: previousSessionParams,
|
|
sessionDisplayId: previousSessionDisplayId,
|
|
lastRunId: failedRun.id,
|
|
lastError: message,
|
|
});
|
|
}
|
|
}
|
|
|
|
await finalizeAgentStatus(agent.id, "failed");
|
|
}
|
|
} catch (outerErr) {
|
|
// Setup code before adapter.execute threw (e.g. ensureRuntimeState, resolveWorkspaceForRun).
|
|
// The inner catch did not fire, so we must record the failure here.
|
|
const message = outerErr instanceof Error ? outerErr.message : "Unknown setup failure";
|
|
logger.error({ err: outerErr, runId }, "heartbeat execution setup failed");
|
|
await setRunStatus(runId, "failed", {
|
|
error: message,
|
|
errorCode: "adapter_failed",
|
|
finishedAt: new Date(),
|
|
}).catch(() => undefined);
|
|
await setWakeupStatus(run.wakeupRequestId, "failed", {
|
|
finishedAt: new Date(),
|
|
error: message,
|
|
}).catch(() => undefined);
|
|
const failedRun = await getRun(runId).catch(() => null);
|
|
if (failedRun) {
|
|
// Emit a run-log event so the failure is visible in the run timeline,
|
|
// consistent with what the inner catch block does for adapter failures.
|
|
await appendRunEvent(failedRun, 1, {
|
|
eventType: "error",
|
|
stream: "system",
|
|
level: "error",
|
|
message,
|
|
}).catch(() => undefined);
|
|
await releaseIssueExecutionAndPromote(failedRun).catch(() => undefined);
|
|
}
|
|
// Ensure the agent is not left stuck in "running" if the inner catch handler's
|
|
// DB calls threw (e.g. a transient DB error in finalizeAgentStatus).
|
|
await finalizeAgentStatus(run.agentId, "failed").catch(() => undefined);
|
|
} finally {
|
|
await releaseRuntimeServicesForRun(run.id).catch(() => undefined);
|
|
activeRunExecutions.delete(run.id);
|
|
await startNextQueuedRunForAgent(run.agentId);
|
|
}
|
|
}
|
|
|
|
async function releaseIssueExecutionAndPromote(run: typeof heartbeatRuns.$inferSelect) {
|
|
const promotedRun = await db.transaction(async (tx) => {
|
|
await tx.execute(
|
|
sql`select id from issues where company_id = ${run.companyId} and execution_run_id = ${run.id} for update`,
|
|
);
|
|
|
|
const issue = await tx
|
|
.select({
|
|
id: issues.id,
|
|
companyId: issues.companyId,
|
|
})
|
|
.from(issues)
|
|
.where(and(eq(issues.companyId, run.companyId), eq(issues.executionRunId, run.id)))
|
|
.then((rows) => rows[0] ?? null);
|
|
|
|
if (!issue) return;
|
|
|
|
await tx
|
|
.update(issues)
|
|
.set({
|
|
executionRunId: null,
|
|
executionAgentNameKey: null,
|
|
executionLockedAt: null,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(issues.id, issue.id));
|
|
|
|
while (true) {
|
|
const deferred = await tx
|
|
.select()
|
|
.from(agentWakeupRequests)
|
|
.where(
|
|
and(
|
|
eq(agentWakeupRequests.companyId, issue.companyId),
|
|
eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
|
sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issue.id}`,
|
|
),
|
|
)
|
|
.orderBy(asc(agentWakeupRequests.requestedAt))
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
|
|
if (!deferred) return null;
|
|
|
|
const deferredAgent = await tx
|
|
.select()
|
|
.from(agents)
|
|
.where(eq(agents.id, deferred.agentId))
|
|
.then((rows) => rows[0] ?? null);
|
|
|
|
if (
|
|
!deferredAgent ||
|
|
deferredAgent.companyId !== issue.companyId ||
|
|
deferredAgent.status === "paused" ||
|
|
deferredAgent.status === "terminated" ||
|
|
deferredAgent.status === "pending_approval"
|
|
) {
|
|
await tx
|
|
.update(agentWakeupRequests)
|
|
.set({
|
|
status: "failed",
|
|
finishedAt: new Date(),
|
|
error: "Deferred wake could not be promoted: agent is not invokable",
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(agentWakeupRequests.id, deferred.id));
|
|
continue;
|
|
}
|
|
|
|
const deferredPayload = parseObject(deferred.payload);
|
|
const deferredContextSeed = parseObject(deferredPayload[DEFERRED_WAKE_CONTEXT_KEY]);
|
|
const promotedContextSeed: Record<string, unknown> = { ...deferredContextSeed };
|
|
const promotedReason = readNonEmptyString(deferred.reason) ?? "issue_execution_promoted";
|
|
const promotedSource =
|
|
(readNonEmptyString(deferred.source) as WakeupOptions["source"]) ?? "automation";
|
|
const promotedTriggerDetail =
|
|
(readNonEmptyString(deferred.triggerDetail) as WakeupOptions["triggerDetail"]) ?? null;
|
|
const promotedPayload = deferredPayload;
|
|
delete promotedPayload[DEFERRED_WAKE_CONTEXT_KEY];
|
|
|
|
const {
|
|
contextSnapshot: promotedContextSnapshot,
|
|
taskKey: promotedTaskKey,
|
|
} = enrichWakeContextSnapshot({
|
|
contextSnapshot: promotedContextSeed,
|
|
reason: promotedReason,
|
|
source: promotedSource,
|
|
triggerDetail: promotedTriggerDetail,
|
|
payload: promotedPayload,
|
|
});
|
|
|
|
const sessionBefore = await resolveSessionBeforeForWakeup(deferredAgent, promotedTaskKey);
|
|
const now = new Date();
|
|
const newRun = await tx
|
|
.insert(heartbeatRuns)
|
|
.values({
|
|
companyId: deferredAgent.companyId,
|
|
agentId: deferredAgent.id,
|
|
invocationSource: promotedSource,
|
|
triggerDetail: promotedTriggerDetail,
|
|
status: "queued",
|
|
wakeupRequestId: deferred.id,
|
|
contextSnapshot: promotedContextSnapshot,
|
|
sessionIdBefore: sessionBefore,
|
|
})
|
|
.returning()
|
|
.then((rows) => rows[0]);
|
|
|
|
await tx
|
|
.update(agentWakeupRequests)
|
|
.set({
|
|
status: "queued",
|
|
reason: "issue_execution_promoted",
|
|
runId: newRun.id,
|
|
claimedAt: null,
|
|
finishedAt: null,
|
|
error: null,
|
|
updatedAt: now,
|
|
})
|
|
.where(eq(agentWakeupRequests.id, deferred.id));
|
|
|
|
await tx
|
|
.update(issues)
|
|
.set({
|
|
executionRunId: newRun.id,
|
|
executionAgentNameKey: normalizeAgentNameKey(deferredAgent.name),
|
|
executionLockedAt: now,
|
|
updatedAt: now,
|
|
})
|
|
.where(eq(issues.id, issue.id));
|
|
|
|
return newRun;
|
|
}
|
|
});
|
|
|
|
if (!promotedRun) return;
|
|
|
|
publishLiveEvent({
|
|
companyId: promotedRun.companyId,
|
|
type: "heartbeat.run.queued",
|
|
payload: {
|
|
runId: promotedRun.id,
|
|
agentId: promotedRun.agentId,
|
|
invocationSource: promotedRun.invocationSource,
|
|
triggerDetail: promotedRun.triggerDetail,
|
|
wakeupRequestId: promotedRun.wakeupRequestId,
|
|
},
|
|
});
|
|
|
|
await startNextQueuedRunForAgent(promotedRun.agentId);
|
|
}
|
|
|
|
async function enqueueWakeup(agentId: string, opts: WakeupOptions = {}) {
|
|
const source = opts.source ?? "on_demand";
|
|
const triggerDetail = opts.triggerDetail ?? null;
|
|
const contextSnapshot: Record<string, unknown> = { ...(opts.contextSnapshot ?? {}) };
|
|
const reason = opts.reason ?? null;
|
|
const payload = opts.payload ?? null;
|
|
const {
|
|
contextSnapshot: enrichedContextSnapshot,
|
|
issueIdFromPayload,
|
|
taskKey,
|
|
wakeCommentId,
|
|
} = enrichWakeContextSnapshot({
|
|
contextSnapshot,
|
|
reason,
|
|
source,
|
|
triggerDetail,
|
|
payload,
|
|
});
|
|
const issueId = readNonEmptyString(enrichedContextSnapshot.issueId) ?? issueIdFromPayload;
|
|
|
|
const agent = await getAgent(agentId);
|
|
if (!agent) throw notFound("Agent not found");
|
|
|
|
if (
|
|
agent.status === "paused" ||
|
|
agent.status === "terminated" ||
|
|
agent.status === "pending_approval"
|
|
) {
|
|
throw conflict("Agent is not invokable in its current state", { status: agent.status });
|
|
}
|
|
|
|
const policy = parseHeartbeatPolicy(agent);
|
|
const writeSkippedRequest = async (reason: string) => {
|
|
await db.insert(agentWakeupRequests).values({
|
|
companyId: agent.companyId,
|
|
agentId,
|
|
source,
|
|
triggerDetail,
|
|
reason,
|
|
payload,
|
|
status: "skipped",
|
|
requestedByActorType: opts.requestedByActorType ?? null,
|
|
requestedByActorId: opts.requestedByActorId ?? null,
|
|
idempotencyKey: opts.idempotencyKey ?? null,
|
|
finishedAt: new Date(),
|
|
});
|
|
};
|
|
|
|
if (source === "timer" && !policy.enabled) {
|
|
await writeSkippedRequest("heartbeat.disabled");
|
|
return null;
|
|
}
|
|
if (source !== "timer" && !policy.wakeOnDemand) {
|
|
await writeSkippedRequest("heartbeat.wakeOnDemand.disabled");
|
|
return null;
|
|
}
|
|
|
|
const bypassIssueExecutionLock =
|
|
reason === "issue_comment_mentioned" ||
|
|
readNonEmptyString(enrichedContextSnapshot.wakeReason) === "issue_comment_mentioned";
|
|
|
|
if (issueId && !bypassIssueExecutionLock) {
|
|
const agentNameKey = normalizeAgentNameKey(agent.name);
|
|
const sessionBefore = await resolveSessionBeforeForWakeup(agent, taskKey);
|
|
|
|
const outcome = await db.transaction(async (tx) => {
|
|
await tx.execute(
|
|
sql`select id from issues where id = ${issueId} and company_id = ${agent.companyId} for update`,
|
|
);
|
|
|
|
const issue = await tx
|
|
.select({
|
|
id: issues.id,
|
|
companyId: issues.companyId,
|
|
executionRunId: issues.executionRunId,
|
|
executionAgentNameKey: issues.executionAgentNameKey,
|
|
})
|
|
.from(issues)
|
|
.where(and(eq(issues.id, issueId), eq(issues.companyId, agent.companyId)))
|
|
.then((rows) => rows[0] ?? null);
|
|
|
|
if (!issue) {
|
|
await tx.insert(agentWakeupRequests).values({
|
|
companyId: agent.companyId,
|
|
agentId,
|
|
source,
|
|
triggerDetail,
|
|
reason: "issue_execution_issue_not_found",
|
|
payload,
|
|
status: "skipped",
|
|
requestedByActorType: opts.requestedByActorType ?? null,
|
|
requestedByActorId: opts.requestedByActorId ?? null,
|
|
idempotencyKey: opts.idempotencyKey ?? null,
|
|
finishedAt: new Date(),
|
|
});
|
|
return { kind: "skipped" as const };
|
|
}
|
|
|
|
let activeExecutionRun = issue.executionRunId
|
|
? await tx
|
|
.select()
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, issue.executionRunId))
|
|
.then((rows) => rows[0] ?? null)
|
|
: null;
|
|
|
|
if (activeExecutionRun && activeExecutionRun.status !== "queued" && activeExecutionRun.status !== "running") {
|
|
activeExecutionRun = null;
|
|
}
|
|
|
|
if (!activeExecutionRun && issue.executionRunId) {
|
|
await tx
|
|
.update(issues)
|
|
.set({
|
|
executionRunId: null,
|
|
executionAgentNameKey: null,
|
|
executionLockedAt: null,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(issues.id, issue.id));
|
|
}
|
|
|
|
if (!activeExecutionRun) {
|
|
const legacyRun = await tx
|
|
.select()
|
|
.from(heartbeatRuns)
|
|
.where(
|
|
and(
|
|
eq(heartbeatRuns.companyId, issue.companyId),
|
|
inArray(heartbeatRuns.status, ["queued", "running"]),
|
|
sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issue.id}`,
|
|
),
|
|
)
|
|
.orderBy(
|
|
sql`case when ${heartbeatRuns.status} = 'running' then 0 else 1 end`,
|
|
asc(heartbeatRuns.createdAt),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
|
|
if (legacyRun) {
|
|
activeExecutionRun = legacyRun;
|
|
const legacyAgent = await tx
|
|
.select({ name: agents.name })
|
|
.from(agents)
|
|
.where(eq(agents.id, legacyRun.agentId))
|
|
.then((rows) => rows[0] ?? null);
|
|
await tx
|
|
.update(issues)
|
|
.set({
|
|
executionRunId: legacyRun.id,
|
|
executionAgentNameKey: normalizeAgentNameKey(legacyAgent?.name),
|
|
executionLockedAt: new Date(),
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(issues.id, issue.id));
|
|
}
|
|
}
|
|
|
|
if (activeExecutionRun) {
|
|
const executionAgent = await tx
|
|
.select({ name: agents.name })
|
|
.from(agents)
|
|
.where(eq(agents.id, activeExecutionRun.agentId))
|
|
.then((rows) => rows[0] ?? null);
|
|
const executionAgentNameKey =
|
|
normalizeAgentNameKey(issue.executionAgentNameKey) ??
|
|
normalizeAgentNameKey(executionAgent?.name);
|
|
const isSameExecutionAgent =
|
|
Boolean(executionAgentNameKey) && executionAgentNameKey === agentNameKey;
|
|
const shouldQueueFollowupForCommentWake =
|
|
Boolean(wakeCommentId) &&
|
|
activeExecutionRun.status === "running" &&
|
|
isSameExecutionAgent;
|
|
|
|
if (isSameExecutionAgent && !shouldQueueFollowupForCommentWake) {
|
|
const mergedContextSnapshot = mergeCoalescedContextSnapshot(
|
|
activeExecutionRun.contextSnapshot,
|
|
enrichedContextSnapshot,
|
|
);
|
|
const mergedRun = await tx
|
|
.update(heartbeatRuns)
|
|
.set({
|
|
contextSnapshot: mergedContextSnapshot,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(heartbeatRuns.id, activeExecutionRun.id))
|
|
.returning()
|
|
.then((rows) => rows[0] ?? activeExecutionRun);
|
|
|
|
await tx.insert(agentWakeupRequests).values({
|
|
companyId: agent.companyId,
|
|
agentId,
|
|
source,
|
|
triggerDetail,
|
|
reason: "issue_execution_same_name",
|
|
payload,
|
|
status: "coalesced",
|
|
coalescedCount: 1,
|
|
requestedByActorType: opts.requestedByActorType ?? null,
|
|
requestedByActorId: opts.requestedByActorId ?? null,
|
|
idempotencyKey: opts.idempotencyKey ?? null,
|
|
runId: mergedRun.id,
|
|
finishedAt: new Date(),
|
|
});
|
|
|
|
return { kind: "coalesced" as const, run: mergedRun };
|
|
}
|
|
|
|
const deferredPayload = {
|
|
...(payload ?? {}),
|
|
issueId,
|
|
[DEFERRED_WAKE_CONTEXT_KEY]: enrichedContextSnapshot,
|
|
};
|
|
|
|
const existingDeferred = await tx
|
|
.select()
|
|
.from(agentWakeupRequests)
|
|
.where(
|
|
and(
|
|
eq(agentWakeupRequests.companyId, agent.companyId),
|
|
eq(agentWakeupRequests.agentId, agentId),
|
|
eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
|
sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issue.id}`,
|
|
),
|
|
)
|
|
.orderBy(asc(agentWakeupRequests.requestedAt))
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
|
|
if (existingDeferred) {
|
|
const existingDeferredPayload = parseObject(existingDeferred.payload);
|
|
const existingDeferredContext = parseObject(existingDeferredPayload[DEFERRED_WAKE_CONTEXT_KEY]);
|
|
const mergedDeferredContext = mergeCoalescedContextSnapshot(
|
|
existingDeferredContext,
|
|
enrichedContextSnapshot,
|
|
);
|
|
const mergedDeferredPayload = {
|
|
...existingDeferredPayload,
|
|
...(payload ?? {}),
|
|
issueId,
|
|
[DEFERRED_WAKE_CONTEXT_KEY]: mergedDeferredContext,
|
|
};
|
|
|
|
await tx
|
|
.update(agentWakeupRequests)
|
|
.set({
|
|
payload: mergedDeferredPayload,
|
|
coalescedCount: (existingDeferred.coalescedCount ?? 0) + 1,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(agentWakeupRequests.id, existingDeferred.id));
|
|
|
|
return { kind: "deferred" as const };
|
|
}
|
|
|
|
await tx.insert(agentWakeupRequests).values({
|
|
companyId: agent.companyId,
|
|
agentId,
|
|
source,
|
|
triggerDetail,
|
|
reason: "issue_execution_deferred",
|
|
payload: deferredPayload,
|
|
status: "deferred_issue_execution",
|
|
requestedByActorType: opts.requestedByActorType ?? null,
|
|
requestedByActorId: opts.requestedByActorId ?? null,
|
|
idempotencyKey: opts.idempotencyKey ?? null,
|
|
});
|
|
|
|
return { kind: "deferred" as const };
|
|
}
|
|
|
|
const wakeupRequest = await tx
|
|
.insert(agentWakeupRequests)
|
|
.values({
|
|
companyId: agent.companyId,
|
|
agentId,
|
|
source,
|
|
triggerDetail,
|
|
reason,
|
|
payload,
|
|
status: "queued",
|
|
requestedByActorType: opts.requestedByActorType ?? null,
|
|
requestedByActorId: opts.requestedByActorId ?? null,
|
|
idempotencyKey: opts.idempotencyKey ?? null,
|
|
})
|
|
.returning()
|
|
.then((rows) => rows[0]);
|
|
|
|
const newRun = await tx
|
|
.insert(heartbeatRuns)
|
|
.values({
|
|
companyId: agent.companyId,
|
|
agentId,
|
|
invocationSource: source,
|
|
triggerDetail,
|
|
status: "queued",
|
|
wakeupRequestId: wakeupRequest.id,
|
|
contextSnapshot: enrichedContextSnapshot,
|
|
sessionIdBefore: sessionBefore,
|
|
})
|
|
.returning()
|
|
.then((rows) => rows[0]);
|
|
|
|
await tx
|
|
.update(agentWakeupRequests)
|
|
.set({
|
|
runId: newRun.id,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(agentWakeupRequests.id, wakeupRequest.id));
|
|
|
|
await tx
|
|
.update(issues)
|
|
.set({
|
|
executionRunId: newRun.id,
|
|
executionAgentNameKey: agentNameKey,
|
|
executionLockedAt: new Date(),
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(issues.id, issue.id));
|
|
|
|
return { kind: "queued" as const, run: newRun };
|
|
});
|
|
|
|
if (outcome.kind === "deferred" || outcome.kind === "skipped") return null;
|
|
if (outcome.kind === "coalesced") return outcome.run;
|
|
|
|
const newRun = outcome.run;
|
|
publishLiveEvent({
|
|
companyId: newRun.companyId,
|
|
type: "heartbeat.run.queued",
|
|
payload: {
|
|
runId: newRun.id,
|
|
agentId: newRun.agentId,
|
|
invocationSource: newRun.invocationSource,
|
|
triggerDetail: newRun.triggerDetail,
|
|
wakeupRequestId: newRun.wakeupRequestId,
|
|
},
|
|
});
|
|
|
|
await startNextQueuedRunForAgent(agent.id);
|
|
return newRun;
|
|
}
|
|
|
|
const activeRuns = await db
|
|
.select()
|
|
.from(heartbeatRuns)
|
|
.where(and(eq(heartbeatRuns.agentId, agentId), inArray(heartbeatRuns.status, ["queued", "running"])))
|
|
.orderBy(desc(heartbeatRuns.createdAt));
|
|
|
|
const sameScopeQueuedRun = activeRuns.find(
|
|
(candidate) => candidate.status === "queued" && isSameTaskScope(runTaskKey(candidate), taskKey),
|
|
);
|
|
const sameScopeRunningRun = activeRuns.find(
|
|
(candidate) => candidate.status === "running" && isSameTaskScope(runTaskKey(candidate), taskKey),
|
|
);
|
|
const shouldQueueFollowupForCommentWake =
|
|
Boolean(wakeCommentId) && Boolean(sameScopeRunningRun) && !sameScopeQueuedRun;
|
|
|
|
const coalescedTargetRun =
|
|
sameScopeQueuedRun ??
|
|
(shouldQueueFollowupForCommentWake ? null : sameScopeRunningRun ?? null);
|
|
|
|
if (coalescedTargetRun) {
|
|
const mergedContextSnapshot = mergeCoalescedContextSnapshot(
|
|
coalescedTargetRun.contextSnapshot,
|
|
contextSnapshot,
|
|
);
|
|
const mergedRun = await db
|
|
.update(heartbeatRuns)
|
|
.set({
|
|
contextSnapshot: mergedContextSnapshot,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(heartbeatRuns.id, coalescedTargetRun.id))
|
|
.returning()
|
|
.then((rows) => rows[0] ?? coalescedTargetRun);
|
|
|
|
await db.insert(agentWakeupRequests).values({
|
|
companyId: agent.companyId,
|
|
agentId,
|
|
source,
|
|
triggerDetail,
|
|
reason,
|
|
payload,
|
|
status: "coalesced",
|
|
coalescedCount: 1,
|
|
requestedByActorType: opts.requestedByActorType ?? null,
|
|
requestedByActorId: opts.requestedByActorId ?? null,
|
|
idempotencyKey: opts.idempotencyKey ?? null,
|
|
runId: mergedRun.id,
|
|
finishedAt: new Date(),
|
|
});
|
|
return mergedRun;
|
|
}
|
|
|
|
const wakeupRequest = await db
|
|
.insert(agentWakeupRequests)
|
|
.values({
|
|
companyId: agent.companyId,
|
|
agentId,
|
|
source,
|
|
triggerDetail,
|
|
reason,
|
|
payload,
|
|
status: "queued",
|
|
requestedByActorType: opts.requestedByActorType ?? null,
|
|
requestedByActorId: opts.requestedByActorId ?? null,
|
|
idempotencyKey: opts.idempotencyKey ?? null,
|
|
})
|
|
.returning()
|
|
.then((rows) => rows[0]);
|
|
|
|
const sessionBefore = await resolveSessionBeforeForWakeup(agent, taskKey);
|
|
|
|
const newRun = await db
|
|
.insert(heartbeatRuns)
|
|
.values({
|
|
companyId: agent.companyId,
|
|
agentId,
|
|
invocationSource: source,
|
|
triggerDetail,
|
|
status: "queued",
|
|
wakeupRequestId: wakeupRequest.id,
|
|
contextSnapshot: enrichedContextSnapshot,
|
|
sessionIdBefore: sessionBefore,
|
|
})
|
|
.returning()
|
|
.then((rows) => rows[0]);
|
|
|
|
await db
|
|
.update(agentWakeupRequests)
|
|
.set({
|
|
runId: newRun.id,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(agentWakeupRequests.id, wakeupRequest.id));
|
|
|
|
publishLiveEvent({
|
|
companyId: newRun.companyId,
|
|
type: "heartbeat.run.queued",
|
|
payload: {
|
|
runId: newRun.id,
|
|
agentId: newRun.agentId,
|
|
invocationSource: newRun.invocationSource,
|
|
triggerDetail: newRun.triggerDetail,
|
|
wakeupRequestId: newRun.wakeupRequestId,
|
|
},
|
|
});
|
|
|
|
await startNextQueuedRunForAgent(agent.id);
|
|
|
|
return newRun;
|
|
}
|
|
|
|
return {
|
|
list: async (companyId: string, agentId?: string, limit?: number) => {
|
|
const query = db
|
|
.select(heartbeatRunListColumns)
|
|
.from(heartbeatRuns)
|
|
.where(
|
|
agentId
|
|
? and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, agentId))
|
|
: eq(heartbeatRuns.companyId, companyId),
|
|
)
|
|
.orderBy(desc(heartbeatRuns.createdAt));
|
|
|
|
const rows = limit ? await query.limit(limit) : await query;
|
|
return rows.map((row) => ({
|
|
...row,
|
|
resultJson: summarizeHeartbeatRunResultJson(row.resultJson),
|
|
}));
|
|
},
|
|
|
|
getRun,
|
|
|
|
getRuntimeState: async (agentId: string) => {
|
|
const state = await getRuntimeState(agentId);
|
|
const agent = await getAgent(agentId);
|
|
if (!agent) return null;
|
|
const ensured = state ?? (await ensureRuntimeState(agent));
|
|
const latestTaskSession = await db
|
|
.select()
|
|
.from(agentTaskSessions)
|
|
.where(and(eq(agentTaskSessions.companyId, agent.companyId), eq(agentTaskSessions.agentId, agent.id)))
|
|
.orderBy(desc(agentTaskSessions.updatedAt))
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
return {
|
|
...ensured,
|
|
sessionDisplayId: latestTaskSession?.sessionDisplayId ?? ensured.sessionId,
|
|
sessionParamsJson: latestTaskSession?.sessionParamsJson ?? null,
|
|
};
|
|
},
|
|
|
|
listTaskSessions: async (agentId: string) => {
|
|
const agent = await getAgent(agentId);
|
|
if (!agent) throw notFound("Agent not found");
|
|
|
|
return db
|
|
.select()
|
|
.from(agentTaskSessions)
|
|
.where(and(eq(agentTaskSessions.companyId, agent.companyId), eq(agentTaskSessions.agentId, agentId)))
|
|
.orderBy(desc(agentTaskSessions.updatedAt), desc(agentTaskSessions.createdAt));
|
|
},
|
|
|
|
resetRuntimeSession: async (agentId: string, opts?: { taskKey?: string | null }) => {
|
|
const agent = await getAgent(agentId);
|
|
if (!agent) throw notFound("Agent not found");
|
|
await ensureRuntimeState(agent);
|
|
const taskKey = readNonEmptyString(opts?.taskKey);
|
|
const clearedTaskSessions = await clearTaskSessions(
|
|
agent.companyId,
|
|
agent.id,
|
|
taskKey ? { taskKey, adapterType: agent.adapterType } : undefined,
|
|
);
|
|
const runtimePatch: Partial<typeof agentRuntimeState.$inferInsert> = {
|
|
sessionId: null,
|
|
lastError: null,
|
|
updatedAt: new Date(),
|
|
};
|
|
if (!taskKey) {
|
|
runtimePatch.stateJson = {};
|
|
}
|
|
|
|
const updated = await db
|
|
.update(agentRuntimeState)
|
|
.set(runtimePatch)
|
|
.where(eq(agentRuntimeState.agentId, agentId))
|
|
.returning()
|
|
.then((rows) => rows[0] ?? null);
|
|
|
|
if (!updated) return null;
|
|
return {
|
|
...updated,
|
|
sessionDisplayId: null,
|
|
sessionParamsJson: null,
|
|
clearedTaskSessions,
|
|
};
|
|
},
|
|
|
|
listEvents: (runId: string, afterSeq = 0, limit = 200) =>
|
|
db
|
|
.select()
|
|
.from(heartbeatRunEvents)
|
|
.where(and(eq(heartbeatRunEvents.runId, runId), gt(heartbeatRunEvents.seq, afterSeq)))
|
|
.orderBy(asc(heartbeatRunEvents.seq))
|
|
.limit(Math.max(1, Math.min(limit, 1000))),
|
|
|
|
readLog: async (runId: string, opts?: { offset?: number; limitBytes?: number }) => {
|
|
const run = await getRun(runId);
|
|
if (!run) throw notFound("Heartbeat run not found");
|
|
if (!run.logStore || !run.logRef) throw notFound("Run log not found");
|
|
|
|
const result = await runLogStore.read(
|
|
{
|
|
store: run.logStore as "local_file",
|
|
logRef: run.logRef,
|
|
},
|
|
opts,
|
|
);
|
|
|
|
return {
|
|
runId,
|
|
store: run.logStore,
|
|
logRef: run.logRef,
|
|
...result,
|
|
content: redactCurrentUserText(result.content),
|
|
};
|
|
},
|
|
|
|
invoke: async (
|
|
agentId: string,
|
|
source: "timer" | "assignment" | "on_demand" | "automation" = "on_demand",
|
|
contextSnapshot: Record<string, unknown> = {},
|
|
triggerDetail: "manual" | "ping" | "callback" | "system" = "manual",
|
|
actor?: { actorType?: "user" | "agent" | "system"; actorId?: string | null },
|
|
) =>
|
|
enqueueWakeup(agentId, {
|
|
source,
|
|
triggerDetail,
|
|
contextSnapshot,
|
|
requestedByActorType: actor?.actorType,
|
|
requestedByActorId: actor?.actorId ?? null,
|
|
}),
|
|
|
|
wakeup: enqueueWakeup,
|
|
|
|
reapOrphanedRuns,
|
|
|
|
resumeQueuedRuns,
|
|
|
|
tickTimers: async (now = new Date()) => {
|
|
const allAgents = await db.select().from(agents);
|
|
let checked = 0;
|
|
let enqueued = 0;
|
|
let skipped = 0;
|
|
|
|
for (const agent of allAgents) {
|
|
if (agent.status === "paused" || agent.status === "terminated" || agent.status === "pending_approval") continue;
|
|
const policy = parseHeartbeatPolicy(agent);
|
|
if (!policy.enabled || policy.intervalSec <= 0) continue;
|
|
|
|
checked += 1;
|
|
const baseline = new Date(agent.lastHeartbeatAt ?? agent.createdAt).getTime();
|
|
const elapsedMs = now.getTime() - baseline;
|
|
if (elapsedMs < policy.intervalSec * 1000) continue;
|
|
|
|
const run = await enqueueWakeup(agent.id, {
|
|
source: "timer",
|
|
triggerDetail: "system",
|
|
reason: "heartbeat_timer",
|
|
requestedByActorType: "system",
|
|
requestedByActorId: "heartbeat_scheduler",
|
|
contextSnapshot: {
|
|
source: "scheduler",
|
|
reason: "interval_elapsed",
|
|
now: now.toISOString(),
|
|
},
|
|
});
|
|
if (run) enqueued += 1;
|
|
else skipped += 1;
|
|
}
|
|
|
|
return { checked, enqueued, skipped };
|
|
},
|
|
|
|
cancelRun: async (runId: string) => {
|
|
const run = await getRun(runId);
|
|
if (!run) throw notFound("Heartbeat run not found");
|
|
if (run.status !== "running" && run.status !== "queued") return run;
|
|
|
|
const running = runningProcesses.get(run.id);
|
|
if (running) {
|
|
running.child.kill("SIGTERM");
|
|
const graceMs = Math.max(1, running.graceSec) * 1000;
|
|
setTimeout(() => {
|
|
if (!running.child.killed) {
|
|
running.child.kill("SIGKILL");
|
|
}
|
|
}, graceMs);
|
|
}
|
|
|
|
const cancelled = await setRunStatus(run.id, "cancelled", {
|
|
finishedAt: new Date(),
|
|
error: "Cancelled by control plane",
|
|
errorCode: "cancelled",
|
|
});
|
|
|
|
await setWakeupStatus(run.wakeupRequestId, "cancelled", {
|
|
finishedAt: new Date(),
|
|
error: "Cancelled by control plane",
|
|
});
|
|
|
|
if (cancelled) {
|
|
await appendRunEvent(cancelled, 1, {
|
|
eventType: "lifecycle",
|
|
stream: "system",
|
|
level: "warn",
|
|
message: "run cancelled",
|
|
});
|
|
await releaseIssueExecutionAndPromote(cancelled);
|
|
}
|
|
|
|
runningProcesses.delete(run.id);
|
|
await finalizeAgentStatus(run.agentId, "cancelled");
|
|
await startNextQueuedRunForAgent(run.agentId);
|
|
return cancelled;
|
|
},
|
|
|
|
cancelActiveForAgent: async (agentId: string) => {
|
|
const runs = await db
|
|
.select()
|
|
.from(heartbeatRuns)
|
|
.where(and(eq(heartbeatRuns.agentId, agentId), inArray(heartbeatRuns.status, ["queued", "running"])));
|
|
|
|
for (const run of runs) {
|
|
await setRunStatus(run.id, "cancelled", {
|
|
finishedAt: new Date(),
|
|
error: "Cancelled due to agent pause",
|
|
errorCode: "cancelled",
|
|
});
|
|
|
|
await setWakeupStatus(run.wakeupRequestId, "cancelled", {
|
|
finishedAt: new Date(),
|
|
error: "Cancelled due to agent pause",
|
|
});
|
|
|
|
const running = runningProcesses.get(run.id);
|
|
if (running) {
|
|
running.child.kill("SIGTERM");
|
|
runningProcesses.delete(run.id);
|
|
}
|
|
await releaseIssueExecutionAndPromote(run);
|
|
}
|
|
|
|
return runs.length;
|
|
},
|
|
|
|
getActiveRunForAgent: async (agentId: string) => {
|
|
const [run] = await db
|
|
.select()
|
|
.from(heartbeatRuns)
|
|
.where(
|
|
and(
|
|
eq(heartbeatRuns.agentId, agentId),
|
|
eq(heartbeatRuns.status, "running"),
|
|
),
|
|
)
|
|
.orderBy(desc(heartbeatRuns.startedAt))
|
|
.limit(1);
|
|
return run ?? null;
|
|
},
|
|
};
|
|
}
|