mirror of
https://github.com/alkimake/paperclip.git
synced 2026-06-15 10:30:37 +09:00
[codex] Add source-scoped recovery actions (#5599)
## Thinking Path > - Paperclip is a control plane for autonomous AI companies, where work must end with a clear disposition rather than ambiguous agent liveness. > - Recovery currently detects stalled or missing-next-step issues, but source issue recovery can become split across child recovery issues, blockers, and comments. > - That makes it harder for operators and agents to see who owns recovery and what exact action is needed on the original issue. > - Source-scoped recovery actions give the original issue a first-class active recovery state with owner, evidence, wake policy, and resolution outcome. > - This pull request adds the recovery-action data model, backend reconciliation and resolution APIs, and board UI indicators/actions. > - The benefit is clearer stalled-work recovery without losing source issue context or relying on comments as the liveness path. ## What Changed - Added the `issue_recovery_actions` schema, shared types/constants/validators, and an idempotent `0084_issue_recovery_actions` migration ordered after current `master` migrations. - Updated stranded/missing-disposition recovery to create source-scoped recovery actions, wake the recovery owner on the source issue, and avoid locking the source issue for recovery-action wakes. - Added API support for reading active recovery actions on issue detail/list surfaces and resolving them with restored, blocked, cancelled, or false-positive outcomes. - Require blocked recovery resolutions to have an unresolved first-class blocker, and removed the UI shortcut that could mark recovery blocked without a blocker selection path. - Surfaced recovery indicators/actions in the issue UI, blocker notices, active run panels, issue rows, and Storybook coverage. - Updated docs and focused tests for recovery semantics, ownership, races, stale comments, and UI behavior. ## Verification - `pnpm exec vitest run server/src/__tests__/issue-recovery-actions.test.ts server/src/__tests__/heartbeat-process-recovery.test.ts ui/src/components/IssueRecoveryActionCard.test.tsx ui/src/components/IssueBlockedNotice.test.tsx ui/src/api/issues.test.ts` — 5 files, 72 tests passed. - `pnpm --filter @paperclipai/shared typecheck` — passed. - `pnpm --filter @paperclipai/db typecheck` — passed, including migration numbering check. - `pnpm --filter @paperclipai/server typecheck` — passed. - `pnpm --filter @paperclipai/ui typecheck` — passed. - Follow-up verification after blocker-resolution guard: `pnpm exec vitest run server/src/__tests__/issue-recovery-actions.test.ts ui/src/components/IssueRecoveryActionCard.test.tsx ui/src/api/issues.test.ts` — 3 files, 27 tests passed. - Follow-up `pnpm --filter @paperclipai/server typecheck` — passed. - Follow-up `pnpm --filter @paperclipai/ui typecheck` — passed. - UI states are available in `ui/storybook/stories/source-issue-recovery.stories.tsx`; screenshot capture helper is `scripts/screenshot-recovery-card.cjs`. ## Risks - Medium: recovery behavior changes from child recovery issue ownership toward source-scoped actions, so operators may see stalled-work state in new places. - Migration risk is mitigated by using the next migration slot after `master` and making the table/constraints/index creation idempotent for anyone who previously applied the old branch-local `0082_dizzy_master_mold` migration. - Existing child recovery issue paths are still guarded for already-created recovery issues, but new source-scoped flows should be watched in CI and Greptile review. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - OpenAI Codex, GPT-5 coding agent, tool use enabled for shell, Git, GitHub, and local test execution. Context window not exposed by the runtime. ## 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) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] 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
c445e59256
commit
0808b388ee
57 changed files with 3947 additions and 224 deletions
|
|
@ -1691,6 +1691,7 @@ function shouldAutoCheckoutIssueForWake(input: {
|
|||
const wakeReason = readNonEmptyString(input.contextSnapshot?.wakeReason);
|
||||
if (!wakeReason) return false;
|
||||
if (wakeReason === "issue_comment_mentioned") return false;
|
||||
if (wakeReason === "source_scoped_recovery_action") return false;
|
||||
if (wakeReason.startsWith("execution_")) return false;
|
||||
|
||||
return true;
|
||||
|
|
@ -5883,8 +5884,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
|||
|
||||
// Fix A (lazy locking): stamp executionRunId now that the run is actually running,
|
||||
// not at queue time. Guard is idempotent — safe if called more than once.
|
||||
const claimedIssueId = readNonEmptyString(parseObject(claimed.contextSnapshot).issueId);
|
||||
if (claimedIssueId) {
|
||||
const claimedContext = parseObject(claimed.contextSnapshot);
|
||||
const claimedIssueId = readNonEmptyString(claimedContext.issueId);
|
||||
const claimedWakeReason = readNonEmptyString(claimedContext.wakeReason);
|
||||
if (claimedIssueId && claimedWakeReason !== "source_scoped_recovery_action") {
|
||||
const claimedAgent = await getAgent(claimed.agentId);
|
||||
await db
|
||||
.update(issues)
|
||||
|
|
@ -7873,7 +7876,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
|||
});
|
||||
const livenessRun = finalizedRun;
|
||||
await refreshContinuationSummaryForRun(livenessRun, agent);
|
||||
if (issueId && outcome === "succeeded") {
|
||||
const skipRunIssueComment = parseObject(livenessRun.contextSnapshot).skipIssueComment === true;
|
||||
if (issueId && outcome === "succeeded" && !skipRunIssueComment) {
|
||||
try {
|
||||
const existingRunComment = await findRunIssueComment(livenessRun.id, livenessRun.companyId, issueId);
|
||||
if (!existingRunComment) {
|
||||
|
|
|
|||
|
|
@ -24,6 +24,7 @@ export { issueThreadInteractionService } from "./issue-thread-interactions.js";
|
|||
export { issueTreeControlService } from "./issue-tree-control.js";
|
||||
export { issueApprovalService } from "./issue-approvals.js";
|
||||
export { issueReferenceService } from "./issue-references.js";
|
||||
export { issueRecoveryActionService } from "./issue-recovery-actions.js";
|
||||
export { goalService } from "./goals.js";
|
||||
export { activityService, type ActivityFilters } from "./activity.js";
|
||||
export { approvalService } from "./approvals.js";
|
||||
|
|
|
|||
295
server/src/services/issue-recovery-actions.ts
Normal file
295
server/src/services/issue-recovery-actions.ts
Normal file
|
|
@ -0,0 +1,295 @@
|
|||
import { and, desc, eq, inArray } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { issueRecoveryActions } from "@paperclipai/db";
|
||||
import type {
|
||||
IssueRecoveryAction,
|
||||
IssueRecoveryActionKind,
|
||||
IssueRecoveryActionOwnerType,
|
||||
IssueRecoveryActionOutcome,
|
||||
IssueRecoveryActionStatus,
|
||||
} from "@paperclipai/shared";
|
||||
|
||||
const ACTIVE_RECOVERY_ACTION_STATUSES = ["active", "escalated"] as const satisfies readonly IssueRecoveryActionStatus[];
|
||||
const MAX_UPSERT_RETRIES = 3;
|
||||
|
||||
type IssueRecoveryActionRow = typeof issueRecoveryActions.$inferSelect;
|
||||
type DbTransaction = Parameters<Parameters<Db["transaction"]>[0]>[0];
|
||||
type DbOrTransaction = Db | DbTransaction;
|
||||
|
||||
export type UpsertIssueRecoveryActionInput = {
|
||||
companyId: string;
|
||||
sourceIssueId: string;
|
||||
recoveryIssueId?: string | null;
|
||||
kind: IssueRecoveryActionKind;
|
||||
ownerType?: IssueRecoveryActionOwnerType;
|
||||
ownerAgentId?: string | null;
|
||||
ownerUserId?: string | null;
|
||||
previousOwnerAgentId?: string | null;
|
||||
returnOwnerAgentId?: string | null;
|
||||
cause: string;
|
||||
fingerprint: string;
|
||||
evidence?: Record<string, unknown>;
|
||||
nextAction: string;
|
||||
wakePolicy?: Record<string, unknown> | null;
|
||||
monitorPolicy?: Record<string, unknown> | null;
|
||||
maxAttempts?: number | null;
|
||||
timeoutAt?: Date | null;
|
||||
lastAttemptAt?: Date | null;
|
||||
};
|
||||
|
||||
export type ResolveIssueRecoveryActionInput = {
|
||||
companyId: string;
|
||||
sourceIssueId: string;
|
||||
actionId?: string | null;
|
||||
status: Extract<IssueRecoveryActionStatus, "resolved" | "cancelled">;
|
||||
outcome: IssueRecoveryActionOutcome;
|
||||
resolutionNote?: string | null;
|
||||
};
|
||||
|
||||
function toReadModel(row: IssueRecoveryActionRow): IssueRecoveryAction {
|
||||
return {
|
||||
id: row.id,
|
||||
companyId: row.companyId,
|
||||
sourceIssueId: row.sourceIssueId,
|
||||
recoveryIssueId: row.recoveryIssueId,
|
||||
kind: row.kind as IssueRecoveryAction["kind"],
|
||||
status: row.status as IssueRecoveryAction["status"],
|
||||
ownerType: row.ownerType as IssueRecoveryAction["ownerType"],
|
||||
ownerAgentId: row.ownerAgentId,
|
||||
ownerUserId: row.ownerUserId,
|
||||
previousOwnerAgentId: row.previousOwnerAgentId,
|
||||
returnOwnerAgentId: row.returnOwnerAgentId,
|
||||
cause: row.cause,
|
||||
fingerprint: row.fingerprint,
|
||||
evidence: row.evidence,
|
||||
nextAction: row.nextAction,
|
||||
wakePolicy: row.wakePolicy,
|
||||
monitorPolicy: row.monitorPolicy,
|
||||
attemptCount: row.attemptCount,
|
||||
maxAttempts: row.maxAttempts,
|
||||
timeoutAt: row.timeoutAt,
|
||||
lastAttemptAt: row.lastAttemptAt,
|
||||
outcome: row.outcome as IssueRecoveryAction["outcome"],
|
||||
resolutionNote: row.resolutionNote,
|
||||
resolvedAt: row.resolvedAt,
|
||||
createdAt: row.createdAt,
|
||||
updatedAt: row.updatedAt,
|
||||
};
|
||||
}
|
||||
|
||||
function isUniqueRecoveryActionConflict(error: unknown) {
|
||||
const maybe = error as { code?: string; constraint?: string; message?: string } | null;
|
||||
return Boolean(
|
||||
maybe &&
|
||||
maybe.code === "23505" &&
|
||||
(
|
||||
maybe.constraint === "issue_recovery_actions_active_source_uq" ||
|
||||
maybe.constraint === "issue_recovery_actions_active_fingerprint_uq" ||
|
||||
typeof maybe.message === "string" && (
|
||||
maybe.message.includes("issue_recovery_actions_active_source_uq") ||
|
||||
maybe.message.includes("issue_recovery_actions_active_fingerprint_uq")
|
||||
)
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
export function issueRecoveryActionService(db: Db) {
|
||||
const upsertQueues = new Map<string, Promise<void>>();
|
||||
|
||||
async function runExclusiveUpsert<T>(
|
||||
input: UpsertIssueRecoveryActionInput,
|
||||
task: () => Promise<T>,
|
||||
): Promise<T> {
|
||||
const key = `${input.companyId}:${input.sourceIssueId}`;
|
||||
const previous = upsertQueues.get(key) ?? Promise.resolve();
|
||||
let release: () => void = () => {};
|
||||
const current = new Promise<void>((resolve) => {
|
||||
release = resolve;
|
||||
});
|
||||
const next = previous.catch(() => undefined).then(() => current);
|
||||
upsertQueues.set(key, next);
|
||||
|
||||
await previous.catch(() => undefined);
|
||||
try {
|
||||
return await task();
|
||||
} finally {
|
||||
release();
|
||||
if (upsertQueues.get(key) === next) {
|
||||
upsertQueues.delete(key);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function getActiveForIssue(companyId: string, sourceIssueId: string): Promise<IssueRecoveryAction | null> {
|
||||
const row = await db
|
||||
.select()
|
||||
.from(issueRecoveryActions)
|
||||
.where(
|
||||
and(
|
||||
eq(issueRecoveryActions.companyId, companyId),
|
||||
eq(issueRecoveryActions.sourceIssueId, sourceIssueId),
|
||||
inArray(issueRecoveryActions.status, [...ACTIVE_RECOVERY_ACTION_STATUSES]),
|
||||
),
|
||||
)
|
||||
.orderBy(desc(issueRecoveryActions.updatedAt))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
return row ? toReadModel(row) : null;
|
||||
}
|
||||
|
||||
async function listActiveForIssues(companyId: string, sourceIssueIds: string[]) {
|
||||
if (sourceIssueIds.length === 0) return new Map<string, IssueRecoveryAction>();
|
||||
const rows = await db
|
||||
.select()
|
||||
.from(issueRecoveryActions)
|
||||
.where(
|
||||
and(
|
||||
eq(issueRecoveryActions.companyId, companyId),
|
||||
inArray(issueRecoveryActions.sourceIssueId, [...new Set(sourceIssueIds)]),
|
||||
inArray(issueRecoveryActions.status, [...ACTIVE_RECOVERY_ACTION_STATUSES]),
|
||||
),
|
||||
)
|
||||
.orderBy(desc(issueRecoveryActions.updatedAt));
|
||||
const result = new Map<string, IssueRecoveryAction>();
|
||||
for (const row of rows) {
|
||||
if (!result.has(row.sourceIssueId)) result.set(row.sourceIssueId, toReadModel(row));
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
async function retryUpsertSourceScoped(
|
||||
input: UpsertIssueRecoveryActionInput,
|
||||
retryCount: number,
|
||||
error?: unknown,
|
||||
): Promise<IssueRecoveryAction> {
|
||||
if (retryCount >= MAX_UPSERT_RETRIES) {
|
||||
if (error) throw error;
|
||||
throw new Error(
|
||||
`Failed to upsert active recovery action for issue ${input.sourceIssueId} after ${MAX_UPSERT_RETRIES} retries`,
|
||||
);
|
||||
}
|
||||
return upsertSourceScopedUnlocked(input, retryCount + 1);
|
||||
}
|
||||
|
||||
async function upsertSourceScopedUnlocked(
|
||||
input: UpsertIssueRecoveryActionInput,
|
||||
retryCount = 0,
|
||||
): Promise<IssueRecoveryAction> {
|
||||
const existing = await getActiveForIssue(input.companyId, input.sourceIssueId);
|
||||
const now = new Date();
|
||||
const ownerType = input.ownerType ?? (input.ownerAgentId ? "agent" : "board");
|
||||
if (existing) {
|
||||
const [updated] = await db
|
||||
.update(issueRecoveryActions)
|
||||
.set({
|
||||
recoveryIssueId: input.recoveryIssueId ?? null,
|
||||
kind: input.kind,
|
||||
status: "active",
|
||||
ownerType,
|
||||
ownerAgentId: input.ownerAgentId ?? null,
|
||||
ownerUserId: input.ownerUserId ?? null,
|
||||
previousOwnerAgentId: input.previousOwnerAgentId ?? existing.previousOwnerAgentId,
|
||||
returnOwnerAgentId: input.returnOwnerAgentId ?? existing.returnOwnerAgentId,
|
||||
cause: input.cause,
|
||||
fingerprint: input.fingerprint,
|
||||
evidence: input.evidence ?? existing.evidence,
|
||||
nextAction: input.nextAction,
|
||||
wakePolicy: input.wakePolicy ?? null,
|
||||
monitorPolicy: input.monitorPolicy ?? null,
|
||||
attemptCount: existing.attemptCount + 1,
|
||||
maxAttempts: input.maxAttempts ?? null,
|
||||
timeoutAt: input.timeoutAt ?? null,
|
||||
lastAttemptAt: input.lastAttemptAt ?? now,
|
||||
outcome: null,
|
||||
resolutionNote: null,
|
||||
resolvedAt: null,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(issueRecoveryActions.id, existing.id),
|
||||
inArray(issueRecoveryActions.status, [...ACTIVE_RECOVERY_ACTION_STATUSES]),
|
||||
),
|
||||
)
|
||||
.returning();
|
||||
if (!updated) {
|
||||
return retryUpsertSourceScoped(input, retryCount);
|
||||
}
|
||||
return toReadModel(updated!);
|
||||
}
|
||||
|
||||
try {
|
||||
const [created] = await db
|
||||
.insert(issueRecoveryActions)
|
||||
.values({
|
||||
companyId: input.companyId,
|
||||
sourceIssueId: input.sourceIssueId,
|
||||
recoveryIssueId: input.recoveryIssueId ?? null,
|
||||
kind: input.kind,
|
||||
status: "active",
|
||||
ownerType,
|
||||
ownerAgentId: input.ownerAgentId ?? null,
|
||||
ownerUserId: input.ownerUserId ?? null,
|
||||
previousOwnerAgentId: input.previousOwnerAgentId ?? null,
|
||||
returnOwnerAgentId: input.returnOwnerAgentId ?? null,
|
||||
cause: input.cause,
|
||||
fingerprint: input.fingerprint,
|
||||
evidence: input.evidence ?? {},
|
||||
nextAction: input.nextAction,
|
||||
wakePolicy: input.wakePolicy ?? null,
|
||||
monitorPolicy: input.monitorPolicy ?? null,
|
||||
attemptCount: 1,
|
||||
maxAttempts: input.maxAttempts ?? null,
|
||||
timeoutAt: input.timeoutAt ?? null,
|
||||
lastAttemptAt: input.lastAttemptAt ?? now,
|
||||
})
|
||||
.returning();
|
||||
return toReadModel(created!);
|
||||
} catch (error) {
|
||||
if (!isUniqueRecoveryActionConflict(error)) throw error;
|
||||
return retryUpsertSourceScoped(input, retryCount, error);
|
||||
}
|
||||
}
|
||||
|
||||
async function upsertSourceScoped(
|
||||
input: UpsertIssueRecoveryActionInput,
|
||||
): Promise<IssueRecoveryAction> {
|
||||
return runExclusiveUpsert(input, () => upsertSourceScopedUnlocked(input));
|
||||
}
|
||||
|
||||
async function resolveActiveForIssue(
|
||||
input: ResolveIssueRecoveryActionInput,
|
||||
dbOrTx: DbOrTransaction = db,
|
||||
): Promise<IssueRecoveryAction | null> {
|
||||
const now = new Date();
|
||||
const predicates = [
|
||||
eq(issueRecoveryActions.companyId, input.companyId),
|
||||
eq(issueRecoveryActions.sourceIssueId, input.sourceIssueId),
|
||||
inArray(issueRecoveryActions.status, [...ACTIVE_RECOVERY_ACTION_STATUSES]),
|
||||
];
|
||||
if (input.actionId) {
|
||||
predicates.push(eq(issueRecoveryActions.id, input.actionId));
|
||||
}
|
||||
|
||||
const [updated] = await dbOrTx
|
||||
.update(issueRecoveryActions)
|
||||
.set({
|
||||
status: input.status,
|
||||
outcome: input.outcome,
|
||||
resolutionNote: input.resolutionNote ?? null,
|
||||
resolvedAt: now,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(and(...predicates))
|
||||
.returning();
|
||||
|
||||
return updated ? toReadModel(updated) : null;
|
||||
}
|
||||
|
||||
return {
|
||||
getActiveForIssue,
|
||||
listActiveForIssues,
|
||||
resolveActiveForIssue,
|
||||
upsertSourceScoped,
|
||||
};
|
||||
}
|
||||
|
|
@ -17,6 +17,7 @@ import {
|
|||
issueAttachments,
|
||||
issueInboxArchives,
|
||||
issueLabels,
|
||||
issueRecoveryActions,
|
||||
issueRelations,
|
||||
issueComments,
|
||||
issueDocuments,
|
||||
|
|
@ -1355,6 +1356,18 @@ async function listIssueBlockerAttentionMap(
|
|||
explicitWaitingIssueIds.add(parsed.issueId);
|
||||
explicitWaitingIssueIds.add(parsed.leafIssueId);
|
||||
}
|
||||
|
||||
const recoveryActionRows: Array<{ sourceIssueId: string }> = await dbOrTx
|
||||
.select({ sourceIssueId: issueRecoveryActions.sourceIssueId })
|
||||
.from(issueRecoveryActions)
|
||||
.where(
|
||||
and(
|
||||
eq(issueRecoveryActions.companyId, companyId),
|
||||
inArray(issueRecoveryActions.status, ["active", "escalated"]),
|
||||
inArray(issueRecoveryActions.sourceIssueId, explicitWaitCandidateIds),
|
||||
),
|
||||
);
|
||||
for (const row of recoveryActionRows) explicitWaitingIssueIds.add(row.sourceIssueId);
|
||||
}
|
||||
|
||||
const agentRows: IssueBlockerAttentionAgentRow[] = agentIds.size > 0
|
||||
|
|
|
|||
|
|
@ -12,10 +12,12 @@ import {
|
|||
agentWakeupRequests,
|
||||
approvals,
|
||||
companies,
|
||||
issueComments,
|
||||
heartbeatRunEvents,
|
||||
heartbeatRunWatchdogDecisions,
|
||||
heartbeatRuns,
|
||||
issueApprovals,
|
||||
issueRecoveryActions,
|
||||
issueRelations,
|
||||
issueThreadInteractions,
|
||||
issues,
|
||||
|
|
@ -29,6 +31,7 @@ import { redactSensitiveText } from "../../redaction.js";
|
|||
import { logActivity } from "../activity-log.js";
|
||||
import { budgetService } from "../budgets.js";
|
||||
import { instanceSettingsService } from "../instance-settings.js";
|
||||
import { issueRecoveryActionService } from "../issue-recovery-actions.js";
|
||||
import { issueTreeControlService } from "../issue-tree-control.js";
|
||||
import { issueService } from "../issues.js";
|
||||
import { getRunLogStore } from "../run-log-store.js";
|
||||
|
|
@ -37,6 +40,7 @@ import {
|
|||
FINISH_SUCCESSFUL_RUN_HANDOFF_REASON,
|
||||
SUCCESSFUL_RUN_MISSING_STATE_REASON,
|
||||
buildSuccessfulRunHandoffExhaustedNotice,
|
||||
noticeMetadataReferencesRecoveryAction,
|
||||
type SuccessfulRunHandoffNotice,
|
||||
} from "./successful-run-handoff.js";
|
||||
import {
|
||||
|
|
@ -386,6 +390,7 @@ function buildLivenessOriginalIssueComment(finding: IssueLivenessFinding, escala
|
|||
|
||||
export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) {
|
||||
const issuesSvc = issueService(db);
|
||||
const recoveryActionsSvc = issueRecoveryActionService(db);
|
||||
const treeControlSvc = issueTreeControlService(db);
|
||||
const budgets = budgetService(db);
|
||||
const instanceSettings = instanceSettingsService(db);
|
||||
|
|
@ -1566,6 +1571,136 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
|||
return recovery;
|
||||
}
|
||||
|
||||
function strandedRecoveryActionKind(cause: StrandedRecoveryCause) {
|
||||
return cause === SUCCESSFUL_RUN_MISSING_STATE_REASON
|
||||
? "missing_disposition" as const
|
||||
: "stranded_assigned_issue" as const;
|
||||
}
|
||||
|
||||
function strandedRecoveryActionFingerprint(input: {
|
||||
issue: typeof issues.$inferSelect;
|
||||
recoveryCause: StrandedRecoveryCause;
|
||||
}) {
|
||||
return [
|
||||
"source_scoped_recovery",
|
||||
input.issue.companyId,
|
||||
input.issue.id,
|
||||
input.recoveryCause,
|
||||
].join(":");
|
||||
}
|
||||
|
||||
function buildStrandedRecoveryActionEvidence(input: {
|
||||
issue: typeof issues.$inferSelect;
|
||||
latestRun: LatestIssueRun;
|
||||
previousStatus: "todo" | "in_progress";
|
||||
recoveryCause: StrandedRecoveryCause;
|
||||
successfulRunHandoffEvidence?: SuccessfulRunHandoffRecoveryEvidence | null;
|
||||
}) {
|
||||
const context = parseObject(input.latestRun?.contextSnapshot);
|
||||
return {
|
||||
sourceIssueId: input.issue.id,
|
||||
sourceIdentifier: input.issue.identifier,
|
||||
previousStatus: input.previousStatus,
|
||||
latestIssueStatus: input.issue.status,
|
||||
latestRunId: input.latestRun?.id ?? null,
|
||||
latestRunStatus: input.latestRun?.status ?? null,
|
||||
latestRunErrorCode: input.latestRun?.errorCode ?? null,
|
||||
retryReason: readNonEmptyString(context.retryReason) ?? null,
|
||||
recoveryCause: input.recoveryCause,
|
||||
sourceRunId: input.successfulRunHandoffEvidence?.sourceRunId ?? null,
|
||||
correctiveRunId: input.successfulRunHandoffEvidence?.correctiveRunId ?? null,
|
||||
missingDisposition: input.successfulRunHandoffEvidence?.missingDisposition ?? null,
|
||||
handoffAttempt: input.successfulRunHandoffEvidence?.handoffAttempt ?? null,
|
||||
maxHandoffAttempts: input.successfulRunHandoffEvidence?.maxHandoffAttempts ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
async function ensureSourceScopedStrandedRecoveryAction(input: {
|
||||
issue: typeof issues.$inferSelect;
|
||||
latestRun: LatestIssueRun;
|
||||
previousStatus: "todo" | "in_progress";
|
||||
recoveryCause?: StrandedRecoveryCause;
|
||||
successfulRunHandoffEvidence?: SuccessfulRunHandoffRecoveryEvidence | null;
|
||||
}) {
|
||||
const recoveryCause = input.recoveryCause ?? "stranded_assigned_issue";
|
||||
const ownerAgentId = await resolveStrandedIssueRecoveryOwnerAgentId(input.issue);
|
||||
const now = new Date();
|
||||
const action = await recoveryActionsSvc.upsertSourceScoped({
|
||||
companyId: input.issue.companyId,
|
||||
sourceIssueId: input.issue.id,
|
||||
kind: strandedRecoveryActionKind(recoveryCause),
|
||||
ownerType: ownerAgentId ? "agent" : "board",
|
||||
ownerAgentId,
|
||||
previousOwnerAgentId: input.issue.assigneeAgentId,
|
||||
returnOwnerAgentId: input.issue.assigneeAgentId,
|
||||
cause: recoveryCause,
|
||||
fingerprint: strandedRecoveryActionFingerprint({
|
||||
issue: input.issue,
|
||||
recoveryCause,
|
||||
}),
|
||||
evidence: buildStrandedRecoveryActionEvidence({
|
||||
issue: input.issue,
|
||||
latestRun: input.latestRun,
|
||||
previousStatus: input.previousStatus,
|
||||
recoveryCause,
|
||||
successfulRunHandoffEvidence: input.successfulRunHandoffEvidence,
|
||||
}),
|
||||
nextAction: recoveryCause === SUCCESSFUL_RUN_MISSING_STATE_REASON
|
||||
? "Choose and record a valid issue disposition without copying transcript content."
|
||||
: "Restore a live execution path, fix the runtime/adapter failure, or record an intentional manual resolution.",
|
||||
wakePolicy: ownerAgentId
|
||||
? {
|
||||
type: "wake_owner",
|
||||
reason: "source_scoped_recovery_action",
|
||||
ownerAgentId,
|
||||
}
|
||||
: {
|
||||
type: "board_escalation",
|
||||
reason: "no_invokable_recovery_owner",
|
||||
},
|
||||
monitorPolicy: null,
|
||||
maxAttempts: null,
|
||||
lastAttemptAt: now,
|
||||
});
|
||||
|
||||
return action;
|
||||
}
|
||||
|
||||
async function enqueueSourceScopedStrandedRecoveryWake(input: {
|
||||
action: Awaited<ReturnType<typeof recoveryActionsSvc.upsertSourceScoped>>;
|
||||
issue: typeof issues.$inferSelect;
|
||||
latestRun: LatestIssueRun;
|
||||
recoveryCause: StrandedRecoveryCause;
|
||||
}) {
|
||||
if (!input.action.ownerAgentId) return;
|
||||
await deps.enqueueWakeup(input.action.ownerAgentId, {
|
||||
source: "assignment",
|
||||
triggerDetail: "system",
|
||||
reason: "source_scoped_recovery_action",
|
||||
idempotencyKey: `source_scoped_recovery_action:${input.action.id}:${input.action.attemptCount}`,
|
||||
payload: withRecoveryModelProfileHint({
|
||||
issueId: input.issue.id,
|
||||
sourceIssueId: input.issue.id,
|
||||
recoveryActionId: input.action.id,
|
||||
strandedRunId: input.latestRun?.id ?? null,
|
||||
recoveryCause: input.recoveryCause,
|
||||
}),
|
||||
requestedByActorType: "system",
|
||||
requestedByActorId: null,
|
||||
contextSnapshot: withRecoveryModelProfileHint({
|
||||
issueId: input.issue.id,
|
||||
taskId: input.issue.id,
|
||||
wakeReason: "source_scoped_recovery_action",
|
||||
skipIssueComment: true,
|
||||
source: "issue_recovery_action",
|
||||
recoveryActionId: input.action.id,
|
||||
sourceIssueId: input.issue.id,
|
||||
strandedRunId: input.latestRun?.id ?? null,
|
||||
recoveryCause: input.recoveryCause,
|
||||
}),
|
||||
});
|
||||
}
|
||||
|
||||
function buildRecoveryIssueInPlaceEscalationComment(input: {
|
||||
issue: typeof issues.$inferSelect;
|
||||
previousStatus: "todo" | "in_progress";
|
||||
|
|
@ -1682,29 +1817,32 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
|||
recoveryCause?: StrandedRecoveryCause;
|
||||
successfulRunHandoffEvidence?: SuccessfulRunHandoffRecoveryEvidence | null;
|
||||
}) {
|
||||
const nestedRecoverySuppressed = isStrandedIssueRecoveryIssue(input.issue);
|
||||
let recoveryIssue: typeof issues.$inferSelect | null = null;
|
||||
if (!nestedRecoverySuppressed) {
|
||||
recoveryIssue = await ensureStrandedIssueRecoveryIssue({
|
||||
if (isStrandedIssueRecoveryIssue(input.issue)) {
|
||||
return escalateStrandedRecoveryIssueInPlace({
|
||||
issue: input.issue,
|
||||
previousStatus: input.previousStatus,
|
||||
latestRun: input.latestRun,
|
||||
recoveryCause: input.recoveryCause,
|
||||
successfulRunHandoffEvidence: input.successfulRunHandoffEvidence,
|
||||
});
|
||||
}
|
||||
|
||||
const recoveryCause = input.recoveryCause ?? "stranded_assigned_issue";
|
||||
const recoveryAction = await ensureSourceScopedStrandedRecoveryAction({
|
||||
issue: input.issue,
|
||||
previousStatus: input.previousStatus,
|
||||
latestRun: input.latestRun,
|
||||
recoveryCause,
|
||||
successfulRunHandoffEvidence: input.successfulRunHandoffEvidence,
|
||||
});
|
||||
const blockerIds = await existingUnresolvedBlockerIssueIds(input.issue.companyId, input.issue.id);
|
||||
const nextBlockerIds = recoveryIssue
|
||||
? [...new Set([...blockerIds, recoveryIssue.id])]
|
||||
: blockerIds;
|
||||
const updated = await issuesSvc.update(input.issue.id, {
|
||||
status: "blocked",
|
||||
blockedByIssueIds: nextBlockerIds,
|
||||
blockedByIssueIds: blockerIds,
|
||||
assigneeAgentId: recoveryAction.ownerAgentId ?? input.issue.assigneeAgentId,
|
||||
});
|
||||
if (!updated) return null;
|
||||
|
||||
const prefix = await getCompanyIssuePrefix(input.issue.companyId);
|
||||
const recoveryOwner = recoveryIssue?.assigneeAgentId ? await getAgent(recoveryIssue.assigneeAgentId) : null;
|
||||
const recoveryOwner = recoveryAction.ownerAgentId ? await getAgent(recoveryAction.ownerAgentId) : null;
|
||||
const sourceAssignee = input.issue.assigneeAgentId ? await getAgent(input.issue.assigneeAgentId) : null;
|
||||
let notice: SuccessfulRunHandoffNotice | null = null;
|
||||
if (input.recoveryCause === SUCCESSFUL_RUN_MISSING_STATE_REASON && input.successfulRunHandoffEvidence) {
|
||||
|
|
@ -1715,39 +1853,60 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
|||
: null,
|
||||
correctiveRun: input.latestRun ? { id: input.latestRun.id, status: input.latestRun.status } : null,
|
||||
sourceAssignee,
|
||||
recoveryIssue,
|
||||
recoveryIssue: null,
|
||||
recoveryActionId: recoveryAction.id,
|
||||
recoveryOwner,
|
||||
latestIssueStatus: input.issue.status,
|
||||
latestHandoffRunStatus: input.latestRun?.status ?? "unknown",
|
||||
missingDisposition: input.successfulRunHandoffEvidence.missingDisposition,
|
||||
});
|
||||
}
|
||||
let recoveryLine: string;
|
||||
if (nestedRecoverySuppressed) {
|
||||
recoveryLine = await buildNestedStrandedRecoveryLine(input.issue, prefix);
|
||||
} else if (recoveryIssue) {
|
||||
recoveryLine = [
|
||||
const recoveryLine = recoveryAction.ownerAgentId
|
||||
? [
|
||||
"",
|
||||
`- Recovery issue: ${issueUiLink({ identifier: recoveryIssue.identifier, id: recoveryIssue.id }, prefix)}`,
|
||||
`- Recovery action: \`${recoveryAction.id}\``,
|
||||
`- Recovery owner: ${agentUiLink(recoveryOwner, prefix)}`,
|
||||
"- Next action: the recovery owner should either restore a live execution path or record the manual resolution, then mark the recovery issue done.",
|
||||
].join("\n");
|
||||
} else {
|
||||
recoveryLine = [
|
||||
"- Next action: the recovery owner should either restore a live execution path or record the manual resolution on the source issue.",
|
||||
].join("\n")
|
||||
: [
|
||||
"",
|
||||
"- Recovery issue: none created because Paperclip could not find an invokable manager, creator, or executive owner with budget available.",
|
||||
`- Recovery action: \`${recoveryAction.id}\``,
|
||||
"- Recovery owner: board escalation, because Paperclip could not find an invokable manager, creator, or executive owner with budget available.",
|
||||
"- Next action: a board operator should assign an invokable recovery owner, fix the agent/runtime state, or record an intentional manual resolution.",
|
||||
].join("\n");
|
||||
}
|
||||
|
||||
if (notice) {
|
||||
await issuesSvc.addComment(input.issue.id, notice.body, {}, {
|
||||
authorType: "system",
|
||||
presentation: notice.presentation,
|
||||
metadata: notice.metadata,
|
||||
});
|
||||
} else {
|
||||
await issuesSvc.addComment(input.issue.id, `${input.comment ?? ""}${recoveryLine}`, {});
|
||||
if (recoveryAction.attemptCount === 1) {
|
||||
const escalationCommentMarker = `Recovery action: \`${recoveryAction.id}\``;
|
||||
|
||||
const hasEscalationComment = await db
|
||||
.select({ id: issueComments.id, body: issueComments.body, metadata: issueComments.metadata })
|
||||
.from(issueComments)
|
||||
.where(
|
||||
and(
|
||||
eq(issueComments.issueId, input.issue.id),
|
||||
eq(issueComments.authorType, "system"),
|
||||
),
|
||||
)
|
||||
.orderBy(desc(issueComments.createdAt))
|
||||
.limit(50)
|
||||
.then((rows) => rows.some((row) =>
|
||||
(row.body ?? "").includes(escalationCommentMarker) ||
|
||||
noticeMetadataReferencesRecoveryAction(row.metadata, recoveryAction.id),
|
||||
));
|
||||
|
||||
if (!hasEscalationComment) {
|
||||
if (notice) {
|
||||
await issuesSvc.addComment(input.issue.id, notice.body, {}, {
|
||||
authorType: "system",
|
||||
presentation: notice.presentation,
|
||||
metadata: notice.metadata,
|
||||
});
|
||||
} else {
|
||||
await issuesSvc.addComment(input.issue.id, `${input.comment ?? ""}${recoveryLine}`, {}, {
|
||||
authorType: "system",
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
await logActivity(db, {
|
||||
|
|
@ -1772,12 +1931,44 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
|||
latestRunId: input.latestRun?.id ?? null,
|
||||
latestRunStatus: input.latestRun?.status ?? null,
|
||||
latestRunErrorCode: input.latestRun?.errorCode ?? null,
|
||||
recoveryIssueId: recoveryIssue?.id ?? null,
|
||||
nestedRecoverySuppressed,
|
||||
blockerIssueIds: nextBlockerIds,
|
||||
recoveryActionId: recoveryAction.id,
|
||||
recoveryOwnerAgentId: recoveryAction.ownerAgentId,
|
||||
previousOwnerAgentId: recoveryAction.previousOwnerAgentId,
|
||||
returnOwnerAgentId: recoveryAction.returnOwnerAgentId,
|
||||
blockerIssueIds: blockerIds,
|
||||
},
|
||||
});
|
||||
|
||||
await enqueueSourceScopedStrandedRecoveryWake({
|
||||
action: recoveryAction,
|
||||
issue: input.issue,
|
||||
latestRun: input.latestRun,
|
||||
recoveryCause,
|
||||
});
|
||||
|
||||
if (recoveryAction.ownerAgentId && recoveryAction.ownerAgentId === input.issue.assigneeAgentId) {
|
||||
const [currentIssue] = await db
|
||||
.select({
|
||||
status: issues.status,
|
||||
assigneeAgentId: issues.assigneeAgentId,
|
||||
})
|
||||
.from(issues)
|
||||
.where(eq(issues.id, input.issue.id))
|
||||
.limit(1);
|
||||
if (
|
||||
currentIssue &&
|
||||
(currentIssue.status !== "blocked" ||
|
||||
currentIssue.assigneeAgentId !== recoveryAction.ownerAgentId)
|
||||
) {
|
||||
const reblocked = await issuesSvc.update(input.issue.id, {
|
||||
status: "blocked",
|
||||
blockedByIssueIds: blockerIds,
|
||||
assigneeAgentId: recoveryAction.ownerAgentId,
|
||||
});
|
||||
if (reblocked) return reblocked;
|
||||
}
|
||||
}
|
||||
|
||||
return updated;
|
||||
}
|
||||
|
||||
|
|
@ -2038,6 +2229,33 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
|||
}
|
||||
|
||||
async function collectIssueGraphLivenessFindings() {
|
||||
const issueRowsPromise = Promise.resolve(db
|
||||
.select({
|
||||
id: issues.id,
|
||||
companyId: issues.companyId,
|
||||
identifier: issues.identifier,
|
||||
title: issues.title,
|
||||
status: issues.status,
|
||||
projectId: issues.projectId,
|
||||
goalId: issues.goalId,
|
||||
parentId: issues.parentId,
|
||||
assigneeAgentId: issues.assigneeAgentId,
|
||||
assigneeUserId: issues.assigneeUserId,
|
||||
createdByAgentId: issues.createdByAgentId,
|
||||
createdByUserId: issues.createdByUserId,
|
||||
executionPolicy: issues.executionPolicy,
|
||||
executionState: issues.executionState,
|
||||
monitorNextCheckAt: issues.monitorNextCheckAt,
|
||||
monitorAttemptCount: issues.monitorAttemptCount,
|
||||
})
|
||||
.from(issues)
|
||||
.where(
|
||||
and(
|
||||
isNull(issues.hiddenAt),
|
||||
notInArray(issues.originKind, [RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation]),
|
||||
),
|
||||
));
|
||||
|
||||
const [
|
||||
issueRows,
|
||||
relationRows,
|
||||
|
|
@ -2048,33 +2266,9 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
|||
interactionRows,
|
||||
approvalRows,
|
||||
recoveryIssueRows,
|
||||
recoveryActionRows,
|
||||
] = await Promise.all([
|
||||
db
|
||||
.select({
|
||||
id: issues.id,
|
||||
companyId: issues.companyId,
|
||||
identifier: issues.identifier,
|
||||
title: issues.title,
|
||||
status: issues.status,
|
||||
projectId: issues.projectId,
|
||||
goalId: issues.goalId,
|
||||
parentId: issues.parentId,
|
||||
assigneeAgentId: issues.assigneeAgentId,
|
||||
assigneeUserId: issues.assigneeUserId,
|
||||
createdByAgentId: issues.createdByAgentId,
|
||||
createdByUserId: issues.createdByUserId,
|
||||
executionPolicy: issues.executionPolicy,
|
||||
executionState: issues.executionState,
|
||||
monitorNextCheckAt: issues.monitorNextCheckAt,
|
||||
monitorAttemptCount: issues.monitorAttemptCount,
|
||||
})
|
||||
.from(issues)
|
||||
.where(
|
||||
and(
|
||||
isNull(issues.hiddenAt),
|
||||
notInArray(issues.originKind, [RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation]),
|
||||
),
|
||||
),
|
||||
issueRowsPromise,
|
||||
db
|
||||
.select({
|
||||
companyId: issueRelations.companyId,
|
||||
|
|
@ -2164,6 +2358,24 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
|||
notInArray(issues.status, ["done", "cancelled"]),
|
||||
),
|
||||
),
|
||||
issueRowsPromise.then((rows) => {
|
||||
const issueIdsUnderAnalysis = rows.map((row) => row.id);
|
||||
return issueIdsUnderAnalysis.length === 0
|
||||
? []
|
||||
: db
|
||||
.select({
|
||||
companyId: issueRecoveryActions.companyId,
|
||||
issueId: issueRecoveryActions.sourceIssueId,
|
||||
status: issueRecoveryActions.status,
|
||||
})
|
||||
.from(issueRecoveryActions)
|
||||
.where(
|
||||
and(
|
||||
inArray(issueRecoveryActions.status, ["active", "escalated"]),
|
||||
inArray(issueRecoveryActions.sourceIssueId, issueIdsUnderAnalysis),
|
||||
),
|
||||
);
|
||||
}),
|
||||
]);
|
||||
|
||||
const openRecoveryIssues = recoveryIssueRows.flatMap((row) => {
|
||||
|
|
@ -2217,7 +2429,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
|||
})),
|
||||
pendingInteractions: interactionRows,
|
||||
pendingApprovals: approvalRows,
|
||||
openRecoveryIssues,
|
||||
openRecoveryIssues: openRecoveryIssues.concat(recoveryActionRows),
|
||||
now: new Date(),
|
||||
});
|
||||
}
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ import {
|
|||
decideSuccessfulRunHandoff,
|
||||
isIdempotentFinishSuccessfulRunHandoffWakeStatus,
|
||||
isSuccessfulRunHandoffRequiredNoticeBody,
|
||||
noticeMetadataReferencesRecoveryAction,
|
||||
} from "./successful-run-handoff.js";
|
||||
|
||||
const run = {
|
||||
|
|
@ -256,6 +257,7 @@ describe("successful run handoff decision", () => {
|
|||
title: "Recover missing next step PAP-1",
|
||||
status: "todo",
|
||||
} as any,
|
||||
recoveryActionId: "77777777-7777-4777-8777-777777777777",
|
||||
recoveryOwner: { id: "66666666-6666-4666-8666-666666666666", name: "CTO" } as any,
|
||||
latestIssueStatus: "in_progress",
|
||||
latestHandoffRunStatus: "failed",
|
||||
|
|
@ -273,7 +275,7 @@ describe("successful run handoff decision", () => {
|
|||
expect.objectContaining({
|
||||
title: "Recovery owner",
|
||||
rows: expect.arrayContaining([
|
||||
expect.objectContaining({ type: "issue_link", identifier: "PAP-2" }),
|
||||
expect.objectContaining({ type: "key_value", label: "Recovery action", value: "77777777-7777-4777-8777-777777777777" }),
|
||||
expect.objectContaining({ type: "agent_link", label: "Recovery owner", name: "CTO" }),
|
||||
]),
|
||||
}),
|
||||
|
|
@ -286,6 +288,8 @@ describe("successful run handoff decision", () => {
|
|||
]),
|
||||
}),
|
||||
]));
|
||||
expect(noticeMetadataReferencesRecoveryAction(notice.metadata, "77777777-7777-4777-8777-777777777777")).toBe(true);
|
||||
expect(noticeMetadataReferencesRecoveryAction(notice.metadata, "88888888-8888-4888-8888-888888888888")).toBe(false);
|
||||
});
|
||||
|
||||
it("recognizes new notices and legacy markdown headings for fallback deduplication", () => {
|
||||
|
|
|
|||
|
|
@ -61,6 +61,19 @@ export type SuccessfulRunHandoffNotice = {
|
|||
metadata: IssueCommentMetadata;
|
||||
};
|
||||
|
||||
export function noticeMetadataReferencesRecoveryAction(
|
||||
metadata: IssueCommentMetadata | null | undefined,
|
||||
recoveryActionId: string,
|
||||
) {
|
||||
return (metadata?.sections ?? []).some((section) =>
|
||||
section.rows.some((row) =>
|
||||
row.type === "key_value" &&
|
||||
row.label === "Recovery action" &&
|
||||
row.value === recoveryActionId,
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
export type SuccessfulRunHandoffDecision =
|
||||
| {
|
||||
kind: "enqueue";
|
||||
|
|
@ -181,6 +194,7 @@ export function buildSuccessfulRunHandoffExhaustedNotice(input: {
|
|||
correctiveRun: NullableNoticeRun;
|
||||
sourceAssignee: NullableNoticeAgent;
|
||||
recoveryIssue: NullableNoticeIssue;
|
||||
recoveryActionId?: string | null;
|
||||
recoveryOwner: NullableNoticeAgent;
|
||||
latestIssueStatus: string;
|
||||
latestHandoffRunStatus: string;
|
||||
|
|
@ -200,7 +214,9 @@ export function buildSuccessfulRunHandoffExhaustedNotice(input: {
|
|||
title: "Recovery owner",
|
||||
rows: [
|
||||
issueLinkRow("Source issue", input.issue),
|
||||
issueLinkRow("Recovery issue", input.recoveryIssue),
|
||||
input.recoveryActionId
|
||||
? keyValueRow("Recovery action", input.recoveryActionId)
|
||||
: issueLinkRow("Recovery issue", input.recoveryIssue),
|
||||
agentLinkRow("Recovery owner", input.recoveryOwner),
|
||||
agentLinkRow("Source assignee", input.sourceAssignee),
|
||||
keyValueRow("Suggested action", "choose and record a valid issue disposition without copying transcript content"),
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue