mirror of
https://github.com/alkimake/paperclip.git
synced 2026-06-15 10:30:37 +09:00
[codex] Harden execution reliability and heartbeat tooling (#3679)
## Thinking Path > - Paperclip orchestrates AI agents for zero-human companies > - Reliable execution depends on heartbeat routing, issue lifecycle semantics, telemetry, and a fast enough local verification loop to keep regressions visible > - The remaining commits on this branch were mostly server/runtime correctness fixes plus test and documentation follow-ups in that area > - Those changes are logically separate from the UI-focused issue-detail and workspace/navigation branches even when they touch overlapping issue APIs > - This pull request groups the execution reliability, heartbeat, telemetry, and tooling changes into one standalone branch > - The benefit is a focused review of the control-plane correctness work, including the follow-up fix that restored the implicit comment-reopen helpers after branch splitting ## What Changed - Hardened issue/heartbeat execution behavior, including self-review stage skipping, deferred mention wakes during active execution, stranded execution recovery, active-run scoping, assignee resolution, and blocked-to-todo wake resumption - Reduced noisy polling/logging overhead by trimming issue run payloads, compacting persisted run logs, silencing high-volume request logs, and capping heartbeat-run queries in dashboard/inbox surfaces - Expanded telemetry and status semantics with adapter/model fields on task completion plus clearer status guidance in docs/onboarding material - Updated test infrastructure and verification defaults with faster route-test module isolation, cheaper default `pnpm test`, e2e isolation from local state, and repo verification follow-ups - Included docs/release housekeeping from the branch and added a small follow-up commit restoring the implicit comment-reopen helpers that were dropped during branch reconstruction ## Verification - `pnpm vitest run server/src/__tests__/issue-comment-reopen-routes.test.ts server/src/__tests__/issue-telemetry-routes.test.ts` - `pnpm vitest run server/src/__tests__/http-log-policy.test.ts server/src/__tests__/heartbeat-run-log.test.ts server/src/__tests__/health.test.ts` - `server/src/__tests__/activity-service.test.ts`, `server/src/__tests__/heartbeat-comment-wake-batching.test.ts`, and `server/src/__tests__/heartbeat-process-recovery.test.ts` were attempted on this host but the embedded Postgres harness reported init-script/data-dir problems and skipped or failed to start, so they are noted as environment-limited ## Risks - Medium: this branch changes core issue/heartbeat routing and reopen/wakeup behavior, so regressions would affect agent execution flow rather than isolated UI polish - Because it also updates verification infrastructure, reviewers should pay attention to whether the new tests are asserting the right failure modes and not just reshaping harness behavior ## Model Used - OpenAI Codex coding agent (GPT-5-class runtime in Codex CLI; exact deployed model ID is not exposed in this environment), reasoning enabled, tool use and local code execution enabled ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [ ] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [ ] If this change affects the UI, I have included before/after screenshots - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
GitHub
parent
e89076148a
commit
7f893ac4ec
106 changed files with 4682 additions and 713 deletions
|
|
@ -11,6 +11,74 @@ export interface ActivityFilters {
|
|||
|
||||
export function activityService(db: Db) {
|
||||
const issueIdAsText = sql<string>`${issues.id}::text`;
|
||||
const summarizedUsageJson = sql<Record<string, unknown> | null>`
|
||||
case
|
||||
when ${heartbeatRuns.usageJson} is null then null
|
||||
else jsonb_strip_nulls(jsonb_build_object(
|
||||
'inputTokens', coalesce(${heartbeatRuns.usageJson} -> 'inputTokens', ${heartbeatRuns.usageJson} -> 'input_tokens'),
|
||||
'input_tokens', coalesce(${heartbeatRuns.usageJson} -> 'input_tokens', ${heartbeatRuns.usageJson} -> 'inputTokens'),
|
||||
'outputTokens', coalesce(${heartbeatRuns.usageJson} -> 'outputTokens', ${heartbeatRuns.usageJson} -> 'output_tokens'),
|
||||
'output_tokens', coalesce(${heartbeatRuns.usageJson} -> 'output_tokens', ${heartbeatRuns.usageJson} -> 'outputTokens'),
|
||||
'cachedInputTokens', coalesce(
|
||||
${heartbeatRuns.usageJson} -> 'cachedInputTokens',
|
||||
${heartbeatRuns.usageJson} -> 'cached_input_tokens',
|
||||
${heartbeatRuns.usageJson} -> 'cache_read_input_tokens'
|
||||
),
|
||||
'cached_input_tokens', coalesce(
|
||||
${heartbeatRuns.usageJson} -> 'cached_input_tokens',
|
||||
${heartbeatRuns.usageJson} -> 'cachedInputTokens',
|
||||
${heartbeatRuns.usageJson} -> 'cache_read_input_tokens'
|
||||
),
|
||||
'cache_read_input_tokens', coalesce(
|
||||
${heartbeatRuns.usageJson} -> 'cache_read_input_tokens',
|
||||
${heartbeatRuns.usageJson} -> 'cached_input_tokens',
|
||||
${heartbeatRuns.usageJson} -> 'cachedInputTokens'
|
||||
),
|
||||
'billingType', coalesce(${heartbeatRuns.usageJson} -> 'billingType', ${heartbeatRuns.usageJson} -> 'billing_type'),
|
||||
'billing_type', coalesce(${heartbeatRuns.usageJson} -> 'billing_type', ${heartbeatRuns.usageJson} -> 'billingType'),
|
||||
'costUsd', coalesce(
|
||||
${heartbeatRuns.usageJson} -> 'costUsd',
|
||||
${heartbeatRuns.usageJson} -> 'cost_usd',
|
||||
${heartbeatRuns.usageJson} -> 'total_cost_usd'
|
||||
),
|
||||
'cost_usd', coalesce(
|
||||
${heartbeatRuns.usageJson} -> 'cost_usd',
|
||||
${heartbeatRuns.usageJson} -> 'costUsd',
|
||||
${heartbeatRuns.usageJson} -> 'total_cost_usd'
|
||||
),
|
||||
'total_cost_usd', coalesce(
|
||||
${heartbeatRuns.usageJson} -> 'total_cost_usd',
|
||||
${heartbeatRuns.usageJson} -> 'cost_usd',
|
||||
${heartbeatRuns.usageJson} -> 'costUsd'
|
||||
)
|
||||
))
|
||||
end
|
||||
`.as("usageJson");
|
||||
const summarizedResultJson = sql<Record<string, unknown> | null>`
|
||||
case
|
||||
when ${heartbeatRuns.resultJson} is null then null
|
||||
else jsonb_strip_nulls(jsonb_build_object(
|
||||
'billingType', coalesce(${heartbeatRuns.resultJson} -> 'billingType', ${heartbeatRuns.resultJson} -> 'billing_type'),
|
||||
'billing_type', coalesce(${heartbeatRuns.resultJson} -> 'billing_type', ${heartbeatRuns.resultJson} -> 'billingType'),
|
||||
'costUsd', coalesce(
|
||||
${heartbeatRuns.resultJson} -> 'costUsd',
|
||||
${heartbeatRuns.resultJson} -> 'cost_usd',
|
||||
${heartbeatRuns.resultJson} -> 'total_cost_usd'
|
||||
),
|
||||
'cost_usd', coalesce(
|
||||
${heartbeatRuns.resultJson} -> 'cost_usd',
|
||||
${heartbeatRuns.resultJson} -> 'costUsd',
|
||||
${heartbeatRuns.resultJson} -> 'total_cost_usd'
|
||||
),
|
||||
'total_cost_usd', coalesce(
|
||||
${heartbeatRuns.resultJson} -> 'total_cost_usd',
|
||||
${heartbeatRuns.resultJson} -> 'cost_usd',
|
||||
${heartbeatRuns.resultJson} -> 'costUsd'
|
||||
)
|
||||
))
|
||||
end
|
||||
`.as("resultJson");
|
||||
|
||||
return {
|
||||
list: (filters: ActivityFilters) => {
|
||||
const conditions = [eq(activityLog.companyId, filters.companyId)];
|
||||
|
|
@ -71,8 +139,8 @@ export function activityService(db: Db) {
|
|||
finishedAt: heartbeatRuns.finishedAt,
|
||||
createdAt: heartbeatRuns.createdAt,
|
||||
invocationSource: heartbeatRuns.invocationSource,
|
||||
usageJson: heartbeatRuns.usageJson,
|
||||
resultJson: heartbeatRuns.resultJson,
|
||||
usageJson: summarizedUsageJson,
|
||||
resultJson: summarizedResultJson,
|
||||
logBytes: heartbeatRuns.logBytes,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
|
|
|
|||
|
|
@ -4388,7 +4388,7 @@ export function companyPortabilityService(db: Db, storage?: StorageService) {
|
|||
billingCode: manifestIssue.billingCode,
|
||||
assigneeAdapterOverrides: manifestIssue.assigneeAdapterOverrides,
|
||||
executionWorkspaceSettings: manifestIssue.executionWorkspaceSettings,
|
||||
labelIds: [],
|
||||
labelIds: manifestIssue.labelIds ?? [],
|
||||
});
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ import {
|
|||
agentRuntimeState,
|
||||
agentTaskSessions,
|
||||
agentWakeupRequests,
|
||||
companySkills as companySkillsTable,
|
||||
heartbeatRunEvents,
|
||||
heartbeatRuns,
|
||||
issueComments,
|
||||
|
|
@ -68,8 +69,14 @@ import {
|
|||
resolveSessionCompactionPolicy,
|
||||
type SessionCompactionPolicy,
|
||||
} from "@paperclipai/adapter-utils";
|
||||
import {
|
||||
readPaperclipSkillSyncPreference,
|
||||
writePaperclipSkillSyncPreference,
|
||||
} from "@paperclipai/adapter-utils/server-utils";
|
||||
import { extractSkillMentionIds } from "@paperclipai/shared";
|
||||
|
||||
const MAX_LIVE_LOG_CHUNK_BYTES = 8 * 1024;
|
||||
const MAX_PERSISTED_LOG_CHUNK_CHARS = 64 * 1024;
|
||||
const HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT = 1;
|
||||
const HEARTBEAT_MAX_CONCURRENT_RUNS_MAX = 10;
|
||||
const DEFERRED_WAKE_CONTEXT_KEY = "_paperclipWakeContext";
|
||||
|
|
@ -84,6 +91,7 @@ const MAX_INLINE_WAKE_COMMENTS = 8;
|
|||
const MAX_INLINE_WAKE_COMMENT_BODY_CHARS = 4_000;
|
||||
const MAX_INLINE_WAKE_COMMENT_BODY_TOTAL_CHARS = 12_000;
|
||||
const execFile = promisify(execFileCallback);
|
||||
const ACTIVE_HEARTBEAT_RUN_STATUSES = ["queued", "running"] as const;
|
||||
const SESSIONED_LOCAL_ADAPTERS = new Set([
|
||||
"claude_local",
|
||||
"codex_local",
|
||||
|
|
@ -92,6 +100,7 @@ const SESSIONED_LOCAL_ADAPTERS = new Set([
|
|||
"opencode_local",
|
||||
"pi_local",
|
||||
]);
|
||||
const INLINE_BASE64_IMAGE_DATA_RE = /("type":"image","source":\{"type":"base64","data":")([A-Za-z0-9+/=]{1024,})(")/g;
|
||||
|
||||
type RuntimeConfigSecretResolver = Pick<
|
||||
ReturnType<typeof secretService>,
|
||||
|
|
@ -123,6 +132,90 @@ export async function resolveExecutionRunAdapterConfig(input: {
|
|||
return { resolvedConfig, secretKeys };
|
||||
}
|
||||
|
||||
export function extractMentionedSkillIdsFromSources(
|
||||
sources: Array<string | null | undefined>,
|
||||
): string[] {
|
||||
const mentionedIds = new Set<string>();
|
||||
for (const source of sources) {
|
||||
if (typeof source !== "string" || source.length === 0) continue;
|
||||
for (const skillId of extractSkillMentionIds(source)) {
|
||||
mentionedIds.add(skillId);
|
||||
}
|
||||
}
|
||||
return [...mentionedIds];
|
||||
}
|
||||
|
||||
export function applyRunScopedMentionedSkillKeys(
|
||||
config: Record<string, unknown>,
|
||||
skillKeys: string[],
|
||||
): Record<string, unknown> {
|
||||
const normalizedSkillKeys = Array.from(
|
||||
new Set(
|
||||
skillKeys
|
||||
.map((value) => value.trim())
|
||||
.filter(Boolean),
|
||||
),
|
||||
);
|
||||
if (normalizedSkillKeys.length === 0) return config;
|
||||
|
||||
const existingPreference = readPaperclipSkillSyncPreference(config);
|
||||
return writePaperclipSkillSyncPreference(config, [
|
||||
...existingPreference.desiredSkills,
|
||||
...normalizedSkillKeys,
|
||||
]);
|
||||
}
|
||||
|
||||
async function resolveRunScopedMentionedSkillKeys(input: {
|
||||
db: Db;
|
||||
companyId: string;
|
||||
issueId: string | null;
|
||||
}): Promise<string[]> {
|
||||
if (!input.issueId) return [];
|
||||
|
||||
const issue = await input.db
|
||||
.select({
|
||||
title: issues.title,
|
||||
description: issues.description,
|
||||
})
|
||||
.from(issues)
|
||||
.where(and(eq(issues.id, input.issueId), eq(issues.companyId, input.companyId)))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!issue) return [];
|
||||
|
||||
const comments = await input.db
|
||||
.select({ body: issueComments.body })
|
||||
.from(issueComments)
|
||||
.where(
|
||||
and(
|
||||
eq(issueComments.issueId, input.issueId),
|
||||
eq(issueComments.companyId, input.companyId),
|
||||
),
|
||||
);
|
||||
const mentionedSkillIds = extractMentionedSkillIdsFromSources([
|
||||
issue.title,
|
||||
issue.description ?? "",
|
||||
...comments.map((comment) => comment.body),
|
||||
]);
|
||||
if (mentionedSkillIds.length === 0) return [];
|
||||
|
||||
const skillRows = await input.db
|
||||
.select({
|
||||
id: companySkillsTable.id,
|
||||
key: companySkillsTable.key,
|
||||
})
|
||||
.from(companySkillsTable)
|
||||
.where(
|
||||
and(
|
||||
eq(companySkillsTable.companyId, input.companyId),
|
||||
inArray(companySkillsTable.id, mentionedSkillIds),
|
||||
),
|
||||
);
|
||||
const skillKeyById = new Map(skillRows.map((row) => [row.id, row.key]));
|
||||
return mentionedSkillIds
|
||||
.map((skillId) => skillKeyById.get(skillId) ?? null)
|
||||
.filter((skillKey): skillKey is string => Boolean(skillKey));
|
||||
}
|
||||
|
||||
export function applyPersistedExecutionWorkspaceConfig(input: {
|
||||
config: Record<string, unknown>;
|
||||
workspaceConfig: ExecutionWorkspaceConfig | null;
|
||||
|
|
@ -323,6 +416,23 @@ function appendExcerpt(prev: string, chunk: string) {
|
|||
return appendWithCap(prev, chunk, MAX_EXCERPT_BYTES);
|
||||
}
|
||||
|
||||
function redactInlineBase64ImageData(chunk: string) {
|
||||
return chunk.replace(INLINE_BASE64_IMAGE_DATA_RE, (_match, prefix: string, data: string, suffix: string) =>
|
||||
`${prefix}[omitted base64 image data: ${data.length} chars]${suffix}`,
|
||||
);
|
||||
}
|
||||
|
||||
export function compactRunLogChunk(chunk: string, maxChars = MAX_PERSISTED_LOG_CHUNK_CHARS) {
|
||||
const normalized = redactInlineBase64ImageData(chunk);
|
||||
if (normalized.length <= maxChars) return normalized;
|
||||
|
||||
const headChars = Math.max(0, Math.floor(maxChars * 0.6));
|
||||
const tailChars = Math.max(0, Math.floor(maxChars * 0.25));
|
||||
const omittedChars = Math.max(0, normalized.length - headChars - tailChars);
|
||||
const marker = `\n[paperclip truncated run log chunk: omitted ${omittedChars} chars]\n`;
|
||||
return `${normalized.slice(0, headChars)}${marker}${normalized.slice(normalized.length - tailChars)}`;
|
||||
}
|
||||
|
||||
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;
|
||||
|
|
@ -2427,7 +2537,7 @@ export function heartbeatService(db: Db) {
|
|||
|
||||
if (isFirstHeartbeat && updated) {
|
||||
const tc = getTelemetryClient();
|
||||
if (tc) trackAgentFirstHeartbeat(tc, { agentRole: updated.role });
|
||||
if (tc) trackAgentFirstHeartbeat(tc, { agentRole: updated.role, agentId: updated.id });
|
||||
}
|
||||
|
||||
if (updated) {
|
||||
|
|
@ -2569,6 +2679,256 @@ export function heartbeatService(db: Db) {
|
|||
}
|
||||
}
|
||||
|
||||
async function getLatestIssueRun(companyId: string, issueId: string) {
|
||||
return db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(
|
||||
and(
|
||||
eq(heartbeatRuns.companyId, companyId),
|
||||
sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`,
|
||||
),
|
||||
)
|
||||
.orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
}
|
||||
|
||||
async function hasActiveExecutionPath(companyId: string, issueId: string) {
|
||||
const [run, deferredWake] = await Promise.all([
|
||||
db
|
||||
.select({ id: heartbeatRuns.id })
|
||||
.from(heartbeatRuns)
|
||||
.where(
|
||||
and(
|
||||
eq(heartbeatRuns.companyId, companyId),
|
||||
inArray(heartbeatRuns.status, [...ACTIVE_HEARTBEAT_RUN_STATUSES]),
|
||||
sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`,
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null),
|
||||
db
|
||||
.select({ id: agentWakeupRequests.id })
|
||||
.from(agentWakeupRequests)
|
||||
.where(
|
||||
and(
|
||||
eq(agentWakeupRequests.companyId, companyId),
|
||||
eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`,
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null),
|
||||
]);
|
||||
|
||||
return Boolean(run || deferredWake);
|
||||
}
|
||||
|
||||
async function enqueueStrandedIssueRecovery(input: {
|
||||
issueId: string;
|
||||
agentId: string;
|
||||
reason: "issue_assignment_recovery" | "issue_continuation_needed";
|
||||
retryReason: "assignment_recovery" | "issue_continuation_needed";
|
||||
source: string;
|
||||
retryOfRunId?: string | null;
|
||||
}) {
|
||||
const queued = await enqueueWakeup(input.agentId, {
|
||||
source: "automation",
|
||||
triggerDetail: "system",
|
||||
reason: input.reason,
|
||||
payload: {
|
||||
issueId: input.issueId,
|
||||
...(input.retryOfRunId ? { retryOfRunId: input.retryOfRunId } : {}),
|
||||
},
|
||||
requestedByActorType: "system",
|
||||
requestedByActorId: null,
|
||||
contextSnapshot: {
|
||||
issueId: input.issueId,
|
||||
taskId: input.issueId,
|
||||
wakeReason: input.reason,
|
||||
retryReason: input.retryReason,
|
||||
source: input.source,
|
||||
...(input.retryOfRunId ? { retryOfRunId: input.retryOfRunId } : {}),
|
||||
},
|
||||
});
|
||||
|
||||
if (queued && input.retryOfRunId) {
|
||||
return db
|
||||
.update(heartbeatRuns)
|
||||
.set({
|
||||
retryOfRunId: input.retryOfRunId,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(heartbeatRuns.id, queued.id))
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? queued);
|
||||
}
|
||||
|
||||
return queued;
|
||||
}
|
||||
|
||||
async function escalateStrandedAssignedIssue(input: {
|
||||
issue: typeof issues.$inferSelect;
|
||||
previousStatus: "todo" | "in_progress";
|
||||
latestRun: typeof heartbeatRuns.$inferSelect | null;
|
||||
comment: string;
|
||||
}) {
|
||||
const updated = await issuesSvc.update(input.issue.id, {
|
||||
status: "blocked",
|
||||
});
|
||||
if (!updated) return null;
|
||||
|
||||
await issuesSvc.addComment(input.issue.id, input.comment, {});
|
||||
|
||||
await logActivity(db, {
|
||||
companyId: input.issue.companyId,
|
||||
actorType: "system",
|
||||
actorId: "system",
|
||||
agentId: null,
|
||||
runId: null,
|
||||
action: "issue.updated",
|
||||
entityType: "issue",
|
||||
entityId: input.issue.id,
|
||||
details: {
|
||||
identifier: input.issue.identifier,
|
||||
status: "blocked",
|
||||
previousStatus: input.previousStatus,
|
||||
source: "heartbeat.reconcile_stranded_assigned_issue",
|
||||
latestRunId: input.latestRun?.id ?? null,
|
||||
latestRunStatus: input.latestRun?.status ?? null,
|
||||
latestRunErrorCode: input.latestRun?.errorCode ?? null,
|
||||
},
|
||||
});
|
||||
|
||||
return updated;
|
||||
}
|
||||
|
||||
async function reconcileStrandedAssignedIssues() {
|
||||
const candidates = await db
|
||||
.select()
|
||||
.from(issues)
|
||||
.where(
|
||||
and(
|
||||
isNull(issues.assigneeUserId),
|
||||
inArray(issues.status, ["todo", "in_progress"]),
|
||||
sql`${issues.assigneeAgentId} is not null`,
|
||||
),
|
||||
);
|
||||
|
||||
const result = {
|
||||
dispatchRequeued: 0,
|
||||
continuationRequeued: 0,
|
||||
escalated: 0,
|
||||
skipped: 0,
|
||||
issueIds: [] as string[],
|
||||
};
|
||||
|
||||
for (const issue of candidates) {
|
||||
const agentId = issue.assigneeAgentId;
|
||||
if (!agentId) {
|
||||
result.skipped += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
const agent = await getAgent(agentId);
|
||||
if (!agent || agent.companyId !== issue.companyId) {
|
||||
result.skipped += 1;
|
||||
continue;
|
||||
}
|
||||
if (agent.status === "paused" || agent.status === "terminated" || agent.status === "pending_approval") {
|
||||
result.skipped += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (await hasActiveExecutionPath(issue.companyId, issue.id)) {
|
||||
result.skipped += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
const latestRun = await getLatestIssueRun(issue.companyId, issue.id);
|
||||
const latestContext = parseObject(latestRun?.contextSnapshot);
|
||||
const latestRetryReason = readNonEmptyString(latestContext.retryReason);
|
||||
|
||||
if (issue.status === "todo") {
|
||||
if (!latestRun || latestRun.status === "succeeded") {
|
||||
result.skipped += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (latestRetryReason === "assignment_recovery") {
|
||||
const updated = await escalateStrandedAssignedIssue({
|
||||
issue,
|
||||
previousStatus: "todo",
|
||||
latestRun,
|
||||
comment:
|
||||
"Paperclip automatically retried dispatch for this assigned `todo` issue after a lost wake/run, " +
|
||||
"but it still has no live execution path. Moving it to `blocked` so it is visible for intervention.",
|
||||
});
|
||||
if (updated) {
|
||||
result.escalated += 1;
|
||||
result.issueIds.push(issue.id);
|
||||
} else {
|
||||
result.skipped += 1;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
const queued = await enqueueStrandedIssueRecovery({
|
||||
issueId: issue.id,
|
||||
agentId,
|
||||
reason: "issue_assignment_recovery",
|
||||
retryReason: "assignment_recovery",
|
||||
source: "issue.assignment_recovery",
|
||||
retryOfRunId: latestRun.id,
|
||||
});
|
||||
if (queued) {
|
||||
result.dispatchRequeued += 1;
|
||||
result.issueIds.push(issue.id);
|
||||
} else {
|
||||
result.skipped += 1;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (latestRetryReason === "issue_continuation_needed") {
|
||||
const updated = await escalateStrandedAssignedIssue({
|
||||
issue,
|
||||
previousStatus: "in_progress",
|
||||
latestRun,
|
||||
comment:
|
||||
"Paperclip automatically retried continuation for this assigned `in_progress` issue after its live " +
|
||||
"execution disappeared, but it still has no live execution path. Moving it to `blocked` so it is " +
|
||||
"visible for intervention.",
|
||||
});
|
||||
if (updated) {
|
||||
result.escalated += 1;
|
||||
result.issueIds.push(issue.id);
|
||||
} else {
|
||||
result.skipped += 1;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
const queued = await enqueueStrandedIssueRecovery({
|
||||
issueId: issue.id,
|
||||
agentId,
|
||||
reason: "issue_continuation_needed",
|
||||
retryReason: "issue_continuation_needed",
|
||||
source: "issue.continuation_recovery",
|
||||
retryOfRunId: latestRun?.id ?? issue.checkoutRunId ?? null,
|
||||
});
|
||||
if (queued) {
|
||||
result.continuationRequeued += 1;
|
||||
result.issueIds.push(issue.id);
|
||||
} else {
|
||||
result.skipped += 1;
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
async function updateRuntimeState(
|
||||
agent: typeof agents.$inferSelect,
|
||||
run: typeof heartbeatRuns.$inferSelect,
|
||||
|
|
@ -2846,9 +3206,18 @@ export function heartbeatService(db: Db) {
|
|||
projectEnv: projectContext?.env ?? null,
|
||||
secretsSvc,
|
||||
});
|
||||
const runScopedMentionedSkillKeys = await resolveRunScopedMentionedSkillKeys({
|
||||
db,
|
||||
companyId: agent.companyId,
|
||||
issueId,
|
||||
});
|
||||
const effectiveResolvedConfig = applyRunScopedMentionedSkillKeys(
|
||||
resolvedConfig,
|
||||
runScopedMentionedSkillKeys,
|
||||
);
|
||||
const runtimeSkillEntries = await companySkills.listRuntimeSkillEntries(agent.companyId);
|
||||
const runtimeConfig = {
|
||||
...resolvedConfig,
|
||||
...effectiveResolvedConfig,
|
||||
paperclipRuntimeSkills: runtimeSkillEntries,
|
||||
};
|
||||
const workspaceOperationRecorder = workspaceOperationsSvc.createRecorder({
|
||||
|
|
@ -3183,7 +3552,9 @@ export function heartbeatService(db: Db) {
|
|||
|
||||
const currentUserRedactionOptions = await getCurrentUserRedactionOptions();
|
||||
const onLog = async (stream: "stdout" | "stderr", chunk: string) => {
|
||||
const sanitizedChunk = redactCurrentUserText(chunk, currentUserRedactionOptions);
|
||||
const sanitizedChunk = compactRunLogChunk(
|
||||
redactCurrentUserText(chunk, currentUserRedactionOptions),
|
||||
);
|
||||
if (stream === "stdout") stdoutExcerpt = appendExcerpt(stdoutExcerpt, sanitizedChunk);
|
||||
if (stream === "stderr") stderrExcerpt = appendExcerpt(stderrExcerpt, sanitizedChunk);
|
||||
const ts = new Date().toISOString();
|
||||
|
|
@ -3214,6 +3585,12 @@ export function heartbeatService(db: Db) {
|
|||
},
|
||||
});
|
||||
};
|
||||
if (runScopedMentionedSkillKeys.length > 0) {
|
||||
await onLog(
|
||||
"stdout",
|
||||
`[paperclip] Enabled run-scoped skills from issue mentions: ${runScopedMentionedSkillKeys.join(", ")}\n`,
|
||||
);
|
||||
}
|
||||
for (const warning of runtimeWorkspaceWarnings) {
|
||||
const logEntry = formatRuntimeWorkspaceWarningLog(warning);
|
||||
await onLog(logEntry.stream, logEntry.chunk);
|
||||
|
|
@ -3234,7 +3611,7 @@ export function heartbeatService(db: Db) {
|
|||
issue: issueRef,
|
||||
workspace: executionWorkspace,
|
||||
executionWorkspaceId: persistedExecutionWorkspace?.id ?? issueRef?.executionWorkspaceId ?? null,
|
||||
config: resolvedConfig,
|
||||
config: effectiveResolvedConfig,
|
||||
adapterEnv,
|
||||
onLog,
|
||||
});
|
||||
|
|
@ -3960,11 +4337,10 @@ export function heartbeatService(db: Db) {
|
|||
return null;
|
||||
}
|
||||
|
||||
const bypassIssueExecutionLock =
|
||||
reason === "issue_comment_mentioned" ||
|
||||
readNonEmptyString(enrichedContextSnapshot.wakeReason) === "issue_comment_mentioned";
|
||||
|
||||
if (issueId && !bypassIssueExecutionLock) {
|
||||
if (issueId) {
|
||||
// Mention-triggered wakes can request input from another agent, but they must
|
||||
// still respect the issue execution lock so a second agent cannot start on the
|
||||
// same issue workspace while the assignee already has a live run.
|
||||
const agentNameKey = normalizeAgentNameKey(agent.name);
|
||||
|
||||
const outcome = await db.transaction(async (tx) => {
|
||||
|
|
@ -4700,6 +5076,8 @@ export function heartbeatService(db: Db) {
|
|||
|
||||
resumeQueuedRuns,
|
||||
|
||||
reconcileStrandedAssignedIssues,
|
||||
|
||||
tickTimers: async (now = new Date()) => {
|
||||
const allAgents = await db.select().from(agents);
|
||||
let checked = 0;
|
||||
|
|
|
|||
|
|
@ -174,6 +174,42 @@ function buildCompletedState(previous: IssueExecutionState | null, currentStage:
|
|||
};
|
||||
}
|
||||
|
||||
function buildStateWithCompletedStages(input: {
|
||||
previous: IssueExecutionState | null;
|
||||
completedStageIds: string[];
|
||||
returnAssignee: IssueExecutionStagePrincipal | null;
|
||||
}): IssueExecutionState {
|
||||
return {
|
||||
status: input.previous?.status ?? PENDING_STATUS,
|
||||
currentStageId: input.previous?.currentStageId ?? null,
|
||||
currentStageIndex: input.previous?.currentStageIndex ?? null,
|
||||
currentStageType: input.previous?.currentStageType ?? null,
|
||||
currentParticipant: input.previous?.currentParticipant ?? null,
|
||||
returnAssignee: input.previous?.returnAssignee ?? input.returnAssignee,
|
||||
completedStageIds: input.completedStageIds,
|
||||
lastDecisionId: input.previous?.lastDecisionId ?? null,
|
||||
lastDecisionOutcome: input.previous?.lastDecisionOutcome ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
function buildSkippedStageCompletedState(input: {
|
||||
previous: IssueExecutionState | null;
|
||||
completedStageIds: string[];
|
||||
returnAssignee: IssueExecutionStagePrincipal | null;
|
||||
}): IssueExecutionState {
|
||||
return {
|
||||
status: COMPLETED_STATUS,
|
||||
currentStageId: null,
|
||||
currentStageIndex: null,
|
||||
currentStageType: null,
|
||||
currentParticipant: null,
|
||||
returnAssignee: input.previous?.returnAssignee ?? input.returnAssignee,
|
||||
completedStageIds: input.completedStageIds,
|
||||
lastDecisionId: input.previous?.lastDecisionId ?? null,
|
||||
lastDecisionOutcome: input.previous?.lastDecisionOutcome ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
function buildPendingState(input: {
|
||||
previous: IssueExecutionState | null;
|
||||
stage: IssueExecutionStage;
|
||||
|
|
@ -236,6 +272,18 @@ function clearExecutionStatePatch(input: {
|
|||
}
|
||||
}
|
||||
|
||||
function canAutoSkipPendingStage(input: {
|
||||
stage: IssueExecutionStage;
|
||||
returnAssignee: IssueExecutionStagePrincipal | null;
|
||||
requestedStatus?: string;
|
||||
}) {
|
||||
if (input.requestedStatus !== "done" || input.stage.type !== "review" || !input.returnAssignee) {
|
||||
return false;
|
||||
}
|
||||
return input.stage.participants.length > 0 &&
|
||||
input.stage.participants.every((participant) => principalsEqual(participant, input.returnAssignee));
|
||||
}
|
||||
|
||||
export function applyIssueExecutionPolicyTransition(input: TransitionInput): TransitionResult {
|
||||
const patch: Record<string, unknown> = {};
|
||||
const existingState = parseIssueExecutionState(input.issue.executionState);
|
||||
|
|
@ -431,27 +479,61 @@ export function applyIssueExecutionPolicyTransition(input: TransitionInput): Tra
|
|||
return { patch };
|
||||
}
|
||||
|
||||
const pendingStage =
|
||||
let pendingStage =
|
||||
existingState?.status === CHANGES_REQUESTED_STATUS && currentStage
|
||||
? currentStage
|
||||
: nextPendingStage(input.policy, existingState);
|
||||
if (!pendingStage) return { patch };
|
||||
|
||||
const returnAssignee = existingState?.returnAssignee ?? currentAssignee;
|
||||
const participant = selectStageParticipant(pendingStage, {
|
||||
const skippedStageIds = [...(existingState?.completedStageIds ?? [])];
|
||||
let participant = selectStageParticipant(pendingStage, {
|
||||
preferred:
|
||||
existingState?.status === CHANGES_REQUESTED_STATUS
|
||||
? explicitAssignee ?? existingState.currentParticipant ?? null
|
||||
: explicitAssignee,
|
||||
exclude: returnAssignee,
|
||||
});
|
||||
while (!participant && canAutoSkipPendingStage({ stage: pendingStage, returnAssignee, requestedStatus })) {
|
||||
skippedStageIds.push(pendingStage.id);
|
||||
pendingStage = nextPendingStage(
|
||||
input.policy,
|
||||
buildStateWithCompletedStages({
|
||||
previous: existingState,
|
||||
completedStageIds: skippedStageIds,
|
||||
returnAssignee,
|
||||
}),
|
||||
);
|
||||
if (!pendingStage) {
|
||||
patch.executionState = buildSkippedStageCompletedState({
|
||||
previous: existingState,
|
||||
completedStageIds: skippedStageIds,
|
||||
returnAssignee,
|
||||
});
|
||||
return { patch };
|
||||
}
|
||||
participant = selectStageParticipant(pendingStage, {
|
||||
preferred:
|
||||
existingState?.status === CHANGES_REQUESTED_STATUS
|
||||
? explicitAssignee ?? existingState.currentParticipant ?? null
|
||||
: explicitAssignee,
|
||||
exclude: returnAssignee,
|
||||
});
|
||||
}
|
||||
if (!participant) {
|
||||
throw unprocessable(`No eligible ${pendingStage.type} participant is configured for this issue`);
|
||||
}
|
||||
|
||||
buildPendingStagePatch({
|
||||
patch,
|
||||
previous: existingState,
|
||||
previous:
|
||||
skippedStageIds.length === (existingState?.completedStageIds ?? []).length
|
||||
? existingState
|
||||
: buildStateWithCompletedStages({
|
||||
previous: existingState,
|
||||
completedStageIds: skippedStageIds,
|
||||
returnAssignee,
|
||||
}),
|
||||
policy: input.policy,
|
||||
stage: pendingStage,
|
||||
participant,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue