mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-15 01:52:55 +01:00
DottaandClaude Opus 5.5
763adbea09
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A budget hard stop pauses the agent, which stops its running work > - In the restart-chaos H1 scenario, the stopped task went `blocked` behind a board action that said "inspect and reconcile" and offered no button, and it stayed there after the board resumed the budget > - A pause is an intentional stop, not a failure, so it must not leave the task with no way forward > - This pull request treats a pause's stop as intentional: the task keeps its status and no incident opens, so the normal Retry works > - The benefit is that a budget or agent pause never strands a task behind a control-less incident ## Linked Issues or Issue Description Refs #20113, #20101 **What happened?** Restart-chaos wave w-20261108T210118-s9802 (H1). At the budget pause, task CHA-1 went `blocked` (21:02:27.870) with an active board `active_run_watchdog` action, cause `native_continuation_requires_reconciliation`, next action "Inspect the original failure and reconcile the previous execution before continuing." After raise_budget_and_resume it stayed blocked, `canRetry=false`, with no Retry. The run was cancelled by the agent lifecycle pause (`cancellation: { reason: "Agent lifecycle stop requested", initiator: system, expected: true }`). The wake-queue release (`recordNativeTerminalRecoveryIfNeeded`) blocks every cancelled native run that is not an acknowledged user Stop. **Expected behavior** A pause's stop keeps the task's status and opens no incident; the task can be continued. **Steps to reproduce** Pause an agent (budget hold or company budget pause) while its native run is active. Before this change, the task is blocked behind a board action with no control. ## What Changed - New `server/src/services/governance-pause-stop.ts`: `isGovernancePauseStop(run)` matches an expected system cancellation whose reason is the budget pause stop, or the agent lifecycle stop with a `pauseStop` stamp. - `wake-queue/adapters/postgres.ts` `recordNativeTerminalRecoveryIfNeeded`: returns early for such a run, so the release keeps the task's status and opens no reconciliation incident. - `agent-lifecycle-driver.ts`: the `pausing` host step stamps `resultJson.pauseStop = { holds, companyPauseReason }` on the agent's active runs before stopping them (one update, company pause reason by subquery). Termination and cleanup share the lifecycle stop reason but write no stamp, so they still open the incident. ## Verification - `wake-queue/adapters/postgres.test.ts`: new cases. A run stopped by the lifecycle pause or by the budget pause keeps the task `in_progress` with no incident (both red on master). Agent termination, and another system stop (company archive), still open the incident. - `agent-lifecycle/__tests__/lifecycle.test.ts`: "records what a pause stopped each active run for". `driver-scope.test.ts` updated for the stamp's single update. - `wake-queue` (107), `lifecycle` (68+), `driver-scope`, `companies-service`: 199 passed. Server typecheck (`typecheck-layers`, low-memory): no errors. - Not run: the H1 chaos rerun (after the follow-up continuation PR). ## Risks - A native run that a pause stopped no longer opens a reconciliation incident. Resuming it is the normal Retry, which continues the conversation. Termination, archive and failures still open one. - One extra update per pause attempt on the agent's active runs. It is idempotent (skips runs that already have the stamp) and not on any turn path. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with tool use (shell, file edits, test execution). Extended reasoning enabled. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [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>
1414 lines
64 KiB
TypeScript
1414 lines
64 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 } 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 { 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 { 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 } from "../../run-dispatch/index.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),
|
|
};
|
|
}
|
|
|
|
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";
|
|
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);
|
|
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 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 }) {
|
|
const updated = await issuesSvc.updateForCompany(issueId, companyId, { status: "todo", 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}`,
|
|
]
|
|
: []),
|
|
...(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);
|
|
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 };
|
|
});
|
|
},
|
|
};
|
|
}
|