mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-06 05:56:19 +01:00
Devin FoleyandPaperclip
332e4c5df9
Preserve a valid saved monitor after its successful source releases execution and checkout. Recheck accepted result identity, receipts, ownership and governance under the task lock. Notify the existing scheduler once and preserve accepted live and recovered replies. Apex 5/5 after fixing its reply-presentation finding. All CI checks, 191 focused tests, 293 broader native tests, workspace typecheck and server build pass. Hosted effects require staging verification. Co-Authored-By: Paperclip <noreply@paperclip.ing>
2306 lines
80 KiB
TypeScript
2306 lines
80 KiB
TypeScript
import { scheduledIssueMonitorWait } from "../issue-monitors.js";
|
|
import { hasPublishedGitHubReview } from "../chat-channels/github/review-wait.js";
|
|
import { pendingNativeGovernance } from "./native-governance.js";
|
|
import { findSupersededExternalReviewRequests, supersedeExternalReviewRequestsInTransaction } from "./superseded-external-review-requests.js";
|
|
import { normalizePrpResultSignals, type PrpStructuredRunResult } from "../../vendor/paperclip-runner/index.js";
|
|
import { readNativeMonitorDelivery, readNativePreservedMonitorWait, type NativeMonitorDelivery, type NativePreservedMonitorWait } from "./native-monitor-delivery.js";
|
|
import { readNativePlanWait, type NativePlanWaitSource } from "./native-plan-wait.js";
|
|
import { activeIssueInteractionCondition } from "../issue-question-context.js";
|
|
import { isConversation } from "../agent-conversations.js";
|
|
import { hasPendingNativeChildCompletion, type NativeChildCompletionRecipient } from "./native-child-completion-delivery.js";
|
|
import { and, asc, desc, eq, inArray, notInArray, sql } from "drizzle-orm";
|
|
import type { Db } from "@paperclipai/db";
|
|
import type { CreateIssueThreadInteraction } from "@paperclipai/shared";
|
|
import {
|
|
agentWakeupRequests,
|
|
agents,
|
|
approvals,
|
|
completionContracts,
|
|
heartbeatRuns,
|
|
issueApprovals,
|
|
issueRecoveryActions,
|
|
issueThreadInteractions,
|
|
issues,
|
|
nativeRunFinalizations,
|
|
nativeRunResults,
|
|
statusDecisionEffects,
|
|
statusDecisions,
|
|
workAssessments,
|
|
workspaceOperations,
|
|
} from "@paperclipai/db";
|
|
import {
|
|
NATIVE_STATUS_ARBITER_POLICY_VERSION,
|
|
type NativeStatusDecision,
|
|
type NativeStatusEffect,
|
|
} from "./status-arbiter.js";
|
|
import { nativeSha256 } from "./canonical.js";
|
|
import { getNativeReviewAssignment, type NativeReviewAssignmentContext } from "./native-review-participant.js";
|
|
import {
|
|
readNativeBoardResponseWaitSource,
|
|
readNativeBoardResponseWaitOrigin,
|
|
type NativeBoardResponseWaitOrigin,
|
|
type NativeBoardResponseWaitSource,
|
|
} from "./native-board-response-wait.js";
|
|
import {
|
|
prepareNativeChatReviewPresentationInTransaction,
|
|
restoreNativeChatReviewPresentationInTransaction,
|
|
} from "./native-chat-review-presentation.js";
|
|
import {
|
|
isExternalChatWaitAuthorizationContention,
|
|
resolveExternalChatResponseWaitAuthorizationInTransaction,
|
|
} from "./chat-attachment-reuse.js";
|
|
import { issueService } from "../issues.js";
|
|
import { issueThreadInteractionService } from "../issue-thread-interactions.js";
|
|
import { issueRecoveryActionService } from "../issue-recovery-actions.js";
|
|
import { buildIssueBlockersResolvedWakeIdempotencyKey } from "../issue-dependency-wakeups.js";
|
|
import {
|
|
persistActivity,
|
|
publishActivity,
|
|
type ActivityPublication,
|
|
} from "../activity-log.js";
|
|
import { emitAgentTaskRun } from "../agent-task-run-telemetry.js";
|
|
import { materializeNativeProviderCapacityRetry } from "./native-provider-capacity-retry.js";
|
|
import { NATIVE_PROVIDER_OVERLOADED_CODE } from "./native-provider-failure.js";
|
|
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
|
|
|
|
export class NativeStatusRaceError extends Error {
|
|
readonly code = "native_status_race" as const;
|
|
constructor() {
|
|
super("native_status_race");
|
|
this.name = "NativeStatusRaceError";
|
|
}
|
|
}
|
|
|
|
/** Reconciliation delivery resumes ledger work only; it never re-arbitrates. */
|
|
export async function dispatchPendingNativeStatusEffects(input: {
|
|
db: Db;
|
|
companyId: string;
|
|
issueId: string;
|
|
decisionId: string;
|
|
runId: string;
|
|
}) {
|
|
return input.db.transaction(async (tx) => {
|
|
const coordinator = await tx
|
|
.select({
|
|
decisionId: nativeRunFinalizations.decisionId,
|
|
})
|
|
.from(nativeRunFinalizations)
|
|
.where(
|
|
and(
|
|
eq(nativeRunFinalizations.runId, input.runId),
|
|
eq(nativeRunFinalizations.companyId, input.companyId),
|
|
eq(nativeRunFinalizations.issueId, input.issueId),
|
|
),
|
|
)
|
|
.for("update")
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!coordinator || coordinator.decisionId !== input.decisionId) {
|
|
throw new Error("native_pending_effect_decision_binding_mismatch");
|
|
}
|
|
const pending = await tx
|
|
.select({
|
|
id: statusDecisionEffects.id,
|
|
effectKind: statusDecisionEffects.effectKind,
|
|
targetType: statusDecisionEffects.targetType,
|
|
targetId: statusDecisionEffects.targetId,
|
|
})
|
|
.from(statusDecisionEffects)
|
|
.where(
|
|
and(
|
|
eq(statusDecisionEffects.companyId, input.companyId),
|
|
eq(statusDecisionEffects.issueId, input.issueId),
|
|
eq(statusDecisionEffects.decisionId, input.decisionId),
|
|
inArray(statusDecisionEffects.deliveryState, ["pending", "failed"]),
|
|
),
|
|
)
|
|
.for("update");
|
|
if (pending.length === 0) return [];
|
|
for (const effect of pending) {
|
|
await assertPendingEffectTarget(tx as unknown as Db, {
|
|
companyId: input.companyId,
|
|
issueId: input.issueId,
|
|
effectKind: effect.effectKind,
|
|
targetType: effect.targetType,
|
|
targetId: effect.targetId,
|
|
});
|
|
}
|
|
await tx
|
|
.update(statusDecisionEffects)
|
|
.set({
|
|
deliveryState: "delivered",
|
|
attemptCount: sql`${statusDecisionEffects.attemptCount} + 1`,
|
|
deliveredAt: new Date(),
|
|
nextAttemptAt: null,
|
|
lastError: null,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(
|
|
inArray(
|
|
statusDecisionEffects.id,
|
|
pending.map((row) => row.id),
|
|
),
|
|
);
|
|
await updateRunEffectState({
|
|
tx: tx as unknown as Db,
|
|
companyId: input.companyId,
|
|
runId: input.runId,
|
|
key: "nativeDispatchedEffectIds",
|
|
value: pending.map((row) => row.id),
|
|
});
|
|
return pending.map((row) => row.id);
|
|
});
|
|
}
|
|
|
|
export type NativeStatusCommitFailpoint =
|
|
| "governance_materialization"
|
|
| "interaction_materialization"
|
|
| "continuation_materialization"
|
|
| "blocker_materialization"
|
|
| "recovery_materialization"
|
|
| "status_projection";
|
|
|
|
export type NativeMaterializedStatusEffect = {
|
|
effectKind: string;
|
|
targetType: string;
|
|
targetId: string | null;
|
|
payload: Record<string, unknown>;
|
|
};
|
|
|
|
function record(value: unknown): Record<string, unknown> {
|
|
return value && typeof value === "object" && !Array.isArray(value)
|
|
? (value as Record<string, unknown>)
|
|
: {};
|
|
}
|
|
|
|
function assertNeverEffect(effect: never): never {
|
|
const kind = record(effect).kind;
|
|
throw new Error(
|
|
`native_status_effect_unimplemented:${String(kind ?? "unknown")}`,
|
|
);
|
|
}
|
|
|
|
async function assertPendingEffectTarget(
|
|
tx: Db,
|
|
input: {
|
|
companyId: string;
|
|
issueId: string;
|
|
effectKind: string;
|
|
targetType: string;
|
|
targetId: string | null;
|
|
},
|
|
) {
|
|
if (!input.targetId)
|
|
throw new Error(`native_pending_effect_target_missing:${input.effectKind}`);
|
|
let found = false;
|
|
if (input.targetType === "agent_wakeup_request") {
|
|
found = await tx
|
|
.select({ id: agentWakeupRequests.id })
|
|
.from(agentWakeupRequests)
|
|
.where(
|
|
and(
|
|
eq(agentWakeupRequests.id, input.targetId),
|
|
eq(agentWakeupRequests.companyId, input.companyId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows.length === 1);
|
|
} else if (input.targetType === "agent") {
|
|
found = await tx
|
|
.select({ id: agents.id })
|
|
.from(agents)
|
|
.where(
|
|
and(
|
|
eq(agents.id, input.targetId),
|
|
eq(agents.companyId, input.companyId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows.length === 1);
|
|
} else if (
|
|
input.targetType === "issue_thread_interaction" ||
|
|
input.targetType === "interaction"
|
|
) {
|
|
found = await tx
|
|
.select({ id: issueThreadInteractions.id })
|
|
.from(issueThreadInteractions)
|
|
.where(
|
|
and(
|
|
eq(issueThreadInteractions.id, input.targetId),
|
|
eq(issueThreadInteractions.companyId, input.companyId),
|
|
eq(issueThreadInteractions.issueId, input.issueId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows.length === 1);
|
|
} else if (
|
|
[
|
|
"delegated_issue",
|
|
"issue",
|
|
"issue_checkout",
|
|
"issue_unblock_descriptor",
|
|
].includes(input.targetType)
|
|
) {
|
|
found = await tx
|
|
.select({ id: issues.id })
|
|
.from(issues)
|
|
.where(
|
|
and(
|
|
eq(issues.id, input.targetId),
|
|
eq(issues.companyId, input.companyId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows.length === 1);
|
|
} else if (input.targetType === "issue_recovery_action") {
|
|
found = await tx
|
|
.select({ id: issueRecoveryActions.id })
|
|
.from(issueRecoveryActions)
|
|
.where(
|
|
and(
|
|
eq(issueRecoveryActions.id, input.targetId),
|
|
eq(issueRecoveryActions.companyId, input.companyId),
|
|
eq(issueRecoveryActions.sourceIssueId, input.issueId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows.length === 1);
|
|
} else if (input.targetType === "heartbeat_run") {
|
|
found = await tx
|
|
.select({ id: heartbeatRuns.id })
|
|
.from(heartbeatRuns)
|
|
.where(
|
|
and(
|
|
eq(heartbeatRuns.id, input.targetId),
|
|
eq(heartbeatRuns.companyId, input.companyId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows.length === 1);
|
|
} else if (input.targetType === "workspace_operation") {
|
|
found = await tx
|
|
.select({ id: workspaceOperations.id })
|
|
.from(workspaceOperations)
|
|
.where(
|
|
and(
|
|
eq(workspaceOperations.id, input.targetId),
|
|
eq(workspaceOperations.companyId, input.companyId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows.length === 1);
|
|
} else if (input.targetType === "completion_contract") {
|
|
found = await tx
|
|
.select({ id: completionContracts.id })
|
|
.from(completionContracts)
|
|
.where(
|
|
and(
|
|
eq(completionContracts.id, input.targetId),
|
|
eq(completionContracts.companyId, input.companyId),
|
|
eq(completionContracts.issueId, input.issueId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows.length === 1);
|
|
} else if (input.targetType === "status_decision") {
|
|
found = await tx
|
|
.select({ id: statusDecisions.id })
|
|
.from(statusDecisions)
|
|
.where(
|
|
and(
|
|
eq(statusDecisions.id, input.targetId),
|
|
eq(statusDecisions.companyId, input.companyId),
|
|
eq(statusDecisions.issueId, input.issueId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows.length === 1);
|
|
} else if (input.targetType === "status_decision_effect") {
|
|
found = await tx
|
|
.select({ id: statusDecisionEffects.id })
|
|
.from(statusDecisionEffects)
|
|
.where(
|
|
and(
|
|
eq(statusDecisionEffects.id, input.targetId),
|
|
eq(statusDecisionEffects.companyId, input.companyId),
|
|
eq(statusDecisionEffects.issueId, input.issueId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows.length === 1);
|
|
} else if (input.targetType === "approval") {
|
|
found = await tx
|
|
.select({ id: approvals.id })
|
|
.from(issueApprovals)
|
|
.innerJoin(approvals, eq(issueApprovals.approvalId, approvals.id))
|
|
.where(
|
|
and(
|
|
eq(approvals.id, input.targetId),
|
|
eq(approvals.companyId, input.companyId),
|
|
eq(issueApprovals.issueId, input.issueId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows.length === 1);
|
|
} else if (input.targetType === "execution_stage") {
|
|
found = await tx
|
|
.select({ executionState: issues.executionState })
|
|
.from(issues)
|
|
.where(
|
|
and(
|
|
eq(issues.id, input.issueId),
|
|
eq(issues.companyId, input.companyId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => record(rows[0]?.executionState).status === "pending");
|
|
} else {
|
|
throw new Error(
|
|
`native_pending_effect_target_unimplemented:${input.targetType}`,
|
|
);
|
|
}
|
|
if (!found)
|
|
throw new Error(`native_pending_effect_target_missing:${input.effectKind}`);
|
|
}
|
|
|
|
async function updateRunEffectState(input: {
|
|
tx: Db;
|
|
companyId: string;
|
|
runId: string;
|
|
key: string;
|
|
value: unknown;
|
|
runnerProfile?: boolean;
|
|
}) {
|
|
const run = await input.tx
|
|
.select({
|
|
id: heartbeatRuns.id,
|
|
resultJson: heartbeatRuns.resultJson,
|
|
runnerProfileJson: heartbeatRuns.runnerProfileJson,
|
|
})
|
|
.from(heartbeatRuns)
|
|
.where(
|
|
and(
|
|
eq(heartbeatRuns.id, input.runId),
|
|
eq(heartbeatRuns.companyId, input.companyId),
|
|
),
|
|
)
|
|
.for("update")
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!run) throw new Error("native_status_run_missing");
|
|
const target = input.runnerProfile
|
|
? record(run.runnerProfileJson)
|
|
: record(run.resultJson);
|
|
await input.tx
|
|
.update(heartbeatRuns)
|
|
.set({
|
|
...(input.runnerProfile
|
|
? { runnerProfileJson: { ...target, [input.key]: input.value } }
|
|
: { resultJson: { ...target, [input.key]: input.value } }),
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(heartbeatRuns.id, input.runId));
|
|
return run.id;
|
|
}
|
|
|
|
function failAt(
|
|
actual: NativeStatusCommitFailpoint,
|
|
requested?: NativeStatusCommitFailpoint,
|
|
) {
|
|
if (actual === requested)
|
|
throw new Error(`native_status_failpoint:${actual}`);
|
|
}
|
|
|
|
async function enqueueWake(input: {
|
|
tx: Db;
|
|
companyId: string;
|
|
issueId: string;
|
|
agentId: string;
|
|
reason: string;
|
|
idempotencyKey: string;
|
|
payload: Record<string, unknown>;
|
|
contextSnapshot?: Record<string, unknown>;
|
|
}) {
|
|
const existing = await input.tx
|
|
.select({ id: agentWakeupRequests.id })
|
|
.from(agentWakeupRequests)
|
|
.where(
|
|
and(
|
|
eq(agentWakeupRequests.companyId, input.companyId),
|
|
eq(agentWakeupRequests.idempotencyKey, input.idempotencyKey),
|
|
inArray(agentWakeupRequests.status, [
|
|
"queued",
|
|
"deferred_issue_execution",
|
|
"claimed",
|
|
"completed",
|
|
]),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (existing) return existing.id;
|
|
const inserted = await input.tx
|
|
.insert(agentWakeupRequests)
|
|
.values({
|
|
companyId: input.companyId,
|
|
agentId: input.agentId,
|
|
source: "automation",
|
|
triggerDetail: "system",
|
|
reason: input.reason,
|
|
payload: {
|
|
issueId: input.issueId,
|
|
taskId: input.issueId,
|
|
...input.payload,
|
|
_paperclipWakeContext: {
|
|
issueId: input.issueId,
|
|
taskId: input.issueId,
|
|
...(input.contextSnapshot ?? {}),
|
|
wakeReason: input.reason,
|
|
source: "native_status_decision",
|
|
statusDecisionSource: "native_status_decision",
|
|
},
|
|
},
|
|
requestedByActorType: "system",
|
|
requestedByActorId: "native-status-committer",
|
|
idempotencyKey: input.idempotencyKey,
|
|
})
|
|
.returning({ id: agentWakeupRequests.id })
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!inserted) throw new Error("native_status_wake_not_persisted");
|
|
return inserted.id;
|
|
}
|
|
|
|
async function validateGovernanceGate(
|
|
tx: Db,
|
|
input: {
|
|
companyId: string;
|
|
issueId: string;
|
|
gate: NonNullable<
|
|
Extract<NativeStatusEffect, { kind: "create_interaction" }>["gate"]
|
|
>;
|
|
executionState: unknown;
|
|
runId: string;
|
|
conversationMode: boolean;
|
|
},
|
|
) {
|
|
if (input.gate.kind === "execution_stage") {
|
|
const state =
|
|
input.executionState && typeof input.executionState === "object"
|
|
? (input.executionState as Record<string, unknown>)
|
|
: {};
|
|
if (state.status !== "pending")
|
|
throw new Error("native_governance_gate_resolved");
|
|
return input.gate.id;
|
|
}
|
|
if (input.gate.kind === "interaction") {
|
|
const interaction = await tx
|
|
.select({ id: issueThreadInteractions.id })
|
|
.from(issueThreadInteractions)
|
|
.where(
|
|
and(
|
|
eq(issueThreadInteractions.id, input.gate.id),
|
|
eq(issueThreadInteractions.companyId, input.companyId),
|
|
eq(issueThreadInteractions.issueId, input.issueId),
|
|
eq(issueThreadInteractions.status, "pending"),
|
|
activeIssueInteractionCondition(input),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!interaction) throw new Error("native_governance_gate_resolved");
|
|
return interaction.id;
|
|
}
|
|
const approval = await tx
|
|
.select({ id: approvals.id })
|
|
.from(issueApprovals)
|
|
.innerJoin(approvals, eq(issueApprovals.approvalId, approvals.id))
|
|
.where(
|
|
and(
|
|
eq(issueApprovals.companyId, input.companyId),
|
|
eq(issueApprovals.issueId, input.issueId),
|
|
eq(approvals.companyId, input.companyId),
|
|
eq(approvals.id, input.gate.id),
|
|
inArray(approvals.status, ["pending", "revision_requested"]),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!approval) throw new Error("native_governance_gate_resolved");
|
|
return approval.id;
|
|
}
|
|
|
|
async function materializeDecisionEffect(input: {
|
|
tx: Db;
|
|
companyId: string;
|
|
issue: typeof issues.$inferSelect;
|
|
runId: string;
|
|
decisionId: string;
|
|
effect: NativeStatusEffect;
|
|
failpoint?: NativeStatusCommitFailpoint;
|
|
preMaterializedEffects?: ReadonlyMap<string, NativeMaterializedStatusEffect>;
|
|
/** Terminal heartbeat_runs rows this effect wrote. The caller emits agent.task_run for each, after its transaction commits. */
|
|
terminalRunsToEmit?: (typeof heartbeatRuns.$inferSelect)[];
|
|
}): Promise<NativeMaterializedStatusEffect> {
|
|
const { effect } = input;
|
|
if (effect.kind === "create_interaction") {
|
|
failAt("governance_materialization", input.failpoint);
|
|
if (effect.gate) {
|
|
const targetId = await validateGovernanceGate(input.tx, {
|
|
companyId: input.companyId,
|
|
issueId: input.issue.id,
|
|
gate: effect.gate,
|
|
executionState: input.issue.executionState,
|
|
runId: input.runId,
|
|
conversationMode: isConversation(input.issue),
|
|
});
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: effect.gate.kind,
|
|
targetId,
|
|
payload: { gate: effect.gate },
|
|
};
|
|
}
|
|
const interaction = await issueThreadInteractionService(input.tx).create(
|
|
input.issue,
|
|
{
|
|
kind: "ask_user_questions",
|
|
idempotencyKey: `native-attention:${input.decisionId}`,
|
|
sourceRunId: input.runId,
|
|
title: "Native attention request",
|
|
summary: "The native runner requires a board response.",
|
|
continuationPolicy: "wake_assignee",
|
|
payload: {
|
|
version: 1,
|
|
questions: [
|
|
{
|
|
id: "response",
|
|
prompt: effect.prompt ?? "Provide the required response.",
|
|
selectionMode: "single",
|
|
options: [{ id: "continue", label: "Continue" }],
|
|
},
|
|
],
|
|
},
|
|
},
|
|
{ systemId: "native-status-committer", runId: input.runId },
|
|
);
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "issue_thread_interaction",
|
|
targetId: interaction.id,
|
|
payload: {
|
|
interactionId: interaction.id,
|
|
interactionKind: interaction.kind,
|
|
},
|
|
};
|
|
}
|
|
if (effect.kind === "bind_reviewer") {
|
|
failAt("interaction_materialization", input.failpoint);
|
|
const reviewInput: Extract<
|
|
CreateIssueThreadInteraction,
|
|
{ kind: "request_confirmation" }
|
|
> = {
|
|
kind: "request_confirmation",
|
|
idempotencyKey: `native-review:${input.decisionId}${effect.requestKey ? `:${effect.requestKey}` : ""}`,
|
|
sourceRunId: input.runId,
|
|
resolverPolicy: effect.ownerAgentId ? "anyone" : "human_only",
|
|
addresseeAgentId: effect.ownerAgentId ?? null,
|
|
addresseeUserId: effect.ownerUserId,
|
|
title: "Review requested",
|
|
summary:
|
|
effect.prompt,
|
|
continuationPolicy: "wake_assignee",
|
|
payload: {
|
|
version: 1,
|
|
prompt: effect.prompt,
|
|
detailsMarkdown: effect.detailsMarkdown ?? null,
|
|
acceptLabel: "Approve completion",
|
|
rejectLabel: "Continue work",
|
|
allowDeclineReason: true,
|
|
rejectRequiresReason: true,
|
|
supersedeOnUserComment: false,
|
|
target: {
|
|
type: "custom",
|
|
key: "native_completion_review",
|
|
revisionId: input.decisionId,
|
|
label: "Completion decision",
|
|
},
|
|
},
|
|
};
|
|
const interaction = await issueThreadInteractionService(input.tx).create(
|
|
input.issue,
|
|
reviewInput,
|
|
{ systemId: "native-status-committer", runId: input.runId },
|
|
);
|
|
// The card and its next actor commit together. The post-commit dispatcher
|
|
// revalidates this exact review before granting a scoped reviewer run.
|
|
const reviewContext = {
|
|
nativeReviewInteractionId: interaction.id,
|
|
nativeReviewDecisionId: input.decisionId,
|
|
};
|
|
const reviewerWakeId = effect.ownerAgentId && interaction.effectiveResolverPolicy !== "human_only"
|
|
? await enqueueWake({
|
|
tx: input.tx,
|
|
companyId: input.companyId,
|
|
issueId: input.issue.id,
|
|
agentId: effect.ownerAgentId,
|
|
reason: "native_completion_review",
|
|
idempotencyKey: `native-review:${interaction.id}`,
|
|
payload: reviewContext,
|
|
contextSnapshot: { ...reviewContext, forceFreshSession: true },
|
|
})
|
|
: null;
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "issue_thread_interaction",
|
|
targetId: interaction.id,
|
|
payload: {
|
|
interactionId: interaction.id,
|
|
interactionKind: interaction.kind,
|
|
ownerUserId: effect.ownerUserId,
|
|
ownerAgentId: effect.ownerAgentId ?? null,
|
|
reviewerWakeId,
|
|
},
|
|
};
|
|
}
|
|
if (effect.kind === "notify_owner") {
|
|
const wakeId = await enqueueWake({
|
|
tx: input.tx,
|
|
companyId: input.companyId,
|
|
issueId: input.issue.id,
|
|
agentId: effect.agentId,
|
|
reason: "issue_status_changed",
|
|
idempotencyKey: `native-status:${input.decisionId}:notify-owner`,
|
|
payload: {
|
|
nativeDecisionId: input.decisionId,
|
|
notificationReason: effect.reason,
|
|
},
|
|
});
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "agent_wakeup_request",
|
|
targetId: wakeId,
|
|
payload: { reason: effect.reason },
|
|
};
|
|
}
|
|
if (effect.kind === "create_delegated_issue") {
|
|
const delegated = await issueService(input.tx).create(input.companyId, {
|
|
projectId: input.issue.projectId,
|
|
projectWorkspaceId: input.issue.projectWorkspaceId,
|
|
goalId: input.issue.goalId,
|
|
parentId: input.issue.id,
|
|
title: effect.summary,
|
|
status: "todo",
|
|
priority: input.issue.priority,
|
|
assigneeAgentId: effect.agentId,
|
|
createdByAgentId: input.issue.assigneeAgentId,
|
|
workMode: "standard",
|
|
idempotencyKey: `native-delegation:${input.decisionId}`,
|
|
inheritExecutionWorkspaceFromIssueId: input.issue.id,
|
|
});
|
|
if (!delegated) throw new Error("native_delegated_issue_not_persisted");
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "delegated_issue",
|
|
targetId: delegated.id,
|
|
payload: { summary: effect.summary, agentId: effect.agentId },
|
|
};
|
|
}
|
|
if (effect.kind === "enqueue_continuation") {
|
|
failAt("continuation_materialization", input.failpoint);
|
|
if (
|
|
effect.idempotencyKey.startsWith(
|
|
`connection-intent:tools:${input.runId}:`,
|
|
)
|
|
) {
|
|
const [refresh] = await input.tx
|
|
.select({ id: agentWakeupRequests.id })
|
|
.from(agentWakeupRequests)
|
|
.where(
|
|
and(
|
|
eq(agentWakeupRequests.companyId, input.companyId),
|
|
eq(agentWakeupRequests.agentId, effect.agentId),
|
|
eq(agentWakeupRequests.idempotencyKey, effect.idempotencyKey),
|
|
notInArray(agentWakeupRequests.status, [
|
|
"skipped",
|
|
"failed",
|
|
"cancelled",
|
|
]),
|
|
),
|
|
)
|
|
.limit(1);
|
|
if (refresh)
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "agent_wakeup_request",
|
|
targetId: refresh.id,
|
|
payload: {
|
|
continuationKind: effect.continuationKind,
|
|
summary: effect.summary,
|
|
},
|
|
};
|
|
}
|
|
if (effect.continuationKind === "monitor") throw new Error("Monitor delivery belongs to the issue monitor scheduler");
|
|
const wakeId = await enqueueWake({
|
|
tx: input.tx,
|
|
companyId: input.companyId,
|
|
issueId: input.issue.id,
|
|
agentId: effect.agentId,
|
|
reason: "issue_status_changed",
|
|
idempotencyKey: `native-status:${input.decisionId}:continuation`,
|
|
payload: {
|
|
nativeDecisionId: input.decisionId,
|
|
continuationKind: effect.continuationKind,
|
|
continuationSummary: effect.summary,
|
|
continuationIdempotencyKey: effect.idempotencyKey,
|
|
},
|
|
});
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "agent_wakeup_request",
|
|
targetId: wakeId,
|
|
payload: {
|
|
continuationKind: effect.continuationKind,
|
|
summary: effect.summary,
|
|
},
|
|
};
|
|
}
|
|
if (effect.kind === "bind_blocker") {
|
|
failAt("blocker_materialization", input.failpoint);
|
|
const [bound] = await input.tx
|
|
.update(issues)
|
|
.set({
|
|
unblockDescriptor: { owner: effect.owner, action: effect.action },
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(
|
|
and(
|
|
eq(issues.id, input.issue.id),
|
|
eq(issues.companyId, input.companyId),
|
|
),
|
|
)
|
|
.returning({ id: issues.id });
|
|
if (!bound) throw new Error("native_blocker_binding_not_persisted");
|
|
let wakeId: string | null = null;
|
|
if (effect.owner !== "board") {
|
|
wakeId = await enqueueWake({
|
|
tx: input.tx,
|
|
companyId: input.companyId,
|
|
issueId: input.issue.id,
|
|
agentId: effect.owner.agentId,
|
|
reason: "issue_status_changed",
|
|
idempotencyKey: `native-status:${input.decisionId}:blocker-owner`,
|
|
payload: {
|
|
nativeDecisionId: input.decisionId,
|
|
unblockAction: effect.action,
|
|
},
|
|
});
|
|
}
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "issue_unblock_descriptor",
|
|
targetId: bound.id,
|
|
payload: { owner: effect.owner, action: effect.action, wakeId },
|
|
};
|
|
}
|
|
if (effect.kind === "schedule_retry") {
|
|
failAt("continuation_materialization", input.failpoint);
|
|
if (effect.cause === NATIVE_PROVIDER_OVERLOADED_CODE) {
|
|
const retryId = await materializeNativeProviderCapacityRetry({
|
|
tx: input.tx as unknown as Db, companyId: input.companyId,
|
|
issueId: input.issue.id, runId: input.runId, agentId: effect.agentId,
|
|
});
|
|
return { effectKind: effect.kind, targetType: "heartbeat_run", targetId: retryId,
|
|
payload: { cause: effect.cause, summary: effect.summary } };
|
|
}
|
|
const wakeId = await enqueueWake({
|
|
tx: input.tx,
|
|
companyId: input.companyId,
|
|
issueId: input.issue.id,
|
|
agentId: effect.agentId,
|
|
reason: "issue_status_changed",
|
|
idempotencyKey: `native-status:${input.decisionId}:retry`,
|
|
payload: {
|
|
nativeDecisionId: input.decisionId,
|
|
retryCause: effect.cause,
|
|
retrySummary: effect.summary,
|
|
},
|
|
});
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "agent_wakeup_request",
|
|
targetId: wakeId,
|
|
payload: { cause: effect.cause, summary: effect.summary },
|
|
};
|
|
}
|
|
if (effect.kind === "record_finalization_error") {
|
|
failAt("recovery_materialization", input.failpoint);
|
|
const recovery = await issueRecoveryActionService(
|
|
input.tx,
|
|
).upsertSourceScoped({
|
|
companyId: input.companyId,
|
|
sourceIssueId: input.issue.id,
|
|
kind: "active_run_watchdog",
|
|
ownerType: "agent",
|
|
ownerAgentId: effect.agentId,
|
|
cause: effect.cause,
|
|
fingerprint: nativeSha256({
|
|
runId: input.runId,
|
|
decisionId: input.decisionId,
|
|
cause: effect.cause,
|
|
}),
|
|
evidence: { runId: input.runId, decisionId: input.decisionId },
|
|
nextAction: effect.nextAction,
|
|
wakePolicy: null,
|
|
maxAttempts: 3,
|
|
});
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "issue_recovery_action",
|
|
targetId: recovery.id,
|
|
payload: { cause: effect.cause, nextAction: effect.nextAction },
|
|
};
|
|
}
|
|
if (effect.kind === "release_run_resources") {
|
|
failAt("recovery_materialization", input.failpoint);
|
|
const [released] = await input.tx
|
|
.update(heartbeatRuns)
|
|
.set({
|
|
processPid: null,
|
|
processGroupId: null,
|
|
processStartedAt: null,
|
|
nativePhase: "committed",
|
|
nativePhaseUpdatedAt: new Date(),
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(
|
|
and(
|
|
eq(heartbeatRuns.id, input.runId),
|
|
eq(heartbeatRuns.companyId, input.companyId),
|
|
),
|
|
)
|
|
.returning({ id: heartbeatRuns.id });
|
|
if (!released) throw new Error("native_run_resources_not_released");
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "heartbeat_run",
|
|
targetId: released.id,
|
|
payload: { processReleased: true },
|
|
};
|
|
}
|
|
if (effect.kind === "record_expiry") {
|
|
failAt("recovery_materialization", input.failpoint);
|
|
if (!input.issue.assigneeAgentId)
|
|
throw new Error("native_recovery_owner_missing");
|
|
const interaction = await input.tx
|
|
.select({ id: issueThreadInteractions.id })
|
|
.from(issueThreadInteractions)
|
|
.where(
|
|
and(
|
|
eq(issueThreadInteractions.companyId, input.companyId),
|
|
eq(issueThreadInteractions.issueId, input.issue.id),
|
|
inArray(issueThreadInteractions.status, ["pending", "expired"]),
|
|
),
|
|
)
|
|
.for("update")
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!interaction) throw new Error("native_attention_expiry_target_missing");
|
|
await input.tx
|
|
.update(issueThreadInteractions)
|
|
.set({
|
|
status: "expired",
|
|
resolvedAt: new Date(),
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(issueThreadInteractions.id, interaction.id));
|
|
const cause = "attention_expired";
|
|
const recovery = await issueRecoveryActionService(
|
|
input.tx,
|
|
).upsertSourceScoped({
|
|
companyId: input.companyId,
|
|
sourceIssueId: input.issue.id,
|
|
kind: "active_run_watchdog",
|
|
ownerType: "agent",
|
|
ownerAgentId: input.issue.assigneeAgentId,
|
|
cause,
|
|
fingerprint: nativeSha256({
|
|
runId: input.runId,
|
|
decisionId: input.decisionId,
|
|
cause,
|
|
}),
|
|
evidence: { runId: input.runId, decisionId: input.decisionId },
|
|
nextAction:
|
|
"Review the expired attention route and choose a new authorized path.",
|
|
wakePolicy: null,
|
|
maxAttempts: 3,
|
|
});
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "issue_thread_interaction",
|
|
targetId: interaction.id,
|
|
payload: { cause, recoveryActionId: recovery.id },
|
|
};
|
|
}
|
|
if (effect.kind === "schedule_reconciliation") {
|
|
failAt("recovery_materialization", input.failpoint);
|
|
if (!input.issue.assigneeAgentId)
|
|
throw new Error("native_recovery_owner_missing");
|
|
const cause = "native_reconciliation_required";
|
|
const recovery = await issueRecoveryActionService(
|
|
input.tx,
|
|
).upsertSourceScoped({
|
|
companyId: input.companyId,
|
|
sourceIssueId: input.issue.id,
|
|
kind: "active_run_watchdog",
|
|
ownerType: "agent",
|
|
ownerAgentId: input.issue.assigneeAgentId,
|
|
cause,
|
|
fingerprint: nativeSha256({
|
|
runId: input.runId,
|
|
decisionId: input.decisionId,
|
|
cause,
|
|
}),
|
|
evidence: { runId: input.runId, decisionId: input.decisionId },
|
|
nextAction:
|
|
"Reconcile the native decision against the latest issue status version.",
|
|
wakePolicy: null,
|
|
maxAttempts: 3,
|
|
});
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "issue_recovery_action",
|
|
targetId: recovery.id,
|
|
payload: { cause },
|
|
};
|
|
}
|
|
if (effect.kind === "accept_replacement_turn") {
|
|
const [run] = await input.tx
|
|
.update(heartbeatRuns)
|
|
.set({
|
|
status: "running",
|
|
continuationAttempt: sql`${heartbeatRuns.continuationAttempt} + 1`,
|
|
nextAction: "Accept a replacement native turn on the existing run.",
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(
|
|
and(
|
|
eq(heartbeatRuns.id, input.runId),
|
|
eq(heartbeatRuns.companyId, input.companyId),
|
|
),
|
|
)
|
|
.returning({
|
|
id: heartbeatRuns.id,
|
|
continuationAttempt: heartbeatRuns.continuationAttempt,
|
|
});
|
|
if (!run) throw new Error("native_replacement_turn_not_accepted");
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "heartbeat_run",
|
|
targetId: run.id,
|
|
payload: { continuationAttempt: run.continuationAttempt },
|
|
};
|
|
}
|
|
if (effect.kind === "cancel_continuations") {
|
|
const wakeRows = await input.tx
|
|
.select({ id: agentWakeupRequests.id })
|
|
.from(agentWakeupRequests)
|
|
.where(
|
|
and(
|
|
eq(agentWakeupRequests.companyId, input.companyId),
|
|
inArray(agentWakeupRequests.status, [
|
|
"queued",
|
|
"deferred_issue_execution",
|
|
"claimed",
|
|
]),
|
|
sql`coalesce(
|
|
${agentWakeupRequests.payload} ->> 'issueId',
|
|
${agentWakeupRequests.payload} ->> 'taskId',
|
|
${agentWakeupRequests.payload} -> '_paperclipWakeContext' ->> 'issueId',
|
|
${agentWakeupRequests.payload} -> '_paperclipWakeContext' ->> 'taskId'
|
|
) = ${input.issue.id}`,
|
|
),
|
|
);
|
|
if (wakeRows.length > 0) {
|
|
await input.tx
|
|
.update(agentWakeupRequests)
|
|
.set({
|
|
status: "cancelled",
|
|
finishedAt: new Date(),
|
|
error: "Cancelled by native issue cancellation",
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(
|
|
inArray(
|
|
agentWakeupRequests.id,
|
|
wakeRows.map((row) => row.id),
|
|
),
|
|
);
|
|
}
|
|
const interactionRows = await input.tx
|
|
.select({ id: issueThreadInteractions.id })
|
|
.from(issueThreadInteractions)
|
|
.where(
|
|
and(
|
|
eq(issueThreadInteractions.companyId, input.companyId),
|
|
eq(issueThreadInteractions.issueId, input.issue.id),
|
|
eq(issueThreadInteractions.status, "pending"),
|
|
),
|
|
);
|
|
if (interactionRows.length > 0) {
|
|
await input.tx
|
|
.update(issueThreadInteractions)
|
|
.set({
|
|
status: "cancelled",
|
|
resolvedAt: new Date(),
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(
|
|
inArray(
|
|
issueThreadInteractions.id,
|
|
interactionRows.map((row) => row.id),
|
|
),
|
|
);
|
|
}
|
|
await notifyDeliveryWork(input.tx, DELIVERY_QUEUES.accounting, input.companyId);
|
|
const [run] = await input.tx
|
|
.update(heartbeatRuns)
|
|
.set({
|
|
status: "cancelled",
|
|
finishedAt: new Date(),
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(
|
|
and(
|
|
eq(heartbeatRuns.id, input.runId),
|
|
eq(heartbeatRuns.companyId, input.companyId),
|
|
),
|
|
)
|
|
.returning();
|
|
if (!run) throw new Error("native_continuation_cancellation_run_missing");
|
|
input.terminalRunsToEmit?.push(run);
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "heartbeat_run",
|
|
targetId: run.id,
|
|
payload: {
|
|
cancelledWakeIds: wakeRows.map((row) => row.id),
|
|
cancelledInteractionIds: interactionRows.map((row) => row.id),
|
|
},
|
|
};
|
|
}
|
|
if (effect.kind === "append_superseding_assessment") {
|
|
const lineage = await input.tx
|
|
.select({
|
|
currentAssessmentId: statusDecisions.assessmentId,
|
|
})
|
|
.from(statusDecisions)
|
|
.where(
|
|
and(
|
|
eq(statusDecisions.id, input.decisionId),
|
|
eq(statusDecisions.companyId, input.companyId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!lineage) throw new Error("native_superseded_assessment_missing");
|
|
if (!input.issue.lastStatusDecisionId) {
|
|
const assessment = await input.tx
|
|
.select({
|
|
id: workAssessments.id,
|
|
supersedesAssessmentId: workAssessments.supersedesAssessmentId,
|
|
})
|
|
.from(workAssessments)
|
|
.where(
|
|
and(
|
|
eq(workAssessments.id, lineage.currentAssessmentId),
|
|
eq(workAssessments.companyId, input.companyId),
|
|
eq(workAssessments.issueId, input.issue.id),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!assessment?.supersedesAssessmentId) {
|
|
throw new Error("native_superseding_assessment_not_linked");
|
|
}
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "status_decision",
|
|
targetId: input.decisionId,
|
|
payload: {
|
|
supersedesDecisionId: null,
|
|
assessmentId: assessment.id,
|
|
supersedesAssessmentId: assessment.supersedesAssessmentId,
|
|
},
|
|
};
|
|
}
|
|
const prior = await input.tx
|
|
.select({
|
|
id: statusDecisions.id,
|
|
})
|
|
.from(statusDecisions)
|
|
.where(
|
|
and(
|
|
eq(statusDecisions.id, input.issue.lastStatusDecisionId),
|
|
eq(statusDecisions.companyId, input.companyId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!prior) throw new Error("native_superseded_assessment_missing");
|
|
const assessment = await input.tx
|
|
.select({
|
|
id: workAssessments.id,
|
|
supersedesAssessmentId: workAssessments.supersedesAssessmentId,
|
|
})
|
|
.from(workAssessments)
|
|
.where(
|
|
and(
|
|
eq(workAssessments.id, lineage.currentAssessmentId),
|
|
eq(workAssessments.companyId, input.companyId),
|
|
eq(workAssessments.issueId, input.issue.id),
|
|
eq(workAssessments.runId, input.runId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!assessment?.supersedesAssessmentId) {
|
|
throw new Error("native_superseding_assessment_not_linked");
|
|
}
|
|
// Status decisions form one issue-wide sequence and may supersede a
|
|
// decision from another run. Work-assessment lineage is deliberately
|
|
// run-local, so retain the parent recorded when this reassessment was
|
|
// created instead of replacing it with the prior decision's assessment.
|
|
const [decision] = await input.tx
|
|
.update(statusDecisions)
|
|
.set({
|
|
supersedesDecisionId: input.issue.lastStatusDecisionId,
|
|
})
|
|
.where(
|
|
and(
|
|
eq(statusDecisions.id, input.decisionId),
|
|
eq(statusDecisions.companyId, input.companyId),
|
|
),
|
|
)
|
|
.returning({ id: statusDecisions.id });
|
|
if (!decision) throw new Error("native_superseding_assessment_not_linked");
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "status_decision",
|
|
targetId: decision.id,
|
|
payload: {
|
|
supersedesDecisionId: input.issue.lastStatusDecisionId,
|
|
assessmentId: assessment.id,
|
|
supersedesAssessmentId: assessment.supersedesAssessmentId,
|
|
},
|
|
};
|
|
}
|
|
if (effect.kind === "dispatch_pending_effect") {
|
|
const pending = await input.tx
|
|
.select({ id: statusDecisionEffects.id })
|
|
.from(statusDecisionEffects)
|
|
.where(
|
|
and(
|
|
eq(statusDecisionEffects.companyId, input.companyId),
|
|
eq(statusDecisionEffects.issueId, input.issue.id),
|
|
inArray(statusDecisionEffects.deliveryState, ["pending", "failed"]),
|
|
),
|
|
)
|
|
.for("update");
|
|
if (pending.length === 0) throw new Error("native_pending_effect_missing");
|
|
await input.tx
|
|
.update(statusDecisionEffects)
|
|
.set({
|
|
deliveryState: "delivered",
|
|
attemptCount: sql`${statusDecisionEffects.attemptCount} + 1`,
|
|
deliveredAt: new Date(),
|
|
nextAttemptAt: null,
|
|
lastError: null,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(
|
|
inArray(
|
|
statusDecisionEffects.id,
|
|
pending.map((row) => row.id),
|
|
),
|
|
);
|
|
await updateRunEffectState({
|
|
tx: input.tx,
|
|
companyId: input.companyId,
|
|
runId: input.runId,
|
|
key: "nativeDispatchedEffectIds",
|
|
value: pending.map((row) => row.id),
|
|
});
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "heartbeat_run",
|
|
targetId: input.runId,
|
|
payload: { dispatchedEffectIds: pending.map((row) => row.id) },
|
|
};
|
|
}
|
|
if (effect.kind === "increment_status_version") {
|
|
const updated = await issueService(input.tx).update(
|
|
input.issue.id,
|
|
{
|
|
statusVersion: Number(input.issue.statusVersion) + 1,
|
|
actorAgentId: null,
|
|
actorUserId: null,
|
|
},
|
|
input.tx,
|
|
);
|
|
if (!updated) throw new Error("native_status_version_not_incremented");
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "issue",
|
|
targetId: updated.id,
|
|
payload: { statusVersion: updated.statusVersion },
|
|
};
|
|
}
|
|
if (effect.kind === "record_shadow_decision") {
|
|
const targetId = await updateRunEffectState({
|
|
tx: input.tx,
|
|
companyId: input.companyId,
|
|
runId: input.runId,
|
|
key: "nativeShadowDecision",
|
|
value: { decisionId: input.decisionId, applicationState: "shadow" },
|
|
});
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "heartbeat_run",
|
|
targetId,
|
|
payload: { decisionId: input.decisionId, applicationState: "shadow" },
|
|
};
|
|
}
|
|
if (effect.kind === "render_four_layers") {
|
|
const targetId = await updateRunEffectState({
|
|
tx: input.tx,
|
|
companyId: input.companyId,
|
|
runId: input.runId,
|
|
key: "nativeOutcomeLayers",
|
|
value: {
|
|
runStatus: "persisted",
|
|
issueStatus: input.issue.status,
|
|
claim: "preserved",
|
|
decisionId: input.decisionId,
|
|
},
|
|
});
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "heartbeat_run",
|
|
targetId,
|
|
payload: { layerCount: 4 },
|
|
};
|
|
}
|
|
if (effect.kind === "materialize_contract") {
|
|
const contract = await input.tx
|
|
.select({
|
|
id: completionContracts.id,
|
|
canonicalSha256: completionContracts.canonicalSha256,
|
|
})
|
|
.from(completionContracts)
|
|
.where(
|
|
and(
|
|
eq(completionContracts.companyId, input.companyId),
|
|
eq(completionContracts.issueId, input.issue.id),
|
|
),
|
|
)
|
|
.orderBy(desc(completionContracts.revision))
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!contract) throw new Error("native_completion_contract_missing");
|
|
await input.tx
|
|
.update(heartbeatRuns)
|
|
.set({
|
|
completionContractId: contract.id,
|
|
completionContractSha256: contract.canonicalSha256,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(
|
|
and(
|
|
eq(heartbeatRuns.id, input.runId),
|
|
eq(heartbeatRuns.companyId, input.companyId),
|
|
),
|
|
);
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "completion_contract",
|
|
targetId: contract.id,
|
|
payload: { canonicalSha256: contract.canonicalSha256 },
|
|
};
|
|
}
|
|
if (effect.kind === "record_mode_labeled_divergence") {
|
|
const targetId = await updateRunEffectState({
|
|
tx: input.tx,
|
|
companyId: input.companyId,
|
|
runId: input.runId,
|
|
key: "nativeLegacyDivergence",
|
|
value: { mode: "native", decisionId: input.decisionId, classified: true },
|
|
});
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "heartbeat_run",
|
|
targetId,
|
|
payload: { mode: "native", classified: true },
|
|
};
|
|
}
|
|
if (effect.kind === "record_mode_native") {
|
|
const [run] = await input.tx
|
|
.update(heartbeatRuns)
|
|
.set({
|
|
runtimeMode: "native",
|
|
runtimeModeResolvedAt: new Date(),
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(
|
|
and(
|
|
eq(heartbeatRuns.id, input.runId),
|
|
eq(heartbeatRuns.companyId, input.companyId),
|
|
),
|
|
)
|
|
.returning({ id: heartbeatRuns.id });
|
|
if (!run) throw new Error("native_runtime_mode_not_recorded");
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "heartbeat_run",
|
|
targetId: run.id,
|
|
payload: { runtimeMode: "native" },
|
|
};
|
|
}
|
|
if (effect.kind === "record_policy_version") {
|
|
const targetId = await updateRunEffectState({
|
|
tx: input.tx,
|
|
companyId: input.companyId,
|
|
runId: input.runId,
|
|
key: "nativeStatusPolicyVersion",
|
|
value: NATIVE_STATUS_ARBITER_POLICY_VERSION,
|
|
runnerProfile: true,
|
|
});
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "heartbeat_run",
|
|
targetId,
|
|
payload: { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION },
|
|
};
|
|
}
|
|
if (effect.kind === "finish_as_native") {
|
|
const targetId = await updateRunEffectState({
|
|
tx: input.tx,
|
|
companyId: input.companyId,
|
|
runId: input.runId,
|
|
key: "nativeKillSwitchDisposition",
|
|
value: "finish_as_native",
|
|
});
|
|
await input.tx
|
|
.update(heartbeatRuns)
|
|
.set({ runtimeMode: "native", updatedAt: new Date() })
|
|
.where(eq(heartbeatRuns.id, input.runId));
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "heartbeat_run",
|
|
targetId,
|
|
payload: { runtimeMode: "native" },
|
|
};
|
|
}
|
|
if (effect.kind === "resume_workspace_operation") {
|
|
const resumed = input.preMaterializedEffects?.get(effect.kind);
|
|
if (
|
|
!resumed ||
|
|
resumed.targetType !== "workspace_operation" ||
|
|
!resumed.targetId
|
|
) {
|
|
throw new Error("native_workspace_operation_not_executed");
|
|
}
|
|
const operation = await input.tx
|
|
.select({ status: workspaceOperations.status })
|
|
.from(workspaceOperations)
|
|
.where(
|
|
and(
|
|
eq(workspaceOperations.id, resumed.targetId),
|
|
eq(workspaceOperations.companyId, input.companyId),
|
|
eq(workspaceOperations.heartbeatRunId, input.runId),
|
|
eq(workspaceOperations.issueId, input.issue.id),
|
|
eq(workspaceOperations.phase, "workspace_finalize"),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (operation?.status !== "succeeded")
|
|
throw new Error("native_workspace_operation_not_succeeded");
|
|
return resumed;
|
|
}
|
|
if (effect.kind === "record_stale_response") {
|
|
const interaction = await input.tx
|
|
.select({ id: issueThreadInteractions.id })
|
|
.from(issueThreadInteractions)
|
|
.where(
|
|
and(
|
|
eq(issueThreadInteractions.companyId, input.companyId),
|
|
eq(issueThreadInteractions.issueId, input.issue.id),
|
|
inArray(issueThreadInteractions.status, [
|
|
"answered",
|
|
"accepted",
|
|
"rejected",
|
|
"expired",
|
|
]),
|
|
),
|
|
)
|
|
.orderBy(desc(issueThreadInteractions.updatedAt))
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!interaction) throw new Error("native_stale_response_missing");
|
|
await input.tx
|
|
.update(issueThreadInteractions)
|
|
.set({
|
|
summary: "Response retained for audit after native supersession.",
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(issueThreadInteractions.id, interaction.id));
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "issue_thread_interaction",
|
|
targetId: interaction.id,
|
|
payload: { status: "superseded" },
|
|
};
|
|
}
|
|
if (effect.kind === "link_canonical_request") {
|
|
const interactions = await input.tx
|
|
.select({ id: issueThreadInteractions.id })
|
|
.from(issueThreadInteractions)
|
|
.where(
|
|
and(
|
|
eq(issueThreadInteractions.companyId, input.companyId),
|
|
eq(issueThreadInteractions.issueId, input.issue.id),
|
|
),
|
|
)
|
|
.orderBy(asc(issueThreadInteractions.createdAt))
|
|
.limit(2);
|
|
if (interactions.length < 2)
|
|
throw new Error("native_attention_canonical_pair_missing");
|
|
const canonical = interactions[0]!;
|
|
const duplicate = interactions[1]!;
|
|
await input.tx
|
|
.update(issueThreadInteractions)
|
|
.set({
|
|
summary: `Canonical native attention request: ${canonical.id}`,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(issueThreadInteractions.id, duplicate.id));
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "issue_thread_interaction",
|
|
targetId: duplicate.id,
|
|
payload: { canonicalRequestId: canonical.id },
|
|
};
|
|
}
|
|
if (effect.kind === "release_checkout") {
|
|
const [released] = await input.tx
|
|
.update(issues)
|
|
.set({
|
|
checkoutRunId: null,
|
|
executionRunId: null,
|
|
executionAgentNameKey: null,
|
|
executionLockedAt: null,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(
|
|
and(
|
|
eq(issues.id, input.issue.id),
|
|
eq(issues.companyId, input.companyId),
|
|
),
|
|
)
|
|
.returning({ id: issues.id });
|
|
if (!released) throw new Error("native_checkout_not_released");
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "issue_checkout",
|
|
targetId: released.id,
|
|
payload: { checkoutRunId: null, executionRunId: null },
|
|
};
|
|
}
|
|
if (effect.kind === "record_recovery") {
|
|
failAt("recovery_materialization", input.failpoint);
|
|
const recovery = await issueRecoveryActionService(
|
|
input.tx,
|
|
).upsertSourceScoped({
|
|
companyId: input.companyId,
|
|
sourceIssueId: input.issue.id,
|
|
kind: "active_run_watchdog",
|
|
ownerType: "agent",
|
|
ownerAgentId: effect.agentId,
|
|
cause: effect.cause,
|
|
fingerprint: nativeSha256({
|
|
runId: input.runId,
|
|
decisionId: input.decisionId,
|
|
cause: effect.cause,
|
|
}),
|
|
evidence: { runId: input.runId, decisionId: input.decisionId },
|
|
nextAction: effect.nextAction,
|
|
wakePolicy: { kind: "resume_native_run", runId: input.runId },
|
|
maxAttempts: 3,
|
|
});
|
|
return {
|
|
effectKind: effect.kind,
|
|
targetType: "issue_recovery_action",
|
|
targetId: recovery.id,
|
|
payload: { cause: effect.cause, nextAction: effect.nextAction },
|
|
};
|
|
}
|
|
return assertNeverEffect(effect);
|
|
}
|
|
|
|
export async function commitNativeStatusDecision(input: {
|
|
db: Db;
|
|
companyId: string;
|
|
issueId: string;
|
|
runId: string;
|
|
assessmentId: string;
|
|
priorStatus: string;
|
|
priorStatusVersion: number;
|
|
priorDecisionId: string | null;
|
|
decision: NativeStatusDecision;
|
|
failpoint?: NativeStatusCommitFailpoint;
|
|
preMaterializedEffects?: NativeMaterializedStatusEffect[];
|
|
supersedesCommittedDecisionId?: string;
|
|
supersededExternalReviewRequests?: { agentId: string; resultId: string; interactionIds: string[] };
|
|
requireExternalChatResponseWaitAuthorization?: { agentId: string };
|
|
requirePlanWaitSource?: NativePlanWaitSource;
|
|
requireMonitorWait?: { agentId: string; nextCheckAt: string };
|
|
requireMonitorDelivery?: NativeMonitorDelivery;
|
|
requirePreservedMonitorWait?: NativePreservedMonitorWait;
|
|
requireNoPendingChildCompletion?: NativeChildCompletionRecipient;
|
|
requireModelRejectionOwner?: { agentId: string; reviewContext: NativeReviewAssignmentContext | null };
|
|
requireProviderFailureOwner?: { agentId: string; reviewContext: NativeReviewAssignmentContext | null };
|
|
requireBoardResponseWaitSource?: NativeBoardResponseWaitSource;
|
|
requireBoardResponseWaitOrigin?: NativeBoardResponseWaitOrigin;
|
|
reviewResponsePresentation?: {
|
|
agentId: string;
|
|
resultId: string;
|
|
gateId: string;
|
|
};
|
|
}) {
|
|
if (input.decision.reasonCode === null) {
|
|
throw new Error("native_status_reason_code_required");
|
|
}
|
|
const reasonCode = input.decision.reasonCode;
|
|
if (reasonCode === "native_provider_model_rejected" && !input.requireModelRejectionOwner) {
|
|
throw new Error("native_model_rejection_owner_required");
|
|
}
|
|
if (reasonCode.startsWith("native_provider_overloaded") && !input.requireProviderFailureOwner) {
|
|
throw new Error("native_provider_failure_owner_required");
|
|
}
|
|
const publications: ActivityPublication[] = [];
|
|
const terminalRunsToEmit: (typeof heartbeatRuns.$inferSelect)[] = [];
|
|
const committed = await input.db.transaction(async (tx) => {
|
|
const coordinator = await tx
|
|
.select()
|
|
.from(nativeRunFinalizations)
|
|
.where(
|
|
and(
|
|
eq(nativeRunFinalizations.runId, input.runId),
|
|
eq(nativeRunFinalizations.companyId, input.companyId),
|
|
eq(nativeRunFinalizations.issueId, input.issueId),
|
|
),
|
|
)
|
|
.for("update")
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!coordinator)
|
|
throw new Error("native_finalization_coordinator_missing");
|
|
const issue = await tx
|
|
.select()
|
|
.from(issues)
|
|
.where(
|
|
and(
|
|
eq(issues.id, input.issueId),
|
|
eq(issues.companyId, input.companyId),
|
|
),
|
|
)
|
|
.for("update")
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!issue) throw new NativeStatusRaceError();
|
|
// A completed model turn cannot close a persistent conversation. Preserve
|
|
// the task here; the response finalizer records its durable waiting state.
|
|
if (issue.conversationAgentId && input.decision.statusAction === "done") {
|
|
input = { ...input, decision: { ...input.decision, statusAction: "preserve", toStatus: issue.status as NativeStatusDecision["toStatus"], effects: [] } };
|
|
}
|
|
if (coordinator.phase === "committed" && coordinator.decisionId) {
|
|
if (input.supersedesCommittedDecisionId) {
|
|
if (
|
|
coordinator.decisionId !== input.supersedesCommittedDecisionId ||
|
|
coordinator.assessmentId === input.assessmentId
|
|
) {
|
|
throw new NativeStatusRaceError();
|
|
}
|
|
} else {
|
|
const existingDecision = await tx
|
|
.select()
|
|
.from(statusDecisions)
|
|
.where(
|
|
and(
|
|
eq(statusDecisions.id, coordinator.decisionId),
|
|
eq(statusDecisions.companyId, input.companyId),
|
|
eq(statusDecisions.issueId, input.issueId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!existingDecision)
|
|
throw new Error("native_committed_decision_missing");
|
|
return { decision: existingDecision, issue, replayed: true };
|
|
}
|
|
}
|
|
if (
|
|
issue.status !== input.priorStatus ||
|
|
Number(issue.statusVersion) !== input.priorStatusVersion ||
|
|
issue.lastStatusDecisionId !== input.priorDecisionId
|
|
) {
|
|
throw new NativeStatusRaceError();
|
|
}
|
|
if (reasonCode === "scheduled_monitor_waiting") {
|
|
const expected = input.requireMonitorWait;
|
|
const publishedReviewWait = input.decision.statusAction === "in_review" && input.decision.toStatus === "in_review" &&
|
|
await hasPublishedGitHubReview(tx as unknown as Db, input);
|
|
if (!expected || issue.workMode !== "standard" || issue.executionRunId !== input.runId ||
|
|
(input.decision.statusAction !== "preserve" && !publishedReviewWait) || input.decision.effects.length !== 1 ||
|
|
input.decision.effects[0]?.kind !== "release_checkout" ||
|
|
scheduledIssueMonitorWait(issue, expected.agentId) !== expected.nextCheckAt) {
|
|
throw new NativeStatusRaceError();
|
|
}
|
|
}
|
|
if (reasonCode === "scheduled_monitor_dispatched") {
|
|
const expected = input.requireMonitorDelivery;
|
|
const actual = expected ? await readNativeMonitorDelivery(tx as unknown as Db, input) : null;
|
|
if (!expected || !actual || actual.wakeId !== expected.wakeId ||
|
|
actual.successorRunId !== expected.successorRunId ||
|
|
issue.executionRunId === input.runId || issue.checkoutRunId === input.runId ||
|
|
input.decision.statusAction !== "preserve" || input.decision.toStatus !== issue.status ||
|
|
input.decision.effects.length !== 0) throw new NativeStatusRaceError();
|
|
}
|
|
if (reasonCode === "scheduled_monitor_preserved") {
|
|
const expected = input.requirePreservedMonitorWait;
|
|
const actual = expected ? await readNativePreservedMonitorWait(tx as unknown as Db, {
|
|
companyId: input.companyId, issueId: input.issueId, runId: input.runId, agentId: expected.agentId,
|
|
}, issue) : null;
|
|
if (!expected || !actual || actual.nextCheckAt !== expected.nextCheckAt || actual.resultId !== expected.resultId ||
|
|
input.decision.statusAction !== "preserve" || input.decision.toStatus !== issue.status ||
|
|
input.decision.effects.length !== 0 || await pendingNativeGovernance({ db: tx as unknown as Db,
|
|
companyId: input.companyId, issueId: input.issueId, runId: input.runId, executionState: record(issue.executionState),
|
|
})) throw new NativeStatusRaceError();
|
|
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
|
|
}
|
|
// Recheck under the parent status lock: a child can finish between the
|
|
// finalizer's read and this commit. Re-arbitrate instead of discarding its wake.
|
|
if (input.requireNoPendingChildCompletion && await hasPendingNativeChildCompletion(
|
|
tx as unknown as Db, input.requireNoPendingChildCompletion,
|
|
)) throw new NativeStatusRaceError();
|
|
if (input.requireModelRejectionOwner || input.requireProviderFailureOwner) {
|
|
const owner = (input.requireProviderFailureOwner ?? input.requireModelRejectionOwner)!;
|
|
// A successor can claim execution without changing statusVersion. Keep
|
|
// this new blocking authority behind the same locked issue snapshot.
|
|
if (issue.executionRunId && issue.executionRunId !== input.runId) throw new NativeStatusRaceError();
|
|
if (owner.reviewContext) {
|
|
const review = await getNativeReviewAssignment(tx as unknown as Db, {
|
|
companyId: input.companyId, issueId: input.issueId, agentId: owner.agentId, contextSnapshot: owner.reviewContext,
|
|
});
|
|
if (review?.interaction.status !== "pending") throw new NativeStatusRaceError();
|
|
} else if (issue.assigneeAgentId !== owner.agentId || issue.assigneeUserId) {
|
|
throw new NativeStatusRaceError();
|
|
}
|
|
}
|
|
if (input.requireExternalChatResponseWaitAuthorization) {
|
|
let authorization;
|
|
try {
|
|
authorization =
|
|
await resolveExternalChatResponseWaitAuthorizationInTransaction(
|
|
tx as unknown as Db,
|
|
{
|
|
companyId: input.companyId,
|
|
issueId: input.issueId,
|
|
runId: input.runId,
|
|
agentId:
|
|
input.requireExternalChatResponseWaitAuthorization.agentId,
|
|
},
|
|
"nonblocking",
|
|
);
|
|
} catch (error) {
|
|
if (isExternalChatWaitAuthorizationContention(error)) {
|
|
throw new NativeStatusRaceError();
|
|
}
|
|
throw error;
|
|
}
|
|
if (authorization !== "authorized") {
|
|
throw new NativeStatusRaceError();
|
|
}
|
|
}
|
|
const passiveBoardResponseWait =
|
|
reasonCode === "board_response_waiting" ||
|
|
reasonCode === "board_response_wait_superseded";
|
|
let planWait: Awaited<ReturnType<typeof readNativePlanWait>> = null;
|
|
let planWaitResult: { resultId: string; resultSha256: string } | null = null;
|
|
if (reasonCode === "native_plan_accepted_waiting_for_continuation") {
|
|
const expected = input.requirePlanWaitSource;
|
|
if (!expected || expected.companyId !== input.companyId || expected.issueId !== input.issueId || expected.runId !== input.runId || input.decision.effects.length !== 0 || input.decision.statusAction !== "in_progress" || input.decision.toStatus !== "in_progress") throw new NativeStatusRaceError();
|
|
try {
|
|
planWait = await readNativePlanWait(tx as unknown as Db, expected, true);
|
|
if (!planWait || nativeSha256(planWait.source) !== nativeSha256(expected)) throw new NativeStatusRaceError();
|
|
// Recheck the committed semantic result too: an arbitrary caller cannot
|
|
// use a genuine plan receipt to authorize a different finalization.
|
|
const [accepted] = await tx.select().from(nativeRunResults).where(and(
|
|
eq(nativeRunResults.id, coordinator!.resultId!), eq(nativeRunResults.companyId, input.companyId),
|
|
eq(nativeRunResults.issueId, input.issueId), eq(nativeRunResults.runId, input.runId), eq(nativeRunResults.schemaStatus, "accepted"),
|
|
)).for("share", { noWait: true });
|
|
const envelope = record(accepted?.resultJson), terminal = record(envelope.terminal);
|
|
if (!accepted || accepted.completionContractId !== planWait.source.contractId || accepted.turnId !== planWait.source.turnId ||
|
|
nativeSha256(envelope.result) !== nativeSha256(planWait.result) ||
|
|
terminal.schema !== "paperclip.prp.terminal.v1" || terminal.runTerminalState !== "succeeded" || terminal.turnTerminalState !== "completed" || terminal.reportedWorkDisposition !== "yielded") throw new NativeStatusRaceError();
|
|
// completeRun hashes its full private binding, not the stored resultJson.
|
|
// Preserve that accepted identity instead of inventing a new digest.
|
|
planWaitResult = { resultId: accepted.id, resultSha256: accepted.canonicalSha256 };
|
|
} catch (error) {
|
|
if (isExternalChatWaitAuthorizationContention(error)) throw new NativeStatusRaceError();
|
|
throw error;
|
|
}
|
|
}
|
|
let boardResponseWaitOrigin: NativeBoardResponseWaitOrigin | null = null;
|
|
if (
|
|
passiveBoardResponseWait ||
|
|
input.requireBoardResponseWaitOrigin
|
|
) {
|
|
const expected = input.requireBoardResponseWaitOrigin;
|
|
if (
|
|
!expected ||
|
|
expected.companyId !== input.companyId ||
|
|
expected.issueId !== input.issueId ||
|
|
expected.runId !== input.runId ||
|
|
(passiveBoardResponseWait && input.decision.effects.length !== 0)
|
|
)
|
|
throw new NativeStatusRaceError();
|
|
try {
|
|
boardResponseWaitOrigin = await readNativeBoardResponseWaitOrigin(
|
|
tx as unknown as Db,
|
|
{
|
|
companyId: input.companyId,
|
|
issueId: input.issueId,
|
|
runId: input.runId,
|
|
agentId: expected.agentId,
|
|
},
|
|
true,
|
|
);
|
|
} catch (error) {
|
|
if (isExternalChatWaitAuthorizationContention(error))
|
|
throw new NativeStatusRaceError();
|
|
throw error;
|
|
}
|
|
if (
|
|
!boardResponseWaitOrigin ||
|
|
nativeSha256(boardResponseWaitOrigin) !== nativeSha256(expected)
|
|
)
|
|
throw new NativeStatusRaceError();
|
|
}
|
|
let boardResponseWait: Awaited<
|
|
ReturnType<typeof readNativeBoardResponseWaitSource>
|
|
> = null;
|
|
// Corrective continuations also carry this precondition. Revalidate it
|
|
// under the same locks before any decision effects can restart old work.
|
|
if (reasonCode === "board_response_waiting" || input.requireBoardResponseWaitSource) {
|
|
const expected = input.requireBoardResponseWaitSource;
|
|
if (
|
|
!expected ||
|
|
(passiveBoardResponseWait && input.decision.effects.length !== 0) ||
|
|
expected.companyId !== input.companyId ||
|
|
expected.issueId !== input.issueId ||
|
|
expected.runId !== input.runId
|
|
)
|
|
throw new NativeStatusRaceError();
|
|
try {
|
|
boardResponseWait = await readNativeBoardResponseWaitSource(
|
|
tx as unknown as Db,
|
|
{
|
|
companyId: input.companyId,
|
|
issueId: input.issueId,
|
|
runId: input.runId,
|
|
agentId: expected.agentId,
|
|
},
|
|
true,
|
|
);
|
|
} catch (error) {
|
|
if (isExternalChatWaitAuthorizationContention(error))
|
|
throw new NativeStatusRaceError();
|
|
throw error;
|
|
}
|
|
if (
|
|
!boardResponseWait ||
|
|
nativeSha256(boardResponseWait.source) !== nativeSha256(expected)
|
|
) {
|
|
throw new NativeStatusRaceError();
|
|
}
|
|
}
|
|
let externalChatReviewPresentation = null;
|
|
if (
|
|
reasonCode === "governed_response_waiting" &&
|
|
input.reviewResponsePresentation &&
|
|
input.decision.effects.length === 1 &&
|
|
input.decision.effects[0]?.kind === "create_interaction" &&
|
|
input.decision.effects[0].gate?.kind === "interaction" &&
|
|
input.decision.effects[0].gate.id ===
|
|
input.reviewResponsePresentation.gateId
|
|
) {
|
|
try {
|
|
externalChatReviewPresentation =
|
|
await prepareNativeChatReviewPresentationInTransaction(
|
|
tx as unknown as Db,
|
|
{
|
|
companyId: input.companyId,
|
|
issueId: input.issueId,
|
|
runId: input.runId,
|
|
assessmentId: input.assessmentId,
|
|
...input.reviewResponsePresentation,
|
|
},
|
|
);
|
|
} catch (error) {
|
|
if (isExternalChatWaitAuthorizationContention(error))
|
|
throw new NativeStatusRaceError();
|
|
throw error;
|
|
}
|
|
}
|
|
const recordCoordinatorDecision = async (decisionId: string) => {
|
|
const finalizationError = input.decision.effects.find(
|
|
effect => effect.kind === "record_finalization_error",
|
|
);
|
|
await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.companyId);
|
|
await tx
|
|
.update(nativeRunFinalizations)
|
|
.set({
|
|
phase: finalizationError ? "retryable_failure" : "committed",
|
|
assessmentId: input.assessmentId,
|
|
decisionId,
|
|
leaseOwner: null,
|
|
leaseExpiresAt: null,
|
|
recoveryState: null,
|
|
recoveryRequestId: null,
|
|
failureCode: finalizationError?.cause ?? null,
|
|
failureDetail: finalizationError
|
|
? {
|
|
originalFailureCode: finalizationError.cause,
|
|
recoveryOwner: {
|
|
kind: "agent",
|
|
agentId: finalizationError.agentId,
|
|
},
|
|
nextAction: finalizationError.nextAction,
|
|
}
|
|
: null,
|
|
nextAttemptAt: finalizationError ? new Date(Date.now() + 30_000) : null,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(nativeRunFinalizations.runId, input.runId));
|
|
};
|
|
const decisionJson = {
|
|
statusAction: input.decision.statusAction,
|
|
toStatus: input.decision.toStatus,
|
|
reasonCode,
|
|
unblockDescriptor: input.decision.unblockDescriptor,
|
|
effects: input.decision.effects,
|
|
...(externalChatReviewPresentation
|
|
? { externalChatReviewPresentation }
|
|
: {}),
|
|
...(boardResponseWait
|
|
? { boardResponseWait: boardResponseWait.source }
|
|
: {}),
|
|
...(boardResponseWaitOrigin ? { boardResponseWaitOrigin } : {}),
|
|
...(planWait ? { planWait: planWait.source, planWaitResult } : {}),
|
|
...(reasonCode === "scheduled_monitor_dispatched" ? { monitorDelivery: input.requireMonitorDelivery } : {}),
|
|
priorStatusVersion: input.priorStatusVersion,
|
|
projectedStatusVersion:
|
|
input.decision.statusAction === "preserve"
|
|
? input.priorStatusVersion
|
|
: input.priorStatusVersion + 1,
|
|
};
|
|
const decisionDigest = nativeSha256({
|
|
issueId: input.issueId,
|
|
assessmentId: input.assessmentId,
|
|
policyVersion: input.decision.policyVersion,
|
|
fromStatus: issue.status,
|
|
priorStatusVersion: input.priorStatusVersion,
|
|
priorDecisionId: input.priorDecisionId,
|
|
decision: decisionJson,
|
|
});
|
|
let decisionRow = await tx
|
|
.select()
|
|
.from(statusDecisions)
|
|
.where(
|
|
and(
|
|
eq(statusDecisions.issueId, input.issueId),
|
|
eq(statusDecisions.decisionDigest, decisionDigest),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (
|
|
decisionRow?.applicationState === "applied" ||
|
|
decisionRow?.applicationState === "proposed"
|
|
) {
|
|
// The effects already committed, but this retry owns a fresh coordinator
|
|
// lease. Release it and restore the retry deadline without replaying effects.
|
|
await recordCoordinatorDecision(decisionRow.id);
|
|
return { decision: decisionRow, issue, replayed: true };
|
|
}
|
|
if (!decisionRow) {
|
|
const latestDecisionVersion = await tx
|
|
.select({
|
|
decisionVersion: statusDecisions.decisionVersion,
|
|
})
|
|
.from(statusDecisions)
|
|
.where(
|
|
and(
|
|
eq(statusDecisions.companyId, input.companyId),
|
|
eq(statusDecisions.issueId, input.issueId),
|
|
),
|
|
)
|
|
.orderBy(desc(statusDecisions.decisionVersion))
|
|
.limit(1)
|
|
.then((rows) => Number(rows[0]?.decisionVersion ?? 0));
|
|
[decisionRow] = await tx
|
|
.insert(statusDecisions)
|
|
.values({
|
|
companyId: input.companyId,
|
|
issueId: input.issueId,
|
|
runId: input.runId,
|
|
assessmentId: input.assessmentId,
|
|
decisionVersion:
|
|
Math.max(input.priorStatusVersion, latestDecisionVersion) + 1,
|
|
policyVersion: input.decision.policyVersion,
|
|
fromStatus: issue.status,
|
|
toStatus: input.decision.toStatus,
|
|
reasonCode,
|
|
decisionJson,
|
|
decisionDigest,
|
|
applicationState: "proposed",
|
|
})
|
|
.returning();
|
|
}
|
|
if (!decisionRow) throw new Error("native_status_decision_not_persisted");
|
|
|
|
if (planWait) {
|
|
await issueService(tx as unknown as Db).addComment(
|
|
input.issueId, planWait.result.summary,
|
|
{ agentId: planWait.source.agentId, runId: input.runId }, undefined, tx,
|
|
);
|
|
}
|
|
if (boardResponseWait && reasonCode === "board_response_waiting") {
|
|
// The answer and passive-wait receipt commit together. In particular,
|
|
// no chat presentation authorization is supplied: this is Board-only.
|
|
await issueService(tx as unknown as Db).addComment(
|
|
input.issueId,
|
|
boardResponseWait.summary,
|
|
{ agentId: boardResponseWait.source.agentId, runId: input.runId },
|
|
undefined,
|
|
tx,
|
|
);
|
|
}
|
|
|
|
if (input.decision.statusAction === "done" && issue.status !== "done" &&
|
|
issue.parentId && issue.originKind !== "task_watchdog") {
|
|
// Serialize child notification creation with the parent's final pending-
|
|
// result check. Lock before child writes can acquire an implicit parent
|
|
// foreign-key lock, so concurrent siblings never upgrade shared locks.
|
|
await tx.select({ id: issues.id }).from(issues).where(and(
|
|
eq(issues.id, issue.parentId), eq(issues.companyId, input.companyId),
|
|
)).for("update");
|
|
}
|
|
|
|
if (input.supersededExternalReviewRequests) {
|
|
const expected = input.supersededExternalReviewRequests;
|
|
const binding = { companyId: input.companyId, issueId: input.issueId, runId: input.runId, agentId: expected.agentId };
|
|
const candidates = await findSupersededExternalReviewRequests(tx as unknown as Db, binding);
|
|
const ids = candidates.map(({ interaction }) => interaction.id);
|
|
const [accepted] = await tx.select().from(nativeRunResults).where(and(
|
|
eq(nativeRunResults.id, coordinator.resultId!), eq(nativeRunResults.companyId, input.companyId),
|
|
eq(nativeRunResults.issueId, input.issueId), eq(nativeRunResults.runId, input.runId), eq(nativeRunResults.schemaStatus, "accepted"),
|
|
));
|
|
const result = record(accepted?.resultJson.result), terminal = record(accepted?.resultJson.terminal);
|
|
if (ids.length !== expected.interactionIds.length || expected.interactionIds.some(id => !ids.includes(id)) ||
|
|
terminal.runTerminalState !== "succeeded" || terminal.turnTerminalState !== "completed" ||
|
|
!["done", "yielded"].includes(String(result.reportedWorkDisposition)) ||
|
|
!accepted || accepted.id !== expected.resultId || normalizePrpResultSignals(accepted.resultJson.result as PrpStructuredRunResult).actionableAttentionRequests.length !== 0 ||
|
|
!["completion_claim_policy_accepted", "completion_contract_satisfied", "external_chat_response_waiting", "scheduled_monitor_waiting"].includes(reasonCode) ||
|
|
await pendingNativeGovernance({ db: tx as unknown as Db, ...binding, executionState: record(issue.executionState), ignoredInteractionIds: ids })) {
|
|
throw new NativeStatusRaceError();
|
|
}
|
|
await supersedeExternalReviewRequestsInTransaction(tx as unknown as Db, binding, candidates, publications);
|
|
}
|
|
|
|
const materialized: NativeMaterializedStatusEffect[] = [];
|
|
const preMaterializedEffects = new Map(
|
|
(input.preMaterializedEffects ?? []).map((effect) => [
|
|
effect.effectKind,
|
|
effect,
|
|
]),
|
|
);
|
|
for (const effect of input.decision.effects) {
|
|
materialized.push(
|
|
await materializeDecisionEffect({
|
|
tx: tx as unknown as Db,
|
|
companyId: input.companyId,
|
|
issue,
|
|
runId: input.runId,
|
|
decisionId: decisionRow.id,
|
|
effect,
|
|
failpoint: input.failpoint,
|
|
preMaterializedEffects,
|
|
terminalRunsToEmit,
|
|
}),
|
|
);
|
|
}
|
|
|
|
let updated: typeof issues.$inferSelect;
|
|
if (input.decision.statusAction === "preserve") {
|
|
updated = (await tx
|
|
.select()
|
|
.from(issues)
|
|
.where(
|
|
and(
|
|
eq(issues.id, input.issueId),
|
|
eq(issues.companyId, input.companyId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null)) as typeof issues.$inferSelect;
|
|
if (!updated) throw new NativeStatusRaceError();
|
|
} else {
|
|
failAt("status_projection", input.failpoint);
|
|
const projected = await issueService(tx as unknown as Db).update(
|
|
input.issueId,
|
|
{
|
|
status: input.decision.toStatus,
|
|
statusVersion: input.priorStatusVersion + 1,
|
|
lastStatusDecisionId: decisionRow.id,
|
|
unblockDescriptor: input.decision.unblockDescriptor,
|
|
actorAgentId: null,
|
|
actorUserId: null,
|
|
},
|
|
tx,
|
|
publications,
|
|
);
|
|
if (!projected) throw new NativeStatusRaceError();
|
|
updated = projected;
|
|
materialized.unshift({
|
|
effectKind: "issue_status_projection",
|
|
targetType: "issue",
|
|
targetId: input.issueId,
|
|
payload: {
|
|
fromStatus: issue.status,
|
|
toStatus: input.decision.toStatus,
|
|
reasonCode,
|
|
},
|
|
});
|
|
}
|
|
|
|
if (input.decision.statusAction === "done" && issue.status !== "done") {
|
|
const issueSvc = issueService(tx as unknown as Db);
|
|
const dependents = await issueSvc.listWakeableBlockedDependents(
|
|
input.issueId,
|
|
);
|
|
const completedResultSummary = await tx
|
|
.select({ resultJson: nativeRunResults.resultJson })
|
|
.from(nativeRunResults)
|
|
.where(
|
|
and(
|
|
eq(nativeRunResults.companyId, input.companyId),
|
|
eq(nativeRunResults.runId, input.runId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => {
|
|
const summary = record(record(rows[0]?.resultJson).result).summary;
|
|
return typeof summary === "string" && summary.trim().length > 0
|
|
? summary.trim()
|
|
: null;
|
|
});
|
|
const parent = issue.parentId
|
|
? await issueSvc.getWakeableParentAfterChildCompletion(issue.parentId, {
|
|
issueId: input.issueId,
|
|
summary: completedResultSummary,
|
|
})
|
|
: null;
|
|
const parentIsDependent = parent
|
|
? dependents.some((dependent) => dependent.id === parent.id)
|
|
: false;
|
|
// Ordinary children can finish new work after reopening. Key that wake to
|
|
// the committed decision; replaying the decision still shares its identity.
|
|
// Watchdog children retain their stable key to avoid feedback loops.
|
|
const childCompletionDecisionId = issue.originKind === "task_watchdog" ? null : decisionRow.id;
|
|
const childCompletionRevision = childCompletionDecisionId ? `:${childCompletionDecisionId}` : "";
|
|
// The dispatcher must retain this new input until a fresh parent turn can
|
|
// read it. Watchdog handoffs keep their existing coalescing semantics.
|
|
const childCompletionDelivery = childCompletionDecisionId
|
|
? { nativeChildCompletionDecisionId: childCompletionDecisionId }
|
|
: {};
|
|
for (const dependent of dependents) {
|
|
const isCompletedChildParent = parent?.id === dependent.id;
|
|
const idempotencyKey = isCompletedChildParent
|
|
? `issue_children_completed:${dependent.id}:${input.issueId}${childCompletionRevision}`
|
|
: buildIssueBlockersResolvedWakeIdempotencyKey({
|
|
dependentIssueId: dependent.id,
|
|
resolvedBlockerIssueId: input.issueId,
|
|
});
|
|
const childCompletionContext =
|
|
isCompletedChildParent && parent
|
|
? {
|
|
completedChildIssueId: input.issueId,
|
|
...childCompletionDelivery,
|
|
childIssueIds: parent.childIssueIds,
|
|
childIssueSummaries: parent.childIssueSummaries,
|
|
onboardingCompletion: parent.onboardingCompletion,
|
|
childIssueSummaryTruncated: parent.childIssueSummaryTruncated,
|
|
}
|
|
: null;
|
|
const wakeId = await enqueueWake({
|
|
tx: tx as unknown as Db,
|
|
companyId: input.companyId,
|
|
issueId: dependent.id,
|
|
agentId: dependent.assigneeAgentId,
|
|
reason: isCompletedChildParent
|
|
? "issue_children_completed"
|
|
: "issue_blockers_resolved",
|
|
idempotencyKey,
|
|
payload: {
|
|
resolvedBlockerIssueId: input.issueId,
|
|
blockerIssueIds: dependent.blockerIssueIds,
|
|
...(childCompletionContext ?? {}),
|
|
},
|
|
contextSnapshot: childCompletionContext ?? undefined,
|
|
});
|
|
materialized.push({
|
|
effectKind: isCompletedChildParent
|
|
? "parent_dependency_wake"
|
|
: "dependency_wake",
|
|
targetType: "agent_wakeup_request",
|
|
targetId: wakeId,
|
|
payload: {
|
|
dependentIssueId: dependent.id,
|
|
resolvedBlockerIssueId: input.issueId,
|
|
...(childCompletionContext ?? {}),
|
|
},
|
|
});
|
|
}
|
|
if (parent && !parentIsDependent) {
|
|
const wakeId = await enqueueWake({
|
|
tx: tx as unknown as Db,
|
|
companyId: input.companyId,
|
|
issueId: parent.id,
|
|
agentId: parent.assigneeAgentId,
|
|
reason: "issue_children_completed",
|
|
idempotencyKey: `issue_children_completed:${parent.id}:${input.issueId}${childCompletionRevision}`,
|
|
payload: {
|
|
completedChildIssueId: input.issueId,
|
|
...childCompletionDelivery,
|
|
childIssueIds: parent.childIssueIds,
|
|
childIssueSummaries: parent.childIssueSummaries,
|
|
onboardingCompletion: parent.onboardingCompletion,
|
|
childIssueSummaryTruncated: parent.childIssueSummaryTruncated,
|
|
},
|
|
contextSnapshot: {
|
|
completedChildIssueId: input.issueId,
|
|
...childCompletionDelivery,
|
|
childIssueIds: parent.childIssueIds,
|
|
childIssueSummaries: parent.childIssueSummaries,
|
|
onboardingCompletion: parent.onboardingCompletion,
|
|
childIssueSummaryTruncated: parent.childIssueSummaryTruncated,
|
|
},
|
|
});
|
|
materialized.push({
|
|
effectKind: "parent_wake",
|
|
targetType: "agent_wakeup_request",
|
|
targetId: wakeId,
|
|
payload: {
|
|
parentIssueId: parent.id,
|
|
completedChildIssueId: input.issueId,
|
|
...childCompletionDelivery,
|
|
childIssueSummaries: parent.childIssueSummaries,
|
|
onboardingCompletion: parent.onboardingCompletion,
|
|
childIssueSummaryTruncated: parent.childIssueSummaryTruncated,
|
|
},
|
|
});
|
|
}
|
|
}
|
|
|
|
for (const [index, effect] of materialized.entries()) {
|
|
await tx
|
|
.insert(statusDecisionEffects)
|
|
.values({
|
|
companyId: input.companyId,
|
|
issueId: input.issueId,
|
|
decisionId: decisionRow.id,
|
|
ordinal: index + 1,
|
|
effectKind: effect.effectKind,
|
|
targetType: effect.targetType,
|
|
targetId: effect.targetId,
|
|
idempotencyKey: `native-status:${decisionRow.id}:${index + 1}`,
|
|
payload: effect.payload,
|
|
deliveryState: "delivered",
|
|
attemptCount: 1,
|
|
deliveredAt: new Date(),
|
|
})
|
|
.onConflictDoNothing();
|
|
}
|
|
const shadowOnly = input.decision.effects.some(
|
|
(effect) => effect.kind === "record_shadow_decision",
|
|
);
|
|
const applicationState = shadowOnly ? "proposed" : "applied";
|
|
await tx
|
|
.update(statusDecisions)
|
|
.set({
|
|
applicationState,
|
|
appliedAt: shadowOnly ? null : new Date(),
|
|
})
|
|
.where(eq(statusDecisions.id, decisionRow.id));
|
|
await recordCoordinatorDecision(decisionRow.id);
|
|
if (externalChatReviewPresentation && input.reviewResponsePresentation) {
|
|
await restoreNativeChatReviewPresentationInTransaction(
|
|
tx as unknown as Db,
|
|
{
|
|
companyId: input.companyId,
|
|
issueId: input.issueId,
|
|
runId: input.runId,
|
|
decisionId: decisionRow.id,
|
|
resultId: input.reviewResponsePresentation.resultId,
|
|
assessmentId: input.assessmentId,
|
|
},
|
|
);
|
|
}
|
|
const { publication } = await persistActivity(tx as unknown as Db, {
|
|
companyId: input.companyId,
|
|
actorType: "system",
|
|
actorId: "native-status-committer",
|
|
action:
|
|
input.decision.statusAction === "preserve"
|
|
? "issue.status_decision_recorded"
|
|
: "issue.updated",
|
|
entityType: "issue",
|
|
entityId: input.issueId,
|
|
issueId: input.issueId,
|
|
runId: input.runId,
|
|
details: {
|
|
source: "native_status_decision",
|
|
assessmentId: input.assessmentId,
|
|
decisionId: decisionRow.id,
|
|
fromStatus: issue.status,
|
|
toStatus: input.decision.toStatus,
|
|
reasonCode,
|
|
effectCount: materialized.length,
|
|
},
|
|
});
|
|
publications.push(publication);
|
|
return {
|
|
decision: { ...decisionRow, applicationState },
|
|
issue: updated,
|
|
replayed: false,
|
|
};
|
|
});
|
|
|
|
for (const publication of publications) publishActivity(publication);
|
|
for (const terminalRun of terminalRunsToEmit)
|
|
await emitAgentTaskRun(input.db, terminalRun);
|
|
return committed;
|
|
}
|