mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-13 07:12:00 +01:00
DottaandClaude Opus 5.5
8a4eaab2c0
A person can answer an agent's question while the agent's turn is still running. The answer then waits on its own deferred wake. If that turn completed the task, release cancelled every deferred assignee wake on the done task, including the answer, even though the person's own queued messages reopen the task. The answer was dropped without any run receiving it. Treat an undelivered person's answer that was saved before the task was completed as owed input: reopen the done task straight to in_progress (an answer continues the assignee's own execution) and promote the answer wake. Answers saved after completion, tool-action outcomes, already-delivered answers, agent-requested wakes and cancelled tasks keep the old behavior. Eval steering.approval-delivery (oracle v7): also accept Acorn work that the worker's settling turn saved after the answer was saved, while still requiring exactly one later turn bound to the answer. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1460 lines
68 KiB
TypeScript
1460 lines
68 KiB
TypeScript
import { notifyStrandedAgentStatus } from "../../../services/live-run-recovery-notifications.js";
|
|
import { notifyDeferredExecutionWork } from "../../../services/deferred-execution-notifications.js";
|
|
import { notifyRunDispatch } from "../../../services/run-dispatch-notifications.js";
|
|
import { notifyStatusCardIssueChange } from "../../../services/status-card-notifications.js";
|
|
import { notifyExternalConversationSettlement } from "../../../services/external-conversation-settlement-notifications.js";
|
|
import { notifyNativeStatusReassessment } from "../../../services/native-status-reassessment-notifications.js";
|
|
import { notifyUnsafeWorkspaceExportRecovery } from "../../../services/unsafe-workspace-export-notifications.js";
|
|
import { notifyChatInboundWakeNotices } from "../../../services/chat-inbound-wake-notice-notifications.js";
|
|
import { notifyChatRunMilestones } from "../../../services/chat-run-milestone-notifications.js";
|
|
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js";
|
|
import { isAcknowledgedNativeReassignmentStop, isAcknowledgedNativeStop } from "../../../services/acknowledged-native-stop.js";
|
|
import { readRunCancellation } from "../../../services/run-cancellation.js";
|
|
import { isCompletedOnboardingHandoffWake } from "../../../services/chat-completion-delivery.js";
|
|
import { instanceSettingsService } from "../../../services/instance-settings.js";
|
|
import { currentConversationCommentCondition } from "../../../services/agent-conversations.js";
|
|
import { getExecutionBlocker, hasPendingRunCleanup } from "../../../services/execution-blocker.js";
|
|
import { nativeManualAdmissionHoldsExecution } from "../../../services/native-runtime/native-runner-ownership.js";
|
|
import { and, asc, eq, inArray, isNull, notInArray, or, sql } from "drizzle-orm";
|
|
import { readReleaseRun, type ReleaseRun } from "./release-run-read.js";
|
|
import { isGovernancePauseStop } from "../../../services/governance-pause-stop.js";
|
|
import type { Db } from "@paperclipai/db";
|
|
import { deliveredQueuedCommentIds, EXTERNAL_CHAT_MESSAGE_WAKE_MUTATION, EXTERNAL_CHAT_MESSAGE_WAKE_REASON } from "../../../services/queued-comment-delivery.js";
|
|
import { extractIssueReferenceIdentifiers, isUuidLike } from "@paperclipai/shared";
|
|
import {
|
|
activityLog,
|
|
agentWakeupRequests,
|
|
agents,
|
|
chatActions,
|
|
heartbeatRuns,
|
|
issueComments,
|
|
issueRecoveryActions,
|
|
issueRelations,
|
|
issues,
|
|
nativeRunFinalizations,
|
|
} from "@paperclipai/db";
|
|
import { hasConversationContinuationPolicy } from "../../../services/conversation-continuation.js";
|
|
import { readQueuedInteractionResponse } from "../../../services/queued-interaction-response.js";
|
|
import { legacyExecutionNeedsReconciliationWithEvidence } from "../../../services/legacy-execution-recovery.js";
|
|
import {
|
|
authorizeFailedChatRunRetryWake,
|
|
FailedChatRunRetryAuthorizationError,
|
|
} from "../../../services/durable-chat-wakeup.js";
|
|
import { isRetiredExternalChatQuestionSource } from "../../../services/question-response-delivery.js";
|
|
import { HttpError } from "../../../errors.js";
|
|
import { evaluateAgentInvokabilityFromDb } from "../../../services/agent-invokability.js";
|
|
import { issueTreeControlService, isVerifiedIssueTreeControlInteractionWake } from "../../../services/issue-tree-control.js";
|
|
import { isAutomaticRecoverySuppressedByPauseHold } from "../../../services/recovery/pause-hold-guard.js";
|
|
import { classifyContinuationFailure } from "../../../services/recovery/service.js";
|
|
import { executionRetryAccounting } from "../../../services/execution-recovery-attempt.js";
|
|
import { issueService } from "../../../services/issues.js";
|
|
import { issueRecoveryActionService } from "../../../services/issue-recovery-actions.js";
|
|
import { readContinuationAttempt } from "../../../services/recovery/run-liveness-continuations.js";
|
|
import { withRecoveryContext } from "../../../services/recovery/status-only-context.js";
|
|
import { parseIssueExecutionState } from "../../../services/issue-execution-policy.js";
|
|
import {
|
|
queuedCommentIdsFromWakePayload,
|
|
queuedCommentIdsFromRunContext,
|
|
withQueuedCommentIdsInWakePayload,
|
|
} from "../../../services/issue-queued-comment-queue.js";
|
|
import { CHAT_CONVERSATION_ID_KEY, extractWakeCommentIds, readWakeChatConversationId } from "../../run-dispatch/index.js";
|
|
import { deferredWakeHoldsNoOtherConversation, otherConversationCommentIds, splitOtherConversationMessages } from "./chat-conversation-batch.js";
|
|
import { hasInteractionContinuationWakeContext } from "../domain/context.js";
|
|
import { decidePreDrain, type PreDrainFacts } from "../domain/policy.js";
|
|
import {
|
|
EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON,
|
|
isConfigurationIncompleteFailedRun,
|
|
isWorkspaceValidationFailedRun,
|
|
parseObject,
|
|
readNonEmptyString,
|
|
} from "../domain/values.js";
|
|
import { requireTransactionScopeTx, TransactionScope } from "../application/ports.js";
|
|
import type {
|
|
DeferredWakeCandidate,
|
|
InvokableAgentSnapshot,
|
|
IssueLockWriter,
|
|
IssueSnapshot,
|
|
LockedIssueExecution,
|
|
ReleaseTransactionResult,
|
|
RunSnapshot,
|
|
WakeAdmissionReader,
|
|
WakeAdmissionWriter,
|
|
WakeQueueHost,
|
|
WakeQueueTransaction,
|
|
} from "../application/ports.js";
|
|
import type { RunSummary } from "../application/types.js";
|
|
|
|
const DEFERRED_WAKE_STATUS = "deferred_issue_execution";
|
|
const DEFERRED_WAKE_CONTEXT_KEY = "_paperclipWakeContext";
|
|
const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const;
|
|
|
|
type HeartbeatRunRow = typeof heartbeatRuns.$inferSelect;
|
|
type IssueRow = typeof issues.$inferSelect;
|
|
|
|
function normalizeAgentNameKey(value: string | null | undefined): string | null {
|
|
if (typeof value !== "string") return null;
|
|
const normalized = value.trim().toLowerCase();
|
|
return normalized.length > 0 ? normalized : null;
|
|
}
|
|
|
|
function toRequestedByActorType(value: string | null): "user" | "agent" | "system" | null {
|
|
// The database column is free text; map any value outside the union to
|
|
// null instead of widening the type back to string.
|
|
return value === "user" || value === "agent" || value === "system" ? value : null;
|
|
}
|
|
|
|
function toRunSnapshot(row: ReleaseRun): RunSnapshot {
|
|
const configurationIncompletePayload = parseObject(parseObject(row.resultJson).configurationIncomplete);
|
|
return {
|
|
id: row.id,
|
|
companyId: row.companyId,
|
|
agentId: row.agentId,
|
|
status: row.status,
|
|
runtimeMode: row.runtimeMode,
|
|
conversationContinuation: row.runtimeMode === "legacy" && hasConversationContinuationPolicy(row.resultJson),
|
|
errorCode: row.errorCode,
|
|
responsibleUserId: row.responsibleUserId,
|
|
contextSnapshot: parseObject(row.contextSnapshot),
|
|
configurationIncompletePayload: Object.keys(configurationIncompletePayload).length > 0 ? configurationIncompletePayload : null,
|
|
operatorStopped: isOperatorStoppedRunRow(row),
|
|
retryableBeforeProviderWork: isRetryableBeforeProviderWorkRunRow(row),
|
|
executionRetryAccounting: carriedRetryAccounting(row),
|
|
};
|
|
}
|
|
|
|
function carriedRetryAccounting(row: ReleaseRun): Record<string, unknown> | null {
|
|
const accounting = executionRetryAccounting(row);
|
|
return accounting.failureRetries > 0 || accounting.maxTurnContinuations > 0 || (accounting.interruptedPreparations ?? 0) > 0
|
|
|| (accounting.storageWaits ?? 0) > 0 || (accounting.agentDirectoryWaits ?? 0) > 0 ? { ...accounting } : null;
|
|
}
|
|
|
|
export function isRetryableBeforeProviderWorkRunRow(row: ReleaseRun): boolean {
|
|
const recovery = parseObject(parseObject(row.resultJson).executionRecovery);
|
|
return ["failed", "timed_out", "interrupted"].includes(row.status)
|
|
&& recovery.kind === "bootstrap" && recovery.providerWorkStarted === false
|
|
&& classifyContinuationFailure(row).kind === "transient_infra";
|
|
}
|
|
|
|
const OPERATOR_STOP_ERROR_CODES = new Set(["operator_interrupted", "agent_paused"]);
|
|
|
|
/** Mirrors recovery's `isOperatorCancelledRun`, plus agent pause, for the release-time recovery tail. */
|
|
export function isOperatorStoppedRunRow(row: Pick<HeartbeatRunRow, "status" | "errorCode" | "resultJson">): boolean {
|
|
if (row.status !== "cancelled") return false;
|
|
const cancellation = readRunCancellation(row.resultJson);
|
|
if (cancellation?.source === "queued_message") return false;
|
|
if (cancellation?.source === "operator" && cancellation.expected) return true;
|
|
// A pause's stop (budget, agent or lifecycle pause) is intentional too: the
|
|
// release must not treat the paused agent's task as stranded work.
|
|
if (isGovernancePauseStop(row)) return true;
|
|
if (row.errorCode && OPERATOR_STOP_ERROR_CODES.has(row.errorCode)) return true;
|
|
const actorType = parseObject(row.resultJson).cancelledByActorType;
|
|
return actorType === "user" || actorType === "board";
|
|
}
|
|
|
|
function toIssueSnapshot(row: IssueRow): IssueSnapshot {
|
|
return {
|
|
conversationAgentId: row.conversationAgentId,
|
|
conversationUserId: row.conversationUserId,
|
|
conversationState: row.conversationState,
|
|
id: row.id,
|
|
companyId: row.companyId,
|
|
identifier: row.identifier ?? "",
|
|
status: row.status,
|
|
assigneeAgentId: row.assigneeAgentId,
|
|
assigneeUserId: row.assigneeUserId,
|
|
hiddenAt: row.hiddenAt,
|
|
originKind: row.originKind,
|
|
monitorNextCheckAt: row.monitorNextCheckAt,
|
|
executionState: (row.executionState as Record<string, unknown> | null) ?? null,
|
|
responsibleUserId: row.responsibleUserId,
|
|
parentId: row.parentId,
|
|
originId: row.originId,
|
|
originRunId: row.originRunId,
|
|
};
|
|
}
|
|
|
|
function toRunSummary(row: HeartbeatRunRow): RunSummary {
|
|
return {
|
|
id: row.id,
|
|
companyId: row.companyId,
|
|
agentId: row.agentId,
|
|
invocationSource: row.invocationSource,
|
|
triggerDetail: row.triggerDetail,
|
|
wakeupRequestId: row.wakeupRequestId,
|
|
};
|
|
}
|
|
|
|
function toDeferredWakeCandidate(row: typeof agentWakeupRequests.$inferSelect): DeferredWakeCandidate {
|
|
const payload = parseObject(row.payload);
|
|
const queuedCommentIds = queuedCommentIdsFromWakePayload(payload);
|
|
const deferredContextSeed = parseObject(payload[DEFERRED_WAKE_CONTEXT_KEY]);
|
|
const deferredCommentIds = extractWakeCommentIds(deferredContextSeed);
|
|
const wakeReason = readNonEmptyString(deferredContextSeed.wakeReason);
|
|
const queuedReason = wakeReason ?? readNonEmptyString(row.reason);
|
|
const queuedWakeIsCommentOnly =
|
|
!queuedReason ||
|
|
queuedReason === "issue_commented" ||
|
|
queuedReason === "issue_reopened_via_comment" ||
|
|
queuedReason === "issue_comment_mentioned" ||
|
|
// An inbound chat message wake carries only its saved messages. When a
|
|
// completed run already received them, nothing is left to start, and a
|
|
// reviewed chat run with no message cannot be attested.
|
|
(queuedReason === EXTERNAL_CHAT_MESSAGE_WAKE_REASON && payload.mutation === EXTERNAL_CHAT_MESSAGE_WAKE_MUTATION);
|
|
const preservesIndependentContinuation =
|
|
hasInteractionContinuationWakeContext(deferredContextSeed) ||
|
|
deferredContextSeed.resumeIntent === true ||
|
|
!queuedWakeIsCommentOnly;
|
|
|
|
return {
|
|
id: row.id,
|
|
companyId: row.companyId,
|
|
agentId: row.agentId,
|
|
reason: row.reason,
|
|
source: row.source,
|
|
triggerDetail: row.triggerDetail,
|
|
requestedByActorType: toRequestedByActorType(row.requestedByActorType),
|
|
requestedByActorId: row.requestedByActorId,
|
|
payload,
|
|
queuedCommentIds,
|
|
preservesIndependentContinuation,
|
|
deferredContextSeed,
|
|
deferredCommentIds,
|
|
wakeReason,
|
|
};
|
|
}
|
|
|
|
export type WakeQueuePostgresAdapterDeps = {
|
|
resolveResponsibleUserId: WakeQueueHost["resolveResponsibleUserId"];
|
|
getRoutineEnv: WakeQueueHost["getRoutineEnv"];
|
|
resolveSessionBeforeForWakeup: WakeQueueHost["resolveSessionBeforeForWakeup"];
|
|
};
|
|
|
|
function buildHost(_tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueHost {
|
|
return {
|
|
resolveResponsibleUserId: deps.resolveResponsibleUserId,
|
|
getRoutineEnv: deps.getRoutineEnv,
|
|
resolveSessionBeforeForWakeup: deps.resolveSessionBeforeForWakeup,
|
|
};
|
|
}
|
|
|
|
function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, run: ReleaseRun): WakeQueueTransaction {
|
|
const treeControlSvc = issueTreeControlService(tx);
|
|
const issuesSvc = issueService(tx);
|
|
const interruptQueueId = run.runtimeMode !== "native" && run.status === "cancelled"
|
|
? readNonEmptyString(run.resultJson?.queuedCommentInterruptQueueId)
|
|
: null;
|
|
|
|
return {
|
|
async findInvokableAgent({ companyId, agentId }): Promise<InvokableAgentSnapshot | null> {
|
|
const agent = await tx
|
|
.select()
|
|
.from(agents)
|
|
.where(and(eq(agents.id, agentId), eq(agents.companyId, companyId)))
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!agent) return null;
|
|
const invokability = await evaluateAgentInvokabilityFromDb(tx, agent);
|
|
return { id: agent.id, companyId: agent.companyId, name: agent.name, invokable: invokability.invokable,
|
|
paused: !invokability.invokable && invokability.reason === "paused" };
|
|
},
|
|
|
|
async findNextDeferredWake({ companyId, issueId, excludedWakeIds }) {
|
|
while (true) {
|
|
const row = await tx
|
|
.select()
|
|
.from(agentWakeupRequests)
|
|
.where(
|
|
and(
|
|
eq(agentWakeupRequests.companyId, companyId),
|
|
eq(agentWakeupRequests.status, DEFERRED_WAKE_STATUS),
|
|
excludedWakeIds?.length ? notInArray(agentWakeupRequests.id, excludedWakeIds) : undefined,
|
|
sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`,
|
|
interruptQueueId ? eq(agentWakeupRequests.id, interruptQueueId) : undefined,
|
|
interruptQueueId ? eq(agentWakeupRequests.agentId, run.agentId) : undefined,
|
|
),
|
|
)
|
|
.orderBy(asc(agentWakeupRequests.requestedAt))
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!row) return null;
|
|
const candidate = toDeferredWakeCandidate(row);
|
|
// Upgrade repair: older code could merge two conversations into one
|
|
// saved chat wake. Split it and read the queue again.
|
|
const conversationId = readWakeChatConversationId(candidate.deferredContextSeed);
|
|
if (conversationId && candidate.queuedCommentIds.length > 0 && await splitOtherConversationMessages(tx, row, {
|
|
companyId, issueId, conversationId, queuedCommentIds: candidate.queuedCommentIds, payload: candidate.payload,
|
|
})) continue;
|
|
try {
|
|
const authorizedFailedChatRetry = await authorizeFailedChatRunRetryWake(
|
|
db,
|
|
tx,
|
|
{
|
|
phase: "promotion",
|
|
wakeupRequestId: row.id,
|
|
companyId,
|
|
agentId: row.agentId,
|
|
issueId,
|
|
contextSnapshot: candidate.deferredContextSeed,
|
|
},
|
|
);
|
|
return { ...candidate, authorizedFailedChatRetry };
|
|
} catch (error) {
|
|
// Retire only a proven denial. Transient failures roll back this
|
|
// transaction instead of discarding otherwise authorized work.
|
|
if (
|
|
!(error instanceof FailedChatRunRetryAuthorizationError) &&
|
|
!(
|
|
error instanceof HttpError &&
|
|
error.status >= 400 &&
|
|
error.status < 500
|
|
)
|
|
) {
|
|
throw error;
|
|
}
|
|
const now = new Date();
|
|
await notifyStrandedAgentStatus(tx, { companyId, agentId: row.agentId });
|
|
await notifyChatInboundWakeNotices(tx, { companyId, issueId });
|
|
await tx
|
|
.update(agentWakeupRequests)
|
|
.set({
|
|
status: "cancelled",
|
|
finishedAt: now,
|
|
error:
|
|
"The exact failed chat request is no longer authorized. Send a new request in the current connected conversation.",
|
|
updatedAt: now,
|
|
})
|
|
.where(
|
|
and(
|
|
eq(agentWakeupRequests.companyId, companyId),
|
|
eq(agentWakeupRequests.id, row.id),
|
|
eq(agentWakeupRequests.status, DEFERRED_WAKE_STATUS),
|
|
),
|
|
);
|
|
}
|
|
}
|
|
},
|
|
|
|
async getQueuedCommentLiveness({ companyId, issueId, wakeAgentId, finishingRunId, finishingRunAgentId, queuedCommentIds }) {
|
|
// Malformed ids (for example an 8-char short form written into a wake
|
|
// payload) can never match a comment row, but Postgres rejects the whole
|
|
// `inArray` statement on the first invalid uuid and blocks every queued
|
|
// wake for the issue. Skip them up front so one bad id cannot take down
|
|
// the batch; the liveness filters below already treat them as not live.
|
|
const queryableCommentIds = queuedCommentIds
|
|
.map((commentId) => commentId.trim())
|
|
.filter((commentId) => isUuidLike(commentId));
|
|
const rows = queryableCommentIds.length === 0 ? [] : await tx
|
|
.select({ id: issueComments.id, deletedAt: issueComments.deletedAt, createdByRunId: issueComments.createdByRunId })
|
|
.from(issueComments)
|
|
.where(and(eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId), inArray(issueComments.id, queryableCommentIds), currentConversationCommentCondition()));
|
|
const targetsFinishingRunAgent = wakeAgentId === finishingRunAgentId;
|
|
const liveAndUnauthored = queryableCommentIds.filter((commentId) => {
|
|
const row = rows.find((candidate) => candidate.id === commentId);
|
|
return Boolean(row && !row.deletedAt && (!targetsFinishingRunAgent || row.createdByRunId !== finishingRunId));
|
|
});
|
|
// Promoting a saved wake whose comments a completed run already
|
|
// received would deliver them twice and can reopen finished work.
|
|
const delivered = await deliveredQueuedCommentIds(tx, { companyId, issueId, agentId: wakeAgentId,
|
|
commentIds: liveAndUnauthored });
|
|
const liveNonSelfCommentIds = liveAndUnauthored.filter(commentId => !delivered.has(commentId));
|
|
const containedSelfAuthoredComment = rows.some(
|
|
(row) => targetsFinishingRunAgent && !row.deletedAt && row.createdByRunId === finishingRunId,
|
|
);
|
|
return { liveNonSelfCommentIds, containedSelfAuthoredComment, deliveredCommentIds: [...delivered] };
|
|
},
|
|
|
|
async cancelDeferredWake({ companyId, wakeId, agentId, reason, now }) {
|
|
await notifyStrandedAgentStatus(tx, { companyId, agentId });
|
|
await notifyChatInboundWakeNotices(tx, { companyId });
|
|
const rows = await tx
|
|
.update(agentWakeupRequests)
|
|
.set({ status: "cancelled", finishedAt: now, error: reason, updatedAt: now })
|
|
.where(
|
|
and(
|
|
eq(agentWakeupRequests.id, wakeId),
|
|
eq(agentWakeupRequests.agentId, agentId),
|
|
eq(agentWakeupRequests.companyId, companyId),
|
|
eq(agentWakeupRequests.status, DEFERRED_WAKE_STATUS),
|
|
),
|
|
)
|
|
.returning({ id: agentWakeupRequests.id });
|
|
return rows.length > 0;
|
|
},
|
|
|
|
async normalizeDeferredWakeCommentIds({ companyId, wakeId, payload, liveCommentIds, now }) {
|
|
const rows = await tx
|
|
.update(agentWakeupRequests)
|
|
.set({ payload: withQueuedCommentIdsInWakePayload(payload, liveCommentIds), updatedAt: now })
|
|
.where(
|
|
and(
|
|
eq(agentWakeupRequests.id, wakeId),
|
|
eq(agentWakeupRequests.companyId, companyId),
|
|
eq(agentWakeupRequests.status, DEFERRED_WAKE_STATUS),
|
|
),
|
|
)
|
|
.returning();
|
|
const row = rows[0];
|
|
return row ? toDeferredWakeCandidate(row) : null;
|
|
},
|
|
|
|
async failDeferredWake({ companyId, wakeId, agentId, now }) {
|
|
await notifyStrandedAgentStatus(tx, { companyId, agentId });
|
|
await notifyChatInboundWakeNotices(tx, { companyId });
|
|
const rows = await tx
|
|
.update(agentWakeupRequests)
|
|
.set({
|
|
status: "failed",
|
|
finishedAt: now,
|
|
error: "Deferred wake could not be promoted: agent is not invokable",
|
|
updatedAt: now,
|
|
})
|
|
.where(
|
|
and(
|
|
eq(agentWakeupRequests.id, wakeId),
|
|
eq(agentWakeupRequests.agentId, agentId),
|
|
eq(agentWakeupRequests.companyId, companyId),
|
|
eq(agentWakeupRequests.status, DEFERRED_WAKE_STATUS),
|
|
),
|
|
)
|
|
.returning({ id: agentWakeupRequests.id });
|
|
return rows.length > 0;
|
|
},
|
|
|
|
async getPauseHoldFacts({ companyId, issueId, wakeAgentId, deferredContextSeed, requestedByActorType, requestedByActorId }) {
|
|
const activePauseHold = await treeControlSvc.getActivePauseHoldGate(companyId, issueId);
|
|
if (!activePauseHold) {
|
|
return {
|
|
activePauseHold: false,
|
|
treeHoldInteractionWake: false,
|
|
holdId: null,
|
|
rootIssueId: null,
|
|
mode: null,
|
|
reason: null,
|
|
releasePolicy: null,
|
|
};
|
|
}
|
|
const treeHoldInteractionWake = await isVerifiedIssueTreeControlInteractionWake(tx, {
|
|
companyId,
|
|
issueId,
|
|
agentId: wakeAgentId,
|
|
contextSnapshot: deferredContextSeed,
|
|
requestedByActorType,
|
|
requestedByActorId,
|
|
});
|
|
return {
|
|
activePauseHold: true,
|
|
treeHoldInteractionWake,
|
|
holdId: activePauseHold.holdId,
|
|
rootIssueId: activePauseHold.rootIssueId,
|
|
mode: activePauseHold.mode,
|
|
reason: activePauseHold.reason,
|
|
releasePolicy: activePauseHold.releasePolicy,
|
|
};
|
|
},
|
|
|
|
async isOwedInteractionAnswer({ companyId, issueId, payload }) {
|
|
const response = await readQueuedInteractionResponse(tx as unknown as Db, companyId, issueId, payload);
|
|
if (!response || response.toolActionOutcome || !response.comment.authorUserId) return false;
|
|
const [task] = await tx.select({ completedAt: issues.completedAt }).from(issues)
|
|
.where(and(eq(issues.companyId, companyId), eq(issues.id, issueId)));
|
|
const answeredAt = new Date(response.comment.createdAt);
|
|
return !!task && (!task.completedAt || answeredAt.getTime() <= task.completedAt.getTime());
|
|
},
|
|
|
|
async getCommentSelfAuthorship({ companyId, issueId, finishingRunId, commentIds }) {
|
|
const rows = await tx
|
|
.select({ createdByRunId: issueComments.createdByRunId })
|
|
.from(issueComments)
|
|
.where(and(eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId), inArray(issueComments.id, commentIds)));
|
|
return { allSelfAuthored: rows.length > 0 && rows.every((row) => row.createdByRunId === finishingRunId) };
|
|
},
|
|
|
|
async isCompletedDelegationMention({ companyId, issueId, finishingRunId, wakeAgentId, commentIds }) {
|
|
if (
|
|
run.companyId !== companyId || run.id !== finishingRunId ||
|
|
readNonEmptyString(parseObject(run.contextSnapshot).issueId) !== issueId
|
|
) return false;
|
|
const uniqueCommentIds = [...new Set(commentIds)];
|
|
if (uniqueCommentIds.length === 0) return false;
|
|
const parent = await tx.select({ status: issues.status, assigneeAgentId: issues.assigneeAgentId, identifier: issues.identifier })
|
|
.from(issues).where(and(eq(issues.companyId, companyId), eq(issues.id, issueId)))
|
|
.then((rows) => rows[0]);
|
|
if (parent?.status !== "done" || parent.assigneeAgentId !== run.agentId) return false;
|
|
|
|
const comments = await tx.select({ body: issueComments.body, createdByRunId: issueComments.createdByRunId,
|
|
authorAgentId: issueComments.authorAgentId })
|
|
.from(issueComments).where(and(
|
|
eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId),
|
|
inArray(issueComments.id, uniqueCommentIds), isNull(issueComments.deletedAt), currentConversationCommentCondition(),
|
|
));
|
|
if (comments.length !== uniqueCommentIds.length || comments.some((comment) =>
|
|
comment.createdByRunId !== run.id || (comment.authorAgentId !== null && comment.authorAgentId !== run.agentId)
|
|
)) return false;
|
|
const referencesByComment = comments.map((comment) => extractIssueReferenceIdentifiers(comment.body)
|
|
.filter((identifier) => identifier !== parent.identifier));
|
|
// A parent link is context; every other reference must identify the
|
|
// single completed child. Never discard unrelated follow-up work.
|
|
if (referencesByComment.some((references) => references.length !== 1)) return false;
|
|
const identifiers = [...new Set(referencesByComment.flat())];
|
|
const children = await tx.select({ identifier: issues.identifier, status: issues.status })
|
|
.from(issues).where(and(
|
|
eq(issues.companyId, companyId), eq(issues.parentId, issueId),
|
|
eq(issues.assigneeAgentId, wakeAgentId), inArray(issues.identifier, identifiers),
|
|
));
|
|
return referencesByComment.every((references) => {
|
|
const referencedChildren = children.filter((child) => child.identifier !== null && references.includes(child.identifier));
|
|
return referencedChildren.length === 1 && referencedChildren[0].status === "done";
|
|
});
|
|
},
|
|
|
|
async isCompletedOnboardingHandoffWake(input) {
|
|
return isCompletedOnboardingHandoffWake(tx, input);
|
|
},
|
|
|
|
async reopenIssue({ companyId, issueId, status = "todo" }) {
|
|
const updated = await issuesSvc.updateForCompany(issueId, companyId, { status, executionState: null }, tx);
|
|
return updated ? toIssueSnapshot(updated as unknown as IssueRow) : null;
|
|
},
|
|
|
|
async claimDeferredWakeForPromotion({ companyId, wakeId, now }) {
|
|
const claimed = await tx
|
|
.update(agentWakeupRequests)
|
|
.set({
|
|
status: "queued",
|
|
reason: "issue_execution_promoted",
|
|
claimedAt: null,
|
|
finishedAt: null,
|
|
error: null,
|
|
updatedAt: now,
|
|
})
|
|
.where(
|
|
and(
|
|
eq(agentWakeupRequests.id, wakeId),
|
|
eq(agentWakeupRequests.companyId, companyId),
|
|
eq(agentWakeupRequests.status, DEFERRED_WAKE_STATUS),
|
|
),
|
|
)
|
|
.returning({ id: agentWakeupRequests.id });
|
|
return claimed.length > 0;
|
|
},
|
|
|
|
async finalizePromotedWake(input) {
|
|
await notifyChatRunMilestones(tx, { companyId: input.companyId, issueId: input.issue.id });
|
|
const newRun = await tx
|
|
.insert(heartbeatRuns)
|
|
.values({
|
|
companyId: input.deferredAgent.companyId,
|
|
agentId: input.deferredAgent.id,
|
|
invocationSource: input.source,
|
|
triggerDetail: input.triggerDetail,
|
|
status: "queued",
|
|
wakeupRequestId: input.wakeId,
|
|
retryOfRunId: input.authorizedFailedChatRetry
|
|
? readNonEmptyString(input.contextSnapshot.retryOfRunId)
|
|
: null,
|
|
contextSnapshot: input.contextSnapshot,
|
|
responsibleUserId: input.responsibleUserId,
|
|
sessionIdBefore: input.sessionBefore,
|
|
continuationAttempt: readContinuationAttempt(input.contextSnapshot.livenessContinuationAttempt),
|
|
})
|
|
.returning()
|
|
.then((rows) => rows[0]);
|
|
|
|
// `claimDeferredWakeForPromotion` already moved this row off
|
|
// `deferred_issue_execution` inside this same transaction, so no
|
|
// concurrent claimer can still match that guard; this extra `runId is
|
|
// null` guard only protects against writing the link twice.
|
|
await tx
|
|
.update(agentWakeupRequests)
|
|
.set({ runId: newRun.id, updatedAt: input.now })
|
|
.where(
|
|
and(
|
|
eq(agentWakeupRequests.id, input.wakeId),
|
|
eq(agentWakeupRequests.companyId, input.companyId),
|
|
isNull(agentWakeupRequests.runId),
|
|
),
|
|
);
|
|
|
|
// Promoted mention wakes are issue-scoped, not issue ownership
|
|
// transfers. The lock-clearing step earlier in this transaction
|
|
// already set `executionRunId` to null for this issue, so the `is
|
|
// null` guard only protects against taking the lock twice.
|
|
await tx
|
|
.update(issues)
|
|
.set({
|
|
executionRunId: newRun.id,
|
|
executionAgentNameKey: normalizeAgentNameKey(input.deferredAgent.name),
|
|
executionLockedAt: input.now,
|
|
updatedAt: input.now,
|
|
})
|
|
.where(
|
|
and(
|
|
eq(issues.id, input.issue.id),
|
|
eq(issues.companyId, input.companyId),
|
|
eq(issues.assigneeAgentId, input.deferredAgent.id),
|
|
isNull(issues.executionRunId),
|
|
),
|
|
);
|
|
|
|
return toRunSummary(newRun);
|
|
},
|
|
|
|
async hasExistingExecutionPath({ companyId, issueId, excludeRunId, agentId }) {
|
|
const row = await tx
|
|
.select({ id: heartbeatRuns.id })
|
|
.from(heartbeatRuns)
|
|
.where(
|
|
and(
|
|
eq(heartbeatRuns.companyId, companyId),
|
|
inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES]),
|
|
sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`,
|
|
sql`${heartbeatRuns.id} <> ${excludeRunId}`,
|
|
agentId ? eq(heartbeatRuns.agentId, agentId) : sql`true`,
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
return row !== null;
|
|
},
|
|
|
|
async hasExplicitBlockerPath({ companyId, issueId }) {
|
|
const row = await tx
|
|
.select({ issueId: issueRelations.issueId })
|
|
.from(issueRelations)
|
|
.innerJoin(issues, eq(issueRelations.issueId, issues.id))
|
|
.where(
|
|
and(
|
|
eq(issueRelations.companyId, companyId),
|
|
eq(issueRelations.relatedIssueId, issueId),
|
|
eq(issueRelations.type, "blocks"),
|
|
eq(issues.companyId, companyId),
|
|
notInArray(issues.status, ["done", "cancelled"]),
|
|
isNull(issues.hiddenAt),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
return row !== null;
|
|
},
|
|
|
|
async isAutomaticRecoverySuppressedByPauseHold({ companyId, issueId }) {
|
|
return isAutomaticRecoverySuppressedByPauseHold(tx, companyId, issueId, treeControlSvc);
|
|
},
|
|
|
|
async isImmediateRecoverySourceBlocked({ companyId, runId }) {
|
|
if (companyId !== run.companyId || runId !== run.id) {
|
|
throw new Error(
|
|
"wake-queue: recovery source does not match the locked execution",
|
|
);
|
|
}
|
|
const failedChatRequestOwner = run.wakeupRequestId
|
|
? await tx
|
|
.select({ id: chatActions.id })
|
|
.from(chatActions)
|
|
.where(
|
|
and(
|
|
eq(chatActions.id, run.wakeupRequestId),
|
|
eq(chatActions.companyId, run.companyId),
|
|
inArray(chatActions.kind, ["inbound_wakeup", "failed_run_retry"]),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null)
|
|
: null;
|
|
return (
|
|
Boolean(failedChatRequestOwner) ||
|
|
run.errorCode === "chat_failed_run_retry_not_authorized" ||
|
|
classifyContinuationFailure(run).kind === "non_retryable"
|
|
);
|
|
},
|
|
|
|
async queueReviewParticipantRecoveryRun({ companyId, issue, finishingRun, recoveryAgent, sessionBefore, now }) {
|
|
await notifyStatusCardIssueChange(tx, companyId);
|
|
const executionState = parseIssueExecutionState(issue.executionState);
|
|
const wakeupRequest = await tx
|
|
.insert(agentWakeupRequests)
|
|
.values({
|
|
companyId,
|
|
agentId: recoveryAgent.id,
|
|
source: "automation",
|
|
triggerDetail: "system",
|
|
reason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON,
|
|
payload: withRecoveryContext(
|
|
{
|
|
issueId: issue.id,
|
|
retryOfRunId: finishingRun.id,
|
|
retryReason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON,
|
|
currentStageId: executionState?.currentStageId ?? null,
|
|
currentStageType: executionState?.currentStageType ?? null,
|
|
},
|
|
"normal_model",
|
|
),
|
|
status: "queued",
|
|
requestedByActorType: "system",
|
|
requestedByActorId: null,
|
|
updatedAt: now,
|
|
})
|
|
.returning()
|
|
.then((rows) => rows[0]);
|
|
|
|
await notifyRunDispatch(tx, { companyId });
|
|
const queuedRun = await tx
|
|
.insert(heartbeatRuns)
|
|
.values({
|
|
companyId,
|
|
agentId: recoveryAgent.id,
|
|
invocationSource: "automation",
|
|
triggerDetail: "system",
|
|
status: "queued",
|
|
wakeupRequestId: wakeupRequest.id,
|
|
contextSnapshot: withRecoveryContext(
|
|
{
|
|
issueId: issue.id,
|
|
taskId: issue.id,
|
|
wakeReason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON,
|
|
retryReason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON,
|
|
source: "issue.execution_review_recovery",
|
|
retryOfRunId: finishingRun.id,
|
|
currentStageId: executionState?.currentStageId ?? null,
|
|
currentStageType: executionState?.currentStageType ?? null,
|
|
reviewRecoveryInstruction:
|
|
"The previous reviewer run ended while this execution-review stage was still pending. Submit the review decision now, or mark the issue blocked with the exact unblock action.",
|
|
},
|
|
"normal_model",
|
|
),
|
|
sessionIdBefore: sessionBefore,
|
|
retryOfRunId: finishingRun.id,
|
|
updatedAt: now,
|
|
})
|
|
.returning()
|
|
.then((rows) => rows[0]);
|
|
|
|
await tx
|
|
.update(agentWakeupRequests)
|
|
.set({ runId: queuedRun.id, updatedAt: now })
|
|
.where(and(eq(agentWakeupRequests.id, wakeupRequest.id), eq(agentWakeupRequests.companyId, companyId)));
|
|
|
|
await tx
|
|
.update(issues)
|
|
.set({
|
|
executionRunId: queuedRun.id,
|
|
executionAgentNameKey: normalizeAgentNameKey(recoveryAgent.name),
|
|
executionLockedAt: now,
|
|
updatedAt: now,
|
|
})
|
|
.where(and(eq(issues.id, issue.id), eq(issues.companyId, companyId)));
|
|
|
|
return toRunSummary(queuedRun);
|
|
},
|
|
|
|
async queueImmediateRecoveryRun({
|
|
companyId,
|
|
issue,
|
|
finishingRun,
|
|
recoveryAgent,
|
|
reason,
|
|
contextSnapshot,
|
|
responsibleUserId,
|
|
sessionBefore,
|
|
now,
|
|
}) {
|
|
await notifyChatRunMilestones(tx, { companyId, issueId: issue.id });
|
|
const wakeupRequest = await tx
|
|
.insert(agentWakeupRequests)
|
|
.values({
|
|
companyId,
|
|
agentId: recoveryAgent.id,
|
|
source: "automation",
|
|
triggerDetail: "system",
|
|
reason,
|
|
payload: withRecoveryContext({ issueId: issue.id, retryOfRunId: finishingRun.id }, "normal_model"),
|
|
status: "queued",
|
|
requestedByActorType: "system",
|
|
requestedByActorId: null,
|
|
updatedAt: now,
|
|
})
|
|
.returning()
|
|
.then((rows) => rows[0]);
|
|
|
|
await notifyRunDispatch(tx, { companyId });
|
|
const queuedRun = await tx
|
|
.insert(heartbeatRuns)
|
|
.values({
|
|
companyId,
|
|
agentId: recoveryAgent.id,
|
|
invocationSource: "automation",
|
|
triggerDetail: "system",
|
|
status: "queued",
|
|
wakeupRequestId: wakeupRequest.id,
|
|
contextSnapshot,
|
|
responsibleUserId,
|
|
sessionIdBefore: sessionBefore,
|
|
retryOfRunId: finishingRun.id,
|
|
updatedAt: now,
|
|
})
|
|
.returning()
|
|
.then((rows) => rows[0]);
|
|
|
|
await tx
|
|
.update(agentWakeupRequests)
|
|
.set({ runId: queuedRun.id, updatedAt: now })
|
|
.where(and(eq(agentWakeupRequests.id, wakeupRequest.id), eq(agentWakeupRequests.companyId, companyId)));
|
|
|
|
await tx
|
|
.update(issues)
|
|
.set({
|
|
executionRunId: queuedRun.id,
|
|
executionAgentNameKey: normalizeAgentNameKey(recoveryAgent.name),
|
|
executionLockedAt: now,
|
|
updatedAt: now,
|
|
})
|
|
.where(and(eq(issues.id, issue.id), eq(issues.companyId, companyId)));
|
|
|
|
return toRunSummary(queuedRun);
|
|
},
|
|
};
|
|
}
|
|
|
|
async function recordNativeTerminalRecoveryIfNeeded(tx: Db, run: ReleaseRun, issue: IssueRow, now: Date): Promise<boolean> {
|
|
const applies =
|
|
run.runtimeMode === "native" &&
|
|
["failed", "timed_out", "interrupted", "cancelled"].includes(run.status) &&
|
|
issue.assigneeAgentId === run.agentId &&
|
|
!["done", "cancelled"].includes(issue.status);
|
|
if (!applies || isAcknowledgedNativeStop(run) || isAcknowledgedNativeReassignmentStop(run)) return false;
|
|
// A pause's stop is intentional: the task keeps its status and resuming
|
|
// continues it. It is never a reconciliation incident with no control. A
|
|
// termination that replaced the pause shares its stop reason, so the agent
|
|
// must still be pausable, not terminating or terminated.
|
|
if (isGovernancePauseStop(run)) {
|
|
const [agent] = await tx.select({ lifecycleState: agents.lifecycleState }).from(agents)
|
|
.where(and(eq(agents.id, run.agentId), eq(agents.companyId, issue.companyId)));
|
|
if (agent && !["terminating", "cleaning_up", "terminated", "rejected"].includes(agent.lifecycleState)) return false;
|
|
}
|
|
|
|
const existing = await tx
|
|
.select({ id: issueRecoveryActions.id, evidence: issueRecoveryActions.evidence })
|
|
.from(issueRecoveryActions)
|
|
.where(
|
|
and(
|
|
eq(issueRecoveryActions.companyId, issue.companyId),
|
|
eq(issueRecoveryActions.sourceIssueId, issue.id),
|
|
or(
|
|
inArray(issueRecoveryActions.status, ["active", "escalated"]),
|
|
sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'runId' = ${run.id}`,
|
|
),
|
|
),
|
|
)
|
|
.limit(1);
|
|
let nativeFailureBlock: { runId: string; statusVersion: number } | undefined;
|
|
if (issue.status !== "blocked") {
|
|
const projected = await issueService(tx).update(issue.id, { status: "blocked" }, tx);
|
|
if (projected) {
|
|
nativeFailureBlock = { runId: run.id, statusVersion: projected.statusVersion };
|
|
await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, `task:${issue.id}`);
|
|
await tx.insert(activityLog).values({
|
|
companyId: issue.companyId, actorType: "system", actorId: "execution-recovery",
|
|
action: "issue.updated", entityType: "issue", entityId: issue.id, runId: run.id,
|
|
details: { status: "blocked", previousStatus: issue.status, reason: "native_continuation_requires_reconciliation" },
|
|
});
|
|
}
|
|
}
|
|
// Status projection is required even when restart/finalization created the
|
|
// incident first. Preserve its owner, cause, retry budget, and prior evidence.
|
|
if (nativeFailureBlock) {
|
|
for (const action of existing) {
|
|
await tx.update(issueRecoveryActions).set({
|
|
evidence: { ...action.evidence, nativeFailureBlock }, updatedAt: now,
|
|
}).where(and(eq(issueRecoveryActions.id, action.id), eq(issueRecoveryActions.companyId, issue.companyId)));
|
|
}
|
|
}
|
|
if (!existing.length) {
|
|
await tx
|
|
.update(nativeRunFinalizations)
|
|
.set({
|
|
phase: "terminal_failure",
|
|
leaseOwner: null,
|
|
leaseExpiresAt: null,
|
|
nextAttemptAt: null,
|
|
recoveryState: "blocked",
|
|
failureCode: "native_continuation_requires_reconciliation",
|
|
updatedAt: now,
|
|
})
|
|
.where(
|
|
and(
|
|
eq(nativeRunFinalizations.companyId, issue.companyId),
|
|
eq(nativeRunFinalizations.runId, run.id),
|
|
isNull(nativeRunFinalizations.resultId),
|
|
),
|
|
);
|
|
await issueRecoveryActionService(tx).upsertSourceScoped({
|
|
companyId: issue.companyId,
|
|
sourceIssueId: issue.id,
|
|
kind: "active_run_watchdog",
|
|
ownerType: "board",
|
|
returnOwnerAgentId: run.agentId,
|
|
cause: "native_continuation_requires_reconciliation",
|
|
fingerprint: `native-continuation:${run.id}`,
|
|
evidence: { runId: run.id, originalFailureCode: run.errorCode, ...(nativeFailureBlock ? { nativeFailureBlock } : {}) },
|
|
nextAction:
|
|
"Inspect the original failure and reconcile the previous execution before continuing. Automatic recovery cannot start another incident.",
|
|
maxAttempts: 3,
|
|
wakePolicy: null,
|
|
supersedeOnIdentityChange: true,
|
|
});
|
|
}
|
|
return true;
|
|
}
|
|
|
|
/**
|
|
* Builds the temporary transaction-scope handle the admission port needs.
|
|
* `heartbeat.ts` calls this through `createWakeQueue`'s own wrapper; it
|
|
* never builds a `TransactionScope` itself.
|
|
*/
|
|
export function createAdmissionTransactionScope(companyId: string, tx: Db): TransactionScope {
|
|
return TransactionScope.create(companyId, tx);
|
|
}
|
|
|
|
function requireAdmissionTx(scope: TransactionScope | null | undefined, companyId: string): Db {
|
|
return requireTransactionScopeTx(scope, companyId) as Db;
|
|
}
|
|
|
|
export function createWakeAdmissionReader(): WakeAdmissionReader {
|
|
return {
|
|
async matchesActiveWakeActor(scope, input) {
|
|
const tx = requireAdmissionTx(scope, input.companyId);
|
|
if (!input.wakeupRequestId) return false;
|
|
const actor = await tx
|
|
.select({
|
|
type: agentWakeupRequests.requestedByActorType,
|
|
id: agentWakeupRequests.requestedByActorId,
|
|
})
|
|
.from(agentWakeupRequests)
|
|
.where(
|
|
and(
|
|
eq(agentWakeupRequests.companyId, input.companyId),
|
|
eq(agentWakeupRequests.id, input.wakeupRequestId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
return (
|
|
actor !== null &&
|
|
actor.type === input.requestedByActorType &&
|
|
actor.id === input.requestedByActorId
|
|
);
|
|
},
|
|
async isSameExecutionAgent(
|
|
scope,
|
|
{
|
|
companyId,
|
|
activeExecutionRunAgentId,
|
|
issueExecutionAgentNameKey,
|
|
agentNameKey,
|
|
},
|
|
) {
|
|
const tx = requireAdmissionTx(scope, companyId);
|
|
const executionAgent = await tx
|
|
.select({ name: agents.name })
|
|
.from(agents)
|
|
.where(
|
|
and(
|
|
eq(agents.id, activeExecutionRunAgentId),
|
|
eq(agents.companyId, companyId),
|
|
),
|
|
)
|
|
.then((rows) => rows[0] ?? null);
|
|
const executionAgentNameKey =
|
|
normalizeAgentNameKey(issueExecutionAgentNameKey) ??
|
|
normalizeAgentNameKey(executionAgent?.name);
|
|
return (
|
|
Boolean(executionAgentNameKey) && executionAgentNameKey === agentNameKey
|
|
);
|
|
},
|
|
|
|
async findExistingDeferredWake(
|
|
scope,
|
|
{ companyId, agentId, issueId, chatConversationId, durableActor },
|
|
) {
|
|
const tx = requireAdmissionTx(scope, companyId);
|
|
const row = await tx
|
|
.select()
|
|
.from(agentWakeupRequests)
|
|
.where(
|
|
and(
|
|
eq(agentWakeupRequests.companyId, companyId),
|
|
eq(agentWakeupRequests.agentId, agentId),
|
|
eq(agentWakeupRequests.status, DEFERRED_WAKE_STATUS),
|
|
sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`,
|
|
// A saved card response (interaction receipt) never absorbs user
|
|
// input. Skip it so a later message joins the existing message
|
|
// wake instead of starting another one behind the receipt.
|
|
sql`${agentWakeupRequests.payload} ->> 'interactionId' is null`,
|
|
sql`${agentWakeupRequests.payload} -> ${DEFERRED_WAKE_CONTEXT_KEY}::text ->> 'interactionId' is null`,
|
|
...(chatConversationId
|
|
? [
|
|
sql`coalesce(${agentWakeupRequests.payload} -> ${DEFERRED_WAKE_CONTEXT_KEY}::text ->> ${CHAT_CONVERSATION_ID_KEY}::text, ${chatConversationId}) = ${chatConversationId}`,
|
|
// A wake with no recorded conversation can still hold
|
|
// another conversation's messages (an outbox-saved wake).
|
|
deferredWakeHoldsNoOtherConversation(companyId, chatConversationId),
|
|
]
|
|
: []),
|
|
...(durableActor
|
|
? [
|
|
durableActor.type === null
|
|
? isNull(agentWakeupRequests.requestedByActorType)
|
|
: eq(
|
|
agentWakeupRequests.requestedByActorType,
|
|
durableActor.type,
|
|
),
|
|
durableActor.id === null
|
|
? isNull(agentWakeupRequests.requestedByActorId)
|
|
: eq(
|
|
agentWakeupRequests.requestedByActorId,
|
|
durableActor.id,
|
|
),
|
|
]
|
|
: []),
|
|
),
|
|
)
|
|
.orderBy(asc(agentWakeupRequests.requestedAt))
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!row) return null;
|
|
const payload = parseObject(row.payload);
|
|
return {
|
|
id: row.id,
|
|
runId: row.runId,
|
|
payload,
|
|
deferredContext: parseObject(payload[DEFERRED_WAKE_CONTEXT_KEY]),
|
|
coalescedCount: row.coalescedCount,
|
|
};
|
|
},
|
|
};
|
|
}
|
|
|
|
export function createWakeAdmissionWriter(): WakeAdmissionWriter {
|
|
return {
|
|
async coalesceIntoActiveExecutionRun(scope, input) {
|
|
const tx = requireAdmissionTx(scope, input.companyId);
|
|
const now = new Date();
|
|
const [candidate] = await tx.select().from(heartbeatRuns).where(and(
|
|
eq(heartbeatRuns.id, input.activeExecutionRunId), eq(heartbeatRuns.companyId, input.companyId),
|
|
)).limit(1);
|
|
if (!candidate) throw new Error("wake-queue: the coalesce target run was not found for this company");
|
|
if (candidate.agentId !== input.agentId) return null;
|
|
const [primary] = candidate.wakeupRequestId ? await tx.select().from(agentWakeupRequests).where(and(
|
|
eq(agentWakeupRequests.id, candidate.wakeupRequestId), eq(agentWakeupRequests.companyId, input.companyId),
|
|
eq(agentWakeupRequests.agentId, input.agentId),
|
|
)).for("update").limit(1) : [];
|
|
const [current] = await tx.select().from(heartbeatRuns).where(and(
|
|
eq(heartbeatRuns.id, candidate.id), eq(heartbeatRuns.companyId, input.companyId),
|
|
eq(heartbeatRuns.agentId, input.agentId),
|
|
)).for("update").limit(1);
|
|
if (!current || current.status !== input.activeExecutionRunStatus || current.wakeupRequestId !== candidate.wakeupRequestId) return null;
|
|
const commentIds = queuedCommentIdsFromRunContext(input.mergedContextSnapshot);
|
|
// A reviewed chat run attests one conversation. Never fold a message
|
|
// into a run that already holds another conversation's messages; the
|
|
// wake is saved behind the run instead.
|
|
const mergedConversationId = readWakeChatConversationId(input.mergedContextSnapshot);
|
|
if (mergedConversationId && commentIds.length > 0 && (await otherConversationCommentIds(tx, {
|
|
companyId: input.companyId, conversationId: mergedConversationId, commentIds })).length > 0) return null;
|
|
const synchronizeBatch = current.status === "queued" && commentIds.length > 0
|
|
&& primary?.status === "queued" && queuedCommentIdsFromWakePayload(primary.payload).length > 0;
|
|
if (current.status === "queued" && commentIds.length && current.wakeupRequestId &&
|
|
(!primary || !["queued", "claimed"].includes(primary.status) || primary.runId !== current.id
|
|
|| (input.durableReceipt && (primary.requestedByActorType !== input.requestedByActorType || primary.requestedByActorId !== input.requestedByActorId))
|
|
|| primary.payload?.issueId !== input.mergedContextSnapshot.issueId)) return null;
|
|
await notifyChatRunMilestones(tx, { companyId: input.companyId, runId: input.activeExecutionRunId });
|
|
if (synchronizeBatch && primary) await tx.update(agentWakeupRequests).set({
|
|
payload: withQueuedCommentIdsInWakePayload(primary.payload, commentIds, { preserveSource: true }), updatedAt: now,
|
|
}).where(and(eq(agentWakeupRequests.id, primary.id), eq(agentWakeupRequests.companyId, input.companyId),
|
|
eq(agentWakeupRequests.agentId, input.agentId), eq(agentWakeupRequests.runId, current.id), eq(agentWakeupRequests.status, primary.status)));
|
|
const mergedRun = await tx
|
|
.update(heartbeatRuns)
|
|
.set({ contextSnapshot: input.mergedContextSnapshot, updatedAt: now })
|
|
.where(
|
|
and(
|
|
eq(heartbeatRuns.id, input.activeExecutionRunId),
|
|
eq(heartbeatRuns.companyId, input.companyId),
|
|
),
|
|
)
|
|
.returning()
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!mergedRun) {
|
|
// The compare-and-set write affected no row. Throw to roll the
|
|
// transaction back instead of recording a coalesced wake against a
|
|
// run this write never touched.
|
|
throw new Error(
|
|
"wake-queue: the coalesce target run was not found for this company",
|
|
);
|
|
}
|
|
await tx.insert(agentWakeupRequests).values({
|
|
...input.durableReceipt,
|
|
companyId: input.companyId,
|
|
agentId: input.agentId,
|
|
source: input.source,
|
|
triggerDetail: input.triggerDetail,
|
|
reason: "issue_execution_same_name",
|
|
payload: input.payload,
|
|
status: "coalesced",
|
|
coalescedCount: 1,
|
|
requestedByActorType: input.requestedByActorType,
|
|
requestedByActorId: input.requestedByActorId,
|
|
idempotencyKey: input.idempotencyKey,
|
|
runId: mergedRun.id,
|
|
finishedAt: now,
|
|
});
|
|
return mergedRun as unknown as Record<string, unknown>;
|
|
},
|
|
|
|
async mergeIntoExistingDeferredWake(scope, input) {
|
|
const tx = requireAdmissionTx(scope, input.companyId);
|
|
if (input.manualUserWakeActorId) await notifyChatInboundWakeNotices(tx, { companyId: input.companyId });
|
|
const [prior] = await tx.select().from(agentWakeupRequests).where(and(
|
|
eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.id, input.existingDeferredWakeId),
|
|
eq(agentWakeupRequests.status, DEFERRED_WAKE_STATUS),
|
|
)).for("update");
|
|
if (!prior) throw new Error("wake-queue: the deferred wake to merge into was not found for this company");
|
|
const priorPayload = parseObject(prior.payload), issueId = readNonEmptyString(priorPayload.issueId);
|
|
const previousComments = new Set(queuedCommentIdsFromWakePayload(priorPayload));
|
|
const newComments = queuedCommentIdsFromWakePayload(input.mergedPayload).some(id => !previousComments.has(id));
|
|
const newInteraction = readNonEmptyString(input.mergedPayload.interactionId);
|
|
if (newComments || newInteraction && newInteraction !== readNonEmptyString(priorPayload.interactionId))
|
|
await notifyDeferredExecutionWork(tx, { companyId: input.companyId, requestId: prior.id });
|
|
const priorCommentId = readNonEmptyString(priorPayload.commentId), priorActor = readNonEmptyString(prior.requestedByActorId);
|
|
const [priorComment] = prior.requestedByActorType === "user" && priorActor && issueId && isUuidLike(issueId) &&
|
|
priorCommentId && isUuidLike(priorCommentId) ? await tx.select({ id: issueComments.id, body: issueComments.body }).from(issueComments).where(and(
|
|
eq(issueComments.companyId, input.companyId), eq(issueComments.issueId, issueId), eq(issueComments.id, priorCommentId),
|
|
eq(issueComments.authorType, "user"), eq(issueComments.authorUserId, priorActor),
|
|
isNull(issueComments.deletedAt), isNull(issueComments.createdByRunId),
|
|
)) : [];
|
|
if ((input.coalescedReceipt || priorComment?.body.trim()) &&
|
|
(readNonEmptyString(input.mergedPayload.issueId) !== issueId || (input.coalescedReceipt &&
|
|
(input.coalescedReceipt.agentId !== prior.agentId || readNonEmptyString(input.coalescedReceipt.payload.issueId) !== issueId))))
|
|
throw new Error("wake-queue: merged receipt does not match the saved queue scope");
|
|
if (priorComment?.body.trim()) {
|
|
const [recorded] = await tx.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests).where(and(
|
|
eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.agentId, prior.agentId),
|
|
eq(agentWakeupRequests.status, "coalesced"), eq(agentWakeupRequests.requestedByActorType, "user"),
|
|
eq(agentWakeupRequests.requestedByActorId, priorActor!),
|
|
prior.runId ? eq(agentWakeupRequests.runId, prior.runId) : isNull(agentWakeupRequests.runId),
|
|
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
|
|
sql`${agentWakeupRequests.payload}->>'commentId' = ${priorComment.id}`,
|
|
sql`${agentWakeupRequests.payload}->>'coalescedIntoWakeupRequestId' = ${prior.id}`,
|
|
)).limit(1);
|
|
if (!recorded) await tx.insert(agentWakeupRequests).values({
|
|
companyId: input.companyId, agentId: prior.agentId, source: prior.source, triggerDetail: prior.triggerDetail,
|
|
reason: prior.reason, payload: { issueId, commentId: priorComment.id, coalescedIntoWakeupRequestId: prior.id },
|
|
status: "coalesced", coalescedCount: 1, requestedByActorType: "user", requestedByActorId: priorActor,
|
|
requestedAt: prior.requestedAt, finishedAt: new Date(), runId: prior.runId,
|
|
});
|
|
}
|
|
const rows = await tx
|
|
.update(agentWakeupRequests)
|
|
.set({
|
|
payload: input.mergedPayload,
|
|
coalescedCount: input.nextCoalescedCount,
|
|
...(input.manualUserWakeActorId ? {
|
|
requestedByActorType: "user",
|
|
requestedByActorId: input.manualUserWakeActorId,
|
|
} : {}),
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(
|
|
and(
|
|
eq(agentWakeupRequests.id, input.existingDeferredWakeId),
|
|
eq(agentWakeupRequests.companyId, input.companyId),
|
|
eq(agentWakeupRequests.status, DEFERRED_WAKE_STATUS),
|
|
),
|
|
)
|
|
.returning({ id: agentWakeupRequests.id });
|
|
if (rows.length === 0) {
|
|
// The compare-and-set write affected no row: a concurrent writer
|
|
// already moved this wake off `deferred_issue_execution`. Roll the
|
|
// transaction back instead of leaving the merge half-applied.
|
|
throw new Error(
|
|
"wake-queue: the deferred wake to merge into was not found for this company",
|
|
);
|
|
}
|
|
if (input.coalescedReceipt) {
|
|
await tx.insert(agentWakeupRequests).values({
|
|
...input.coalescedReceipt,
|
|
companyId: input.companyId,
|
|
agentId: prior.agentId,
|
|
runId: prior.runId,
|
|
payload: { ...input.coalescedReceipt.payload, coalescedIntoWakeupRequestId: prior.id },
|
|
status: "coalesced",
|
|
coalescedCount: 1,
|
|
finishedAt: new Date(),
|
|
});
|
|
}
|
|
},
|
|
|
|
async insertNewDeferredWake(scope, input) {
|
|
const tx = requireAdmissionTx(scope, input.companyId);
|
|
await notifyDeferredExecutionWork(tx, { companyId: input.companyId });
|
|
await tx.insert(agentWakeupRequests).values({
|
|
...input.durableReceipt,
|
|
companyId: input.companyId,
|
|
agentId: input.agentId,
|
|
source: input.source,
|
|
triggerDetail: input.triggerDetail,
|
|
reason: "issue_execution_deferred",
|
|
payload: input.payload,
|
|
status: DEFERRED_WAKE_STATUS,
|
|
requestedByActorType: input.requestedByActorType,
|
|
requestedByActorId: input.requestedByActorId,
|
|
idempotencyKey: input.idempotencyKey,
|
|
});
|
|
},
|
|
};
|
|
}
|
|
|
|
export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAdapterDeps): IssueLockWriter {
|
|
return {
|
|
async withIssueExecutionLock(input, fn): Promise<ReleaseTransactionResult & { run: RunSnapshot }> {
|
|
return db.transaction(async (rawTx) => {
|
|
const tx = rawTx as unknown as Db;
|
|
const evidence = await readReleaseRun(tx, input);
|
|
const run = evidence.run;
|
|
if (!run) {
|
|
throw new Error(`wake-queue: run ${input.runId} was not found while releasing issue execution`);
|
|
}
|
|
const runSnapshot = toRunSnapshot(run);
|
|
if (input.resumeStrandedQueue && (!run.finishedAt || !["succeeded", "failed", "cancelled", "timed_out", "interrupted"].includes(run.status))) {
|
|
return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot };
|
|
}
|
|
const contextIssueId = readNonEmptyString(parseObject(run.contextSnapshot).issueId);
|
|
|
|
// Lock the context issue (if any) and every issue that still references this
|
|
// run, in id order, so two concurrent finalizations can never deadlock on
|
|
// each other's row-lock acquisition order.
|
|
await tx.execute(
|
|
contextIssueId
|
|
? sql`
|
|
select id from issues
|
|
where company_id = ${input.companyId}
|
|
and (
|
|
id = ${contextIssueId}
|
|
or execution_run_id = ${run.id}
|
|
or checkout_run_id = ${run.id}
|
|
)
|
|
order by id
|
|
for update
|
|
`
|
|
: sql`
|
|
select id from issues
|
|
where company_id = ${input.companyId}
|
|
and (execution_run_id = ${run.id} or checkout_run_id = ${run.id})
|
|
order by id
|
|
for update
|
|
`,
|
|
);
|
|
|
|
if (input.resumeStrandedQueue && run.runtimeMode === "native") {
|
|
const [finalized] = contextIssueId && run.nativeIssueId === contextIssueId
|
|
? await tx.select({ runId: nativeRunFinalizations.runId }).from(nativeRunFinalizations).where(and(
|
|
eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, run.id),
|
|
eq(nativeRunFinalizations.issueId, run.nativeIssueId), eq(nativeRunFinalizations.phase, "committed"),
|
|
)).limit(1)
|
|
: [];
|
|
if (!finalized) return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot };
|
|
}
|
|
if (await nativeManualAdmissionHoldsExecution(tx, { companyId: input.companyId, runId: run.id })) {
|
|
return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot };
|
|
}
|
|
if (input.resumeStrandedQueue && await hasPendingRunCleanup(tx, input.companyId, run.id)) {
|
|
return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot };
|
|
}
|
|
|
|
const candidateIssues = await tx
|
|
.select()
|
|
.from(issues)
|
|
.where(
|
|
and(
|
|
eq(issues.companyId, input.companyId),
|
|
contextIssueId
|
|
? or(eq(issues.id, contextIssueId), eq(issues.executionRunId, run.id), eq(issues.checkoutRunId, run.id))
|
|
: or(eq(issues.executionRunId, run.id), eq(issues.checkoutRunId, run.id)),
|
|
),
|
|
)
|
|
.orderBy(asc(issues.id));
|
|
|
|
for (const issue of candidateIssues) {
|
|
if (issue.executionRunId === run.id || issue.checkoutRunId === run.id) {
|
|
await notifyExternalConversationSettlement(tx, { companyId: input.companyId, issueId: issue.id });
|
|
await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: issue.id });
|
|
await notifyNativeStatusReassessment(tx, { companyId: input.companyId, issueId: issue.id });
|
|
}
|
|
}
|
|
// Two separate updates: a retry can move `executionRunId` to a new run
|
|
// while `checkoutRunId` still points at this one finishing.
|
|
await tx
|
|
.update(issues)
|
|
.set({ executionRunId: null, executionAgentNameKey: null, executionLockedAt: null, updatedAt: input.now })
|
|
.where(and(eq(issues.companyId, input.companyId), eq(issues.executionRunId, run.id)));
|
|
await tx
|
|
.update(issues)
|
|
.set({ checkoutRunId: null, updatedAt: input.now })
|
|
.where(and(eq(issues.companyId, input.companyId), eq(issues.checkoutRunId, run.id)));
|
|
|
|
const issueRow =
|
|
(contextIssueId ? candidateIssues.find((candidate) => candidate.id === contextIssueId) : candidateIssues[0]) ?? null;
|
|
|
|
// A queue interrupt authorizes only its original pending queue. Replays
|
|
// after dispatch or deleting the final message cannot launch other work.
|
|
const interruptQueueId = run.runtimeMode !== "native"
|
|
? readNonEmptyString(run.resultJson?.queuedCommentInterruptQueueId)
|
|
: null;
|
|
const [interruptedQueue] = interruptQueueId && issueRow
|
|
? await tx.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests).where(and(
|
|
eq(agentWakeupRequests.id, interruptQueueId),
|
|
eq(agentWakeupRequests.companyId, run.companyId),
|
|
eq(agentWakeupRequests.agentId, run.agentId),
|
|
eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
|
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueRow.id}`,
|
|
)).limit(1)
|
|
: [];
|
|
const cancellation = readRunCancellation(run.resultJson);
|
|
const queuedOperatorStop = run.runtimeMode === "legacy" && run.status === "cancelled" &&
|
|
run.runtimeModeResolvedAt === null && run.startedAt === null && run.processStartedAt === null &&
|
|
run.processPid === null && run.processGroupId === null &&
|
|
cancellation?.source === "operator" && cancellation.expected;
|
|
const preDrainFacts: PreDrainFacts = {
|
|
issueRowPresent: issueRow !== null,
|
|
executionRunIdMatchesRun: !issueRow || !issueRow.executionRunId || issueRow.executionRunId === run.id,
|
|
isWorkspaceValidationFailedRun: isWorkspaceValidationFailedRun(run),
|
|
isConfigurationIncompleteFailedRun: isConfigurationIncompleteFailedRun(run),
|
|
issueStatus: issueRow?.status ?? "",
|
|
hasAssigneeUser: Boolean(issueRow?.assigneeUserId),
|
|
assigneeAgentMatchesRunAgent: issueRow?.assigneeAgentId === run.agentId,
|
|
legacyExecutionNeedsReconciliation: evidence.kind === "complete" &&
|
|
await legacyExecutionNeedsReconciliationWithEvidence(tx as unknown as Db, evidence.run),
|
|
// An operator stop never promotes old queued work by itself. The
|
|
// next explicit wake adopts those messages atomically when it
|
|
// queues a run.
|
|
executionCancellationAcknowledged:
|
|
isAcknowledgedNativeReassignmentStop(run) ||
|
|
(run.status === "cancelled" &&
|
|
(queuedOperatorStop || parseObject(run.resultJson?.executionCancellation).state === "acknowledged" || isAcknowledgedNativeStop(run)) &&
|
|
!interruptedQueue),
|
|
};
|
|
const preDrain = decidePreDrain(preDrainFacts);
|
|
// A deliberate stop does not promote the stopped agent's queued work.
|
|
// When a reassignment committed while that stop was still settling, the
|
|
// new assignee's wake parked behind this run; it is explicit new work
|
|
// and must run now, or the task is left with no run at all.
|
|
const handoffAssigneeAgentId = preDrain.kind === "released" &&
|
|
decidePreDrain({ ...preDrainFacts, executionCancellationAcknowledged: false }).kind === "proceed" &&
|
|
issueRow?.assigneeAgentId && !issueRow.assigneeUserId && issueRow.assigneeAgentId !== run.agentId
|
|
? issueRow.assigneeAgentId : undefined;
|
|
|
|
if (preDrain.kind === "released" && !handoffAssigneeAgentId) {
|
|
return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot };
|
|
}
|
|
|
|
// decidePreDrain only returns "blocked" or "proceed" when the issue row is present.
|
|
if (!issueRow) {
|
|
throw new Error(`wake-queue: pre-drain decision ${preDrain.kind} reached without an issue row`);
|
|
}
|
|
|
|
// Enqueue does not stamp executionRunId until dispatch. A concurrent
|
|
// queued successor still owns the next turn, including during a late
|
|
// finalization/stranded-queue retry under this issue lock. Another
|
|
// agent's review participation retains its separate recovery path.
|
|
const [successor] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
|
eq(heartbeatRuns.companyId, input.companyId),
|
|
input.resumeStrandedQueue ? undefined : eq(heartbeatRuns.agentId, run.agentId),
|
|
input.resumeStrandedQueue ? undefined : sql`${heartbeatRuns.id} <> ${run.id}`,
|
|
or(eq(heartbeatRuns.nativeIssueId, issueRow.id),
|
|
sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueRow.id}`),
|
|
or(inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]),
|
|
input.resumeStrandedQueue ? sql`(${heartbeatRuns.createdAt}, ${heartbeatRuns.id}) > (
|
|
SELECT source.created_at, source.id FROM heartbeat_runs source
|
|
WHERE source.company_id = ${input.companyId} AND source.id = ${run.id})` : undefined),
|
|
)).limit(1);
|
|
if (successor) return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot };
|
|
|
|
if (preDrain.kind === "blocked") {
|
|
return {
|
|
outcome: {
|
|
kind: "blocked",
|
|
issue: toIssueSnapshot(issueRow),
|
|
previousStatus: issueRow.status as "todo" | "in_progress",
|
|
noticeKind: preDrain.noticeKind,
|
|
},
|
|
postCommitEffects: [],
|
|
run: runSnapshot,
|
|
};
|
|
}
|
|
|
|
// The durable external answer owns the sole successor of this retired
|
|
// question source; release locks without creating another incident.
|
|
if (
|
|
run.runtimeMode === "native" &&
|
|
run.status === "cancelled" &&
|
|
run.errorCode === "external_chat_continuation" &&
|
|
(await isRetiredExternalChatQuestionSource(tx, {
|
|
companyId: run.companyId,
|
|
issueId: issueRow.id,
|
|
agentId: run.agentId,
|
|
runId: run.id,
|
|
}))
|
|
) {
|
|
return {
|
|
outcome: { kind: "released" },
|
|
postCommitEffects: [],
|
|
run: runSnapshot,
|
|
};
|
|
}
|
|
|
|
// A native run blocked for a rejected model or incomplete configuration waits for that
|
|
// change and a retry. The provider refused the request, so there is no execution to
|
|
// reconcile; a reconciliation incident would also hide the run's Retry.
|
|
if (!handoffAssigneeAgentId && run.runtimeMode === "native" && isConfigurationIncompleteFailedRun(run)
|
|
&& issueRow.status === "blocked" && issueRow.assigneeAgentId === run.agentId) {
|
|
return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot };
|
|
}
|
|
if (!handoffAssigneeAgentId && await recordNativeTerminalRecoveryIfNeeded(tx, run, issueRow, input.now)) {
|
|
return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot };
|
|
}
|
|
|
|
// A release must leave deferred messages intact while execution is held.
|
|
// The finishing conversation may still own its lease until finally cleanup.
|
|
// Allow only bounded retry planning in that case; admission stays gated.
|
|
const executionBlocker = await getExecutionBlocker(tx, issueRow.companyId, issueRow.id);
|
|
const recoveryOnly = Boolean(executionBlocker &&
|
|
executionBlocker.cause === "execution_owner_active" && executionBlocker.runId === run.id &&
|
|
runSnapshot.conversationContinuation && ["failed", "timed_out", "interrupted"].includes(run.status));
|
|
if (executionBlocker && !recoveryOnly) {
|
|
return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot };
|
|
}
|
|
|
|
// Releases still settle while Agent Chat is disabled, but no deferred
|
|
// turn or recovery successor may be created. Check here in the shared
|
|
// transaction so cleanup retries and restart sweeps use the same gate.
|
|
if (issueRow.conversationAgentId && !(await instanceSettingsService(tx).getExperimental()).enableAgentChat) {
|
|
return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot };
|
|
}
|
|
|
|
const locked: LockedIssueExecution = { primaryIssue: toIssueSnapshot(issueRow), run: runSnapshot,
|
|
recoveryOnly: recoveryOnly && !handoffAssigneeAgentId, ...(handoffAssigneeAgentId ? { handoffAssigneeAgentId } : {}) };
|
|
const result = await fn(locked, { host: buildHost(tx, deps), transaction: buildTransaction(tx, deps, db, run) });
|
|
// Explicit retry admission requires the exact source's task claim.
|
|
// Preserve it only when the existing queue-first recovery policy
|
|
// requests that retry. The clear/restore stays inside this task lock.
|
|
if (run.runtimeMode === "legacy" && ["failed", "timed_out"].includes(run.status) &&
|
|
run.contextSnapshot?.explicitUserContinuation && issueRow.executionRunId === run.id &&
|
|
result.postCommitEffects.some(effect => effect.kind === "conversation_retry_requested" && effect.runId === run.id)) {
|
|
await tx.update(issues).set({ executionRunId: run.id,
|
|
executionAgentNameKey: issueRow.executionAgentNameKey,
|
|
executionLockedAt: issueRow.executionLockedAt,
|
|
}).where(and(eq(issues.companyId, input.companyId), eq(issues.id, issueRow.id),
|
|
sql`${issues.executionRunId} is null`));
|
|
}
|
|
return { ...result, run: runSnapshot };
|
|
});
|
|
},
|
|
};
|
|
}
|