mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-05 21:02:34 +01:00
Merge branch 'codex/claude-runner-default' into codex/claude-runner-platforms
* codex/claude-runner-default: fix(adapter-utils): preserve SSH command positional arguments feat(runner): qualify native Codex platforms and cache hosted builds docs: establish paired feature and agent access maps (#16025) fix: retain matching-host runner development overrides fix(acpx): record turn activity at the moment a stop fires (#15962) refactor(chat): extract channel status, receipts, drafts and authority workflows (#15995) fix: preserve saved setup and match sandbox runner targets feat: add portable verification skills and refresh the feature map (#15983)
This commit is contained in:
Dotta committed
commit
02ad4267d4
120 files changed
+11143
-3797
No files matched your search
@@ -657,6 +657,78 @@ describe("agent routes adapter validation", () => {
|
||||
expect(res.status, JSON.stringify(res.body)).toBe(200);
|
||||
});
|
||||
|
||||
it.each(["pass", "fail"] as const)("tests saved legacy Codex on a disabled adapter and retains its %s runtime verdict", async (status) => {
|
||||
const adapterConfig = { model: "gpt-5.4", extraArgs: ["--legacy-setting"], env: { CODEX_HOME: "/saved-codex-home" } };
|
||||
mockAgentService.getById.mockResolvedValue({ ...(await mockAgentService.getById()), adapterConfig });
|
||||
mockAdapterPluginStore.getDisabledAdapterTypes.mockReturnValue(["codex_local"]);
|
||||
const probe = vi.fn(async () => ({ adapterType: "codex_local", status, testedAt: new Date(0).toISOString(),
|
||||
checks: status === "fail" ? [{ code: "adapter_auth_missing", level: "error" as const, message: "Sign in to the saved Codex account." }] : [] }));
|
||||
const { registerServerAdapter } = await import("../adapters/index.js");
|
||||
registerServerAdapter({ ...externalAdapter, type: "codex_local", testEnvironment: probe });
|
||||
try {
|
||||
const app = await createApp();
|
||||
const res = await requestApp(app, baseUrl => request(baseUrl)
|
||||
.post("/api/companies/company-1/adapters/codex_local/test-environment")
|
||||
.send({ agentId: "11111111-1111-4111-8111-111111111111", adapterConfig: {
|
||||
...adapterConfig, env: { CODEX_HOME: { type: "plain", value: "***REDACTED***" } },
|
||||
} }));
|
||||
expect(res.status, JSON.stringify(res.body)).toBe(200);
|
||||
expect(res.body).toMatchObject({ adapterType: "codex_local", status });
|
||||
expect(probe).toHaveBeenCalledOnce();
|
||||
expect(probe).toHaveBeenCalledWith(expect.objectContaining({ adapterType: "codex_local", config: adapterConfig }));
|
||||
expect(mockSecretService.normalizeAdapterConfigForPersistence).toHaveBeenCalledWith("company-1", adapterConfig, expect.objectContaining({ adapterType: "codex_local" }));
|
||||
expect(mockAgentService.create).not.toHaveBeenCalled();
|
||||
expect(mockAgentService.update).not.toHaveBeenCalled();
|
||||
} finally { await unregisterTestAdapter("codex_local"); }
|
||||
});
|
||||
|
||||
it.each(["draft", "explicit legacy", "explicit auto", "explicit native", "changed harness"] as const)("rejects a disabled Codex setup selection for %s", async (mode) => {
|
||||
mockAdapterPluginStore.getDisabledAdapterTypes.mockReturnValue(["codex_local"]);
|
||||
if (mode === "changed harness") mockAgentService.getById.mockResolvedValue({ ...(await mockAgentService.getById()), adapterType: "external_test" });
|
||||
const probe = vi.fn();
|
||||
const { registerServerAdapter } = await import("../adapters/index.js");
|
||||
registerServerAdapter({ ...externalAdapter, type: "codex_local", testEnvironment: probe });
|
||||
try {
|
||||
const runner = mode === "explicit legacy" ? "legacy" : mode === "explicit auto" ? "auto" : mode === "explicit native" ? "paperclip" : undefined;
|
||||
const app = await createApp();
|
||||
const res = await requestApp(app, baseUrl => request(baseUrl)
|
||||
.post("/api/companies/company-1/adapters/codex_local/test-environment")
|
||||
.send({ adapterConfig: {}, ...(mode === "draft" ? {} : { agentId: "11111111-1111-4111-8111-111111111111" }), ...(runner ? { runner } : {}) }));
|
||||
expect(res.status, JSON.stringify(res.body)).toBe(422);
|
||||
expect(res.body.details).toMatchObject({ code: "agent_runner_unavailable" });
|
||||
expect(probe).not.toHaveBeenCalled();
|
||||
expect(mockSecretService.normalizeAdapterConfigForPersistence).not.toHaveBeenCalled();
|
||||
} finally { await unregisterTestAdapter("codex_local"); }
|
||||
});
|
||||
|
||||
it("still rejects new Codex creation when its adapter is disabled", async () => {
|
||||
mockAdapterPluginStore.getDisabledAdapterTypes.mockReturnValue(["codex_local"]);
|
||||
const app = await createApp();
|
||||
const res = await requestApp(app, baseUrl => request(baseUrl)
|
||||
.post("/api/companies/company-1/agents").send({ name: "New Codex", adapterType: "codex_local" }));
|
||||
expect(res.status, JSON.stringify(res.body)).toBe(422);
|
||||
expect(res.body.details).toMatchObject({ code: "agent_runner_unavailable" });
|
||||
expect(mockAgentService.create).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("rechecks selection policy when a saved native adapter switches its underlying harness", async () => {
|
||||
mockAgentService.getById.mockResolvedValue({ ...(await mockAgentService.getById()), adapterType: "paperclip_runner", adapterConfig: { provider: "codex" } });
|
||||
mockInstanceSettingsService.getExperimental.mockResolvedValue({ enableNativeRunner: false });
|
||||
const probe = vi.fn();
|
||||
const { registerServerAdapter } = await import("../adapters/index.js");
|
||||
registerServerAdapter({ ...externalAdapter, type: "paperclip_runner", testEnvironment: probe });
|
||||
try {
|
||||
const app = await createApp();
|
||||
const res = await requestApp(app, baseUrl => request(baseUrl)
|
||||
.post("/api/companies/company-1/adapters/paperclip_runner/test-environment")
|
||||
.send({ agentId: "11111111-1111-4111-8111-111111111111", adapterConfig: { provider: "acpx", acpxAgent: "claude" } }));
|
||||
expect(res.status, JSON.stringify(res.body)).toBe(422);
|
||||
expect(res.body.details).toMatchObject({ code: "paperclip_runner_rollout_disabled" });
|
||||
expect(probe).not.toHaveBeenCalled();
|
||||
expect(mockSecretService.normalizeAdapterConfigForPersistence).not.toHaveBeenCalled();
|
||||
} finally { await unregisterTestAdapter("paperclip_runner"); }
|
||||
});
|
||||
|
||||
it("still creates an agent on an adapter that is registered and enabled", async () => {
|
||||
const { registerServerAdapter } = await import("../adapters/index.js");
|
||||
registerServerAdapter(externalAdapter);
|
||||
|
||||
@@ -3142,15 +3142,21 @@ export function agentRoutes(
|
||||
? restoreRedactedAgentEnv(inputAdapterConfig, savedAgent.adapterConfig)
|
||||
: inputAdapterConfig;
|
||||
}
|
||||
const selection = await resolveNewAgentRunnerForCompany(db, companyId, {
|
||||
adapterType: type, adapterConfig: adapterConfigForTest,
|
||||
runner: req.body.runner ?? (savedAgent && savedAgent.adapterType === type ? (type === "paperclip_runner" ? undefined : "legacy") : undefined),
|
||||
defaultEnvironmentId: requestedEnvironmentId,
|
||||
});
|
||||
const retainsSavedExecution = savedAgent && req.body.runner === undefined
|
||||
&& savedAgent.adapterType === type
|
||||
&& agentHarnessType(type, adapterConfigForTest) === agentHarnessType(savedAgent.adapterType, savedAgent.adapterConfig);
|
||||
// Preserve saved execution; the adapter probe below still validates runtime readiness.
|
||||
const selection = retainsSavedExecution
|
||||
? { adapterType: type, adapterConfig: adapterConfigForTest }
|
||||
: await resolveNewAgentRunnerForCompany(db, companyId, {
|
||||
adapterType: type, adapterConfig: adapterConfigForTest,
|
||||
runner: req.body.runner,
|
||||
defaultEnvironmentId: requestedEnvironmentId,
|
||||
});
|
||||
type = selection.adapterType;
|
||||
adapterConfigForTest = selection.adapterConfig;
|
||||
const adapter = requireServerAdapter(type);
|
||||
if (!savedAgent || req.body.runner !== undefined || type !== savedAgent.adapterType) await assertSelectableAdapterType(type, adapterConfigForTest);
|
||||
if (!retainsSavedExecution) await assertSelectableAdapterType(type, adapterConfigForTest);
|
||||
if (poolBinding?.mode === "router") {
|
||||
await validateManagedAgentBinding(req, companyId, savedAgentId ?? "", type, adapterConfigForTest, poolBinding, requestedEnvironmentId, false, !savedAgentId);
|
||||
res.json({ adapterType: type, status: "warn", testedAt: new Date().toISOString(), checks: [{ code: "ai_connection_pool_task_test_required", level: "warn", message: "Pool configuration is available. Run a task to verify the selected account and harness; this preview does not allocate a task or contact a provider." }] });
|
||||
|
||||
@@ -323,11 +323,29 @@ describe("run failure diagnostics", () => {
|
||||
expect(JSON.stringify(result)).not.toContain("private");
|
||||
});
|
||||
|
||||
it("includes stop-time ACP activity and closed pending tool-kind labels", () => {
|
||||
const result = sanitizeRunFailureDiagnostics(collectRunFailureDiagnostics(run({ resultJson: {
|
||||
acpLastEventAgeMs: 700, acpLastEventAgeAtStopMs: 14_000_000, acpOldestPendingToolAgeAtStopMs: 14_100_000,
|
||||
acpPendingToolKinds: "execute,unknown",
|
||||
} }), {}));
|
||||
expect(result.execution).toEqual({
|
||||
acpLastEventAgeMs: 700, acpLastEventAgeAtStopMs: 14_000_000, acpOldestPendingToolAgeAtStopMs: 14_100_000,
|
||||
acpPendingToolKinds: "execute,unknown",
|
||||
});
|
||||
});
|
||||
|
||||
it.each(["execute,private command", "", "execute,", "Execute", ["execute"], 1])(
|
||||
"omits pending tool kinds that are not closed labels (%j)", (value) => {
|
||||
const result = collectRunFailureDiagnostics(run({ resultJson: { acpPendingToolKinds: value } }), {});
|
||||
expect(result.execution).toEqual({});
|
||||
},
|
||||
);
|
||||
|
||||
it.each([null, -1, Infinity, NaN, 1.5, Number.MAX_SAFE_INTEGER + 1, "private", {}])(
|
||||
"omits invalid ACP activity fields (%j)", (value) => {
|
||||
const result = collectRunFailureDiagnostics(run({ resultJson: {
|
||||
acpLastEventAgeMs: value, acpObservedEventCount: value, acpPendingToolCount: value,
|
||||
acpToolInventoryComplete: value,
|
||||
acpToolInventoryComplete: value, acpLastEventAgeAtStopMs: value, acpOldestPendingToolAgeAtStopMs: value,
|
||||
} }), {});
|
||||
expect(result.execution).toEqual({});
|
||||
},
|
||||
|
||||
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,903 @@
|
||||
import type { EndpointRow, DbTransaction, RuntimeContext, LifecycleRuntimeFence } from "../../chat-channels/runtime-types.js";
|
||||
|
||||
import { notifyChatPublicationWork } from "../../chat-work-notifications.js";
|
||||
|
||||
import { nativeSha256 } from "../../native-runtime/canonical.js";
|
||||
|
||||
import { projectSafeChatPublicationText } from "../../chat-publication-projection.js";
|
||||
import { PhotonAnswerValidationError, nativePhotonInteraction, photonResponseCommand, parsePhotonQuestionAnswer, type PhotonPromptReceipt, type PhotonInteractionBinding, type PhotonDraft } from "../../photon/interactions.js";
|
||||
import { validateNativeQuestionResponseInput } from "../../native-runtime/native-question-bridge.js";
|
||||
import type { AskUserQuestionsAnswer, AskUserQuestionsInteraction, IssueThreadInteraction } from "@paperclipai/shared";
|
||||
import { PhotonError, photonFailure } from "../../photon/cloud.js";
|
||||
import { PhotonChatAdapter, photonThreadId } from "../../photon/adapter.js";
|
||||
|
||||
import type { LiveEvent as PhotonEvent } from "@photon-ai/advanced-imessage";
|
||||
|
||||
import {
|
||||
and,
|
||||
eq,
|
||||
inArray,
|
||||
isNotNull,
|
||||
sql,
|
||||
} from "drizzle-orm";
|
||||
|
||||
import type { Db } from "@paperclipai/db";
|
||||
|
||||
import {
|
||||
chatActions,
|
||||
chatConversations,
|
||||
chatEndpointResources,
|
||||
chatEndpoints,
|
||||
chatExternalPrincipals,
|
||||
chatPublications,
|
||||
issues,
|
||||
toolConnections,
|
||||
} from "@paperclipai/db";
|
||||
|
||||
import {
|
||||
HttpError,
|
||||
forbidden,
|
||||
} from "../../../errors.js";
|
||||
|
||||
import {
|
||||
type ChatSdkMessageCallbackEvent,
|
||||
} from "../../chat-sdk-runtime.js";
|
||||
|
||||
import {
|
||||
logActivity,
|
||||
} from "../../activity-log.js";
|
||||
|
||||
import { issueThreadInteractionService } from "../../issue-thread-interactions.js";
|
||||
import {
|
||||
questionResponseDeliveryService,
|
||||
} from "../../question-response-delivery.js";
|
||||
|
||||
import type {
|
||||
Author,
|
||||
} from "chat";
|
||||
|
||||
import type { EndpointRecord } from "../runtime-types.js";
|
||||
|
||||
type PhotonEventsOptions = {
|
||||
runtimeCallbackRecord: (endpointId: string, context: RuntimeContext, statuses: EndpointRow["status"][]) => Promise<EndpointRecord | null>;
|
||||
runtimeCallbackEndpoint: (tx: DbTransaction, endpointId: string, context: LifecycleRuntimeFence, statuses: EndpointRow["status"][]) => Promise<EndpointRow | null>;
|
||||
invalidateRuntime: (endpointId: string) => Promise<boolean>;
|
||||
ensurePrincipal: (endpoint: EndpointRow, author: Author, raw?: unknown) => Promise<{ principal: Pick<typeof chatExternalPrincipals.$inferSelect, "id" | "kind" | "isBot">; userId: string | null; linkedDenied: boolean }>;
|
||||
requireCurrentExternalActionAuthorization: (tx: DbTransaction, input: { conversationId: string; endpointId: string; expectedUserId: string; principalId: string; runtimeContext: LifecycleRuntimeFence }) => Promise<void>;
|
||||
isExternalActionAuthorizationChange: (error: unknown) => boolean;
|
||||
scheduleMessageProcessing: (task: () => Promise<void>) => void;
|
||||
processPendingPublications: () => Promise<unknown>;
|
||||
questionResponses: Pick<ReturnType<typeof questionResponseDeliveryService>, "deliver">;
|
||||
redactError: (error: unknown) => string;
|
||||
renewDiscordGatewayOwnershipForMessageAdmission: (tx: DbTransaction, endpointId: string, context: RuntimeContext) => Promise<{ kind: "owned" | "lost" }>;
|
||||
handleSdkMessage: (event: ChatSdkMessageCallbackEvent, runtimeContext?: RuntimeContext, messageOptions?: { receiptReactionSupported?: boolean; slackSlashControl?: boolean }) => Promise<void>;
|
||||
};
|
||||
|
||||
export function createPhotonEvents(db: Db, options: PhotonEventsOptions) {
|
||||
const { runtimeCallbackRecord, runtimeCallbackEndpoint, invalidateRuntime, ensurePrincipal, requireCurrentExternalActionAuthorization, isExternalActionAuthorizationChange, scheduleMessageProcessing, processPendingPublications, questionResponses, redactError, renewDiscordGatewayOwnershipForMessageAdmission, handleSdkMessage } = options;
|
||||
|
||||
|
||||
async function handlePhotonFailure(
|
||||
endpointId: string,
|
||||
error: unknown,
|
||||
context: RuntimeContext,
|
||||
) {
|
||||
if (error instanceof PhotonError && error.code === "attachment_not_ready")
|
||||
return;
|
||||
error = photonFailure(error);
|
||||
const fatal =
|
||||
error instanceof PhotonError &&
|
||||
[
|
||||
"credentials",
|
||||
"line_unavailable",
|
||||
"history_gap",
|
||||
"invalid_response",
|
||||
].includes(error.code);
|
||||
const message =
|
||||
error instanceof PhotonError
|
||||
? error.message
|
||||
: "Photon connection interrupted; reconnecting";
|
||||
await db.transaction(async (tx) => {
|
||||
const endpoint = await runtimeCallbackEndpoint(tx, endpointId, context, [
|
||||
"verifying",
|
||||
"active",
|
||||
"attention",
|
||||
]);
|
||||
if (!endpoint) return;
|
||||
await tx
|
||||
.update(chatEndpoints)
|
||||
.set({
|
||||
status: "attention",
|
||||
healthMessage: message,
|
||||
lastError: message,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(chatEndpoints.id, endpointId));
|
||||
if (fatal)
|
||||
await tx
|
||||
.update(toolConnections)
|
||||
.set({
|
||||
enabled: false,
|
||||
healthStatus: "error",
|
||||
healthMessage: message,
|
||||
})
|
||||
.where(eq(toolConnections.id, endpoint.connectionId));
|
||||
});
|
||||
// Retirement is queued outside the receiver callback to avoid self-joining.
|
||||
void invalidateRuntime(endpointId).catch(() => undefined);
|
||||
}
|
||||
|
||||
async function processPhotonResponse(
|
||||
endpoint: EndpointRow,
|
||||
event: PhotonEvent,
|
||||
threadId: string,
|
||||
adapter: PhotonChatAdapter,
|
||||
context: RuntimeContext,
|
||||
): Promise<boolean> {
|
||||
const command =
|
||||
event.type === "message.received"
|
||||
? photonResponseCommand(event.message.content.text ?? "")
|
||||
: null;
|
||||
const replyGuid =
|
||||
event.type === "message.received" ? event.message.replyTargetGuid : null;
|
||||
const prompt =
|
||||
event.type === "poll.changed"
|
||||
? await adapter.state.read<PhotonPromptReceipt>(
|
||||
`poll-message:${event.pollMessageGuid}`,
|
||||
)
|
||||
: replyGuid
|
||||
? ((await adapter.state.read<PhotonPromptReceipt>(
|
||||
`prompt-message:${replyGuid}`,
|
||||
)) ??
|
||||
(await adapter.state.read<PhotonPromptReceipt>(
|
||||
`poll-message:${replyGuid}`,
|
||||
)))
|
||||
: null;
|
||||
const reference = command?.reference ?? prompt?.reference;
|
||||
if (!reference) {
|
||||
if (
|
||||
!event.isFromMe &&
|
||||
(replyGuid ||
|
||||
(event.type === "poll.changed" &&
|
||||
(event.delta.type === "voted" || event.delta.type === "unvoted")))
|
||||
) {
|
||||
// A vote can beat the local poll-binding write. Keep it behind the
|
||||
// checkpoint while a publication in this chat is in flight.
|
||||
const pending = await db
|
||||
.select({ id: chatPublications.id })
|
||||
.from(chatPublications)
|
||||
.innerJoin(
|
||||
chatConversations,
|
||||
eq(chatConversations.id, chatPublications.conversationId),
|
||||
)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.endpointId, endpoint.id),
|
||||
eq(chatConversations.externalThreadId, threadId),
|
||||
inArray(chatPublications.state, [
|
||||
"pending",
|
||||
"retry",
|
||||
"streaming",
|
||||
"delivery_unknown",
|
||||
]),
|
||||
isNotNull(sql`${chatPublications.payload}->>'interactionId'`),
|
||||
),
|
||||
)
|
||||
.limit(1);
|
||||
if (pending.length)
|
||||
throw new PhotonError(
|
||||
"attachment_not_ready",
|
||||
"Waiting for Photon poll binding",
|
||||
);
|
||||
}
|
||||
return (
|
||||
event.type === "poll.changed" ||
|
||||
(event.type === "message.received" &&
|
||||
/^\/(answer|submit)\b/i.test(event.message.content.text ?? ""))
|
||||
);
|
||||
}
|
||||
const action = await db
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.companyId, endpoint.companyId),
|
||||
eq(chatActions.endpointId, endpoint.id),
|
||||
eq(chatActions.kind, "photon_interaction"),
|
||||
eq(chatActions.providerActionId, `photon:${reference}`),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0]);
|
||||
if (!action || action.status !== "issued" || !action.conversationId)
|
||||
return true;
|
||||
const binding = action.payload as unknown as PhotonInteractionBinding;
|
||||
if (
|
||||
binding.version !== 1 ||
|
||||
binding.reference !== reference ||
|
||||
!Number.isFinite(Date.parse(binding.expiresAt)) ||
|
||||
Date.parse(binding.expiresAt) <= Date.now()
|
||||
)
|
||||
return true;
|
||||
const conversation = await db
|
||||
.select()
|
||||
.from(chatConversations)
|
||||
.where(
|
||||
and(
|
||||
eq(chatConversations.companyId, endpoint.companyId),
|
||||
eq(chatConversations.endpointId, endpoint.id),
|
||||
eq(chatConversations.id, action.conversationId),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0]);
|
||||
if (
|
||||
!conversation ||
|
||||
!["active", "waiting"].includes(conversation.state) ||
|
||||
conversation.sessionGeneration !== binding.sessionGeneration ||
|
||||
conversation.externalThreadId !== threadId
|
||||
)
|
||||
return true;
|
||||
const publication = await db
|
||||
.select()
|
||||
.from(chatPublications)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.companyId, endpoint.companyId),
|
||||
eq(chatPublications.endpointId, endpoint.id),
|
||||
eq(chatPublications.id, binding.publicationId),
|
||||
eq(chatPublications.conversationId, conversation.id),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0]);
|
||||
if (
|
||||
!publication ||
|
||||
publication.payload.interactionId !== binding.interactionId ||
|
||||
new Date(event.occurredAt).getTime() < action.createdAt.getTime()
|
||||
)
|
||||
return true;
|
||||
if (
|
||||
["pending", "retry", "streaming", "delivery_unknown"].includes(
|
||||
publication.state,
|
||||
)
|
||||
)
|
||||
throw new PhotonError(
|
||||
"attachment_not_ready",
|
||||
"Waiting for Photon prompt publication",
|
||||
);
|
||||
if (publication.state !== "published") return true;
|
||||
const interaction = (
|
||||
await issueThreadInteractionService(db).listForIssue(conversation.issueId)
|
||||
).find((candidate) => candidate.id === binding.interactionId);
|
||||
if (
|
||||
!interaction ||
|
||||
interaction.status !== "pending" ||
|
||||
interaction.companyId !== endpoint.companyId ||
|
||||
interaction.createdByAgentId !== endpoint.assignedAgentId ||
|
||||
!nativePhotonInteraction(interaction)
|
||||
)
|
||||
return true;
|
||||
const actor =
|
||||
event.type === "message.received" ? event.message.sender : event.actor;
|
||||
if (event.isFromMe || !actor?.address || actor.service !== "iMessage")
|
||||
return true;
|
||||
const chat = await adapter.chatInfo(threadId);
|
||||
if (
|
||||
chat.isArchived ||
|
||||
!chat.participants.some(
|
||||
(person) =>
|
||||
person.address === actor.address && person.service === actor.service,
|
||||
)
|
||||
)
|
||||
return true;
|
||||
const principal = await ensurePrincipal(
|
||||
endpoint,
|
||||
{
|
||||
userId: `${actor.service}:${actor.address}`,
|
||||
userName: actor.address,
|
||||
fullName: actor.address,
|
||||
isMe: false,
|
||||
isBot: false,
|
||||
},
|
||||
event,
|
||||
);
|
||||
if (
|
||||
!principal.userId ||
|
||||
principal.linkedDenied ||
|
||||
principal.principal.kind !== "user" ||
|
||||
principal.principal.isBot
|
||||
)
|
||||
return true;
|
||||
const assertCurrent = async () =>
|
||||
db.transaction((tx) =>
|
||||
requireCurrentExternalActionAuthorization(tx, {
|
||||
conversationId: conversation.id,
|
||||
endpointId: endpoint.id,
|
||||
expectedUserId: principal.userId!,
|
||||
principalId: principal.principal.id,
|
||||
runtimeContext: context,
|
||||
}),
|
||||
);
|
||||
try {
|
||||
await assertCurrent();
|
||||
} catch (error) {
|
||||
if (isExternalActionAuthorizationChange(error)) return true;
|
||||
throw error;
|
||||
}
|
||||
const draftKey = `draft:${reference}:${principal.principal.id}:${principal.userId}`;
|
||||
let draft = (await adapter.state.read<PhotonDraft>(draftKey)) ?? {
|
||||
schema: 1,
|
||||
interactionId: interaction.id,
|
||||
principalId: principal.principal.id,
|
||||
userId: principal.userId,
|
||||
answers: [],
|
||||
lastSequence: 0,
|
||||
};
|
||||
if (draft.lastSequence > event.sequence) return true;
|
||||
const questionIndex = command?.questionIndex ?? prompt?.questionIndex ?? 0;
|
||||
let answerValue =
|
||||
command?.value ??
|
||||
(event.type === "message.received"
|
||||
? (event.message.content.text ?? "")
|
||||
: "");
|
||||
let pollChoice: string | undefined;
|
||||
if (event.type === "poll.changed") {
|
||||
if (!prompt || event.pollMessageGuid !== prompt.pollMessageGuid)
|
||||
return true;
|
||||
if (event.delta.type !== "voted" && event.delta.type !== "unvoted")
|
||||
return true;
|
||||
pollChoice = prompt.options[event.delta.optionIdentifier];
|
||||
if (!pollChoice) return true;
|
||||
if (event.delta.type === "unvoted") {
|
||||
const questionId =
|
||||
interaction.kind === "ask_user_questions"
|
||||
? interaction.payload.questions[questionIndex]?.id
|
||||
: null;
|
||||
await adapter.state.update<PhotonDraft>(draftKey, () => ({
|
||||
...draft,
|
||||
answers: draft.answers.filter(
|
||||
(answer) =>
|
||||
answer.questionId !== questionId ||
|
||||
!answer.optionIds.includes(pollChoice!),
|
||||
),
|
||||
decision: draft.decision === pollChoice ? undefined : draft.decision,
|
||||
lastSequence: event.sequence,
|
||||
}));
|
||||
return true;
|
||||
}
|
||||
}
|
||||
const enqueueResponse = async (key: string, text: string) => {
|
||||
await assertCurrent();
|
||||
await db.transaction(async (tx) => {
|
||||
await tx
|
||||
.insert(chatActions)
|
||||
.values({
|
||||
companyId: endpoint.companyId,
|
||||
endpointId: endpoint.id,
|
||||
conversationId: conversation.id,
|
||||
kind: "photon_response_notice",
|
||||
providerActionId: `photon-notice:${key}`,
|
||||
payload: {
|
||||
reference,
|
||||
questionIndex,
|
||||
publicationId: binding.publicationId,
|
||||
},
|
||||
status: "processed",
|
||||
})
|
||||
.onConflictDoNothing();
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.insert(chatPublications)
|
||||
.values({
|
||||
companyId: endpoint.companyId,
|
||||
endpointId: endpoint.id,
|
||||
conversationId: conversation.id,
|
||||
issueId: conversation.issueId,
|
||||
idempotencyKey: key,
|
||||
payload: {
|
||||
...publication.payload,
|
||||
card: undefined,
|
||||
text: projectSafeChatPublicationText(text),
|
||||
},
|
||||
state: "pending",
|
||||
})
|
||||
.onConflictDoNothing();
|
||||
});
|
||||
scheduleMessageProcessing(() =>
|
||||
processPendingPublications().then(() => undefined),
|
||||
);
|
||||
};
|
||||
const notice = (text: string) =>
|
||||
enqueueResponse(`photon-response:${reference}:${event.sequence}`, text);
|
||||
try {
|
||||
if (interaction.kind === "ask_user_questions") {
|
||||
if (command?.command !== "submit") {
|
||||
let answer: AskUserQuestionsAnswer;
|
||||
if (pollChoice) {
|
||||
const question = interaction.payload.questions[questionIndex];
|
||||
if (
|
||||
!question ||
|
||||
!question.options.some((option) => option.id === pollChoice)
|
||||
)
|
||||
return true;
|
||||
answer = { questionId: question.id, optionIds: [pollChoice] };
|
||||
} else
|
||||
answer = parsePhotonQuestionAnswer(
|
||||
interaction,
|
||||
questionIndex,
|
||||
answerValue,
|
||||
);
|
||||
draft = {
|
||||
...draft,
|
||||
answers: [
|
||||
...draft.answers.filter(
|
||||
(entry) => entry.questionId !== answer.questionId,
|
||||
),
|
||||
answer,
|
||||
],
|
||||
lastSequence: event.sequence,
|
||||
};
|
||||
}
|
||||
const answers = interaction.payload.questions.flatMap((question) =>
|
||||
draft.answers.filter((answer) => answer.questionId === question.id),
|
||||
);
|
||||
if (
|
||||
interaction.payload.questions.length > 1 &&
|
||||
command?.command !== "submit"
|
||||
) {
|
||||
await adapter.state.update<PhotonDraft>(draftKey, () => draft);
|
||||
const next = interaction.payload.questions.findIndex(
|
||||
(question) =>
|
||||
!answers.some((answer) => answer.questionId === question.id),
|
||||
);
|
||||
if (next >= 0)
|
||||
await enqueueResponse(
|
||||
`photon-question:${reference}:${next}`,
|
||||
"Input needed",
|
||||
);
|
||||
else
|
||||
await notice(
|
||||
`Your answers are saved. Send /submit ${reference} to submit them.`,
|
||||
);
|
||||
return true;
|
||||
}
|
||||
await validateNativeQuestionResponseInput(interaction, { answers });
|
||||
// Canonical service rechecks required fields, options, audience, and
|
||||
// pending -> answered atomically with the response-delivery receipt.
|
||||
const issue = await db
|
||||
.select()
|
||||
.from(issues)
|
||||
.where(
|
||||
and(
|
||||
eq(issues.companyId, endpoint.companyId),
|
||||
eq(issues.id, conversation.issueId),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0]);
|
||||
if (!issue) return true;
|
||||
const answered = await issueThreadInteractionService(
|
||||
db,
|
||||
).answerQuestions(
|
||||
issue,
|
||||
interaction.id,
|
||||
{ answers },
|
||||
{ userId: principal.userId },
|
||||
{
|
||||
beforeResolveInTransaction: async (tx) => {
|
||||
await requireCurrentExternalActionAuthorization(tx, {
|
||||
conversationId: conversation.id,
|
||||
endpointId: endpoint.id,
|
||||
expectedUserId: principal.userId!,
|
||||
principalId: principal.principal.id,
|
||||
runtimeContext: context,
|
||||
});
|
||||
const current = await tx
|
||||
.select()
|
||||
.from(chatConversations)
|
||||
.where(eq(chatConversations.id, conversation.id))
|
||||
.then((rows) => rows[0]);
|
||||
if (
|
||||
current?.sessionGeneration !== binding.sessionGeneration ||
|
||||
Date.parse(binding.expiresAt) <= Date.now()
|
||||
)
|
||||
throw forbidden(
|
||||
"Photon prompt expired or its task generation changed",
|
||||
);
|
||||
},
|
||||
afterResolveInTransaction: async (tx, resolved) => {
|
||||
const claimed = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
principalId: principal.principal.id,
|
||||
status: "processed",
|
||||
result: {
|
||||
code: "photon_question_answered",
|
||||
interactionId: resolved.id,
|
||||
interactionStatus: resolved.status,
|
||||
answersSha256: nativeSha256(
|
||||
(resolved as AskUserQuestionsInteraction).result?.answers,
|
||||
),
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, action.id),
|
||||
eq(chatActions.status, "issued"),
|
||||
),
|
||||
)
|
||||
.returning({ id: chatActions.id });
|
||||
if (!claimed.length)
|
||||
throw forbidden("Photon prompt has already been resolved");
|
||||
await logActivity(tx as unknown as Db, {
|
||||
companyId: endpoint.companyId,
|
||||
actorType: "user",
|
||||
actorId: principal.userId!,
|
||||
action: "issue.thread_interaction_answered",
|
||||
entityType: "issue",
|
||||
entityId: issue.id,
|
||||
details: {
|
||||
source: "external_chat",
|
||||
provider: endpoint.provider,
|
||||
endpointId: endpoint.id,
|
||||
conversationId: conversation.id,
|
||||
interactionId: resolved.id,
|
||||
},
|
||||
});
|
||||
},
|
||||
},
|
||||
);
|
||||
scheduleMessageProcessing(async () => {
|
||||
await questionResponses.deliver(answered.id);
|
||||
await processPendingPublications();
|
||||
});
|
||||
} else {
|
||||
if (command?.command === "submit") return true;
|
||||
const matched = /^(accept|reject|1|2)(?:\s+([\s\S]+))?$/i.exec(
|
||||
answerValue.trim(),
|
||||
);
|
||||
const decision =
|
||||
pollChoice === "accept" || pollChoice === "reject"
|
||||
? pollChoice
|
||||
: matched
|
||||
? /^(accept|1)$/i.test(matched[1])
|
||||
? "accept"
|
||||
: "reject"
|
||||
: draft.decision;
|
||||
if (!decision)
|
||||
throw new PhotonAnswerValidationError(
|
||||
`Send /answer ${reference} Accept or /answer ${reference} Reject <reason>.`,
|
||||
);
|
||||
const reason =
|
||||
matched?.[2]?.trim() ??
|
||||
(draft.decision === "reject" && !pollChoice
|
||||
? answerValue.trim()
|
||||
: "");
|
||||
if (
|
||||
decision === "reject" &&
|
||||
interaction.payload.rejectRequiresReason &&
|
||||
!reason
|
||||
) {
|
||||
await adapter.state.update<PhotonDraft>(draftKey, () => ({
|
||||
...draft,
|
||||
decision,
|
||||
lastSequence: event.sequence,
|
||||
}));
|
||||
await notice(
|
||||
`Give a rejection reason with /answer ${reference} Reject <reason>.`,
|
||||
);
|
||||
return true;
|
||||
}
|
||||
const issue = await db
|
||||
.select()
|
||||
.from(issues)
|
||||
.where(
|
||||
and(
|
||||
eq(issues.companyId, endpoint.companyId),
|
||||
eq(issues.id, conversation.issueId),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0]);
|
||||
if (!issue) return true;
|
||||
const mutation = {
|
||||
beforeResolveInTransaction: async (tx: DbTransaction) => {
|
||||
await requireCurrentExternalActionAuthorization(tx, {
|
||||
conversationId: conversation.id,
|
||||
endpointId: endpoint.id,
|
||||
expectedUserId: principal.userId!,
|
||||
principalId: principal.principal.id,
|
||||
runtimeContext: context,
|
||||
});
|
||||
const current = await tx
|
||||
.select()
|
||||
.from(chatConversations)
|
||||
.where(eq(chatConversations.id, conversation.id))
|
||||
.then((rows) => rows[0]);
|
||||
if (
|
||||
current?.sessionGeneration !== binding.sessionGeneration ||
|
||||
Date.parse(binding.expiresAt) <= Date.now()
|
||||
)
|
||||
throw forbidden(
|
||||
"Photon prompt expired or its task generation changed",
|
||||
);
|
||||
},
|
||||
afterResolveInTransaction: async (
|
||||
tx: DbTransaction,
|
||||
resolved: IssueThreadInteraction,
|
||||
) => {
|
||||
const claimed = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
principalId: principal.principal.id,
|
||||
status: "processed",
|
||||
result: {
|
||||
code: "photon_confirmation_resolved",
|
||||
interactionId: resolved.id,
|
||||
interactionStatus: resolved.status,
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, action.id),
|
||||
eq(chatActions.status, "issued"),
|
||||
),
|
||||
)
|
||||
.returning({ id: chatActions.id });
|
||||
if (!claimed.length)
|
||||
throw forbidden("Photon prompt has already been resolved");
|
||||
await logActivity(tx as unknown as Db, {
|
||||
companyId: endpoint.companyId,
|
||||
actorType: "user",
|
||||
actorId: principal.userId!,
|
||||
action:
|
||||
decision === "accept"
|
||||
? "issue.thread_interaction_accepted"
|
||||
: "issue.thread_interaction_rejected",
|
||||
entityType: "issue",
|
||||
entityId: issue.id,
|
||||
details: {
|
||||
source: "external_chat",
|
||||
provider: endpoint.provider,
|
||||
endpointId: endpoint.id,
|
||||
conversationId: conversation.id,
|
||||
interactionId: resolved.id,
|
||||
},
|
||||
});
|
||||
},
|
||||
};
|
||||
if (decision === "accept")
|
||||
await issueThreadInteractionService(db).acceptInteraction(
|
||||
issue,
|
||||
interaction.id,
|
||||
{},
|
||||
{ userId: principal.userId },
|
||||
mutation,
|
||||
);
|
||||
else
|
||||
await issueThreadInteractionService(db).rejectInteraction(
|
||||
issue,
|
||||
interaction.id,
|
||||
{ reason },
|
||||
{ userId: principal.userId },
|
||||
mutation,
|
||||
);
|
||||
scheduleMessageProcessing(async () => {
|
||||
await processPendingPublications();
|
||||
});
|
||||
}
|
||||
} catch (error) {
|
||||
if (error instanceof PhotonError) throw error;
|
||||
if (isExternalActionAuthorizationChange(error)) return true;
|
||||
if (error instanceof HttpError && [403, 404, 409].includes(error.status))
|
||||
return true;
|
||||
// Only validator errors become correction copy. Infrastructure errors
|
||||
// leave the checkpoint unchanged and retry through durable recovery.
|
||||
if (
|
||||
(error instanceof HttpError && error.status === 422) ||
|
||||
error instanceof PhotonAnswerValidationError
|
||||
) {
|
||||
const missingIndex = interaction.kind === "ask_user_questions" && command?.command === "submit"
|
||||
? interaction.payload.questions.findIndex((question) =>
|
||||
question.required !== false && !draft.answers.some((answer) => answer.questionId === question.id))
|
||||
: -1;
|
||||
await notice(
|
||||
`${redactError(error)} Send /answer ${reference}.${(missingIndex >= 0 ? missingIndex : questionIndex) + 1} <answer> to correct it.`,
|
||||
);
|
||||
return true;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
async function handlePhotonEvent(
|
||||
endpointId: string,
|
||||
event: PhotonEvent,
|
||||
context: RuntimeContext,
|
||||
) {
|
||||
const record = await runtimeCallbackRecord(endpointId, context, [
|
||||
"verifying",
|
||||
"active",
|
||||
"attention",
|
||||
]);
|
||||
if (!record)
|
||||
throw new Error("Photon receiver no longer owns this endpoint");
|
||||
const adapter = context.endpointRuntime!.getProviderAdapter();
|
||||
if (!(adapter instanceof PhotonChatAdapter))
|
||||
throw new Error("Photon adapter unavailable");
|
||||
if (adapter.authentication.identity.allocation === "shared" && event.type === "group.changed") return;
|
||||
if (
|
||||
event.type === "group.changed" &&
|
||||
(event.change.type === "participantRemoved" ||
|
||||
event.change.type === "participantLeft") &&
|
||||
event.change.participant.address === record.endpoint.botExternalId
|
||||
) {
|
||||
const removedThreadId = photonThreadId({
|
||||
lineId: adapter.authentication.identity.lineId,
|
||||
chatGuid: event.chatGuid,
|
||||
isGroup: true,
|
||||
});
|
||||
await db.transaction(async (tx) => {
|
||||
if (
|
||||
!(await runtimeCallbackEndpoint(tx, endpointId, context, [
|
||||
"verifying",
|
||||
"active",
|
||||
]))
|
||||
)
|
||||
throw new Error("Photon receiver is no longer current");
|
||||
if (
|
||||
(
|
||||
await renewDiscordGatewayOwnershipForMessageAdmission(
|
||||
tx,
|
||||
endpointId,
|
||||
context,
|
||||
)
|
||||
).kind !== "owned"
|
||||
)
|
||||
throw new Error("Photon receiver lease was replaced");
|
||||
// Removal can be the first retained event after activation. Preserve a
|
||||
// tombstone even when the group has never been discovered locally.
|
||||
await tx
|
||||
.insert(chatEndpointResources)
|
||||
.values({
|
||||
companyId: record.endpoint.companyId,
|
||||
endpointId,
|
||||
providerResourceId: removedThreadId,
|
||||
type: "group_chat",
|
||||
label: "Unavailable iMessage group",
|
||||
enabled: false,
|
||||
availability: "unavailable",
|
||||
metadata: { photonRemoved: true },
|
||||
})
|
||||
.onConflictDoUpdate({
|
||||
target: [
|
||||
chatEndpointResources.endpointId,
|
||||
chatEndpointResources.type,
|
||||
chatEndpointResources.providerResourceId,
|
||||
],
|
||||
set: {
|
||||
availability: "unavailable",
|
||||
metadata: sql`jsonb_set(${chatEndpointResources.metadata}, '{photonRemoved}', 'true'::jsonb)`,
|
||||
updatedAt: new Date(),
|
||||
},
|
||||
});
|
||||
});
|
||||
await adapter.endTyping(removedThreadId);
|
||||
return;
|
||||
}
|
||||
const groupIdentity = photonThreadId({ lineId: adapter.authentication.identity.lineId, chatGuid: event.chatGuid, isGroup: true });
|
||||
const removedGroup = await db.select({ metadata: chatEndpointResources.metadata }).from(chatEndpointResources).where(and(eq(chatEndpointResources.companyId, record.endpoint.companyId), eq(chatEndpointResources.endpointId, endpointId), eq(chatEndpointResources.providerResourceId, groupIdentity))).limit(1).then((rows) => rows[0]);
|
||||
const readded = event.type === "group.changed" && event.change.type === "participantAdded" && event.change.participant.address === record.endpoint.botExternalId;
|
||||
// Late events from a removed chat must still advance recovery without a
|
||||
// now-forbidden chat lookup holding every other conversation behind them.
|
||||
if (removedGroup?.metadata.photonRemoved === true && !readded) return;
|
||||
const chat = await adapter.client.chats
|
||||
.get(event.chatGuid)
|
||||
.catch((error) => {
|
||||
throw photonFailure(error);
|
||||
});
|
||||
if (chat.guid !== event.chatGuid || chat.service !== "iMessage" || (chat.isGroup && adapter.authentication.identity.allocation === "shared")) return;
|
||||
const threadId = photonThreadId({
|
||||
lineId: adapter.authentication.identity.lineId,
|
||||
chatGuid: chat.guid,
|
||||
isGroup: chat.isGroup,
|
||||
});
|
||||
const removed =
|
||||
event.type === "group.changed" &&
|
||||
(event.change.type === "participantRemoved" ||
|
||||
event.change.type === "participantLeft") &&
|
||||
event.change.participant.address === record.endpoint.botExternalId;
|
||||
const available = await db.transaction(async (tx) => {
|
||||
if (
|
||||
!(await runtimeCallbackEndpoint(tx, endpointId, context, [
|
||||
"verifying",
|
||||
"active",
|
||||
]))
|
||||
)
|
||||
throw new Error("Photon receiver is no longer current");
|
||||
const ownership = await renewDiscordGatewayOwnershipForMessageAdmission(
|
||||
tx,
|
||||
endpointId,
|
||||
context,
|
||||
);
|
||||
if (ownership.kind !== "owned")
|
||||
throw new Error("Photon receiver lease was replaced");
|
||||
const existing = await tx
|
||||
.select()
|
||||
.from(chatEndpointResources)
|
||||
.where(
|
||||
and(
|
||||
eq(chatEndpointResources.endpointId, endpointId),
|
||||
eq(chatEndpointResources.providerResourceId, threadId),
|
||||
eq(
|
||||
chatEndpointResources.type,
|
||||
chat.isGroup ? "group_chat" : "direct_message",
|
||||
),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0]);
|
||||
const added =
|
||||
event.type === "group.changed" &&
|
||||
event.change.type === "participantAdded" &&
|
||||
event.change.participant.address === record.endpoint.botExternalId;
|
||||
const stillRemoved =
|
||||
removed || (existing?.metadata.photonRemoved === true && !added);
|
||||
const availability =
|
||||
stillRemoved || chat.isArchived ? "unavailable" : "available";
|
||||
const metadata = {
|
||||
photonRemoved: stillRemoved,
|
||||
participants: chat.participants
|
||||
.slice(0, 256)
|
||||
.map((p) => ({ address: p.address, service: p.service })),
|
||||
};
|
||||
const label = (
|
||||
chat.displayName || chat.participants.map((p) => p.address).join(", ")
|
||||
).slice(0, 512);
|
||||
await tx
|
||||
.insert(chatEndpointResources)
|
||||
.values({
|
||||
companyId: record.endpoint.companyId,
|
||||
endpointId,
|
||||
providerResourceId: threadId,
|
||||
type: chat.isGroup ? "group_chat" : "direct_message",
|
||||
label,
|
||||
enabled: !chat.isGroup,
|
||||
availability,
|
||||
metadata,
|
||||
})
|
||||
.onConflictDoUpdate({
|
||||
target: [
|
||||
chatEndpointResources.endpointId,
|
||||
chatEndpointResources.type,
|
||||
chatEndpointResources.providerResourceId,
|
||||
],
|
||||
set: { label, availability, metadata, updatedAt: new Date() },
|
||||
});
|
||||
return availability === "available";
|
||||
});
|
||||
if (!available) return;
|
||||
if (
|
||||
!event.isFromMe &&
|
||||
(await processPhotonResponse(
|
||||
record.endpoint,
|
||||
event,
|
||||
threadId,
|
||||
adapter,
|
||||
context,
|
||||
))
|
||||
)
|
||||
return;
|
||||
if (event.type === "message.received" && !event.isFromMe) {
|
||||
await handleSdkMessage(
|
||||
{
|
||||
provider: "imessage-photon",
|
||||
endpointId,
|
||||
thread: context.endpointRuntime!.thread(threadId),
|
||||
message: adapter.normalize(event.message, chat),
|
||||
trigger: chat.isGroup ? "unaddressed_message" : "direct_message",
|
||||
},
|
||||
context,
|
||||
);
|
||||
}
|
||||
}
|
||||
return { handlePhotonFailure, handlePhotonEvent };
|
||||
}
|
||||
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,1000 @@
|
||||
import type { EndpointRow, DbTransaction, DbOrTransaction, LifecycleRuntimeFence } from "../../chat-channels/runtime-types.js";
|
||||
|
||||
import { notifyChatPublicationWork } from "../../chat-work-notifications.js";
|
||||
|
||||
import type { ChatEndpointRuntime } from "../../voice/voice-runtime.js";
|
||||
|
||||
import { chatCredentialMutationLease, type CredentialMutationLeaseGuard } from "../../chat-credential-mutation-lease.js";
|
||||
|
||||
import {
|
||||
randomUUID,
|
||||
} from "node:crypto";
|
||||
import {
|
||||
and,
|
||||
asc,
|
||||
eq,
|
||||
inArray,
|
||||
isNull,
|
||||
lte,
|
||||
notExists,
|
||||
or,
|
||||
sql,
|
||||
} from "drizzle-orm";
|
||||
|
||||
import type { Db } from "@paperclipai/db";
|
||||
|
||||
import {
|
||||
chatActions,
|
||||
chatConversations,
|
||||
chatEndpointResources,
|
||||
chatEndpoints,
|
||||
chatMessageLinks,
|
||||
chatPublications,
|
||||
} from "@paperclipai/db";
|
||||
|
||||
import {
|
||||
isUuidLike,
|
||||
} from "@paperclipai/shared";
|
||||
|
||||
import {
|
||||
type SlackFileUploadAcceptedReceipt,
|
||||
} from "../../chat-sdk-runtime.js";
|
||||
|
||||
import {
|
||||
stageSlackSessionSync,
|
||||
} from "../../chat-slack-session-outbox.js";
|
||||
|
||||
import { classifyChatPublicationError } from "../../chat-publication-errors.js";
|
||||
|
||||
import type { EndpointRecord } from "../runtime-types.js";
|
||||
|
||||
type SlackFileUploadReceiptPayload = {
|
||||
version: 1;
|
||||
publicationId: string;
|
||||
publicationAttempt: number;
|
||||
threadId: string;
|
||||
fileIds: string[];
|
||||
botExternalId: string;
|
||||
runtimeGeneration: number;
|
||||
credentialFingerprint: string;
|
||||
};
|
||||
|
||||
function slackFileUploadReceiptPayload(
|
||||
payload: Record<string, unknown>,
|
||||
): SlackFileUploadReceiptPayload | null {
|
||||
if (
|
||||
payload.version !== 1 ||
|
||||
typeof payload.publicationId !== "string" ||
|
||||
!isUuidLike(payload.publicationId) ||
|
||||
typeof payload.publicationAttempt !== "number" ||
|
||||
!Number.isSafeInteger(payload.publicationAttempt) ||
|
||||
payload.publicationAttempt < 1 ||
|
||||
typeof payload.threadId !== "string" ||
|
||||
payload.threadId.length < 1 ||
|
||||
payload.threadId.length > 2_048 ||
|
||||
!payload.threadId.startsWith("slack:") ||
|
||||
typeof payload.botExternalId !== "string" ||
|
||||
payload.botExternalId.length < 1 ||
|
||||
payload.botExternalId.length > 255 ||
|
||||
typeof payload.runtimeGeneration !== "number" ||
|
||||
!Number.isSafeInteger(payload.runtimeGeneration) ||
|
||||
payload.runtimeGeneration < 0 ||
|
||||
typeof payload.credentialFingerprint !== "string" ||
|
||||
!/^[a-f0-9]{64}$/.test(payload.credentialFingerprint) ||
|
||||
!Array.isArray(payload.fileIds) ||
|
||||
payload.fileIds.length < 1 ||
|
||||
payload.fileIds.length > 20 ||
|
||||
payload.fileIds.some(
|
||||
(fileId) =>
|
||||
typeof fileId !== "string" || !/^F[A-Z0-9]{1,254}$/.test(fileId),
|
||||
) ||
|
||||
new Set(payload.fileIds).size !== payload.fileIds.length
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
return payload as SlackFileUploadReceiptPayload;
|
||||
}
|
||||
|
||||
function operatorConfirmedSlackFileReceipt(
|
||||
result: Record<string, unknown> | null,
|
||||
payload: SlackFileUploadReceiptPayload,
|
||||
): boolean {
|
||||
return (
|
||||
result?.operatorConfirmedPublicationId === payload.publicationId &&
|
||||
result.operatorConfirmedAttempt === payload.publicationAttempt &&
|
||||
typeof result.operatorConfirmedAt === "string" &&
|
||||
Number.isFinite(Date.parse(result.operatorConfirmedAt))
|
||||
);
|
||||
}
|
||||
|
||||
function sameSlackFileUploadReceipt(
|
||||
left: SlackFileUploadReceiptPayload,
|
||||
right: SlackFileUploadReceiptPayload,
|
||||
): boolean {
|
||||
return (
|
||||
left.publicationId === right.publicationId &&
|
||||
left.publicationAttempt === right.publicationAttempt &&
|
||||
left.threadId === right.threadId &&
|
||||
left.botExternalId === right.botExternalId &&
|
||||
left.runtimeGeneration === right.runtimeGeneration &&
|
||||
left.credentialFingerprint === right.credentialFingerprint &&
|
||||
left.fileIds.length === right.fileIds.length &&
|
||||
left.fileIds.every((fileId, index) => fileId === right.fileIds[index])
|
||||
);
|
||||
}
|
||||
const SLACK_FILE_RECEIPT_MAX_AGE_MS = 24 * 60 * 60_000;
|
||||
const SLACK_FILE_RECEIPT_MAX_ATTEMPTS = 12;
|
||||
const SLACK_FILE_RECEIPT_STALE_MS = 60_000;
|
||||
type ConversationRow = typeof chatConversations.$inferSelect;
|
||||
type SlackFileReceiptsOptions = {
|
||||
endpointRecord: (id: string) => Promise<EndpointRecord | null>;
|
||||
withCredentialMutationLease: ReturnType<typeof chatCredentialMutationLease>;
|
||||
runtimeCallbackEndpoint: (tx: DbTransaction, endpointId: string, context: LifecycleRuntimeFence, statuses: EndpointRow["status"][]) => Promise<EndpointRow | null>;
|
||||
nonDirectDestinationAllowed: (endpoint: EndpointRow, resource: typeof chatEndpointResources.$inferSelect | null | undefined) => boolean;
|
||||
runtimeFor: (endpoint: EndpointRow) => Promise<ChatEndpointRuntime>;
|
||||
isExplicitOperatorPublication: (publication: typeof chatPublications.$inferSelect) => boolean;
|
||||
recordExactOutboundPublicationLink: (tx: DbOrTransaction, input: { publication: typeof chatPublications.$inferSelect; providerMessageId: string }) => Promise<boolean>;
|
||||
};
|
||||
|
||||
export function createSlackFileReceipts(db: Db, options: SlackFileReceiptsOptions) {
|
||||
const { endpointRecord, withCredentialMutationLease, runtimeCallbackEndpoint, nonDirectDestinationAllowed, runtimeFor, isExplicitOperatorPublication, recordExactOutboundPublicationLink } = options;
|
||||
|
||||
|
||||
async function recordSlackFileUploadReceipt(input: {
|
||||
credentialLease: CredentialMutationLeaseGuard;
|
||||
conversation: ConversationRow;
|
||||
endpoint: EndpointRow;
|
||||
publication: typeof chatPublications.$inferSelect;
|
||||
receipt: SlackFileUploadAcceptedReceipt;
|
||||
runtimeContext: LifecycleRuntimeFence;
|
||||
}): Promise<string> {
|
||||
const publicationAttempt = input.publication.attempts + 1;
|
||||
const botExternalId = input.endpoint.botExternalId;
|
||||
const expectedThreadId = `slack:${input.receipt.channelId}:${input.receipt.threadTs ?? ""}`;
|
||||
if (
|
||||
input.endpoint.provider !== "slack" ||
|
||||
!botExternalId ||
|
||||
input.receipt.version !== 1 ||
|
||||
expectedThreadId !== input.conversation.externalThreadId ||
|
||||
input.receipt.fileIds.length !==
|
||||
(input.publication.payload.attachmentIds?.length ?? 0)
|
||||
) {
|
||||
throw new Error(
|
||||
"Slack file upload receipt did not match its publication",
|
||||
);
|
||||
}
|
||||
const payload: SlackFileUploadReceiptPayload = {
|
||||
version: 1,
|
||||
publicationId: input.publication.id,
|
||||
publicationAttempt,
|
||||
threadId: expectedThreadId,
|
||||
fileIds: [...input.receipt.fileIds],
|
||||
botExternalId,
|
||||
runtimeGeneration: input.runtimeContext.generation,
|
||||
credentialFingerprint: input.runtimeContext.credentialFingerprint,
|
||||
};
|
||||
if (!slackFileUploadReceiptPayload(payload)) {
|
||||
throw new Error("Slack file upload receipt was malformed");
|
||||
}
|
||||
const providerActionId = `slack-file-receipt:${input.publication.id}:${publicationAttempt}`;
|
||||
const recordedAt = new Date();
|
||||
return db.transaction(async (tx) => {
|
||||
await input.credentialLease.assertOwned(tx);
|
||||
const publication = await tx
|
||||
.select({ id: chatPublications.id })
|
||||
.from(chatPublications)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.id, input.publication.id),
|
||||
eq(chatPublications.companyId, input.publication.companyId),
|
||||
eq(chatPublications.endpointId, input.publication.endpointId),
|
||||
eq(
|
||||
chatPublications.conversationId,
|
||||
input.publication.conversationId,
|
||||
),
|
||||
eq(chatPublications.state, "streaming"),
|
||||
eq(chatPublications.attempts, publicationAttempt),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const endpoint = publication
|
||||
? await runtimeCallbackEndpoint(
|
||||
tx,
|
||||
input.endpoint.id,
|
||||
input.runtimeContext,
|
||||
["verifying", "active"],
|
||||
)
|
||||
: null;
|
||||
const conversation = endpoint
|
||||
? await tx
|
||||
.select({
|
||||
externalThreadId: chatConversations.externalThreadId,
|
||||
id: chatConversations.id,
|
||||
})
|
||||
.from(chatConversations)
|
||||
.where(
|
||||
and(
|
||||
eq(chatConversations.id, input.conversation.id),
|
||||
eq(chatConversations.companyId, input.publication.companyId),
|
||||
eq(chatConversations.endpointId, input.endpoint.id),
|
||||
inArray(chatConversations.state, [
|
||||
"active",
|
||||
"waiting",
|
||||
"completed",
|
||||
]),
|
||||
),
|
||||
)
|
||||
.for("no key update")
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
if (
|
||||
!publication ||
|
||||
endpoint?.provider !== "slack" ||
|
||||
endpoint.botExternalId !== botExternalId ||
|
||||
conversation?.externalThreadId !== expectedThreadId
|
||||
) {
|
||||
throw new Error(
|
||||
"Slack file upload ownership changed before receipt persistence",
|
||||
);
|
||||
}
|
||||
await notifyChatPublicationWork(tx);
|
||||
const inserted = await tx
|
||||
.insert(chatActions)
|
||||
.values({
|
||||
companyId: input.publication.companyId,
|
||||
endpointId: input.publication.endpointId,
|
||||
conversationId: input.publication.conversationId,
|
||||
kind: "slack_file_upload_receipt",
|
||||
providerActionId,
|
||||
payload,
|
||||
status: "received",
|
||||
result: { code: "slack_file_upload_identity_pending", attempts: 0 },
|
||||
createdAt: recordedAt,
|
||||
updatedAt: recordedAt,
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
.returning({ id: chatActions.id });
|
||||
if (inserted[0]) {
|
||||
await input.credentialLease.assertOwned(tx);
|
||||
return inserted[0].id;
|
||||
}
|
||||
const existing = await tx
|
||||
.select({ id: chatActions.id })
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.endpointId, input.publication.endpointId),
|
||||
eq(chatActions.providerActionId, providerActionId),
|
||||
eq(chatActions.kind, "slack_file_upload_receipt"),
|
||||
sql`${chatActions.payload} = ${JSON.stringify(payload)}::jsonb`,
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!existing) {
|
||||
throw new Error(
|
||||
"Slack file upload receipt conflicted with durable state",
|
||||
);
|
||||
}
|
||||
await input.credentialLease.assertOwned(tx);
|
||||
return existing.id;
|
||||
});
|
||||
}
|
||||
|
||||
async function stageSlackFilePublicationSuccess(
|
||||
tx: DbTransaction,
|
||||
input: {
|
||||
committedAt: Date;
|
||||
endpoint: EndpointRow;
|
||||
publication: typeof chatPublications.$inferSelect;
|
||||
providerMessageId: string;
|
||||
runtimeContext: LifecycleRuntimeFence;
|
||||
stageSessionSync: boolean;
|
||||
},
|
||||
): Promise<void> {
|
||||
if (
|
||||
!(await recordExactOutboundPublicationLink(tx, {
|
||||
publication: input.publication,
|
||||
providerMessageId: input.providerMessageId,
|
||||
}))
|
||||
) {
|
||||
throw new Error(
|
||||
"Slack file upload message identity conflicts with another publication",
|
||||
);
|
||||
}
|
||||
await tx
|
||||
.update(chatEndpoints)
|
||||
.set({
|
||||
lastPublicationAt: input.committedAt,
|
||||
updatedAt: input.committedAt,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatEndpoints.id, input.endpoint.id),
|
||||
inArray(chatEndpoints.status, ["verifying", "active"]),
|
||||
sql`coalesce((${chatEndpoints.setup}->>'runtimeGeneration')::integer, 0) = ${input.runtimeContext.generation}`,
|
||||
),
|
||||
);
|
||||
if (input.stageSessionSync) {
|
||||
await stageSlackSessionSync(tx, {
|
||||
companyId: input.publication.companyId,
|
||||
endpointId: input.publication.endpointId,
|
||||
conversationId: input.publication.conversationId,
|
||||
runtimeGeneration: input.runtimeContext.generation,
|
||||
credentialFingerprint: input.runtimeContext.credentialFingerprint,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async function lockSlackFileReceiptAuthorization(
|
||||
tx: DbTransaction,
|
||||
input: {
|
||||
actionId: string;
|
||||
payload: SlackFileUploadReceiptPayload;
|
||||
},
|
||||
): Promise<
|
||||
| {
|
||||
action: typeof chatActions.$inferSelect;
|
||||
conversation: ConversationRow;
|
||||
endpoint: EndpointRow;
|
||||
mode: "delivery_unknown" | "operator_confirmed";
|
||||
publication: typeof chatPublications.$inferSelect;
|
||||
}
|
||||
| { action: typeof chatActions.$inferSelect; mode: "defer" | "settled" }
|
||||
| null
|
||||
> {
|
||||
const publication = await tx
|
||||
.select()
|
||||
.from(chatPublications)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.id, input.payload.publicationId),
|
||||
eq(chatPublications.attempts, input.payload.publicationAttempt),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const action = publication
|
||||
? await tx
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, input.actionId),
|
||||
eq(chatActions.companyId, publication.companyId),
|
||||
eq(chatActions.endpointId, publication.endpointId),
|
||||
eq(chatActions.conversationId, publication.conversationId),
|
||||
eq(chatActions.kind, "slack_file_upload_receipt"),
|
||||
eq(
|
||||
chatActions.providerActionId,
|
||||
`slack-file-receipt:${publication.id}:${publication.attempts}`,
|
||||
),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
const currentPayload = action
|
||||
? slackFileUploadReceiptPayload(action.payload)
|
||||
: null;
|
||||
if (
|
||||
!publication ||
|
||||
!action ||
|
||||
!publication.payload.attachmentIds?.length ||
|
||||
publication.payload.card ||
|
||||
publication.payload.interactionId ||
|
||||
publication.idempotencyKey.startsWith("control:") ||
|
||||
!currentPayload ||
|
||||
!sameSlackFileUploadReceipt(currentPayload, input.payload)
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
if (publication.state === "streaming") {
|
||||
return { action, mode: "defer" };
|
||||
}
|
||||
if (publication.state === "published" && publication.providerMessageId) {
|
||||
const exactLink = await tx
|
||||
.select({ id: chatMessageLinks.id })
|
||||
.from(chatMessageLinks)
|
||||
.where(
|
||||
and(
|
||||
eq(chatMessageLinks.companyId, publication.companyId),
|
||||
eq(chatMessageLinks.endpointId, publication.endpointId),
|
||||
eq(chatMessageLinks.conversationId, publication.conversationId),
|
||||
eq(chatMessageLinks.publicationId, publication.id),
|
||||
publication.commentId
|
||||
? eq(chatMessageLinks.commentId, publication.commentId)
|
||||
: isNull(chatMessageLinks.commentId),
|
||||
eq(
|
||||
chatMessageLinks.providerMessageId,
|
||||
publication.providerMessageId,
|
||||
),
|
||||
eq(chatMessageLinks.direction, "outbound"),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
return exactLink ? { action, mode: "settled" } : null;
|
||||
}
|
||||
const mode =
|
||||
publication.state === "delivery_unknown"
|
||||
? "delivery_unknown"
|
||||
: publication.state === "published" &&
|
||||
!publication.providerMessageId &&
|
||||
operatorConfirmedSlackFileReceipt(action.result, input.payload)
|
||||
? "operator_confirmed"
|
||||
: null;
|
||||
if (!mode) return null;
|
||||
const endpoint = await runtimeCallbackEndpoint(
|
||||
tx,
|
||||
publication.endpointId,
|
||||
{
|
||||
generation: input.payload.runtimeGeneration,
|
||||
credentialFingerprint: input.payload.credentialFingerprint,
|
||||
},
|
||||
["verifying", "active"],
|
||||
);
|
||||
const conversation = endpoint
|
||||
? await tx
|
||||
.select()
|
||||
.from(chatConversations)
|
||||
.where(
|
||||
and(
|
||||
eq(chatConversations.id, publication.conversationId),
|
||||
eq(chatConversations.companyId, publication.companyId),
|
||||
eq(chatConversations.endpointId, publication.endpointId),
|
||||
eq(chatConversations.externalThreadId, input.payload.threadId),
|
||||
inArray(chatConversations.state, [
|
||||
"active",
|
||||
"waiting",
|
||||
"completed",
|
||||
]),
|
||||
),
|
||||
)
|
||||
.for("no key update")
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
const resource =
|
||||
endpoint && conversation?.resourceId && !conversation.isDirectMessage
|
||||
? await tx
|
||||
.select()
|
||||
.from(chatEndpointResources)
|
||||
.where(
|
||||
and(
|
||||
eq(chatEndpointResources.id, conversation.resourceId),
|
||||
eq(chatEndpointResources.companyId, endpoint.companyId),
|
||||
eq(chatEndpointResources.endpointId, endpoint.id),
|
||||
),
|
||||
)
|
||||
.for("no key update")
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
if (
|
||||
endpoint?.provider !== "slack" ||
|
||||
endpoint.botExternalId !== input.payload.botExternalId ||
|
||||
!conversation ||
|
||||
!(conversation.isDirectMessage
|
||||
? endpoint.allowDirectMessages
|
||||
: nonDirectDestinationAllowed(endpoint, resource))
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
return { action, conversation, endpoint, mode, publication };
|
||||
}
|
||||
|
||||
async function processPendingSlackFileUploadReceipts(limit = 25) {
|
||||
const selectedAt = new Date();
|
||||
const actions = await db
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(
|
||||
slackReceiptEligibility(selectedAt),
|
||||
)
|
||||
.orderBy(asc(chatActions.updatedAt), asc(chatActions.id))
|
||||
.limit(limit);
|
||||
let processed = 0;
|
||||
for (const selected of actions) {
|
||||
const payload = slackFileUploadReceiptPayload(selected.payload);
|
||||
if (!payload) {
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatPublicationWork(tx);
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "failed",
|
||||
result: {
|
||||
code: "slack_file_upload_receipt_payload_invalid",
|
||||
retryable: false,
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, selected.id),
|
||||
eq(chatActions.kind, selected.kind),
|
||||
eq(chatActions.status, selected.status),
|
||||
sql`${chatActions.payload} = ${JSON.stringify(selected.payload)}::jsonb`,
|
||||
selected.result === null
|
||||
? isNull(chatActions.result)
|
||||
: sql`${chatActions.result} = ${JSON.stringify(selected.result)}::jsonb`,
|
||||
),
|
||||
);
|
||||
});
|
||||
continue;
|
||||
}
|
||||
const record = await endpointRecord(selected.endpointId);
|
||||
if (!record) {
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatPublicationWork(tx);
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
result: { code: "slack_file_upload_receipt_endpoint_removed" },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, selected.id),
|
||||
eq(chatActions.kind, selected.kind),
|
||||
eq(chatActions.status, selected.status),
|
||||
sql`${chatActions.payload} = ${JSON.stringify(selected.payload)}::jsonb`,
|
||||
selected.result === null
|
||||
? isNull(chatActions.result)
|
||||
: sql`${chatActions.result} = ${JSON.stringify(selected.result)}::jsonb`,
|
||||
),
|
||||
);
|
||||
});
|
||||
continue;
|
||||
}
|
||||
if (["paused", "attention"].includes(record.endpoint.status)) continue;
|
||||
const ownerToken = randomUUID();
|
||||
let attempt = 1;
|
||||
try {
|
||||
await withCredentialMutationLease(record.endpoint, async (lease) => {
|
||||
const claim = await db.transaction(async (tx) => {
|
||||
await lease.assertOwned(tx);
|
||||
const locked = await lockSlackFileReceiptAuthorization(tx, {
|
||||
actionId: selected.id,
|
||||
payload,
|
||||
});
|
||||
const decisionAt = new Date();
|
||||
const priorAttempts =
|
||||
typeof locked?.action.result?.attempts === "number" &&
|
||||
Number.isSafeInteger(locked.action.result.attempts)
|
||||
? locked.action.result.attempts
|
||||
: 0;
|
||||
attempt = priorAttempts + 1;
|
||||
const retryAt =
|
||||
typeof locked?.action.result?.retryAt === "string"
|
||||
? Date.parse(locked.action.result.retryAt)
|
||||
: Number.NaN;
|
||||
const eligible = locked
|
||||
? locked.action.status === "received" ||
|
||||
(locked.action.status === "failed" &&
|
||||
locked.action.result?.retryable === true &&
|
||||
Number.isFinite(retryAt) &&
|
||||
retryAt <= decisionAt.getTime()) ||
|
||||
(locked.action.status === "processing" &&
|
||||
locked.action.updatedAt.getTime() <=
|
||||
decisionAt.getTime() - SLACK_FILE_RECEIPT_STALE_MS)
|
||||
: false;
|
||||
if (locked && !eligible) return null;
|
||||
if (locked?.mode === "defer") {
|
||||
if (locked.action.status === "processing") {
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "received",
|
||||
result: {
|
||||
code: "slack_file_upload_identity_pending",
|
||||
attempts: priorAttempts,
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(chatActions.id, locked.action.id));
|
||||
}
|
||||
return null;
|
||||
}
|
||||
if (locked?.mode === "settled") {
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "processed",
|
||||
result: {
|
||||
code: "slack_file_upload_identity_confirmed",
|
||||
attempts: priorAttempts,
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(chatActions.id, locked.action.id));
|
||||
return null;
|
||||
}
|
||||
if (
|
||||
!locked ||
|
||||
decisionAt.getTime() -
|
||||
(locked?.action.createdAt ?? selected.createdAt).getTime() >
|
||||
SLACK_FILE_RECEIPT_MAX_AGE_MS ||
|
||||
attempt > SLACK_FILE_RECEIPT_MAX_ATTEMPTS
|
||||
) {
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
result: {
|
||||
code: locked
|
||||
? "slack_file_upload_receipt_expired"
|
||||
: "slack_file_upload_receipt_authorization_changed",
|
||||
attempts: priorAttempts,
|
||||
},
|
||||
updatedAt: decisionAt,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, selected.id),
|
||||
inArray(chatActions.status, [
|
||||
"received",
|
||||
"failed",
|
||||
"processing",
|
||||
]),
|
||||
),
|
||||
);
|
||||
return null;
|
||||
}
|
||||
if (
|
||||
!["received", "failed", "processing"].includes(
|
||||
locked.action.status,
|
||||
)
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
await notifyChatPublicationWork(tx);
|
||||
const [claimed] = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "processing",
|
||||
result: {
|
||||
attempts: attempt,
|
||||
ownerToken,
|
||||
...(locked.mode === "operator_confirmed"
|
||||
? {
|
||||
operatorConfirmedPublicationId:
|
||||
locked.action.result?.operatorConfirmedPublicationId,
|
||||
operatorConfirmedAttempt:
|
||||
locked.action.result?.operatorConfirmedAttempt,
|
||||
operatorConfirmedAt:
|
||||
locked.action.result?.operatorConfirmedAt,
|
||||
}
|
||||
: {}),
|
||||
},
|
||||
updatedAt: decisionAt,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, locked.action.id),
|
||||
eq(chatActions.status, locked.action.status),
|
||||
),
|
||||
)
|
||||
.returning({ id: chatActions.id });
|
||||
await lease.assertOwned(tx);
|
||||
return claimed ? locked : null;
|
||||
});
|
||||
if (!claim || !("endpoint" in claim)) return;
|
||||
processed += 1;
|
||||
const endpointRuntime = await runtimeFor(claim.endpoint);
|
||||
await lease.assertOwned();
|
||||
const providerMessageId =
|
||||
await endpointRuntime.resolveSlackFileUploadReceipt(
|
||||
payload.threadId,
|
||||
payload.fileIds,
|
||||
);
|
||||
if (!providerMessageId) {
|
||||
await db.transaction(async (tx) => {
|
||||
await lease.assertOwned(tx);
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "failed",
|
||||
result: sql`coalesce(${chatActions.result}, '{}'::jsonb) || ${JSON.stringify(
|
||||
{
|
||||
attempts: attempt,
|
||||
code: "slack_file_upload_identity_pending",
|
||||
retryable: true,
|
||||
retryAt: new Date(
|
||||
Date.now() + Math.min(60_000, 1_000 * 2 ** attempt),
|
||||
).toISOString(),
|
||||
},
|
||||
)}::jsonb`,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, selected.id),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`,
|
||||
),
|
||||
);
|
||||
});
|
||||
return;
|
||||
}
|
||||
await db.transaction(async (tx) => {
|
||||
await lease.assertOwned(tx);
|
||||
const locked = await lockSlackFileReceiptAuthorization(tx, {
|
||||
actionId: selected.id,
|
||||
payload,
|
||||
});
|
||||
if (
|
||||
!locked ||
|
||||
locked.mode === "defer" ||
|
||||
locked.mode === "settled" ||
|
||||
!("publication" in locked) ||
|
||||
locked.action.status !== "processing" ||
|
||||
locked.action.result?.ownerToken !== ownerToken
|
||||
) {
|
||||
if (!locked) {
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
result: {
|
||||
attempts: attempt,
|
||||
code: "slack_file_upload_receipt_authorization_changed",
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, selected.id),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`,
|
||||
),
|
||||
);
|
||||
} else if (locked.mode === "defer") {
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "received",
|
||||
result: {
|
||||
attempts: attempt,
|
||||
code: "slack_file_upload_identity_pending",
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, selected.id),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`,
|
||||
),
|
||||
);
|
||||
} else {
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status:
|
||||
locked.mode === "settled" ? "processed" : "cancelled",
|
||||
result: {
|
||||
attempts: attempt,
|
||||
code:
|
||||
locked.mode === "settled"
|
||||
? "slack_file_upload_identity_confirmed"
|
||||
: "slack_file_upload_receipt_superseded",
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, selected.id),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`,
|
||||
),
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
const committedAt = new Date();
|
||||
if (
|
||||
!(await recordExactOutboundPublicationLink(tx, {
|
||||
publication: locked.publication,
|
||||
providerMessageId,
|
||||
}))
|
||||
) {
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "failed",
|
||||
result: {
|
||||
attempts: attempt,
|
||||
code: "slack_file_upload_receipt_message_conflict",
|
||||
retryable: false,
|
||||
},
|
||||
updatedAt: committedAt,
|
||||
})
|
||||
.where(eq(chatActions.id, selected.id));
|
||||
return;
|
||||
}
|
||||
if (locked.mode === "delivery_unknown") {
|
||||
await notifyChatPublicationWork(tx);
|
||||
const [settled] = await tx
|
||||
.update(chatPublications)
|
||||
.set({
|
||||
state: "published",
|
||||
providerMessageId,
|
||||
publishedAt: committedAt,
|
||||
nextAttemptAt: null,
|
||||
redactedError: null,
|
||||
updatedAt: committedAt,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.id, locked.publication.id),
|
||||
eq(chatPublications.state, "delivery_unknown"),
|
||||
eq(chatPublications.attempts, payload.publicationAttempt),
|
||||
),
|
||||
)
|
||||
.returning({ id: chatPublications.id });
|
||||
if (!settled) {
|
||||
throw new Error(
|
||||
"Slack file upload publication changed before receipt commit",
|
||||
);
|
||||
}
|
||||
await stageSlackFilePublicationSuccess(tx, {
|
||||
committedAt,
|
||||
endpoint: locked.endpoint,
|
||||
publication: locked.publication,
|
||||
providerMessageId,
|
||||
runtimeContext: {
|
||||
generation: payload.runtimeGeneration,
|
||||
credentialFingerprint: payload.credentialFingerprint,
|
||||
},
|
||||
stageSessionSync:
|
||||
!isExplicitOperatorPublication(locked.publication) &&
|
||||
!locked.publication.idempotencyKey.startsWith(
|
||||
"control:status:",
|
||||
),
|
||||
});
|
||||
} else {
|
||||
await notifyChatPublicationWork(tx);
|
||||
const [enriched] = await tx
|
||||
.update(chatPublications)
|
||||
.set({ providerMessageId })
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.id, locked.publication.id),
|
||||
eq(chatPublications.state, "published"),
|
||||
eq(chatPublications.attempts, payload.publicationAttempt),
|
||||
isNull(chatPublications.providerMessageId),
|
||||
),
|
||||
)
|
||||
.returning({ id: chatPublications.id });
|
||||
if (!enriched) {
|
||||
throw new Error(
|
||||
"Slack file upload publication changed before identity enrichment",
|
||||
);
|
||||
}
|
||||
}
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "processed",
|
||||
result: {
|
||||
attempts: attempt,
|
||||
code: "slack_file_upload_identity_confirmed",
|
||||
},
|
||||
updatedAt: committedAt,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, selected.id),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`,
|
||||
),
|
||||
);
|
||||
await lease.assertOwned(tx);
|
||||
});
|
||||
});
|
||||
} catch (error) {
|
||||
const disposition = classifyChatPublicationError(error, attempt);
|
||||
const retryable =
|
||||
disposition.kind === "retry" ||
|
||||
disposition.kind === "delivery_unknown";
|
||||
const retryMs =
|
||||
disposition.kind === "retry"
|
||||
? disposition.retryAfterMs
|
||||
: Math.min(60_000, 1_000 * 2 ** Math.min(attempt, 8));
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatPublicationWork(tx);
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: retryable ? "failed" : "cancelled",
|
||||
result: sql`coalesce(${chatActions.result}, '{}'::jsonb) || ${JSON.stringify(
|
||||
{
|
||||
attempts: attempt,
|
||||
code: retryable
|
||||
? "slack_file_upload_receipt_lookup_retry"
|
||||
: "slack_file_upload_receipt_lookup_rejected",
|
||||
retryable,
|
||||
...(retryable
|
||||
? { retryAt: new Date(Date.now() + retryMs).toISOString() }
|
||||
: {}),
|
||||
},
|
||||
)}::jsonb`,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, selected.id),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`,
|
||||
),
|
||||
);
|
||||
});
|
||||
}
|
||||
}
|
||||
return processed;
|
||||
}
|
||||
|
||||
function slackReceiptEligibility(selectedAt?: Date) {
|
||||
return and(
|
||||
eq(chatActions.kind, "slack_file_upload_receipt"),
|
||||
notExists(
|
||||
db
|
||||
.select({ id: chatEndpoints.id })
|
||||
.from(chatEndpoints)
|
||||
.where(
|
||||
and(
|
||||
eq(chatEndpoints.id, chatActions.endpointId),
|
||||
inArray(chatEndpoints.status, ["paused", "attention"]),
|
||||
),
|
||||
),
|
||||
),
|
||||
// The normal publication owner is still polling Slack. Receipt
|
||||
// recovery must neither contend with it nor let its not-yet-due
|
||||
// action occupy the bounded recovery page.
|
||||
notExists(
|
||||
db
|
||||
.select({ id: chatPublications.id })
|
||||
.from(chatPublications)
|
||||
.where(
|
||||
and(
|
||||
sql`${chatPublications.id}::text = ${chatActions.payload}->>'publicationId'`,
|
||||
sql`${chatPublications.attempts}::text = ${chatActions.payload}->>'publicationAttempt'`,
|
||||
eq(chatPublications.state, "streaming"),
|
||||
),
|
||||
),
|
||||
),
|
||||
or(
|
||||
eq(chatActions.status, "received"),
|
||||
and(
|
||||
eq(chatActions.status, "failed"),
|
||||
sql`coalesce(${chatActions.result}->>'retryable', 'false') = 'true'`,
|
||||
selectedAt ? sql`(${chatActions.result}->>'retryAt')::timestamptz <= ${selectedAt.toISOString()}::timestamptz` : sql`${chatActions.result}->>'retryAt' is not null`,
|
||||
),
|
||||
and(
|
||||
eq(chatActions.status, "processing"),
|
||||
selectedAt ? lte(chatActions.updatedAt, new Date(selectedAt.getTime() - SLACK_FILE_RECEIPT_STALE_MS)) : undefined,
|
||||
),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
async function nextSlackReceiptAt(): Promise<number | null> {
|
||||
const [row] = await db.select({ at: sql<string | null>`min(case
|
||||
when ${chatActions.status} = 'processing' then ${chatActions.updatedAt} + ${SLACK_FILE_RECEIPT_STALE_MS} * interval '1 millisecond'
|
||||
when ${chatActions.status} = 'failed' then (${chatActions.result}->>'retryAt')::timestamptz
|
||||
else now() end)` }).from(chatActions).where(slackReceiptEligibility());
|
||||
return row?.at == null ? null : new Date(row.at).getTime();
|
||||
}
|
||||
return { recordSlackFileUploadReceipt, stageSlackFilePublicationSuccess, processPendingSlackFileUploadReceipts, nextSlackReceiptAt };
|
||||
}
|
||||
@@ -0,0 +1,408 @@
|
||||
import type { EndpointRow, DbTransaction, DbOrTransaction, LifecycleRuntimeFence } from "../../chat-channels/runtime-types.js";
|
||||
|
||||
import { dueChatAction } from "../../chat-action-work.js";
|
||||
|
||||
import { notifyChatActionWork } from "../../chat-work-notifications.js";
|
||||
|
||||
import { chatCredentialMutationLease } from "../../chat-credential-mutation-lease.js";
|
||||
|
||||
import {
|
||||
randomUUID,
|
||||
} from "node:crypto";
|
||||
import {
|
||||
and,
|
||||
asc,
|
||||
desc,
|
||||
eq,
|
||||
gt,
|
||||
inArray,
|
||||
lte,
|
||||
or,
|
||||
sql,
|
||||
} from "drizzle-orm";
|
||||
|
||||
import type { Db } from "@paperclipai/db";
|
||||
|
||||
import {
|
||||
chatActions,
|
||||
chatConversations,
|
||||
chatEndpointResources,
|
||||
chatEndpoints,
|
||||
chatPublications,
|
||||
heartbeatRuns,
|
||||
} from "@paperclipai/db";
|
||||
|
||||
import { logger } from "../../../middleware/logger.js";
|
||||
|
||||
import {
|
||||
setSlackSessionStatus,
|
||||
slackSessionStatusForPublication,
|
||||
} from "../../chat-slack-sessions.js";
|
||||
import {
|
||||
slackSessionSyncPayload,
|
||||
} from "../../chat-slack-session-outbox.js";
|
||||
|
||||
import { classifyChatPublicationError } from "../../chat-publication-errors.js";
|
||||
|
||||
import type { EndpointRecord } from "../runtime-types.js";
|
||||
|
||||
type SlackSessionStatusOptions = {
|
||||
endpointRecord: (id: string) => Promise<EndpointRecord | null>;
|
||||
withCredentialMutationLease: ReturnType<typeof chatCredentialMutationLease>;
|
||||
runtimeCallbackEndpoint: (tx: DbTransaction, endpointId: string, context: LifecycleRuntimeFence, statuses: EndpointRow["status"][]) => Promise<EndpointRow | null>;
|
||||
nonDirectDestinationAllowed: (endpoint: EndpointRow, resource: typeof chatEndpointResources.$inferSelect | null | undefined) => boolean;
|
||||
runIdFromMilestonePublication: (publication: typeof chatPublications.$inferSelect) => string | null;
|
||||
resolveCredentials: (endpoint: EndpointRow, database?: DbOrTransaction) => Promise<Record<string, string>>;
|
||||
fetchImpl: typeof globalThis.fetch;
|
||||
redactError: (error: unknown) => string;
|
||||
slackSessionSyncSelectionBarrier?: () => Promise<void>;
|
||||
};
|
||||
|
||||
export function createSlackSessionStatus(db: Db, options: SlackSessionStatusOptions) {
|
||||
const { endpointRecord, withCredentialMutationLease, runtimeCallbackEndpoint, nonDirectDestinationAllowed, runIdFromMilestonePublication, resolveCredentials, fetchImpl, redactError } = options;
|
||||
|
||||
|
||||
async function processPendingSlackSessionSyncs(
|
||||
limit = 25,
|
||||
onlyActionId?: string,
|
||||
) {
|
||||
const now = new Date();
|
||||
const actions = await db
|
||||
.select({ action: chatActions })
|
||||
.from(chatActions)
|
||||
.leftJoin(chatEndpoints, and(eq(chatEndpoints.id, chatActions.endpointId), eq(chatEndpoints.companyId, chatActions.companyId)))
|
||||
.where(and(
|
||||
onlyActionId ? eq(chatActions.id, onlyActionId) : undefined,
|
||||
dueChatAction("slack_session_sync", now.getTime()),
|
||||
))
|
||||
.orderBy(asc(chatActions.updatedAt))
|
||||
.limit(limit)
|
||||
.then(rows => rows.map(row => row.action));
|
||||
await options.slackSessionSyncSelectionBarrier?.();
|
||||
for (const action of actions) {
|
||||
const payload = slackSessionSyncPayload(action.payload);
|
||||
if (!payload || !action.conversationId) {
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "failed",
|
||||
result: { code: "slack_session_payload_invalid" },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, action.id),
|
||||
eq(chatActions.updatedAt, action.updatedAt),
|
||||
),
|
||||
);
|
||||
});
|
||||
continue;
|
||||
}
|
||||
const revisionWhere = and(
|
||||
eq(chatActions.id, action.id),
|
||||
sql`(${chatActions.payload}->>'revision')::bigint = ${payload.revision}`,
|
||||
);
|
||||
const selectedSnapshotWhere = and(
|
||||
revisionWhere,
|
||||
eq(chatActions.status, action.status),
|
||||
eq(chatActions.updatedAt, action.updatedAt),
|
||||
);
|
||||
const record = await endpointRecord(action.endpointId);
|
||||
if (record && ["paused", "attention"].includes(record.endpoint.status)) continue;
|
||||
if (!record || record.endpoint.provider !== "slack") {
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
result: { code: "slack_session_endpoint_unavailable" },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(selectedSnapshotWhere);
|
||||
});
|
||||
continue;
|
||||
}
|
||||
const attempts =
|
||||
(typeof action.result?.attempts === "number"
|
||||
? action.result.attempts
|
||||
: 0) + 1;
|
||||
const ownerToken = randomUUID();
|
||||
let ownsAttempt = false;
|
||||
try {
|
||||
await withCredentialMutationLease(record.endpoint, async (lease) => {
|
||||
const claim = await db.transaction(async (tx) => {
|
||||
await lease.assertOwned(tx);
|
||||
const endpoint = await runtimeCallbackEndpoint(
|
||||
tx,
|
||||
action.endpointId,
|
||||
{
|
||||
generation: payload.runtimeGeneration,
|
||||
credentialFingerprint: payload.credentialFingerprint,
|
||||
},
|
||||
["active", "verifying"],
|
||||
);
|
||||
const conversation = endpoint
|
||||
? await tx
|
||||
.select()
|
||||
.from(chatConversations)
|
||||
.where(
|
||||
and(
|
||||
eq(chatConversations.id, action.conversationId!),
|
||||
eq(chatConversations.endpointId, endpoint.id),
|
||||
eq(chatConversations.companyId, endpoint.companyId),
|
||||
inArray(chatConversations.state, [
|
||||
"active",
|
||||
"waiting",
|
||||
"completed",
|
||||
]),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
const resource =
|
||||
endpoint && conversation?.resourceId
|
||||
? await tx
|
||||
.select()
|
||||
.from(chatEndpointResources)
|
||||
.where(
|
||||
and(
|
||||
eq(chatEndpointResources.id, conversation.resourceId),
|
||||
eq(chatEndpointResources.companyId, endpoint.companyId),
|
||||
eq(chatEndpointResources.endpointId, endpoint.id),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
const superseded = conversation
|
||||
? await tx
|
||||
.select({ id: chatConversations.id })
|
||||
.from(chatConversations)
|
||||
.where(
|
||||
and(
|
||||
eq(chatConversations.companyId, conversation.companyId),
|
||||
eq(chatConversations.endpointId, conversation.endpointId),
|
||||
eq(
|
||||
chatConversations.externalThreadId,
|
||||
conversation.externalThreadId,
|
||||
),
|
||||
gt(
|
||||
chatConversations.sessionGeneration,
|
||||
conversation.sessionGeneration,
|
||||
),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
.then((rows) => rows.length > 0)
|
||||
: false;
|
||||
if (
|
||||
!endpoint ||
|
||||
!conversation ||
|
||||
superseded ||
|
||||
!(conversation.isDirectMessage
|
||||
? endpoint.allowDirectMessages
|
||||
: nonDirectDestinationAllowed(endpoint, resource))
|
||||
) {
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
result: { code: "slack_session_destination_changed" },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(selectedSnapshotWhere);
|
||||
return null;
|
||||
}
|
||||
// Recompute at dispatch: a delayed processing retry must not
|
||||
// overwrite a newer final response or a later question card.
|
||||
const publication = await tx
|
||||
.select()
|
||||
.from(chatPublications)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.companyId, endpoint.companyId),
|
||||
eq(chatPublications.endpointId, endpoint.id),
|
||||
eq(chatPublications.conversationId, conversation.id),
|
||||
eq(chatPublications.state, "published"),
|
||||
sql`${chatPublications.idempotencyKey} not like 'explicit%'`,
|
||||
sql`${chatPublications.idempotencyKey} not like 'control:status:%'`,
|
||||
sql`${chatPublications.idempotencyKey} not like 'wake:%'`,
|
||||
),
|
||||
)
|
||||
.orderBy(
|
||||
desc(chatPublications.publishedAt),
|
||||
desc(chatPublications.createdAt),
|
||||
desc(chatPublications.id),
|
||||
)
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!publication) {
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
result: { code: "slack_session_no_published_state" },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(selectedSnapshotWhere);
|
||||
return null;
|
||||
}
|
||||
let status = slackSessionStatusForPublication(
|
||||
publication.payload,
|
||||
conversation.state === "completed",
|
||||
);
|
||||
const runId = runIdFromMilestonePublication(publication);
|
||||
if (status === "processing" && runId) {
|
||||
const run = await tx
|
||||
.select({ status: heartbeatRuns.status })
|
||||
.from(heartbeatRuns)
|
||||
.where(
|
||||
and(
|
||||
eq(heartbeatRuns.id, runId),
|
||||
eq(heartbeatRuns.companyId, endpoint.companyId),
|
||||
eq(heartbeatRuns.agentId, endpoint.assignedAgentId),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
// Stop may finish before its confirmation publication. Never
|
||||
// resurrect processing from that run's earlier working receipt.
|
||||
if (!run || !["queued", "running"].includes(run.status))
|
||||
status = "active";
|
||||
}
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
const claimed = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "processing",
|
||||
result: { attempts, ownerToken },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
selectedSnapshotWhere,
|
||||
or(
|
||||
and(
|
||||
eq(chatActions.status, "received"),
|
||||
sql`(${chatActions.result}->>'retryAt' is null or (${chatActions.result}->>'retryAt')::timestamptz <= ${new Date().toISOString()}::timestamptz)`,
|
||||
),
|
||||
and(
|
||||
eq(chatActions.status, "processing"),
|
||||
lte(chatActions.updatedAt, new Date(Date.now() - 60_000)),
|
||||
),
|
||||
),
|
||||
),
|
||||
)
|
||||
.returning({ id: chatActions.id });
|
||||
await lease.assertOwned(tx);
|
||||
return claimed.length ? { endpoint, conversation, status } : null;
|
||||
});
|
||||
if (!claim) return;
|
||||
ownsAttempt = true;
|
||||
const credentials = await resolveCredentials(claim.endpoint);
|
||||
await lease.assertOwned();
|
||||
const outcome = await setSlackSessionStatus({
|
||||
botToken: credentials.botToken,
|
||||
threadId: claim.conversation.externalThreadId,
|
||||
status: claim.status,
|
||||
fetch: fetchImpl,
|
||||
});
|
||||
await db.transaction(async (tx) => {
|
||||
await lease.assertOwned(tx);
|
||||
// Refresh long work before Slack's one-hour processing timeout.
|
||||
// Unsupported sessions settle until a new publication restages
|
||||
// them; completed threads must not create perpetual polling work.
|
||||
const retryMs =
|
||||
outcome === "updated" && claim.status === "processing"
|
||||
? 30 * 60_000
|
||||
: null;
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: retryMs === null ? "processed" : "received",
|
||||
result: {
|
||||
attempts,
|
||||
outcome,
|
||||
sessionStatus: claim.status,
|
||||
...(retryMs === null
|
||||
? {}
|
||||
: {
|
||||
retryAt: new Date(Date.now() + retryMs).toISOString(),
|
||||
}),
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
revisionWhere,
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`,
|
||||
),
|
||||
);
|
||||
});
|
||||
});
|
||||
} catch (error) {
|
||||
const disposition = classifyChatPublicationError(error, attempts);
|
||||
// Unlike posting a message, repeating setStatus cannot duplicate any
|
||||
// content. Ambiguous transport/receipt failures remain retryable, but
|
||||
// a definite scope/auth/destination rejection waits for new activity.
|
||||
const retryable =
|
||||
disposition.kind === "retry" ||
|
||||
disposition.kind === "delivery_unknown";
|
||||
const retryMs =
|
||||
disposition.kind === "retry"
|
||||
? disposition.retryAfterMs
|
||||
: Math.min(5 * 60_000, 1_000 * 2 ** Math.min(attempts, 8));
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: retryable ? "received" : "failed",
|
||||
result: {
|
||||
attempts,
|
||||
code: retryable
|
||||
? "slack_session_sync_retry"
|
||||
: "slack_session_sync_rejected",
|
||||
...(retryable
|
||||
? { retryAt: new Date(Date.now() + retryMs).toISOString() }
|
||||
: {}),
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
revisionWhere,
|
||||
ownsAttempt
|
||||
? and(
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`,
|
||||
)
|
||||
: and(
|
||||
eq(chatActions.status, "received"),
|
||||
eq(chatActions.updatedAt, action.updatedAt),
|
||||
),
|
||||
),
|
||||
);
|
||||
});
|
||||
logger.warn(
|
||||
{
|
||||
endpointId: action.endpointId,
|
||||
actionId: action.id,
|
||||
retryable,
|
||||
error: redactError(error),
|
||||
},
|
||||
"Slack session status update did not complete; message delivery is unchanged",
|
||||
);
|
||||
}
|
||||
}
|
||||
return actions.length;
|
||||
}
|
||||
return { processPendingSlackSessionSyncs };
|
||||
}
|
||||
@@ -0,0 +1,461 @@
|
||||
import type { EndpointRow, DbTransaction, DbOrTransaction, RuntimeContext, LifecycleRuntimeFence } from "../../chat-channels/runtime-types.js";
|
||||
|
||||
import { notifyChatPublicationWork } from "../../chat-work-notifications.js";
|
||||
|
||||
import { type ChatRuntime } from "../../chat-runtime.js";
|
||||
import type { ChatEndpointRuntime } from "../../voice/voice-runtime.js";
|
||||
|
||||
import { type CredentialMutationLeaseGuard } from "../../chat-credential-mutation-lease.js";
|
||||
|
||||
import {
|
||||
and,
|
||||
eq,
|
||||
sql,
|
||||
} from "drizzle-orm";
|
||||
|
||||
import type { Db } from "@paperclipai/db";
|
||||
|
||||
import {
|
||||
chatActions,
|
||||
chatConversations,
|
||||
chatPublications,
|
||||
toolConnections,
|
||||
} from "@paperclipai/db";
|
||||
import type {
|
||||
ToolCredentialSecretRef,
|
||||
} from "@paperclipai/shared";
|
||||
|
||||
import {
|
||||
TELEGRAM_DRAFT_ACTION_KIND,
|
||||
TELEGRAM_DRAFT_STOPPED_REASON,
|
||||
parseTelegramDraftBinding,
|
||||
telegramDraftTextSha256,
|
||||
telegramGenerationStoppedReceipt,
|
||||
telegramPrivateDraftDestination,
|
||||
type TelegramDraftControl,
|
||||
type TelegramGenerationStoppedProof,
|
||||
} from "../../chat-telegram-draft-stop.js";
|
||||
|
||||
import {
|
||||
type ChatSdkCallbackEvent,
|
||||
type ChatSdkRuntime,
|
||||
} from "../../chat-sdk-runtime.js";
|
||||
|
||||
import {
|
||||
renderPublicationTransportText,
|
||||
} from "../../chat-publication-text-parts.js";
|
||||
|
||||
import type { EndpointRecord } from "../runtime-types.js";
|
||||
|
||||
type ConversationRow = typeof chatConversations.$inferSelect;
|
||||
type TelegramPublicationDraftsOptions = {
|
||||
runtimeCallbackRecord: (endpointId: string, context: RuntimeContext, statuses: EndpointRow["status"][]) => Promise<EndpointRecord | null>;
|
||||
runtimeCallbackEndpoint: (tx: DbTransaction, endpointId: string, context: LifecycleRuntimeFence, statuses: EndpointRow["status"][]) => Promise<EndpointRow | null>;
|
||||
runtimeFor: (endpoint: EndpointRow) => Promise<ChatEndpointRuntime>;
|
||||
runtime: Pick<ChatSdkRuntime | ChatRuntime, "get">;
|
||||
runtimeContexts: WeakMap<object, RuntimeContext>;
|
||||
credentialFingerprint: (refs: ToolCredentialSecretRef[]) => string;
|
||||
authorizeRetainedChatSourcePublication: (tx: DbOrTransaction, publication: typeof chatPublications.$inferSelect) => Promise<boolean>;
|
||||
};
|
||||
|
||||
export function createTelegramPublicationDrafts(db: Db, options: TelegramPublicationDraftsOptions) {
|
||||
const { runtimeCallbackRecord, runtimeCallbackEndpoint, runtimeFor, runtime, runtimeContexts, credentialFingerprint, authorizeRetainedChatSourcePublication } = options;
|
||||
|
||||
|
||||
async function handleTelegramGenerationStopped(
|
||||
callback: ChatSdkCallbackEvent<TelegramGenerationStoppedProof>,
|
||||
context: RuntimeContext,
|
||||
): Promise<void> {
|
||||
const receipt = telegramGenerationStoppedReceipt(callback.event);
|
||||
if (callback.provider !== "telegram" || !receipt) return;
|
||||
const record = await runtimeCallbackRecord(callback.endpointId, context, [
|
||||
"verifying",
|
||||
"active",
|
||||
]);
|
||||
if (
|
||||
record?.endpoint.provider !== "telegram" ||
|
||||
record.endpoint.botExternalId !== receipt.botUserId
|
||||
)
|
||||
return;
|
||||
const candidates = await db
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.companyId, record.endpoint.companyId),
|
||||
eq(chatActions.endpointId, callback.endpointId),
|
||||
eq(chatActions.kind, TELEGRAM_DRAFT_ACTION_KIND),
|
||||
sql`${chatActions.payload}->>'botUserId' = ${receipt.botUserId}`,
|
||||
sql`${chatActions.payload}->>'draftId' = ${String(receipt.draftId)}`,
|
||||
),
|
||||
)
|
||||
.limit(2);
|
||||
if (candidates.length !== 1) return;
|
||||
const candidate = candidates[0]!;
|
||||
const binding = parseTelegramDraftBinding(candidate.payload);
|
||||
if (
|
||||
!binding ||
|
||||
binding.chatId !== receipt.chatId ||
|
||||
binding.messageThreadId !== receipt.messageThreadId ||
|
||||
binding.runtimeGeneration !== context.generation ||
|
||||
binding.credentialFingerprint !== context.credentialFingerprint
|
||||
)
|
||||
return;
|
||||
await db.transaction(async (tx) => {
|
||||
// The sender holds a renewable credential lease during draft HTTP. Do
|
||||
// not acquire that lease here: a Stop can arrive inside the first RPC.
|
||||
// Both Stop and final-send use the publication→endpoint→draft lock order.
|
||||
const publication = await tx
|
||||
.select()
|
||||
.from(chatPublications)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.companyId, record.endpoint.companyId),
|
||||
eq(chatPublications.endpointId, callback.endpointId),
|
||||
eq(chatPublications.conversationId, binding.conversationId),
|
||||
eq(chatPublications.id, binding.publicationId),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const endpoint = await runtimeCallbackEndpoint(
|
||||
tx,
|
||||
callback.endpointId,
|
||||
context,
|
||||
["verifying", "active"],
|
||||
);
|
||||
if (
|
||||
!publication ||
|
||||
!endpoint ||
|
||||
endpoint.provider !== "telegram" ||
|
||||
endpoint.botExternalId !== receipt.botUserId ||
|
||||
runtime.get(endpoint.id) !== context.endpointRuntime ||
|
||||
publication.attempts !== binding.publicationAttempt ||
|
||||
telegramDraftTextSha256(
|
||||
renderPublicationTransportText(publication.payload),
|
||||
) !== binding.textSha256
|
||||
)
|
||||
return;
|
||||
const connection = await tx
|
||||
.select()
|
||||
.from(toolConnections)
|
||||
.where(
|
||||
and(
|
||||
eq(toolConnections.companyId, endpoint.companyId),
|
||||
eq(toolConnections.id, endpoint.connectionId),
|
||||
),
|
||||
)
|
||||
.for("share")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (
|
||||
!connection?.enabled ||
|
||||
connection.status !== "active" ||
|
||||
credentialFingerprint(connection.credentialSecretRefs) !==
|
||||
binding.credentialFingerprint
|
||||
)
|
||||
return;
|
||||
const conversation = await tx
|
||||
.select()
|
||||
.from(chatConversations)
|
||||
.where(
|
||||
and(
|
||||
eq(chatConversations.companyId, endpoint.companyId),
|
||||
eq(chatConversations.endpointId, endpoint.id),
|
||||
eq(chatConversations.id, binding.conversationId),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (
|
||||
!conversation ||
|
||||
!conversation.isDirectMessage ||
|
||||
conversation.sessionGeneration !== binding.sessionGeneration ||
|
||||
conversation.externalThreadId !==
|
||||
`telegram:${binding.chatId}${binding.messageThreadId === null ? "" : `:${binding.messageThreadId}`}`
|
||||
)
|
||||
return;
|
||||
const action = await tx
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(eq(chatActions.id, candidate.id))
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (
|
||||
!action ||
|
||||
JSON.stringify(action.payload) !== JSON.stringify(candidate.payload)
|
||||
)
|
||||
return;
|
||||
// final_sending is an irreversible local I/O claim, not a delivery
|
||||
// receipt. A late Stop must not claim that an in-flight final was undone.
|
||||
if (
|
||||
action.result?.phase !== "drafting" ||
|
||||
action.status !== "processing" ||
|
||||
!["streaming", "delivery_unknown"].includes(publication.state)
|
||||
)
|
||||
return;
|
||||
const stoppedAt = new Date();
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
result: {
|
||||
phase: "stopped",
|
||||
updateId: receipt.updateId,
|
||||
stoppedAt: stoppedAt.toISOString(),
|
||||
},
|
||||
updatedAt: stoppedAt,
|
||||
})
|
||||
.where(eq(chatActions.id, action.id));
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.update(chatPublications)
|
||||
.set({
|
||||
state: "cancelled",
|
||||
nextAttemptAt: null,
|
||||
redactedError: TELEGRAM_DRAFT_STOPPED_REASON,
|
||||
updatedAt: stoppedAt,
|
||||
})
|
||||
.where(eq(chatPublications.id, publication.id));
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
result: {
|
||||
code: "telegram_draft_presentation_stopped",
|
||||
attempts: publication.attempts,
|
||||
},
|
||||
updatedAt: stoppedAt,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.companyId, publication.companyId),
|
||||
eq(chatActions.endpointId, publication.endpointId),
|
||||
eq(
|
||||
chatActions.providerActionId,
|
||||
`task-control-authorization:${publication.id}`,
|
||||
),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`${chatActions.result}->>'attempts' = ${String(publication.attempts)}`,
|
||||
),
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
async function reserveTelegramPublicationDraft(input: {
|
||||
endpoint: EndpointRow;
|
||||
conversation: ConversationRow;
|
||||
publication: typeof chatPublications.$inferSelect;
|
||||
text: string;
|
||||
credentialLease: CredentialMutationLeaseGuard;
|
||||
runtimeContext: RuntimeContext;
|
||||
}): Promise<{ actionId: string; control: TelegramDraftControl }> {
|
||||
const endpointRuntime = await runtimeFor(input.endpoint);
|
||||
const registeredContext = runtimeContexts.get(endpointRuntime as object);
|
||||
if (
|
||||
!registeredContext ||
|
||||
registeredContext.generation !== input.runtimeContext.generation ||
|
||||
registeredContext.credentialFingerprint !==
|
||||
input.runtimeContext.credentialFingerprint ||
|
||||
registeredContext.version !== input.runtimeContext.version
|
||||
)
|
||||
throw new Error("Telegram draft runtime authority changed");
|
||||
const draftRuntimeContext = { ...input.runtimeContext, endpointRuntime };
|
||||
const destination = telegramPrivateDraftDestination(
|
||||
input.conversation.externalThreadId,
|
||||
);
|
||||
if (
|
||||
!destination ||
|
||||
!input.conversation.isDirectMessage ||
|
||||
!input.endpoint.botExternalId
|
||||
)
|
||||
throw new Error("Telegram private draft destination is invalid");
|
||||
const bindingBase = {
|
||||
version: 1 as const,
|
||||
publicationId: input.publication.id,
|
||||
publicationAttempt: input.publication.attempts + 1,
|
||||
conversationId: input.conversation.id,
|
||||
sessionGeneration: input.conversation.sessionGeneration,
|
||||
runtimeGeneration: input.runtimeContext.generation,
|
||||
credentialFingerprint: input.runtimeContext.credentialFingerprint,
|
||||
botUserId: input.endpoint.botExternalId,
|
||||
...destination,
|
||||
textSha256: telegramDraftTextSha256(input.text),
|
||||
};
|
||||
const reserved = await db.transaction(async (tx) => {
|
||||
await input.credentialLease.assertOwned(tx);
|
||||
const publication = await tx
|
||||
.select()
|
||||
.from(chatPublications)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.id, input.publication.id),
|
||||
eq(chatPublications.state, "streaming"),
|
||||
eq(chatPublications.attempts, bindingBase.publicationAttempt),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!publication)
|
||||
throw new Error("Telegram draft publication ownership changed");
|
||||
const providerActionId = `telegram-draft-publication:${publication.id}:${publication.attempts}`;
|
||||
const existing = await tx
|
||||
.select({ id: chatActions.id })
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.endpointId, input.endpoint.id),
|
||||
eq(chatActions.providerActionId, providerActionId),
|
||||
),
|
||||
)
|
||||
.limit(1);
|
||||
// A second invocation of the same attempt cannot create another draft or
|
||||
// replay an uncertain final. A new explicit publication attempt gets a
|
||||
// fresh, never-before-used bot-global ID.
|
||||
if (existing.length)
|
||||
throw new Error("Telegram draft attempt already exists");
|
||||
// NO CYCLE and no company/endpoint ownership: rollback, endpoint deletion
|
||||
// and bot re-binding cannot recycle a provider-visible draft identity.
|
||||
// Exhaustion is a refusal, never a fallback to a process-local/random ID.
|
||||
const allocated = await tx.execute(
|
||||
sql`select nextval('chat_telegram_draft_ids')::integer as "draftId"`,
|
||||
);
|
||||
const binding = parseTelegramDraftBinding({
|
||||
...bindingBase,
|
||||
draftId: allocated[0]?.draftId,
|
||||
});
|
||||
if (!binding)
|
||||
throw new Error("Telegram draft identity allocation was invalid");
|
||||
const [action] = await tx
|
||||
.insert(chatActions)
|
||||
.values({
|
||||
companyId: input.endpoint.companyId,
|
||||
endpointId: input.endpoint.id,
|
||||
conversationId: input.conversation.id,
|
||||
kind: TELEGRAM_DRAFT_ACTION_KIND,
|
||||
providerActionId,
|
||||
payload: binding,
|
||||
status: "issued",
|
||||
result: { phase: "reserved" },
|
||||
})
|
||||
.returning();
|
||||
return { action: action!, binding };
|
||||
});
|
||||
const before = (final: boolean): Promise<boolean> =>
|
||||
db.transaction(async (tx) => {
|
||||
await input.credentialLease.assertOwned(tx);
|
||||
const publication = await tx
|
||||
.select()
|
||||
.from(chatPublications)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.companyId, input.endpoint.companyId),
|
||||
eq(chatPublications.id, input.publication.id),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const endpoint = await runtimeCallbackEndpoint(
|
||||
tx,
|
||||
input.endpoint.id,
|
||||
draftRuntimeContext,
|
||||
["verifying", "active"],
|
||||
);
|
||||
if (
|
||||
!endpoint ||
|
||||
endpoint.provider !== "telegram" ||
|
||||
!endpoint.allowDirectMessages ||
|
||||
endpoint.botExternalId !== reserved.binding.botUserId ||
|
||||
runtime.get(endpoint.id) !== endpointRuntime
|
||||
)
|
||||
throw new Error("Telegram draft endpoint authority changed");
|
||||
const connection = await tx
|
||||
.select()
|
||||
.from(toolConnections)
|
||||
.where(
|
||||
and(
|
||||
eq(toolConnections.companyId, endpoint.companyId),
|
||||
eq(toolConnections.id, endpoint.connectionId),
|
||||
),
|
||||
)
|
||||
.for("share")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (
|
||||
!connection?.enabled ||
|
||||
connection.status !== "active" ||
|
||||
credentialFingerprint(connection.credentialSecretRefs) !==
|
||||
reserved.binding.credentialFingerprint
|
||||
)
|
||||
throw new Error("Telegram draft credential authority changed");
|
||||
const conversation = await tx
|
||||
.select()
|
||||
.from(chatConversations)
|
||||
.where(
|
||||
and(
|
||||
eq(chatConversations.companyId, endpoint.companyId),
|
||||
eq(chatConversations.endpointId, endpoint.id),
|
||||
eq(chatConversations.id, input.conversation.id),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const action = await tx
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(eq(chatActions.id, reserved.action.id))
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (
|
||||
!action ||
|
||||
JSON.stringify(action.payload) !==
|
||||
JSON.stringify(reserved.action.payload) ||
|
||||
!publication ||
|
||||
publication.attempts !== reserved.binding.publicationAttempt
|
||||
)
|
||||
throw new Error("Telegram draft attempt authority changed");
|
||||
if (
|
||||
action.status === "cancelled" &&
|
||||
action.result?.phase === "stopped" &&
|
||||
publication.state === "cancelled"
|
||||
)
|
||||
return false;
|
||||
if (
|
||||
publication.state !== "streaming" ||
|
||||
!conversation ||
|
||||
telegramDraftTextSha256(
|
||||
renderPublicationTransportText(publication.payload),
|
||||
) !== reserved.binding.textSha256 ||
|
||||
conversation.sessionGeneration !==
|
||||
reserved.binding.sessionGeneration ||
|
||||
conversation.externalThreadId !==
|
||||
input.conversation.externalThreadId ||
|
||||
!["active", "waiting", "completed"].includes(conversation.state) ||
|
||||
!["reserved", "drafting"].includes(String(action.result?.phase)) ||
|
||||
!["issued", "processing"].includes(action.status) ||
|
||||
!(await authorizeRetainedChatSourcePublication(tx, publication))
|
||||
)
|
||||
throw new Error("Telegram draft source authority changed");
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "processing",
|
||||
result: {
|
||||
phase: final ? "final_sending" : "drafting",
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(chatActions.id, action.id));
|
||||
await input.credentialLease.assertOwned(tx);
|
||||
return true;
|
||||
});
|
||||
return {
|
||||
actionId: reserved.action.id,
|
||||
control: {
|
||||
version: 1,
|
||||
draftId: reserved.binding.draftId,
|
||||
beforeDraft: () => before(false),
|
||||
beforeFinal: () => before(true),
|
||||
},
|
||||
};
|
||||
}
|
||||
return { handleTelegramGenerationStopped, reserveTelegramPublicationDraft };
|
||||
}
|
||||
@@ -12,7 +12,7 @@ import {
|
||||
resolveRemotePaperclipRunnerBinary,
|
||||
} from "./native-codex-runner.js";
|
||||
|
||||
afterEach(() => vi.restoreAllMocks());
|
||||
afterEach(() => { vi.restoreAllMocks(); vi.unstubAllGlobals(); vi.unstubAllEnvs(); });
|
||||
|
||||
describe("installed runner selection", () => {
|
||||
it("uses the runner's platform-aware packaged selection", async () => {
|
||||
@@ -46,7 +46,7 @@ describe("installed runner selection", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("selected SSH daemon target", () => {
|
||||
describe("selected remote daemon target", () => {
|
||||
function header(target: string) {
|
||||
const bytes = Buffer.alloc(64);
|
||||
if (target.startsWith("darwin")) {
|
||||
@@ -72,10 +72,72 @@ describe("selected SSH daemon target", () => {
|
||||
expect(target.execute).toHaveBeenCalledExactlyOnceWith({ command: "sh", args: ["-c", "uname -s; uname -m"], cwd: "/selected-workspace", bypassSession: true, timeoutMs: 10_000 });
|
||||
} finally { await rm(root, { recursive: true, force: true }); }
|
||||
});
|
||||
it.each(["arm64", "x64"].flatMap(controllerArch => ["arm64", "x64"].map(targetArch => ({ controllerArch, targetArch }))))(
|
||||
"selects the actual Linux $targetArch sandbox bytes on a Mac $controllerArch controller", async ({ controllerArch, targetArch }) => {
|
||||
const root = await mkdtemp(join(tmpdir(), "paperclip-sandbox-daemon-"));
|
||||
const binary = join(root, "paperclip-runnerd");
|
||||
try {
|
||||
vi.stubGlobal("process", { ...process, platform: "darwin", arch: controllerArch });
|
||||
await writeFile(binary, header(`linux-${targetArch}`));
|
||||
const controllerBinary = join(root, "controller-runnerd");
|
||||
await writeFile(controllerBinary, header(`darwin-${controllerArch}`), { mode: 0o700 });
|
||||
vi.stubEnv("PAPERCLIP_RUNNER_BINARY", controllerBinary);
|
||||
const select = vi.spyOn(runner, "defaultCapabilityRunnerdBinary").mockReturnValue(binary);
|
||||
const target = remote(`Linux\n${targetArch === "arm64" ? "aarch64" : "x86_64"}\n`);
|
||||
expect(await resolveRemotePaperclipRunnerBinary({ runner: target, remoteCwd: "/sandbox-workspace" })).toBe(binary);
|
||||
expect(select).toHaveBeenCalledExactlyOnceWith("linux", targetArch);
|
||||
expect(target.execute).toHaveBeenCalledExactlyOnceWith({ command: "sh", args: ["-c", "uname -s; uname -m"], cwd: "/sandbox-workspace", bypassSession: true, timeoutMs: 10_000 });
|
||||
} finally { await rm(root, { recursive: true, force: true }); }
|
||||
},
|
||||
);
|
||||
it.each(["linux", "darwin"].flatMap(platform => ["arm64", "x64"].map(architecture => ({ platform, architecture }))))(
|
||||
"prefers the supervisor's fresh $platform/$architecture override for the same-host target", async ({ platform, architecture }) => {
|
||||
const root = await mkdtemp(join(tmpdir(), "paperclip-same-host-daemon-"));
|
||||
try {
|
||||
vi.stubGlobal("process", { ...process, platform, arch: architecture });
|
||||
const fresh = join(root, "fresh-runnerd"), stale = join(root, "stale-packaged-runnerd");
|
||||
await writeFile(fresh, Buffer.concat([header(`${platform}-${architecture}`), Buffer.from("fresh")]), { mode: 0o700 });
|
||||
await writeFile(stale, Buffer.concat([header(`${platform}-${architecture}`), Buffer.from("stale")]), { mode: 0o700 });
|
||||
vi.stubEnv("PAPERCLIP_RUNNER_BINARY", fresh);
|
||||
const select = vi.spyOn(runner, "defaultCapabilityRunnerdBinary").mockReturnValue(stale);
|
||||
const target = remote(`${platform === "linux" ? "Linux" : "Darwin"}\n${architecture === "arm64" ? "arm64" : "x86_64"}\n`);
|
||||
expect(await resolveRemotePaperclipRunnerBinary({ runner: target, remoteCwd: "/workspace" })).toBe(fresh);
|
||||
expect(select).not.toHaveBeenCalled();
|
||||
} finally { await rm(root, { recursive: true, force: true }); }
|
||||
},
|
||||
);
|
||||
it.each(["missing", "relative", "malformed", "wrong-architecture"])("retains same-host override policy and target validation (%s)", async scenario => {
|
||||
const root = await mkdtemp(join(tmpdir(), "paperclip-same-host-daemon-"));
|
||||
try {
|
||||
vi.stubGlobal("process", { ...process, platform: "linux", arch: "x64" });
|
||||
const override = join(root, "configured-runnerd"), fallback = join(root, "packaged-runnerd");
|
||||
if (scenario !== "missing") await writeFile(override, scenario === "malformed" ? Buffer.alloc(64) : header("linux-arm64"), { mode: 0o700 });
|
||||
await writeFile(fallback, header("linux-x64"), { mode: 0o700 });
|
||||
vi.stubEnv("PAPERCLIP_RUNNER_BINARY", scenario === "relative" ? "relative-runnerd" : override);
|
||||
vi.spyOn(runner, "defaultCapabilityRunnerdBinary").mockReturnValue(fallback);
|
||||
const request = resolveRemotePaperclipRunnerBinary({ runner: remote("Linux\nx86_64\n"), remoteCwd: "/workspace" });
|
||||
if (scenario === "missing") expect(await request).toBe(fallback);
|
||||
else await expect(request).rejects.toThrow(scenario === "relative" ? "PAPERCLIP_RUNNER_BINARY must be an absolute path" : "runner_remote_artifact_platform_mismatch");
|
||||
} finally { await rm(root, { recursive: true, force: true }); }
|
||||
});
|
||||
it("ignores a relative controller override when the actual target differs", async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), "paperclip-cross-host-daemon-"));
|
||||
try {
|
||||
vi.stubGlobal("process", { ...process, platform: "darwin", arch: "arm64" });
|
||||
vi.stubEnv("PAPERCLIP_RUNNER_BINARY", "relative-controller-runnerd");
|
||||
const binary = join(root, "linux-runnerd");
|
||||
await writeFile(binary, header("linux-arm64"));
|
||||
const select = vi.spyOn(runner, "defaultCapabilityRunnerdBinary").mockReturnValue(binary);
|
||||
expect(await resolveRemotePaperclipRunnerBinary({ runner: remote("Linux\naarch64\n"), remoteCwd: "/workspace" })).toBe(binary);
|
||||
expect(select).toHaveBeenCalledExactlyOnceWith("linux", "arm64");
|
||||
} finally { await rm(root, { recursive: true, force: true }); }
|
||||
});
|
||||
it.each(["matching", "wrong-architecture", "malformed", "missing"])("verifies an explicit controller-owned remote artifact (%s)", async scenario => {
|
||||
const root = await mkdtemp(join(tmpdir(), "paperclip-ssh-daemon-"));
|
||||
const binary = join(root, "configured-runnerd");
|
||||
try {
|
||||
vi.stubGlobal("process", { ...process, platform: "linux", arch: "arm64" });
|
||||
vi.stubEnv("PAPERCLIP_RUNNER_BINARY", "invalid-relative-controller-override");
|
||||
if (scenario !== "missing") await writeFile(binary, scenario === "malformed" ? Buffer.alloc(64) : header(scenario === "matching" ? "linux-arm64" : "linux-x64"));
|
||||
const select = vi.spyOn(runner, "defaultCapabilityRunnerdBinary");
|
||||
const request = resolveRemotePaperclipRunnerBinary({ runner: remote("Linux\naarch64\n"), remoteCwd: "/workspace", configuredPath: binary });
|
||||
|
||||
@@ -208,7 +208,7 @@ export function resolvePaperclipRunnerBinary(
|
||||
);
|
||||
}
|
||||
|
||||
/** Select controller-owned bytes for the selected SSH host, never its controller. */
|
||||
/** Select controller-owned bytes for the actual remote host, never its controller. */
|
||||
export async function resolveRemotePaperclipRunnerBinary(input: {
|
||||
runner: CommandManagedRuntimeRunner;
|
||||
remoteCwd: string;
|
||||
@@ -220,11 +220,17 @@ export async function resolveRemotePaperclipRunnerBinary(input: {
|
||||
const platform = os === "Linux" ? "linux" : os === "Darwin" ? "darwin" : null;
|
||||
const architecture = arch === "x86_64" ? "x64" : arch === "aarch64" || arch === "arm64" ? "arm64" : null;
|
||||
if (probe.exitCode !== 0 || probe.timedOut || !platform || !architecture) {
|
||||
throw new Error("runner_remote_artifact_platform_mismatch: could not identify a supported remote OS and architecture; check the SSH host and retry");
|
||||
throw new Error("runner_remote_artifact_platform_mismatch: could not identify a supported remote OS and architecture; check the selected execution environment and retry");
|
||||
}
|
||||
const configured = input.configuredPath?.trim();
|
||||
if (configured && !isAbsolute(configured)) throw new Error("PAPERCLIP_RUNNER_REMOTE_BINARY_PATH must be an absolute path");
|
||||
const binary = configured || defaultCapabilityRunnerdBinary(platform, architecture);
|
||||
// A same-host sandbox can use the supervisor's fresh development artifact.
|
||||
// Cross-platform targets must never inherit the controller's override.
|
||||
const controllerOverride = platform === process.platform && architecture === process.arch
|
||||
? process.env.PAPERCLIP_RUNNER_BINARY : undefined;
|
||||
const binary = configured || (controllerOverride
|
||||
? resolvePaperclipRunnerBinary(controllerOverride)
|
||||
: defaultCapabilityRunnerdBinary(platform, architecture));
|
||||
let bytes: Buffer;
|
||||
try { bytes = readFileSync(binary); }
|
||||
catch (cause) { throw new Error(`runner_remote_artifact_unavailable: install the Paperclip release containing the ${platform}-${architecture} daemon or configure PAPERCLIP_RUNNER_REMOTE_BINARY_PATH`, { cause }); }
|
||||
|
||||
@@ -12256,10 +12256,8 @@ describe("runnerd provider runtime wiring", () => {
|
||||
// The mocked remote executes metadata probes; artifact staging only needs bytes.
|
||||
// Keep this regression independent of a locally compiled Rust runner binary.
|
||||
const controllerArtifact = join(isolatedStateDirectory, "paperclip-runnerd");
|
||||
if (needsReplacement || retained || image === "preinstalled-exact") {
|
||||
await writeFile(controllerArtifact, "fixture runner artifact");
|
||||
state.resolveRunnerBinary.mockReturnValueOnce(controllerArtifact);
|
||||
}
|
||||
await writeFile(controllerArtifact, "fixture runner artifact");
|
||||
state.resolveRemoteRunnerBinary.mockResolvedValue(controllerArtifact);
|
||||
const onLog = vi.fn(async () => undefined);
|
||||
const syncIn = vi.fn(async () => undefined);
|
||||
const remoteExecute = vi.fn(
|
||||
@@ -12468,21 +12466,27 @@ describe("runnerd provider runtime wiring", () => {
|
||||
|
||||
});
|
||||
|
||||
it.each(["explicit", "linux-arm64", "linux-x64"])("binds an SSH launch to the selected controller-owned runner artifact (%s)", async mode => {
|
||||
it.each(["ssh", "sandbox"].flatMap(transport => ["explicit", "linux-arm64", "linux-x64", "missing", "mismatch"].map(mode => ({ transport, mode }))))
|
||||
("binds a $transport launch to the selected controller-owned runner artifact ($mode)", async ({ transport, mode }) => {
|
||||
const remoteCwd = "/home/daytona/paperclip-workspace";
|
||||
const controllerArtifact = `/controller/artifacts/${mode}/paperclip-runnerd`;
|
||||
state.resolveRemoteRunnerBinary.mockResolvedValue(controllerArtifact);
|
||||
const explicit = mode === "explicit" || mode === "missing" || mode === "mismatch";
|
||||
const error = mode === "missing" ? "runner_remote_artifact_unavailable" : mode === "mismatch" ? "runner_remote_artifact_platform_mismatch" : null;
|
||||
if (error) state.resolveRemoteRunnerBinary.mockRejectedValue(new Error(error));
|
||||
else state.resolveRemoteRunnerBinary.mockResolvedValue(controllerArtifact);
|
||||
const remoteExecution = {
|
||||
...execution,
|
||||
binding: { ...execution.binding, runId: "run-remote-runner-artifact" },
|
||||
workspace: { ...execution.workspace, cwd: "/host/paperclip-workspace" },
|
||||
} as NativeExecutionInputV1;
|
||||
|
||||
await createRunnerdBackend({
|
||||
const backend = createRunnerdBackend({
|
||||
db: leaseDb(remoteExecution),
|
||||
execution: remoteExecution,
|
||||
runnerInstanceId: "runner",
|
||||
runnerExecutionTarget: {
|
||||
runnerExecutionTarget: transport === "sandbox" ? {
|
||||
kind: "remote", transport: "sandbox", providerKey: "test", remoteCwd, runner: { execute: vi.fn() },
|
||||
} : {
|
||||
kind: "remote",
|
||||
transport: "ssh",
|
||||
remoteCwd,
|
||||
@@ -12497,15 +12501,25 @@ describe("runnerd provider runtime wiring", () => {
|
||||
strictHostKeyChecking: true,
|
||||
},
|
||||
},
|
||||
...(mode === "explicit" ? { runnerRemoteBinaryPath: controllerArtifact } : {}),
|
||||
...(explicit ? { runnerRemoteBinaryPath: controllerArtifact } : {}),
|
||||
});
|
||||
|
||||
if (error) {
|
||||
await expect(backend).rejects.toThrow(error);
|
||||
expect(state.resolveRemoteRunnerBinary).toHaveBeenCalledWith(expect.objectContaining({ remoteCwd, configuredPath: controllerArtifact }));
|
||||
expect(state.resolveRunnerBinary).not.toHaveBeenCalled();
|
||||
expect(state.createBackend).not.toHaveBeenCalled();
|
||||
expect(state.createTransport).not.toHaveBeenCalled();
|
||||
return;
|
||||
}
|
||||
await backend;
|
||||
|
||||
state.createTransport.mockClear();
|
||||
state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!();
|
||||
expect(state.createTransport).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ runnerBinary: controllerArtifact }),
|
||||
);
|
||||
expect(state.resolveRemoteRunnerBinary).toHaveBeenCalledWith(expect.objectContaining({ remoteCwd, configuredPath: mode === "explicit" ? controllerArtifact : undefined }));
|
||||
expect(state.resolveRemoteRunnerBinary).toHaveBeenCalledWith(expect.objectContaining({ remoteCwd, configuredPath: explicit ? controllerArtifact : undefined }));
|
||||
expect(state.resolveRunnerBinary).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
|
||||
@@ -11575,7 +11575,7 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
// filesystem; the remote launcher separately owns the sandbox command path.
|
||||
// When an explicit remote artifact is configured, prepareRemoteRunner stages
|
||||
// these exact bytes at remoteBinary before launch.
|
||||
const targetMatchedCodexArtifact = remoteTarget?.transport === "ssh" && input.execution.provider.kind === "codex";
|
||||
const targetMatchedCodexArtifact = remoteTarget !== null && input.execution.provider.kind === "codex";
|
||||
const controllerRunnerBinary = targetMatchedCodexArtifact
|
||||
? await resolveRemotePaperclipRunnerBinary({ runner: remoteCommandRunner!, remoteCwd: remoteTarget.remoteCwd, configuredPath: input.runnerRemoteBinaryPath })
|
||||
: remoteTarget
|
||||
|
||||
@@ -35,6 +35,7 @@ describe("Codex selected native account verification", () => {
|
||||
vi.stubEnv("PAPERCLIP_RUNNER_REMOTE_BINARY_PATH", "");
|
||||
});
|
||||
afterEach(async () => {
|
||||
vi.unstubAllGlobals();
|
||||
vi.unstubAllEnvs();
|
||||
for (const [input] of nativeProbe.mock.calls) if (input.runtimeDirectory) await rm(input.runtimeDirectory, { recursive: true, force: true });
|
||||
});
|
||||
@@ -238,7 +239,12 @@ describe("Codex selected native account verification", () => {
|
||||
if (confirmed) await expect(stat(root)).rejects.toThrow();
|
||||
else expect((await stat(root)).isDirectory()).toBe(true);
|
||||
});
|
||||
it("uses the same preparation and controller PRP for a sandbox, without a guest provider pack", async () => {
|
||||
it.each(["arm64", "x64"].flatMap(controllerArch => ["arm64", "x64"].flatMap(targetArch => [false, true].map(explicit => ({ controllerArch, targetArch, explicit })))))
|
||||
("uses target $targetArch artifacts and controller PRP from Mac $controllerArch for sandbox setup (override=$explicit)", async ({ controllerArch, targetArch, explicit }) => {
|
||||
vi.stubGlobal("process", { ...process, platform: "darwin", arch: controllerArch });
|
||||
const selectedBinary = `/controller/artifacts/linux-${targetArch}/paperclip-runnerd`;
|
||||
selectedRemoteRunner.mockResolvedValue(selectedBinary);
|
||||
if (explicit) vi.stubEnv("PAPERCLIP_RUNNER_REMOTE_BINARY_PATH", selectedBinary);
|
||||
nativeProbe.mockImplementation(async input => {
|
||||
const registration = await input.transportOptions.controlPlaneRegistration({});
|
||||
expect(registration.connection.mode).toBe("connect");
|
||||
@@ -248,10 +254,23 @@ describe("Codex selected native account verification", () => {
|
||||
const target = { ...context.executionTarget, runner: { execute: sshExecute } };
|
||||
expect((await testNativeRunnerAuthentication({ ...context, executionTarget: target }, "codex", "gpt-6.1-sol")).status).toBe("pass");
|
||||
expect(sshRunner).not.toHaveBeenCalled();
|
||||
expect(nativeArtifacts).toHaveBeenCalledWith(expect.objectContaining({ target, runner: target.runner }));
|
||||
expect(selectedRemoteRunner).toHaveBeenCalledExactlyOnceWith({ runner: target.runner, remoteCwd: target.remoteCwd, configuredPath: explicit ? selectedBinary : null });
|
||||
expect(selectedRunner).not.toHaveBeenCalled();
|
||||
expect(nativeArtifacts).toHaveBeenCalledWith(expect.objectContaining({ target, runner: target.runner, controllerRunnerBinary: selectedBinary, targetPlatformArtifactVerified: true }));
|
||||
expect(nativeProbe.mock.calls[0][0].transportOptions.runnerBinary).toBe(selectedBinary);
|
||||
expect(nativeArtifacts.mock.results[0].value.prepare).toHaveBeenCalledWith("dial_wss");
|
||||
expect(execute).not.toHaveBeenCalled();
|
||||
});
|
||||
it.each(["runner_remote_artifact_unavailable", "runner_remote_artifact_platform_mismatch"])("fails sandbox %s before setup or a controller fallback", async code => {
|
||||
selectedRemoteRunner.mockRejectedValue(new Error(code));
|
||||
const target = { ...context.executionTarget, runner: { execute: sshExecute } };
|
||||
const result = await testNativeRunnerAuthentication({ ...context, executionTarget: target }, "codex", "gpt-6.1-sol");
|
||||
expect(result.status).toBe("fail");
|
||||
expect(result.checks[0].message).toContain(code);
|
||||
expect(selectedRunner).not.toHaveBeenCalled(); expect(nativeArtifacts).not.toHaveBeenCalled();
|
||||
expect(nativeProbe).not.toHaveBeenCalled(); expect(sshExecute).not.toHaveBeenCalled();
|
||||
expect(execute).not.toHaveBeenCalled();
|
||||
});
|
||||
it("prepares listener artifacts before acquiring sandbox ingress and owns its release", async () => {
|
||||
const order: string[] = [];
|
||||
const close = vi.fn();
|
||||
|
||||
@@ -63,11 +63,9 @@ async function probeRemoteNativeCodex(options: {
|
||||
const remoteCodexNpmSpec = process.env.PAPERCLIP_RUNNER_REMOTE_CODEX_NPM_SPEC?.trim() || null;
|
||||
const remoteCodexBinary = remoteCodexNpmSpec ? posix.join(remoteRoot, "harnesses", "codex", "node_modules", ".bin", "codex") : posix.join(remoteRoot, "bin", "codex");
|
||||
const explicitRunner = process.env.PAPERCLIP_RUNNER_REMOTE_BINARY_PATH?.trim() || null;
|
||||
const controllerRunnerBinary = target.transport === "ssh"
|
||||
? await resolveRemotePaperclipRunnerBinary({ runner, remoteCwd: target.remoteCwd, configuredPath: explicitRunner })
|
||||
: explicitRunner || resolvePaperclipRunnerBinary();
|
||||
const controllerRunnerBinary = await resolveRemotePaperclipRunnerBinary({ runner, remoteCwd: target.remoteCwd, configuredPath: explicitRunner });
|
||||
const artifacts = createRemoteNativeArtifactPreparation({ target, runner, remoteBinary, controllerRunnerBinary, remoteRuntimeRoot: remoteRoot, remoteCodexBinary,
|
||||
targetPlatformArtifactVerified: target.transport === "ssh",
|
||||
targetPlatformArtifactVerified: true,
|
||||
runnerRemoteBinaryPath: explicitRunner, runnerRemoteCodexPath: process.env.PAPERCLIP_RUNNER_REMOTE_CODEX_PATH?.trim(), runnerRemoteCodexNpmSpec: remoteCodexNpmSpec, model: options.model });
|
||||
const requireSuccess = (result: { timedOut: boolean; exitCode: number | null }, message: string) => {
|
||||
if (result.timedOut || result.exitCode !== 0) throw new Error(message);
|
||||
|
||||
@@ -6,6 +6,7 @@ import { readProcessLossDiagnostic } from "./process-loss-diagnostics.js";
|
||||
import { WORKSPACE_RESTORE_FAILURE_CODES } from "@paperclipai/shared";
|
||||
import { redactDiagnosticText } from "@paperclipai/adapter-utils/command-redaction";
|
||||
import { sanitizeWorkspaceRestoreDiagnostic } from "@paperclipai/adapter-utils/workspace-restore-diagnostics";
|
||||
import { ACP_TOOL_KINDS } from "@paperclipai/adapter-utils/acpx-engine/constants";
|
||||
import { redactCurrentUserText } from "../log-redaction.js";
|
||||
import { redactSensitiveText, REDACTED_EVENT_VALUE } from "../redaction.js";
|
||||
import { readNativeModelRejectionDiagnostic } from "./native-runtime/native-provider-failure.js";
|
||||
@@ -41,6 +42,15 @@ export interface RunFailureDiagnostics {
|
||||
truncatedFields: string[];
|
||||
}
|
||||
|
||||
const ACP_PENDING_TOOL_KIND_LABELS: ReadonlySet<string> = new Set([...ACP_TOOL_KINDS, "unknown"]);
|
||||
|
||||
/** Closed ACP tool-kind labels only; any other content drops the whole field. */
|
||||
function readAcpPendingToolKinds(value: unknown): string | null {
|
||||
if (typeof value !== "string" || value.length === 0 || value.length > 200) return null;
|
||||
const kinds = value.split(",");
|
||||
return kinds.every((kind) => ACP_PENDING_TOOL_KIND_LABELS.has(kind)) ? value : null;
|
||||
}
|
||||
|
||||
function read(value: unknown, key: string): unknown {
|
||||
if (!value || typeof value !== "object") return undefined;
|
||||
try {
|
||||
@@ -197,10 +207,15 @@ export function collectRunFailureDiagnostics(run: Run, options: RunFailureReport
|
||||
"mode", "stopReason", "timeoutFired", "timeoutSource", "timeoutConfigured",
|
||||
"effectiveTimeoutSec", "errorFamily",
|
||||
]));
|
||||
for (const field of ["acpLastEventAgeMs", "acpObservedEventCount", "acpPendingToolCount"]) {
|
||||
for (const field of [
|
||||
"acpLastEventAgeMs", "acpObservedEventCount", "acpPendingToolCount",
|
||||
"acpLastEventAgeAtStopMs", "acpOldestPendingToolAgeAtStopMs",
|
||||
]) {
|
||||
const value = read(result, field);
|
||||
if (typeof value === "number" && Number.isSafeInteger(value) && value >= 0) execution[field] = value;
|
||||
}
|
||||
const pendingToolKinds = readAcpPendingToolKinds(read(result, "acpPendingToolKinds"));
|
||||
if (pendingToolKinds) execution.acpPendingToolKinds = pendingToolKinds;
|
||||
const toolInventoryComplete = read(result, "acpToolInventoryComplete");
|
||||
if (typeof toolInventoryComplete === "boolean") execution.acpToolInventoryComplete = toolInventoryComplete;
|
||||
const restoreFailure = read(result, "workspaceRestoreFailure");
|
||||
|
||||
Reference in new issue
Block a user