mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 16:02:36 +01:00
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A task interrupted by a restart or deploy must resume automatically, or offer a Retry that works on first use > - A board Retry saves an intent first. Dispatch then rechecks the requester's authority and saved source before it starts a run > - When that recheck refused the intent, dispatch returned a bare "deferred". The intent stayed saved with no reason, and the task still showed Retry > - Each later click with a new key joined that same intent, so Retry returned 202 and never started a run > - This pull request names the refusal on the intent and lets a new click replace an intent that can never start > - The benefit is that a Retry button that is shown starts one run, or explains why it cannot ## Linked Issues or Issue Description **What happened?** A native task in a managed sandbox lost its runner twice. Saved-work recovery recovered the files. The task showed "Saved work recovered. Retry or send a message to continue." and `canRetry: true`. A Retry click returned 202 and logged `issue.retry_requested`, but no run started. A second click with a new idempotency key returned the same request ID and `state: failed`. No run started. **Expected behavior** If Retry is offered, a click starts exactly one run. If the Retry cannot start, the task names the reason, and a later click is a fresh attempt. A Retry accepted during the pre-deploy drain starts after the restart. **Steps to reproduce** 1. Recover the files of a failed native run so the hold offers Retry. 2. Save a Retry intent whose dispatch-time authority recheck fails (for example, the requester lost access). 3. Dispatch it: no run starts, and the intent has no wait reason. 4. Click Retry again with a new key: the response returns the first request ID, and no run starts. **Paperclip version or commit** Reproduced on `b400914345` (staging served `4d49a31286`; the affected code is the same). Related: #19399 (automatic Retry after a lost runner), #19422. ## What Changed - `heartbeat/queue.ts`: when dispatch declines a task Retry because the requester can no longer change the task, or its saved board message changed, record a named wait (`retry_authority_denied`, `retry_source_changed`) and log it. Before, it returned "deferred" with no record. - `task-retry.ts`: a new click with a new key replaces an unadmitted intent that dispatch declined for its requester's access (`retry_authority_denied`). The old intent becomes `skipped` / `task_retry_replaced`, and activity records `replacedRequestId`. Waits on shared state (cleanup, recovery, an active run, preservation) still coalesce. The restart hook never replaces a user's click. Chat-owned intents keep their own owner. - `task-interruption-presentation.ts`: fixed public messages for the two new wait reasons. `retry_authority_denied` keeps Retry, because a new click is a fresh attempt. `retry_source_changed` is terminal: an edited saved message cannot be replayed, so Retry is hidden and the user sends a new message. - `doc/execution-semantics.md`: document the rule. ## Verification - New regressions in `server/src/services/workspace-recovery.test.ts` use the real recovery, Retry and dispatch services on embedded Postgres. The fixture runs the automatic Retry from a lost runner, then that Retry's own run loses its runner, and recovery saves its files: - `starts exactly one run when Retry is pressed after the automatic Retry's own run lost its runner` (also checks that a replayed dispatch and a second click start no second run). - `does not let a Retry that could not start swallow a later Retry`: red before the fix in two steps. With no fix, the declined intent has no wait reason. With only the wait fix, the second click returned the first request ID. Green after the fix: one run, bound to the second click. - `dispatches a Retry accepted during the pre-deploy drain once the restarted server resumes`: green before and after. The drain does not lose a saved Retry. - `npx vitest run src/services/workspace-recovery.test.ts src/services/task-retry.test.ts src/services/task-interruption-presentation.test.ts src/services/quarantined-task-retry.test.ts src/services/explicit-native-continuation.test.ts`: 892 passed. - Not verified on staging. After deploy, an intent that dispatch refuses shows a named `retryWait` reason. ## Risks - Low risk. Only intents that dispatch declined for their requester's access are replaced. All other coalescing stays. - A replaced intent's late dispatch finds it `skipped` and starts nothing, so a double click still starts at most one run. - The exact gate that held the staging intent is not confirmed from records. If it was a shared-state wait, the new code names it only if it is one of the two reasons above. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, through Claude Code with tool use (shell, file edits, local test runs). ## 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 searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
373 lines
28 KiB
TypeScript
373 lines
28 KiB
TypeScript
import { and, desc, eq, inArray, isNull, sql, type SQL } from "drizzle-orm";
|
|
import { agents, agentWakeupRequests, approvals, chatActions, companies, heartbeatRuns, issueApprovals, issueRecoveryActions,
|
|
issues, issueTreeHoldMembers, issueTreeHolds, projects, type Db } from "@paperclipai/db";
|
|
import { unionAll } from "drizzle-orm/pg-core";
|
|
import { HEARTBEAT_RUN_STATUSES, readProviderFailure, type TaskInterruptionPresentation } from "@paperclipai/shared";
|
|
import { captureTaskInterruptionFailure } from "../sentry.js";
|
|
import { isAiAuthenticationBlocked } from "./ai-auth-failure.js";
|
|
import { readSystemRecovery } from "./system-recovery-state.js";
|
|
import { PROJECT_ACCESS_DENIED } from "./project-access-denial.js";
|
|
import { authorizationService } from "./authorization.js";
|
|
import { taskWorkspaceProjectRequirements } from "./task-workspace-access.js";
|
|
import { isAcknowledgedNativeStop } from "./acknowledged-native-stop.js";
|
|
import { createRunSecretRedactionRegistry } from "./run-secret-redaction.js";
|
|
import { instanceSettingsService } from "./instance-settings.js";
|
|
import { redactCurrentUserText } from "../log-redaction.js";
|
|
import { logger } from "../middleware/logger.js";
|
|
import { canonicalExecutionUuid } from "./execution-identity.js";
|
|
import { chatActionDeadline } from "./chat-action-work.js";
|
|
import { isQueuedDependencyWait } from "./run-cancellation.js";
|
|
import { AI_CONNECTION_BUSY_RETRY_REASON, AI_CONNECTION_POOL_WAIT_RETRY_REASON, RUNNER_REVISION_WAIT_RETRY_REASON,
|
|
WORKSPACE_BUSY_RETRY_REASON } from "../modules/run-dispatch/index.js";
|
|
|
|
// These scheduled runs wait for a shared resource; none records a failed
|
|
// attempt. Their dispatcher re-checks the resource, so the run can stay
|
|
// scheduled past its retry time. Retry cannot free the resource either.
|
|
const resourceWaitRetryReasons = new Set<string>([WORKSPACE_BUSY_RETRY_REASON, AI_CONNECTION_BUSY_RETRY_REASON,
|
|
AI_CONNECTION_POOL_WAIT_RETRY_REASON, RUNNER_REVISION_WAIT_RETRY_REASON]);
|
|
|
|
type ProjectionIssue = Pick<typeof issues.$inferSelect, "id" | "companyId" | "status" | "assigneeAgentId" | "projectId">;
|
|
type ProjectionRun = Pick<typeof heartbeatRuns.$inferSelect,
|
|
"id" | "status" | "scheduledRetryAt" | "errorCode" | "livenessState" | "resultJson"> &
|
|
Partial<Pick<typeof heartbeatRuns.$inferSelect, "agentId" | "error" | "companyId" | "nativeIssueId" | "runtimeMode" | "startedAt"
|
|
| "scheduledRetryReason">>;
|
|
type ProjectionWait = { id: string; requestedAt: Date; sourceRunId?: string | null; executionWaitReason?: string | null;
|
|
preparationStatus?: string | null; preparationRetryAt?: number | string | null };
|
|
type ProjectionState = {
|
|
run?: ProjectionRun;
|
|
action?: typeof issueRecoveryActions.$inferSelect;
|
|
wait?: ProjectionWait;
|
|
retryWait?: ProjectionWait;
|
|
agent?: Pick<typeof agents.$inferSelect, "status" | "pauseReason">;
|
|
projectAccessAllowed?: boolean;
|
|
projectAccessProjectId?: string | null;
|
|
workspaceSelectionAvailable?: boolean;
|
|
project?: Pick<typeof projects.$inferSelect, "pausedAt" | "pauseReason">;
|
|
company?: Pick<typeof companies.$inferSelect, "status" | "pauseReason">;
|
|
paused: boolean;
|
|
approval: boolean;
|
|
};
|
|
|
|
function boundedJsonString(value: SQL, limit: number) {
|
|
return sql<string | null>`case when jsonb_typeof(${value}) = 'string' then left(${value} #>> '{}', ${limit}) else null end`;
|
|
}
|
|
|
|
// Persisted wait messages can contain diagnostics. Ordinary task views expose
|
|
// only these stable categories, never arbitrary payloads or source identifiers.
|
|
const retryWaitMessages: Record<string, string> = {
|
|
workspace_repair_required: "Retry is waiting for Paperclip to recover the saved files. Your message is saved.",
|
|
remote_cleanup: "Retry is waiting for the previous environment to stop. Your message is saved.",
|
|
local_cleanup: "Retry is waiting for the previous execution to finish cleanup. Your message is saved.",
|
|
cleanup_pending: "Retry is waiting for the previous execution to finish cleanup. Your message is saved.",
|
|
execution_settling: "Retry is waiting for the previous execution to stop. Your message is saved.",
|
|
controller_settling: "Retry is waiting for the previous execution to finish recovery. Your message is saved.",
|
|
process_identity_missing: "Retry is waiting for Paperclip to verify that the previous execution stopped. Your message is saved.",
|
|
process_running: "Retry is waiting for the previous execution to stop. Your message is saved.",
|
|
execution_active: "Retry is waiting for the current run to finish. Your message is saved.",
|
|
decision_pending: "Retry is waiting for a pending approval or question to be resolved. Your message is saved.",
|
|
finalizing_saved_result: "Paperclip is finishing the saved result before Retry can continue.",
|
|
cleanup_quarantined: "Retry cannot continue this attempt. Send a new message after the previous execution has stopped.",
|
|
cancelled_by_operator: "This attempt was stopped. Send a new message to continue from saved work.",
|
|
// A new click replaces a Retry declined for its requester's access with a
|
|
// fresh attempt under the clicker's access, so Retry stays available.
|
|
retry_authority_denied: "Retry could not start with the requesting user's access. Press Retry to start it with yours.",
|
|
// An edited saved message can never be replayed; only new input continues.
|
|
retry_source_changed: "Retry cannot continue because its saved message changed. Send a new message to continue from saved work.",
|
|
};
|
|
|
|
/** Recognize old persisted failures without displaying arbitrary run error text. */
|
|
function deniedProjectId(issue: ProjectionIssue, run?: ProjectionRun): string | null | undefined {
|
|
const historical = run?.errorCode === "setup_failed" && run.error === "Project is outside this low-trust boundary.";
|
|
if (!run || run.status !== "failed" || run.agentId !== issue.assigneeAgentId
|
|
|| (run.errorCode !== PROJECT_ACCESS_DENIED && !historical)) return undefined;
|
|
const detail = run.resultJson?.projectAccessDenied;
|
|
return detail && typeof detail === "object" && "projectId" in detail && typeof detail.projectId === "string"
|
|
? detail.projectId : null;
|
|
}
|
|
|
|
/** Presentation never grants authority. Dispatch repeats all gates under its
|
|
* locks. The same projection is used in detail, lists and attention surfaces. */
|
|
export function projectTaskInterruption(issue: ProjectionIssue, input: ProjectionState, now = Date.now()): TaskInterruptionPresentation | null {
|
|
const { run, wait, action, agent, project, company } = input;
|
|
const retryWait = input.retryWait ?? wait;
|
|
if (!run) return null;
|
|
const system = readSystemRecovery(action?.evidence);
|
|
const base: TaskInterruptionPresentation = { intentId: wait?.id ?? null, responsibility: "system",
|
|
state: "resolved", nextRetryAt: null, canRetry: false, userAction: null,
|
|
sourceRunIds: [run.id], observedRunId: run.id };
|
|
if (["done", "cancelled"].includes(issue.status) || (run.status === "succeeded" && !wait)) return base;
|
|
const providerFailure = run.status === "failed" || run.status === "scheduled_retry"
|
|
? readProviderFailure(run.resultJson?.providerFailure) : null;
|
|
if (providerFailure) base.providerFailure = providerFailure;
|
|
if (["failed", "scheduled_retry"].includes(run.status) && run.errorCode === "workspace_validation_failed"
|
|
&& typeof run.error === "string" && run.error.trim()) base.workspaceFailure = { message: run.error };
|
|
const userAction = !issue.assigneeAgentId ? "assign_agent" : input.approval || run.errorCode === "native_provider_approval_required" ? "approval"
|
|
: isAiAuthenticationBlocked(run) ? "reconnect"
|
|
: run.errorCode === "native_provider_model_rejected" ? "choose_model"
|
|
: [agent?.pauseReason, project?.pauseReason, company?.pauseReason].some(reason => reason === "budget" || reason === "budget_exceeded")
|
|
? "budget" : input.paused || project?.pausedAt || agent?.status === "paused" || company?.status === "paused" ? "resume" : null;
|
|
if (userAction) return { ...base, responsibility: "user", state: "waiting_for_user", userAction };
|
|
// The task's existing dependency UI owns this wait. Keep actual recovery and
|
|
// Retry preparation failures authoritative; this does not change admission.
|
|
if (!action && !retryWait?.preparationStatus && isQueuedDependencyWait(run)) return base;
|
|
// Intentional Stop is not an unsuccessful attempt. This is presentation only;
|
|
// physical containment and pending cleanup keep their execution-side gates.
|
|
if (!wait && run.runtimeMode === "native" && run.companyId === issue.companyId && run.nativeIssueId === issue.id
|
|
&& isAcknowledgedNativeStop({ ...run, companyId: run.companyId, nativeIssueId: run.nativeIssueId })) return base;
|
|
if (["running", "queued"].includes(run.status)) return { ...base, state: "working" };
|
|
const deniedProject = deniedProjectId(issue, run);
|
|
let projectAccess: TaskInterruptionPresentation["projectAccess"];
|
|
if (deniedProject !== undefined) {
|
|
// Permissions cannot repair a stale or unavailable workspace selection.
|
|
if (input.workspaceSelectionAvailable === false) return { ...base, state: "failed" };
|
|
const ready = input.projectAccessAllowed === true;
|
|
const projectId = input.projectAccessProjectId === undefined ? deniedProject : input.projectAccessProjectId;
|
|
projectAccess = { agentId: run.agentId!, projectId: projectId === issue.projectId ? projectId : null, ready };
|
|
if (!ready) return { ...base, responsibility: "user", state: "waiting_for_user",
|
|
userAction: "project_access", projectAccess };
|
|
}
|
|
const recoveryStep: TaskInterruptionPresentation["recoveryStep"] = !system ? undefined
|
|
: system.phase === "verify_execution" ? "checking_previous_run"
|
|
: system.phase === "preserve_files" ? "saving_previous_files" : "finishing_recovery";
|
|
const activeObservation = action?.evidence.systemRecoveryActiveAt;
|
|
const recoveryRunning = system?.state === "running" && typeof activeObservation === "number"
|
|
&& activeObservation > now - 45_000;
|
|
const recoveryScheduled = Boolean(system?.nextAttemptAt && system.nextAttemptAt > now);
|
|
const preparationDeadline = retryWait?.preparationRetryAt == null ? NaN : Number(retryWait.preparationRetryAt);
|
|
// A pending Retry preparation is already a Retry; never offer another one.
|
|
const preparationPending = retryWait?.sourceRunId === run.id && Boolean(retryWait.preparationStatus)
|
|
&& ["issued", "processing"].includes(retryWait.preparationStatus!) && Number.isFinite(preparationDeadline);
|
|
if (retryWait?.sourceRunId === run.id && retryWait.preparationStatus) {
|
|
const deadline = preparationDeadline;
|
|
const pending = preparationPending;
|
|
// Saved-work recovery gates a pending preparation, so it names the wait.
|
|
// Failed or cancelled preparation stays authoritative.
|
|
if (pending && !recoveryRunning && !recoveryScheduled)
|
|
return { ...base, state: "retry_scheduled", nextRetryAt: new Date(deadline).toISOString() };
|
|
if (!pending) return { ...base, state: "failed", canRetry: retryWait.preparationStatus !== "cancelled" };
|
|
}
|
|
if (run.status === "scheduled_retry" && resourceWaitRetryReasons.has(run.scheduledRetryReason ?? "") && !system)
|
|
return { ...base, state: "working" };
|
|
const retryAt = system?.nextAttemptAt ?? run.scheduledRetryAt?.getTime();
|
|
if (retryAt && retryAt > now) return { ...base, state: "retry_scheduled",
|
|
nextRetryAt: new Date(retryAt).toISOString(), canRetry: !preparationPending,
|
|
...(system?.nextAttemptAt && recoveryStep ? { recoveryStep } : {}) };
|
|
if (recoveryRunning) return { ...base, state: "working", ...(recoveryStep ? { recoveryStep } : {}) };
|
|
if (retryWait?.sourceRunId === run.id && retryWait.executionWaitReason) {
|
|
const reason = Object.hasOwn(retryWaitMessages, retryWait.executionWaitReason) ? retryWait.executionWaitReason : "execution_recovery";
|
|
// Keep the latest input receipt so message submission can settle independently.
|
|
return { ...base, state: "failed",
|
|
canRetry: !["cleanup_quarantined", "cancelled_by_operator", "retry_source_changed"].includes(reason), retryWait: { reason,
|
|
message: retryWaitMessages[reason] ?? "Retry is waiting for execution checks. Your message is saved." } };
|
|
}
|
|
if (wait && wait.requestedAt.getTime() > now - 10_000 && system?.state !== "exhausted")
|
|
return { ...base, state: "working" };
|
|
return { ...base, state: "failed", canRetry: true, ...(projectAccess ? { projectAccess } : {}) };
|
|
}
|
|
|
|
/** A scheduled retry has no outcome yet, so its provider failure is read from
|
|
* the predecessor it retries: the countdown keeps its reason.
|
|
* Batch list reads rather than doing seven queries per row. Keep run logs and
|
|
* large result bodies out of this projection. */
|
|
export async function taskInterruptionPresentations(db: Db, companyId: string, input: ProjectionIssue[]) {
|
|
const selected = input.filter(issue => issue.companyId === companyId);
|
|
const result = new Map<string, TaskInterruptionPresentation | null>();
|
|
if (!selected.length) return result;
|
|
const ids = selected.map(issue => issue.id);
|
|
const requested = sql`unnest(${sql.param([...new Set(ids)])}::text[]) as requested_interruption_issues(requested_issue_id)`;
|
|
const requestedIssueId = sql<string>`"requested_interruption_issues"."requested_issue_id"`;
|
|
const requestedUuid = sql<string>`case when ${requestedIssueId} ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'
|
|
then ${requestedIssueId}::uuid else null end`;
|
|
const legacyIssue = sql<string>`${heartbeatRuns.contextSnapshot}->>'issueId'`;
|
|
const candidate = (condition: ReturnType<typeof and>) => db.select({ id: heartbeatRuns.id, createdAt: heartbeatRuns.createdAt })
|
|
.from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), condition))
|
|
.orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)).limit(1);
|
|
const candidates = unionAll(
|
|
candidate(eq(heartbeatRuns.nativeIssueId, requestedUuid)),
|
|
candidate(and(isNull(heartbeatRuns.nativeIssueId), eq(heartbeatRuns.issueId, requestedUuid))),
|
|
candidate(and(isNull(heartbeatRuns.nativeIssueId), isNull(heartbeatRuns.issueId), eq(legacyIssue, requestedIssueId))),
|
|
).as("interruption_run_candidates");
|
|
const latestRuns = db.select({ id: candidates.id }).from(candidates)
|
|
.orderBy(desc(candidates.createdAt), desc(candidates.id)).limit(1).as("latest_interruption_runs");
|
|
const wakeIssue = sql<string>`${agentWakeupRequests.payload}->>'issueId'`;
|
|
const isRetryWait = sql<boolean>`coalesce(jsonb_typeof(${agentWakeupRequests.payload}#>'{taskRetry,sourceRunId}') = 'string', false)`;
|
|
const agentIds = selected.flatMap(issue => issue.assigneeAgentId ? [issue.assigneeAgentId] : []);
|
|
const projectIds = selected.flatMap(issue => issue.projectId ? [issue.projectId] : []);
|
|
const [runs, waits, agentRows, projectRows, pauseRows, approvalRows, companyRows] = await Promise.all([
|
|
db.select({ issueId: requestedIssueId, id: heartbeatRuns.id, status: heartbeatRuns.status,
|
|
agentId: heartbeatRuns.agentId, companyId: heartbeatRuns.companyId,
|
|
nativeIssueId: heartbeatRuns.nativeIssueId, runtimeMode: heartbeatRuns.runtimeMode, startedAt: heartbeatRuns.startedAt,
|
|
agentAdapter: agents.adapterType,
|
|
error: sql<string | null>`case when ${heartbeatRuns.errorCode} = 'setup_failed'
|
|
and ${heartbeatRuns.error} = 'Project is outside this low-trust boundary.'
|
|
or ${heartbeatRuns.errorCode} = 'workspace_validation_failed' then ${heartbeatRuns.error} else null end`,
|
|
errorCode: heartbeatRuns.errorCode, livenessState: heartbeatRuns.livenessState, scheduledRetryAt: heartbeatRuns.scheduledRetryAt,
|
|
scheduledRetryReason: heartbeatRuns.scheduledRetryReason,
|
|
// The run whose resultJson supplied providerFailure. A scheduled retry
|
|
// shows its predecessor's reason, so redaction must use that run's registry.
|
|
providerFailureRunId: sql<string>`case when ${heartbeatRuns.resultJson}->'providerFailure' is null
|
|
and ${heartbeatRuns.status} = 'scheduled_retry' and ${heartbeatRuns.retryOfRunId} is not null
|
|
then ${heartbeatRuns.retryOfRunId} else ${heartbeatRuns.id} end`,
|
|
resultJson: sql<Record<string, unknown>>`jsonb_build_object('configurationIncomplete', ${heartbeatRuns.resultJson}->'configurationIncomplete',
|
|
'stopReason', ${boundedJsonString(sql`${heartbeatRuns.resultJson}->'stopReason'`, 80)},
|
|
'timeoutSource', ${boundedJsonString(sql`${heartbeatRuns.resultJson}->'timeoutSource'`, 80)},
|
|
'timeoutFired', coalesce(${heartbeatRuns.resultJson}->'timeoutFired' = 'true'::jsonb, false),
|
|
'projectAccessDenied', ${heartbeatRuns.resultJson}->'projectAccessDenied',
|
|
'cancelledByActorType', ${boundedJsonString(sql`${heartbeatRuns.resultJson}->'cancelledByActorType'`, 32)},
|
|
'cancelledByUserId', ${boundedJsonString(sql`${heartbeatRuns.resultJson}->'cancelledByUserId'`, 160)},
|
|
'nativeCancellation', jsonb_build_object(
|
|
'schema', ${boundedJsonString(sql`${heartbeatRuns.resultJson}->'nativeCancellation'->'schema'`, 64)},
|
|
'runId', ${boundedJsonString(sql`${heartbeatRuns.resultJson}->'nativeCancellation'->'runId'`, 160)},
|
|
'companyId', ${boundedJsonString(sql`${heartbeatRuns.resultJson}->'nativeCancellation'->'companyId'`, 160)},
|
|
'issueId', ${boundedJsonString(sql`${heartbeatRuns.resultJson}->'nativeCancellation'->'issueId'`, 160)},
|
|
'scope', ${boundedJsonString(sql`${heartbeatRuns.resultJson}->'nativeCancellation'->'scope'`, 32)},
|
|
'reasonCode', ${boundedJsonString(sql`${heartbeatRuns.resultJson}->'nativeCancellation'->'reasonCode'`, 64)},
|
|
'dispatchState', ${boundedJsonString(sql`${heartbeatRuns.resultJson}->'nativeCancellation'->'dispatchState'`, 32)},
|
|
'dispatched', coalesce(${heartbeatRuns.resultJson}->'nativeCancellation'->'dispatched' = 'true'::jsonb, false),
|
|
'intentAuditId', ${boundedJsonString(sql`${heartbeatRuns.resultJson}->'nativeCancellation'->'intentAuditId'`, 160)},
|
|
'acknowledgementAuditId', ${boundedJsonString(sql`${heartbeatRuns.resultJson}->'nativeCancellation'->'acknowledgementAuditId'`, 160)}),
|
|
'providerFailure', coalesce(${heartbeatRuns.resultJson}->'providerFailure', case when ${heartbeatRuns.status} = 'scheduled_retry'
|
|
then (select predecessor.result_json->'providerFailure' from ${heartbeatRuns} predecessor
|
|
where predecessor.company_id = ${heartbeatRuns.companyId} and predecessor.id = ${heartbeatRuns.retryOfRunId}) end))`,
|
|
}).from(requested).innerJoinLateral(latestRuns, sql`true`).innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, latestRuns.id)))
|
|
.leftJoin(agents, and(eq(agents.companyId, heartbeatRuns.companyId), eq(agents.id, heartbeatRuns.agentId)))
|
|
.orderBy(requestedIssueId),
|
|
// Keep at most two candidates per task: latest Retry and latest other input.
|
|
// A new message must not hide an existing Retry's admission explanation.
|
|
db.selectDistinctOn([wakeIssue, isRetryWait], { issueId: wakeIssue, id: agentWakeupRequests.id, requestedAt: agentWakeupRequests.requestedAt,
|
|
sourceRunId: boundedJsonString(sql`${agentWakeupRequests.payload}#>'{taskRetry,sourceRunId}'`, 160),
|
|
executionWaitReason: boundedJsonString(sql`${agentWakeupRequests.payload}#>'{executionWait,reason}'`, 80),
|
|
preparationStatus: chatActions.status, preparationRetryAt: chatActionDeadline("failed_run_retry") })
|
|
.from(agentWakeupRequests).leftJoin(chatActions, and(eq(chatActions.companyId, agentWakeupRequests.companyId),
|
|
eq(chatActions.id, canonicalExecutionUuid(sql`${agentWakeupRequests.payload}#>>'{taskRetry,chatActionId}'`)),
|
|
eq(chatActions.kind, "failed_run_retry"), sql`${chatActions.payload}->>'taskRetryIntentId' = ${agentWakeupRequests.id}::text`,
|
|
sql`${chatActions.payload}->>'issueId' = ${wakeIssue}`,
|
|
sql`${chatActions.payload}->>'agentId' = ${agentWakeupRequests.agentId}::text`,
|
|
sql`${chatActions.payload}->>'failedRunId' = ${agentWakeupRequests.payload}#>>'{taskRetry,sourceRunId}'`,
|
|
inArray(chatActions.status, ["issued", "processing", "failed", "cancelled"])))
|
|
.where(and(eq(agentWakeupRequests.companyId, companyId), inArray(wakeIssue, ids),
|
|
eq(agentWakeupRequests.status, "deferred_issue_execution"), eq(agentWakeupRequests.requestedByActorType, "user")))
|
|
.orderBy(wakeIssue, isRetryWait, desc(agentWakeupRequests.requestedAt), desc(agentWakeupRequests.id)),
|
|
agentIds.length ? db.select({ id: agents.id, status: agents.status, pauseReason: agents.pauseReason }).from(agents)
|
|
.where(and(eq(agents.companyId, companyId), inArray(agents.id, agentIds))) : [],
|
|
projectIds.length ? db.select({ id: projects.id, pausedAt: projects.pausedAt, pauseReason: projects.pauseReason }).from(projects)
|
|
.where(and(eq(projects.companyId, companyId), inArray(projects.id, projectIds))) : [],
|
|
db.select({ issueId: issueTreeHoldMembers.issueId }).from(issueTreeHoldMembers).innerJoin(issueTreeHolds, and(
|
|
eq(issueTreeHolds.companyId, issueTreeHoldMembers.companyId), eq(issueTreeHolds.id, issueTreeHoldMembers.holdId)))
|
|
.where(and(eq(issueTreeHoldMembers.companyId, companyId), inArray(issueTreeHoldMembers.issueId, ids),
|
|
eq(issueTreeHolds.status, "active"), eq(issueTreeHolds.mode, "pause"))),
|
|
db.select({ issueId: issueApprovals.issueId }).from(issueApprovals).innerJoin(approvals, and(
|
|
eq(approvals.companyId, issueApprovals.companyId), eq(approvals.id, issueApprovals.approvalId)))
|
|
.where(and(eq(issueApprovals.companyId, companyId), inArray(issueApprovals.issueId, ids),
|
|
inArray(approvals.status, ["pending", "revision_requested"]))),
|
|
db.select({ status: companies.status, pauseReason: companies.pauseReason }).from(companies).where(eq(companies.id, companyId)),
|
|
]);
|
|
const runIds = runs.map(run => run.id);
|
|
const actions = runIds.length ? await db.selectDistinctOn([issueRecoveryActions.sourceIssueId]).from(issueRecoveryActions).where(and(
|
|
eq(issueRecoveryActions.companyId, companyId), inArray(issueRecoveryActions.sourceIssueId, ids),
|
|
inArray(sql`${issueRecoveryActions.evidence}->>'runId'`, runIds),
|
|
sql`${issueRecoveryActions.evidence} ? 'systemRecovery'`,
|
|
)).orderBy(issueRecoveryActions.sourceIssueId, desc(issueRecoveryActions.createdAt)) : [];
|
|
const byIssue = <T extends { issueId: string }>(rows: T[]) => new Map(rows.map(row => [row.issueId, row]));
|
|
const runMap = byIssue(runs), waitMap = new Map<string, typeof waits[number]>();
|
|
const retryWaitMap = byIssue(waits.filter(wait => wait.sourceRunId));
|
|
for (const wait of waits) {
|
|
const current = waitMap.get(wait.issueId);
|
|
if (!current || wait.requestedAt > current.requestedAt ||
|
|
(wait.requestedAt.getTime() === current.requestedAt.getTime() && wait.id > current.id)) waitMap.set(wait.issueId, wait);
|
|
}
|
|
// Recheck current authority, including standard trust, root-issue grants and
|
|
// project policy. Do not reuse the failed run's frozen policy for a new retry.
|
|
// Only denied tasks need these reads; deduplicate projects and batch per agent.
|
|
// A task's organizational project can differ from its retained source. Old
|
|
// failures did not record the denied project, and even typed failures can
|
|
// become obsolete after a workspace selection changes.
|
|
const deniedIssues = selected.filter(issue => issue.assigneeAgentId && !["done", "cancelled"].includes(issue.status)
|
|
&& deniedProjectId(issue, runMap.get(issue.id)) !== undefined);
|
|
const requirements = await taskWorkspaceProjectRequirements(db, companyId, deniedIssues.map(issue => issue.id));
|
|
const deniedByAgent = new Map<string, Set<string>>();
|
|
for (const issue of deniedIssues) {
|
|
const ids = deniedByAgent.get(issue.assigneeAgentId!) ?? new Set<string>();
|
|
for (const projectId of requirements.get(issue.id)?.projectIds ?? []) ids.add(projectId);
|
|
if (ids.size) deniedByAgent.set(issue.assigneeAgentId!, ids);
|
|
}
|
|
const currentAccess = new Map<string, boolean>();
|
|
const authorization = authorizationService(db);
|
|
for (const [agentId, ids] of deniedByAgent) {
|
|
const projectIds = [...ids];
|
|
const decisions = await authorization.decideReads({ type: "agent", agentId, companyId, source: "agent_key" },
|
|
projectIds.map(projectId => ({ type: "project" as const, companyId, projectId })));
|
|
projectIds.forEach((projectId, index) => currentAccess.set(`${agentId}:${projectId}`, decisions[index]?.allowed === true));
|
|
}
|
|
const actionMap = new Map(actions.map(action => [action.sourceIssueId, action]));
|
|
const agentMap = new Map(agentRows.map(agent => [agent.id, agent]));
|
|
const projectMap = new Map(projectRows.map(project => [project.id, project]));
|
|
const pauses = new Set(pauseRows.map(row => row.issueId)), approvalsSet = new Set(approvalRows.map(row => row.issueId));
|
|
for (const issue of selected) {
|
|
const required = requirements.get(issue.id);
|
|
const denied = required?.projectIds.filter(projectId => currentAccess.get(`${issue.assigneeAgentId}:${projectId}`) !== true);
|
|
const presentation = projectTaskInterruption(issue, {
|
|
run: runMap.get(issue.id), action: actionMap.get(issue.id), wait: waitMap.get(issue.id), retryWait: retryWaitMap.get(issue.id),
|
|
projectAccessAllowed: denied?.length === 0,
|
|
projectAccessProjectId: denied?.includes(issue.projectId ?? "") ? issue.projectId : null,
|
|
workspaceSelectionAvailable: required?.available ?? false,
|
|
agent: agentMap.get(issue.assigneeAgentId ?? ""), project: projectMap.get(issue.projectId ?? ""),
|
|
company: companyRows[0], paused: pauses.has(issue.id), approval: approvalsSet.has(issue.id),
|
|
});
|
|
result.set(issue.id, presentation);
|
|
const run = runMap.get(issue.id);
|
|
const runStatus = HEARTBEAT_RUN_STATUSES.find(status => status === run?.status);
|
|
if (presentation?.responsibility === "system" && presentation.state === "failed" && run && runStatus) {
|
|
captureTaskInterruptionFailure({ companyId, taskId: issue.id, agentId: run.agentId,
|
|
runId: run.id, runStatus, intentId: presentation.intentId,
|
|
agentAdapter: run.agentAdapter ?? "unknown" });
|
|
}
|
|
}
|
|
// Redact each explanation with the registry of the run that recorded it: the
|
|
// workspace failure is the observed run's, while a scheduled retry's provider
|
|
// failure may be its predecessor's.
|
|
const disclosures = [...result.entries()].flatMap(([issueId, presentation]) =>
|
|
(presentation?.providerFailure || presentation?.workspaceFailure) && presentation.observedRunId
|
|
? [{ id: presentation.observedRunId, providerFailureRunId: runMap.get(issueId)?.providerFailureRunId ?? presentation.observedRunId,
|
|
providerFailure: presentation.providerFailure, workspaceFailure: presentation.workspaceFailure, presentation }] : []);
|
|
if (disclosures.length) {
|
|
try {
|
|
const [redacted, settings] = await Promise.all([
|
|
createRunSecretRedactionRegistry(db).redactForRuns(companyId, disclosures.flatMap(({ id, providerFailureRunId, providerFailure, workspaceFailure }) => [
|
|
{ id: providerFailureRunId, providerFailure, workspaceFailure: undefined },
|
|
{ id, providerFailure: undefined, workspaceFailure },
|
|
])),
|
|
instanceSettingsService(db).getGeneral(),
|
|
]);
|
|
const safe = disclosures.map((_, index) => ({ providerFailure: redacted[2 * index]?.providerFailure,
|
|
workspaceFailure: redacted[2 * index + 1]?.workspaceFailure }));
|
|
const safeMessage = (value: unknown) => typeof value === "string"
|
|
? [...redactCurrentUserText(value, { enabled: settings.censorUsernameInLogs })].slice(0, 4000).join("") : "";
|
|
disclosures.forEach(({ presentation }, index) => {
|
|
if (presentation.providerFailure) {
|
|
const value = safe[index]?.providerFailure;
|
|
const failure = readProviderFailure({ ...value, message: safeMessage(value?.message) });
|
|
if (failure) presentation.providerFailure = failure;
|
|
else delete presentation.providerFailure;
|
|
}
|
|
if (presentation.workspaceFailure) {
|
|
const message = safeMessage(safe[index]?.workspaceFailure?.message);
|
|
if (message.trim()) presentation.workspaceFailure = { message };
|
|
else delete presentation.workspaceFailure;
|
|
}
|
|
});
|
|
} catch {
|
|
for (const { presentation } of disclosures) {
|
|
delete presentation.providerFailure;
|
|
delete presentation.workspaceFailure;
|
|
}
|
|
logger.warn({ companyId, runCount: disclosures.length }, "Could not redact run failure display; explanations omitted");
|
|
}
|
|
}
|
|
return result;
|
|
}
|
|
|
|
export async function taskInterruptionPresentation(db: Db, issue: ProjectionIssue) {
|
|
return (await taskInterruptionPresentations(db, issue.companyId, [issue])).get(issue.id) ?? null;
|
|
}
|