mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 02:32:07 +01:00
feat(slack): add Require at-mention setting (#15612)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Slack messages can create tasks and continue existing tasks. > - A reply in a subscribed thread can start agent work without a new mention. > - Some users want to control each reply with an explicit bot mention. > - This pull request adds a Require at-mention setting to Slack app settings. > - The setting checks new messages and queued messages before they start work. ## Linked Issues or Issue Description **What existing behavior does this improve?** Slack app settings and the rules that admit inbound messages. **Subsystem affected** Server, UI, database, and shared API contracts. **Current behavior** After the bot joins a thread, an ordinary reply can wake the agent. Direct messages do not need a bot mention. **Proposed behavior** Add Require at-mention below Allow direct messages. When enabled, each channel message, thread reply, and direct message must mention the bot. Direct messages must also be allowed. **Reason and benefit** Users can prevent ordinary conversation from starting agent work. **Breaking changes** New Slack connections require a bot mention by default. Existing connections keep their saved setting. Work that has already started is not cancelled. **Additional context** A search for existing at-mention PR titles found no duplicate. ## What Changed - Add the endpoint field, schema validation, API response, and UI toggle. New Slack connections enable the setting. - Add an idempotent database migration and generated metadata. Renumber it to 0321 after current master. - Check the current setting at inbound admission, task mutation, and the first heartbeat queue claim. - Keep Slack tool access and recovery for already-running work when the setting changes. - Cancel queued batches that contain an unmentioned input, including coalesced messages. - Dispatch promoted follow-ups after the scheduler releases its agent start lock. Preserve these checks in the new queue and run-control modules. - Filter unmentioned messages without retaining their content. - Keep provider-message deduplication and existing access checks. - Add thread, DM, duplicate-delivery, queued-message, and new-connection default regression tests. - Document the setting and its scope. ## Verification - Previous head `16791bb7`: typecheck, build, all server shards, Runner checks, and E2E passed. Two chat shards failed in three fixtures that used unmentioned Slack messages. - Update setup follow-ups to mention the Slack bot. Disable the mention rule explicitly in the callback ordering test, which tests ordinary subscribed replies. - Previous local checks passed: six Slack integration cases, all 46 durable chat scheduler tests, all 24 queue/run-control tests, migration safety, and UI token gates. - Greptile reviewed `16791bb7` with 5/5. All review threads are resolved. - Head `53163c20f`: all five focused integration cases passed. These cover all three prior CI failures and both default-on channel/DM cases. Fresh CI and Greptile are running. - Full local recursive typecheck, build, and repository tests were not repeated for this fixture-only change. CI supplies these checks. - Manual check: create a Slack connection. Require at-mention must be on. An ordinary thread reply or DM must not wake the agent. A bot mention must wake it once. Disable the setting and confirm ordinary replies work again. ## Risks - The migration adds a non-null boolean with a false default. Apply it before the new server runs. - The setting also applies to DMs. The UI help text states this. - Enabling it blocks queued unmentioned work but does not cancel a run that has already started. - Full local verification is limited by the environment constraints above; latest-head CI must pass before merge. ## Model Used OpenAI GPT-6 (Codex agent), with reasoning, tool use, and code execution. The runtime does not expose a more specific model version or context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
1602512820
commit
3fc64c5156
17 files changed
+56798
-26
No files matched your search
@@ -197,3 +197,17 @@ apps without that scope require reinstalling with the updated manifest. Other
|
||||
people's bot DMs remain inaccessible. Scheduling uses ordinary Paperclip routines,
|
||||
not a Slack-specific timer. Routine results are sent explicitly through the tool;
|
||||
ordinary task finals are not automatically broadcast to Slack.
|
||||
|
||||
## Require an at-mention for inbound messages
|
||||
|
||||
Slack app settings include **Require at-mention** directly below **Allow direct
|
||||
messages**. It defaults to on for new Slack connections. Existing connections
|
||||
keep their saved setting and thread and DM behavior.
|
||||
When enabled, each inbound message must explicitly @mention the connected bot,
|
||||
including replies in existing channel threads and direct messages. DMs must also
|
||||
be enabled with **Allow direct messages**. Unmentioned messages do not create
|
||||
or continue a task. Queued messages are checked again before starting agent work.
|
||||
If a queued run combines several messages, an unmentioned input cancels that
|
||||
queued run when this setting is enabled. Send a new mention to start it again.
|
||||
This setting does not cancel work that has already started or revoke its Slack
|
||||
tool access.
|
||||
@@ -0,0 +1 @@
|
||||
ALTER TABLE "chat_endpoints" ADD COLUMN IF NOT EXISTS "require_at_mention" boolean DEFAULT false NOT NULL;
|
||||
+56431
File diff suppressed because it is too large.
Load diff
+7
@@ -2234,6 +2234,13 @@
|
||||
"when": 1791497222457,
|
||||
"tag": "0320_daily_onslaught",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 321,
|
||||
"version": "7",
|
||||
"when": 1791568503477,
|
||||
"tag": "0321_acoustic_human_fly",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -68,6 +68,7 @@ export const chatEndpoints = pgTable(
|
||||
allowDirectMessages: boolean("allow_direct_messages")
|
||||
.notNull()
|
||||
.default(true),
|
||||
requireAtMention: boolean("require_at_mention").notNull().default(false),
|
||||
allowGroupChats: boolean("allow_group_chats").notNull().default(false),
|
||||
allowUnlinkedPeople: boolean("allow_unlinked_people")
|
||||
.notNull()
|
||||
|
||||
@@ -158,6 +158,7 @@ export interface ChatEndpointBehaviorPolicy {
|
||||
/** Defaults to queue and is not exposed in the initial settings UI. */
|
||||
concurrency: ChatConcurrencyPolicy;
|
||||
allowDirectMessages: boolean;
|
||||
requireAtMention: boolean;
|
||||
allowGroupChats: boolean;
|
||||
allowUnlinkedPeople: boolean;
|
||||
}
|
||||
@@ -253,6 +254,7 @@ export interface ChatEndpoint {
|
||||
botAvatarUrl?: string | null;
|
||||
photonAllocation?: "dedicated" | "shared";
|
||||
allowDirectMessages: boolean;
|
||||
requireAtMention: boolean;
|
||||
allowGroupChats: boolean;
|
||||
allowUnlinkedPeople: boolean;
|
||||
replyMode: "subscribed";
|
||||
@@ -493,6 +495,7 @@ export interface UpdateChatEndpointInput {
|
||||
slackApp?: SlackAppConfiguration;
|
||||
communicationInstructions?: string;
|
||||
allowDirectMessages?: boolean;
|
||||
requireAtMention?: boolean;
|
||||
allowGroupChats?: boolean;
|
||||
allowUnlinkedPeople?: boolean;
|
||||
}
|
||||
|
||||
@@ -92,6 +92,7 @@ export const updateChatEndpointSchema = z
|
||||
slackApp: slackAppConfigurationSchema.optional(),
|
||||
slackSetupMethod: z.enum(["automatic", "manual", "existing"]).optional(),
|
||||
allowDirectMessages: z.boolean().optional(),
|
||||
requireAtMention: z.boolean().optional(),
|
||||
allowGroupChats: z.boolean().optional(),
|
||||
allowUnlinkedPeople: z.boolean().optional(),
|
||||
})
|
||||
|
||||
@@ -1594,6 +1594,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
message: makeMessage({
|
||||
id: setupFollowUpMessageId,
|
||||
text: "Setup follow-up",
|
||||
mentioned: endpoint.provider === "slack",
|
||||
userId,
|
||||
}),
|
||||
trigger:
|
||||
@@ -1630,7 +1631,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
|
||||
async function configuredSlackEndpoint(
|
||||
fixture: Awaited<ReturnType<typeof seedCompany>>,
|
||||
overrides: { allowUnlinkedPeople?: boolean; linkedBoardUser?: boolean } & Partial<
|
||||
overrides: { allowUnlinkedPeople?: boolean; linkedBoardUser?: boolean; requireAtMention?: boolean } & Partial<
|
||||
Pick<
|
||||
ChatChannelServiceOptions,
|
||||
| "credentialMutationLeaseRenewalIntervalMs"
|
||||
@@ -1673,6 +1674,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
// Slack creation defaults to linked accounts only (covered separately).
|
||||
await context.service.update(endpoint.id, {
|
||||
allowUnlinkedPeople: overrides.allowUnlinkedPeople ?? true,
|
||||
// Existing transport tests explicitly exercise the optional unmentioned mode.
|
||||
requireAtMention: overrides.requireAtMention ?? false,
|
||||
}, "owner-user");
|
||||
await context.service.configure(
|
||||
endpoint.id,
|
||||
@@ -5331,6 +5334,105 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
await service.shutdown();
|
||||
});
|
||||
|
||||
it.each([false, true])("requires at-mentions by default for new Slack connections (DM=%s)", async (isDM) => {
|
||||
const fixture = await seedCompany();
|
||||
const { callbacks, endpoint, service, wakeup } = await configuredSlackEndpoint(fixture, { requireAtMention: true });
|
||||
// The returned endpoint is the original creation result, before fixture updates.
|
||||
expect(endpoint.requireAtMention).toBe(true);
|
||||
expect((await service.get(endpoint.id)).requireAtMention).toBe(true);
|
||||
const thread = makeThread({ channelId: isDM ? "D-DEFAULT" : "C-DEFAULT", id: `slack:${isDM ? "D-DEFAULT" : "C-DEFAULT"}:7500.1`, isDM });
|
||||
const send = (id: string, mentioned: boolean) => deliverMessage({ callbacks, endpointId: endpoint.id, thread: thread.thread,
|
||||
message: makeMessage({ id, text: mentioned ? "@maya start" : "Start", mentioned }), trigger: isDM ? "direct_message" : mentioned ? "mention" : "subscribed_message" });
|
||||
await send("7500.1", false);
|
||||
expect(wakeup).not.toHaveBeenCalled();
|
||||
await send("7500.2", true);
|
||||
expect(wakeup).toHaveBeenCalledTimes(1);
|
||||
await service.update(endpoint.id, { requireAtMention: false });
|
||||
await send("7500.3", false);
|
||||
expect(wakeup).toHaveBeenCalledTimes(2);
|
||||
const other = await service.create(fixture.companyId, { provider: "github", assignedAgentId: fixture.assignedAgentId });
|
||||
expect(other.requireAtMention).toBe(false);
|
||||
expect((await service.get(endpoint.id)).requireAtMention).toBe(false);
|
||||
await service.shutdown();
|
||||
});
|
||||
|
||||
it("preserves running Slack tool authority after Require at-mention is enabled", async () => {
|
||||
await instanceSettingsService(db).updateExperimental({ enableChatConnectors: true });
|
||||
const { resolveSlackTaskAuthority } = await import("../services/connectors/slack-authority.js");
|
||||
const fixture = await seedCompany();
|
||||
const { callbacks, endpoint, service } = await configuredSlackEndpoint(fixture, { linkedBoardUser: true, allowUnlinkedPeople: false });
|
||||
const thread = makeThread({ channelId: "D-RUNNING", id: "slack:D-RUNNING:7400.1", isDM: true });
|
||||
await deliverMessage({ callbacks, endpointId: endpoint.id, thread: thread.thread,
|
||||
message: makeMessage({ id: "7400.1", text: "Read this conversation", userId: "UBOARD" }), trigger: "direct_message" });
|
||||
const [conversation] = await db.select().from(chatConversations).where(eq(chatConversations.endpointId, endpoint.id));
|
||||
const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, endpoint.id), eq(chatActions.kind, "inbound_wakeup")));
|
||||
const runId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId: fixture.companyId, agentId: fixture.assignedAgentId,
|
||||
status: "running", invocationSource: "assignment", responsibleUserId: "owner-user", contextSnapshot: { issueId: conversation.issueId } });
|
||||
await initializeRunIdentity(db, { companyId: fixture.companyId, runId, issueId: conversation.issueId,
|
||||
responsibleUserId: "owner-user", messageIds: [String(action.payload.commentId)], cause: "instruction" });
|
||||
const binding = { companyId: fixture.companyId, agentId: fixture.assignedAgentId, runId, issueId: conversation.issueId };
|
||||
await expect(resolveSlackTaskAuthority(db, binding)).resolves.toMatchObject({ slackUserId: "UBOARD" });
|
||||
await service.update(endpoint.id, { requireAtMention: true });
|
||||
await expect(resolveSlackTaskAuthority(db, binding)).resolves.toMatchObject({ slackUserId: "UBOARD" });
|
||||
});
|
||||
|
||||
it.each([false, true])("requires Slack at-mentions for every message when enabled (DM=%s)", async (isDM) => {
|
||||
const fixture = await seedCompany();
|
||||
const { callbacks, endpoint, service, wakeup } = await configuredSlackEndpoint(fixture);
|
||||
expect((await service.get(endpoint.id)).requireAtMention).toBe(false);
|
||||
const thread = makeThread({ channelId: isDM ? "D-MENTION" : "C-MENTION", id: `slack:${isDM ? "D-MENTION" : "C-MENTION"}:7100.1`, isDM });
|
||||
const send = (id: string, mentioned: boolean, trigger: ChatSdkMessageTrigger = isDM ? "direct_message" : "subscribed_message") =>
|
||||
deliverMessage({ callbacks, endpointId: endpoint.id, thread: thread.thread,
|
||||
message: makeMessage({ id, text: mentioned ? "@maya continue" : "Continue", mentioned }), trigger });
|
||||
// Disabled mode preserves unmentioned DMs and subscribed thread replies.
|
||||
await send("7100.1", true, isDM ? "direct_message" : "mention");
|
||||
await send("7100.2", false);
|
||||
expect(wakeup).toHaveBeenCalledTimes(2);
|
||||
expect((await service.update(endpoint.id, { requireAtMention: true })).requireAtMention).toBe(true);
|
||||
await send("7100.3", false);
|
||||
expect(wakeup).toHaveBeenCalledTimes(2);
|
||||
const [filtered] = await db.select().from(chatDeliveries).where(and(
|
||||
eq(chatDeliveries.endpointId, endpoint.id), eq(chatDeliveries.state, "filtered")));
|
||||
expect(filtered.redactedError).toBe("Message did not @mention the bot");
|
||||
expect(filtered.normalizedEvent).toMatchObject({ filtering: { contentRetained: false } });
|
||||
expect(filtered.normalizedEvent).not.toHaveProperty("message.text");
|
||||
// Both Slack callback kinds can carry an explicit mention. Retries cannot
|
||||
// produce a second wakeup for the same provider message.
|
||||
await send("7100.4", true);
|
||||
await send("7100.4", true, "mention");
|
||||
expect(wakeup).toHaveBeenCalledTimes(3);
|
||||
const conversations = await db.select().from(chatConversations).where(eq(chatConversations.endpointId, endpoint.id));
|
||||
expect(conversations).toHaveLength(1);
|
||||
const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, conversations[0].issueId));
|
||||
expect(comments).toHaveLength(3);
|
||||
await service.update(endpoint.id, { requireAtMention: false });
|
||||
await send("7100.5", false);
|
||||
expect(wakeup).toHaveBeenCalledTimes(4);
|
||||
if (isDM) {
|
||||
await service.update(endpoint.id, { requireAtMention: true, allowDirectMessages: false });
|
||||
await send("7100.6", true);
|
||||
expect(wakeup).toHaveBeenCalledTimes(4);
|
||||
}
|
||||
});
|
||||
|
||||
it("filters a queued Slack message when Require at-mention is enabled before drain", async () => {
|
||||
const fixture = await seedCompany();
|
||||
const { callbacks, endpoint, service, wakeup } = await configuredSlackEndpoint(fixture, {
|
||||
deferWebhookProcessing: true, scheduleDeferredWork: () => undefined,
|
||||
});
|
||||
const thread = makeThread({ channelId: "D-QUEUED-MENTION", id: "slack:D-QUEUED-MENTION:7300.1", isDM: true });
|
||||
await deliverMessage({ callbacks, endpointId: endpoint.id, thread: thread.thread,
|
||||
message: makeMessage({ id: "7300.1", text: "Start work" }), trigger: "direct_message" });
|
||||
await service.update(endpoint.id, { requireAtMention: true });
|
||||
const [delivery] = await db.select().from(chatDeliveries).where(eq(chatDeliveries.endpointId, endpoint.id));
|
||||
await db.update(chatDeliveries).set({ nextAttemptAt: new Date(0) }).where(eq(chatDeliveries.id, delivery.id));
|
||||
await service.processPendingDeliveries(25, delivery.id);
|
||||
expect(wakeup).not.toHaveBeenCalled();
|
||||
expect(await db.select().from(chatConversations).where(eq(chatConversations.endpointId, endpoint.id))).toHaveLength(0);
|
||||
expect((await db.select().from(chatDeliveries).where(eq(chatDeliveries.id, delivery.id)))[0].state).toBe("filtered");
|
||||
});
|
||||
|
||||
it("captures initial Slack communication guidance once per task, ignoring forged message configuration", async () => {
|
||||
const fixture = await seedCompany();
|
||||
const { callbacks, endpoint, service } = await configuredSlackEndpoint(fixture);
|
||||
@@ -16764,7 +16866,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
{ provider: "slack", assignedAgentId: fixture.assignedAgentId },
|
||||
"owner-user",
|
||||
);
|
||||
await service.update(endpoint.id, { allowUnlinkedPeople: true }, "owner-user");
|
||||
await service.update(endpoint.id, { allowUnlinkedPeople: true, requireAtMention: false }, "owner-user");
|
||||
await service.configure(
|
||||
endpoint.id,
|
||||
{
|
||||
|
||||
@@ -17,6 +17,7 @@ import {
|
||||
agentWakeupRequests,
|
||||
chatConversations,
|
||||
chatActions,
|
||||
chatDeliveries,
|
||||
chatEndpoints,
|
||||
chatMessageLinks,
|
||||
chatPublications,
|
||||
@@ -190,6 +191,84 @@ describe("durable inbound chat scheduler receipts", () => {
|
||||
};
|
||||
}
|
||||
|
||||
it.each(["unmentioned", "mentioned", "coalesced", "recovery", "deferred"])("checks Slack mentions after a real heartbeat queue wait (%s)", async (scenario) => {
|
||||
const mentioned = scenario === "mentioned" || scenario === "coalesced";
|
||||
const allowed = scenario === "mentioned" || scenario === "recovery";
|
||||
const f = await fixture();
|
||||
await db.update(agents).set({ adapterType: "durable_chat_retry_test" }).where(eq(agents.id, f.agentId));
|
||||
const applicationId = randomUUID(), connectionId = randomUUID(), endpointId = randomUUID();
|
||||
await db.insert(toolApplications).values({ id: applicationId, companyId: f.companyId, name: "Mention gate", type: "chat" });
|
||||
await db.insert(toolConnections).values({ id: connectionId, companyId: f.companyId, applicationId,
|
||||
name: "Mention gate", uid: randomUUID(), connectionPurpose: "channel", transport: "chat_sdk", status: "active" });
|
||||
await db.insert(chatEndpoints).values({ id: endpointId, companyId: f.companyId, connectionId,
|
||||
provider: "slack", publicId: randomUUID(), assignedAgentId: f.agentId, status: "active" });
|
||||
const request = f.request();
|
||||
await db.insert(issueComments).values({ id: request.commentId, companyId: f.companyId, issueId: f.issueId,
|
||||
authorUserId: "board-user", body: mentioned ? "@bot work" : "Work" });
|
||||
const [delivery] = await db.insert(chatDeliveries).values({ companyId: f.companyId, endpointId,
|
||||
providerEventId: randomUUID(), deduplicationKey: randomUUID(), eventKind: "direct_message", state: "processed",
|
||||
normalizedEvent: { trigger: "direct_message", message: { mentionedBot: mentioned } } }).returning();
|
||||
await db.insert(chatActions).values({ id: request.id, companyId: f.companyId, endpointId,
|
||||
deliveryId: delivery.id, kind: "inbound_wakeup", providerActionId: randomUUID(), status: "processed",
|
||||
payload: { issueId: f.issueId, commentId: request.commentId } });
|
||||
await f.wake(request);
|
||||
const [receipt] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.id));
|
||||
expect(receipt.status).toBe("queued");
|
||||
const [queuedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, receipt.runId!));
|
||||
expect(queuedRun.startedAt).toBeNull();
|
||||
const originalStartedAt = new Date(Date.now() - 60_000);
|
||||
if (scenario === "recovery") {
|
||||
// Same-run retries retain their first claim time, even when queued again.
|
||||
await db.update(heartbeatRuns).set({ startedAt: originalStartedAt })
|
||||
.where(eq(heartbeatRuns.id, receipt.runId!));
|
||||
}
|
||||
let deferredRequestId: string | undefined;
|
||||
if (scenario === "coalesced" || scenario === "deferred") {
|
||||
const second = f.request();
|
||||
await db.insert(issueComments).values({ id: second.commentId, companyId: f.companyId, issueId: f.issueId,
|
||||
authorUserId: "board-user", body: scenario === "deferred" ? "@bot follow-up" : "Unmentioned follow-up" });
|
||||
const [secondDelivery] = await db.insert(chatDeliveries).values({ companyId: f.companyId, endpointId,
|
||||
providerEventId: randomUUID(), deduplicationKey: randomUUID(), eventKind: "direct_message", state: "processed",
|
||||
normalizedEvent: { trigger: "direct_message", message: { mentionedBot: scenario === "deferred" } } }).returning();
|
||||
await db.insert(chatActions).values({ id: second.id, companyId: f.companyId, endpointId,
|
||||
deliveryId: secondDelivery.id, kind: "inbound_wakeup", providerActionId: randomUUID(), status: "processed",
|
||||
payload: { issueId: f.issueId, commentId: second.commentId } });
|
||||
await db.update(issues).set({ executionRunId: receipt.runId, executionAgentNameKey: "maya", executionLockedAt: new Date() })
|
||||
.where(eq(issues.id, f.issueId));
|
||||
if (scenario === "deferred") {
|
||||
// Seed the persisted queue state left by a deferred follow-up before
|
||||
// its predecessor is returned to the unstarted queue.
|
||||
await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, receipt.runId!));
|
||||
runningProcesses.set(receipt.runId!, { child: {} as never, graceSec: 0, processGroupId: null });
|
||||
liveRunIds.add(receipt.runId!);
|
||||
}
|
||||
await f.wake(second);
|
||||
const [coalesced] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, second.id));
|
||||
if (scenario === "deferred") {
|
||||
expect(coalesced).toMatchObject({ status: "deferred_issue_execution", runId: null });
|
||||
deferredRequestId = second.id;
|
||||
runningProcesses.delete(receipt.runId!);
|
||||
liveRunIds.delete(receipt.runId!);
|
||||
await db.update(heartbeatRuns).set({ status: "queued" }).where(eq(heartbeatRuns.id, receipt.runId!));
|
||||
} else {
|
||||
expect(coalesced).toMatchObject({ status: "coalesced", runId: receipt.runId });
|
||||
}
|
||||
}
|
||||
await db.update(chatEndpoints).set({ requireAtMention: true }).where(eq(chatEndpoints.id, endpointId));
|
||||
await f.heartbeat.cancelRun(f.activeRunId, "Fixture slot released", { suppressImmediateRecovery: true });
|
||||
await f.heartbeat.drainActiveRunExecutions();
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, receipt.runId!));
|
||||
expect(run.status).toBe(allowed ? "succeeded" : "cancelled");
|
||||
expect(execute).toHaveBeenCalledTimes(allowed || scenario === "deferred" ? 1 : 0);
|
||||
if (deferredRequestId) {
|
||||
const [promoted] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, deferredRequestId));
|
||||
const [followup] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, promoted.runId!));
|
||||
expect(followup.status).toBe("succeeded");
|
||||
}
|
||||
if (!allowed) expect(run.errorCode).toBe("chat_mention_required");
|
||||
if (scenario === "recovery") expect(run.startedAt).toEqual(originalStartedAt);
|
||||
});
|
||||
|
||||
async function executionHold(f: { companyId: string; issueId: string; agentId: string }) {
|
||||
const [action] = await db.insert(issueRecoveryActions).values({
|
||||
companyId: f.companyId, sourceIssueId: f.issueId, kind: "active_run_watchdog",
|
||||
|
||||
@@ -5844,6 +5844,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
communicationInstructions: endpoint.communicationInstructions,
|
||||
...(endpoint.provider === "imessage-photon" && endpoint.botExternalId ? { photonAllocation: endpoint.botExternalId.startsWith("photon-project:") ? "shared" as const : "dedicated" as const } : {}),
|
||||
allowDirectMessages: endpoint.allowDirectMessages,
|
||||
requireAtMention: endpoint.requireAtMention,
|
||||
allowGroupChats: endpoint.allowGroupChats,
|
||||
allowUnlinkedPeople: endpoint.allowUnlinkedPeople,
|
||||
replyMode: "subscribed",
|
||||
@@ -6027,6 +6028,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
// destination toggle. Keep it closed until the operator explicitly
|
||||
// enables that surface, even when this process is running against a
|
||||
// database created before the column default was hardened.
|
||||
requireAtMention: input.provider === "slack",
|
||||
allowGroupChats: input.provider !== "microsoft-teams",
|
||||
allowUnlinkedPeople: !["slack", "imessage-photon"].includes(input.provider),
|
||||
capabilities: CAPABILITIES[input.provider],
|
||||
@@ -6109,6 +6111,11 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
}
|
||||
if (input.allowDirectMessages !== undefined)
|
||||
values.allowDirectMessages = input.allowDirectMessages;
|
||||
if (input.requireAtMention !== undefined) {
|
||||
if (existing.endpoint.provider !== "slack")
|
||||
throw unprocessable("Require at-mention only applies to Slack connections");
|
||||
values.requireAtMention = input.requireAtMention;
|
||||
}
|
||||
if (input.allowGroupChats && existing.endpoint.provider === "imessage-photon" && (!existing.endpoint.botExternalId || existing.endpoint.botExternalId.startsWith("photon-project:")))
|
||||
throw unprocessable("Photon shared channels support direct messages only; groups require a dedicated channel");
|
||||
if (input.allowGroupChats !== undefined)
|
||||
@@ -14088,6 +14095,24 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
);
|
||||
}
|
||||
|
||||
async function authorizeInboundWorkStart(
|
||||
tx: DbOrTransaction,
|
||||
action: typeof chatActions.$inferSelect,
|
||||
) {
|
||||
const current = await authorizeInboundWakeup(tx, action);
|
||||
if (
|
||||
current.endpoint.provider === "slack" &&
|
||||
current.endpoint.requireAtMention &&
|
||||
current.delivery.normalizedEvent.trigger !== "mention" &&
|
||||
(current.delivery.normalizedEvent.message as { mentionedBot?: boolean } | undefined)?.mentionedBot !== true
|
||||
) {
|
||||
throw forbidden("Message did not @mention the bot", {
|
||||
code: "chat_action_authorization_changed",
|
||||
});
|
||||
}
|
||||
return current;
|
||||
}
|
||||
|
||||
async function processInboundWakeup(deliveryId: string): Promise<boolean> {
|
||||
const now = new Date();
|
||||
const candidate = await db
|
||||
@@ -14202,7 +14227,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
requestedByActorId: String(claimed.payload.requestedByActorId),
|
||||
requestedAt: claimed.createdAt,
|
||||
authorize: async (tx) => {
|
||||
const current = await authorizeInboundWakeup(tx, claimed);
|
||||
const current = await authorizeInboundWorkStart(tx, claimed);
|
||||
if (current.delivery.state !== "processed")
|
||||
throw new Error("chat_inbound_wakeup_acceptance_not_committed");
|
||||
},
|
||||
@@ -14221,7 +14246,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
return true;
|
||||
}
|
||||
const context = await db.transaction((tx) =>
|
||||
authorizeInboundWakeup(tx, claimed),
|
||||
authorizeInboundWorkStart(tx, claimed),
|
||||
);
|
||||
if (context.delivery.state !== "processed")
|
||||
throw new Error("chat_inbound_wakeup_acceptance_not_committed");
|
||||
@@ -14623,6 +14648,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
status: chatEndpoints.status,
|
||||
setup: chatEndpoints.setup,
|
||||
allowDirectMessages: chatEndpoints.allowDirectMessages,
|
||||
requireAtMention: chatEndpoints.requireAtMention,
|
||||
allowGroupChats: chatEndpoints.allowGroupChats,
|
||||
})
|
||||
.from(chatEndpoints)
|
||||
@@ -14786,8 +14812,14 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
resource.availability === "available" &&
|
||||
currentEndpoint.allowGroupChats;
|
||||
}
|
||||
const accepting = endpointAccepting && destinationAccepting;
|
||||
const mentionRequired =
|
||||
endpoint.provider === "slack" &&
|
||||
currentEndpoint.requireAtMention &&
|
||||
trigger !== "mention" &&
|
||||
message.isMention !== true;
|
||||
const accepting = endpointAccepting && destinationAccepting && !mentionRequired;
|
||||
const redactDestinationDelivery =
|
||||
mentionRequired ||
|
||||
(!accepting && thread.isDM) ||
|
||||
(!thread.isDM &&
|
||||
(endpoint.provider === "microsoft-teams" ||
|
||||
@@ -14799,9 +14831,11 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
? staleActivation
|
||||
? "Connection activation changed before admission"
|
||||
: "Connection is not active"
|
||||
: endpoint.provider === "telegram" && !thread.isDM && !addressed
|
||||
? "Message did not address the agent"
|
||||
: "Destination is not enabled in Paperclip";
|
||||
: mentionRequired
|
||||
? "Message did not @mention the bot"
|
||||
: endpoint.provider === "telegram" && !thread.isDM && !addressed
|
||||
? "Message did not address the agent"
|
||||
: "Destination is not enabled in Paperclip";
|
||||
let candidate = admittedDeliveryId
|
||||
? await tx
|
||||
.select()
|
||||
@@ -16185,8 +16219,14 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
const destinationStillAllowed = thread.isDM
|
||||
? currentEndpoint.allowDirectMessages
|
||||
: nonDirectDestinationAllowed(currentEndpoint, currentResource);
|
||||
const mentionStillAllowed =
|
||||
currentEndpoint.provider !== "slack" ||
|
||||
!currentEndpoint.requireAtMention ||
|
||||
trigger === "mention" ||
|
||||
message.isMention === true;
|
||||
if (
|
||||
!endpointStillAllowed ||
|
||||
!mentionStillAllowed ||
|
||||
!destinationStillAllowed ||
|
||||
!currentPrincipalAuthorization.allowed
|
||||
) {
|
||||
@@ -16199,15 +16239,17 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
principalId: null,
|
||||
nextAttemptAt: null,
|
||||
processedAt: filteredAt,
|
||||
redactedError: endpointStillAllowed
|
||||
? destinationStillAllowed
|
||||
? currentPrincipalAuthorization.linkedDenied
|
||||
? "Linked Paperclip account is not currently permitted"
|
||||
: currentEndpoint.allowUnlinkedPeople
|
||||
? "Endpoint sponsor can no longer authorize external guests"
|
||||
: "External identity must be linked to a Paperclip account"
|
||||
: "Destination is not enabled in Paperclip"
|
||||
: "Connection is not active",
|
||||
redactedError: !mentionStillAllowed
|
||||
? "Message did not @mention the bot"
|
||||
: endpointStillAllowed
|
||||
? destinationStillAllowed
|
||||
? currentPrincipalAuthorization.linkedDenied
|
||||
? "Linked Paperclip account is not currently permitted"
|
||||
: currentEndpoint.allowUnlinkedPeople
|
||||
? "Endpoint sponsor can no longer authorize external guests"
|
||||
: "External identity must be linked to a Paperclip account"
|
||||
: "Destination is not enabled in Paperclip"
|
||||
: "Connection is not active",
|
||||
updatedAt: filteredAt,
|
||||
})
|
||||
.where(
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { chatActions, type Db } from "@paperclipai/db";
|
||||
import { eq, sql, type SQLWrapper } from "drizzle-orm";
|
||||
import { agentWakeupRequests, chatActions, chatDeliveries, chatEndpoints, type Db } from "@paperclipai/db";
|
||||
import { and, eq, or, sql, type SQLWrapper } from "drizzle-orm";
|
||||
import { HttpError } from "../errors.js";
|
||||
|
||||
export interface CommittedChatResponseAuthorizationInput {
|
||||
@@ -287,3 +287,41 @@ export function assertDurableChatWakeupReceipt(
|
||||
throw new Error("chat_inbound_wakeup_receipt_conflict");
|
||||
}
|
||||
}
|
||||
|
||||
/** Work-start policy only: ongoing tool and publication authorization must not
|
||||
* revoke an already-started run when the operator enables mention gating. */
|
||||
export async function slackMentionAllowsRunStart(
|
||||
db: Db,
|
||||
run: { id: string; companyId: string; agentId: string; wakeupRequestId: string | null },
|
||||
): Promise<boolean> {
|
||||
if (!run.wakeupRequestId) return true;
|
||||
const inputs = await db.select({ event: chatDeliveries.normalizedEvent })
|
||||
.from(agentWakeupRequests)
|
||||
.innerJoin(chatActions, and(
|
||||
eq(chatActions.id, agentWakeupRequests.id),
|
||||
eq(chatActions.companyId, run.companyId),
|
||||
eq(chatActions.kind, "inbound_wakeup"),
|
||||
))
|
||||
.innerJoin(chatEndpoints, and(
|
||||
eq(chatEndpoints.id, chatActions.endpointId),
|
||||
eq(chatEndpoints.companyId, run.companyId),
|
||||
eq(chatEndpoints.provider, "slack"),
|
||||
eq(chatEndpoints.requireAtMention, true),
|
||||
))
|
||||
.innerJoin(chatDeliveries, and(
|
||||
eq(chatDeliveries.id, chatActions.deliveryId),
|
||||
eq(chatDeliveries.endpointId, chatEndpoints.id),
|
||||
eq(chatDeliveries.companyId, run.companyId),
|
||||
))
|
||||
.where(and(
|
||||
eq(agentWakeupRequests.companyId, run.companyId),
|
||||
eq(agentWakeupRequests.agentId, run.agentId),
|
||||
or(
|
||||
eq(agentWakeupRequests.runId, run.id),
|
||||
eq(agentWakeupRequests.id, run.wakeupRequestId),
|
||||
sql`${agentWakeupRequests.payload}->>'coalescedIntoWakeupRequestId' = ${run.wakeupRequestId}`,
|
||||
),
|
||||
));
|
||||
return inputs.every(({ event }) => event.trigger === "mention" ||
|
||||
(event.message as { mentionedBot?: boolean } | undefined)?.mentionedBot === true);
|
||||
}
|
||||
@@ -1823,6 +1823,7 @@ export function heartbeatService(
|
||||
});
|
||||
|
||||
const heartbeatQueue = createHeartbeatQueue(db, {
|
||||
applyWakeQueuePostCommitEffects,
|
||||
setRunStatus,
|
||||
setWakeupStatus,
|
||||
appendRunEvent,
|
||||
|
||||
@@ -48,6 +48,7 @@ function callbacks(db: Db) {
|
||||
filterZombieCoalesceTarget: <T extends { id: string; status: string }>(run: T | null) => run,
|
||||
getSchedulingSuppression: vi.fn(async () => ({ suppressed: false, reason: null } as Awaited<ReturnType<HeartbeatQueueDependencies["getSchedulingSuppression"]>>)),
|
||||
executeRun: vi.fn(async () => undefined),
|
||||
applyWakeQueuePostCommitEffects: vi.fn(async () => undefined),
|
||||
releaseIssueExecutionAndPromote: vi.fn(async () => undefined),
|
||||
publishRunLifecyclePluginEvent: vi.fn(),
|
||||
cancelRunInternal: vi.fn(async () => undefined),
|
||||
|
||||
@@ -41,6 +41,7 @@ import { claimQueuedNativeReviewRun } from "../native-runtime/native-review-disp
|
||||
import { adapterExecutionControls } from "../adapter-execution-control.js";
|
||||
import { executionFailureRetryCount } from "../execution-recovery-attempt.js";
|
||||
import {
|
||||
slackMentionAllowsRunStart,
|
||||
assertDurableChatWakeupReceipt,
|
||||
assertDurableChatWakeupRequest,
|
||||
authorizeFailedChatRunRetryWake,
|
||||
@@ -162,7 +163,7 @@ import type { issueTreeControlService } from "../issue-tree-control.js";
|
||||
import type { budgetService } from "../budgets.js";
|
||||
import type { instanceSettingsService } from "../instance-settings.js";
|
||||
import type { createRunDispatch } from "../../modules/run-dispatch/index.js";
|
||||
import type { createWakeQueue } from "../../modules/wake-queue/index.js";
|
||||
import type { PostCommitEffect as WakeQueuePostCommitEffect, createWakeQueue } from "../../modules/wake-queue/index.js";
|
||||
|
||||
type HeartbeatRun = typeof heartbeatRuns.$inferSelect;
|
||||
type HeartbeatRunState = ReturnType<typeof createHeartbeatRunState>;
|
||||
@@ -196,7 +197,8 @@ export interface HeartbeatQueueDependencies extends Pick<HeartbeatRetryDependenc
|
||||
filterZombieCoalesceTarget: <T extends { id: string; status: string }>(run: T | null, live: { has(id: string): boolean }) => T | null;
|
||||
getSchedulingSuppression: () => Promise<{ suppressed: boolean; reason: "worktree_instance" | "database_restore_in_progress" | "task_drain" | null }>;
|
||||
executeRun: (runId: string) => Promise<void>;
|
||||
releaseIssueExecutionAndPromote: (run: Pick<HeartbeatRun, "id" | "companyId">, options?: { suppressImmediateRecovery?: boolean }) => Promise<unknown>;
|
||||
releaseIssueExecutionAndPromote: (run: Pick<HeartbeatRun, "id" | "companyId">, options?: { suppressImmediateRecovery?: boolean; deferredPostCommitEffects?: WakeQueuePostCommitEffect[] }) => Promise<unknown>;
|
||||
applyWakeQueuePostCommitEffects: (effects: WakeQueuePostCommitEffect[]) => Promise<void>;
|
||||
publishRunLifecyclePluginEvent: (run: HeartbeatRun) => void;
|
||||
cancelRunInternal: (runId: string, reason?: string, options?: { errorCode?: string }) => Promise<unknown>;
|
||||
cancelActiveForAgentInternal: (agentId: string, reason: string) => Promise<unknown>;
|
||||
@@ -343,6 +345,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
setWakeupStatus,
|
||||
appendRunEvent,
|
||||
releaseIssueExecutionAndPromote,
|
||||
applyWakeQueuePostCommitEffects,
|
||||
issuesSvc,
|
||||
options,
|
||||
publishRunLifecyclePluginEvent,
|
||||
@@ -933,8 +936,34 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
async function claimQueuedRun(
|
||||
run: typeof heartbeatRuns.$inferSelect,
|
||||
companyAgents?: AgentOrgRow[],
|
||||
deferredPostCommitEffects?: WakeQueuePostCommitEffect[],
|
||||
) {
|
||||
if (run.status !== "queued") return run;
|
||||
// Claim is the first execution boundary. startedAt survives same-run
|
||||
// retries; restart recovery may also enter executeRun already running.
|
||||
// Neither may reinterpret an admitted, already-started message as new work.
|
||||
if (!run.startedAt && !(await slackMentionAllowsRunStart(db, run))) {
|
||||
const now = new Date();
|
||||
const reason = "Message did not @mention the bot";
|
||||
// This runs under the agent start lock. Do not use cancelRunInternal,
|
||||
// which recursively starts the next run and reacquires that lock.
|
||||
const cancelled = await setRunStatus(run.id, "cancelled", {
|
||||
finishedAt: now,
|
||||
error: reason,
|
||||
errorCode: "chat_mention_required",
|
||||
});
|
||||
if (cancelled) {
|
||||
await setWakeupStatus(run.wakeupRequestId, "skipped", { finishedAt: now, error: reason });
|
||||
await appendRunEvent(cancelled, {
|
||||
eventType: "lifecycle", stream: "system", level: "warn", message: reason,
|
||||
});
|
||||
await releaseIssueExecutionAndPromote(cancelled, {
|
||||
suppressImmediateRecovery: true, deferredPostCommitEffects,
|
||||
});
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
const agent = await getAgent(run.agentId);
|
||||
if (!agent) {
|
||||
await cancelRunInternal(
|
||||
@@ -1873,6 +1902,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
// Cancelled after the start lock is released: cancelRunInternal promotes the
|
||||
// agent's next queued run, which takes this same lock.
|
||||
const rejectedClaims: Array<{ run: typeof heartbeatRuns.$inferSelect; err: HttpError }> = [];
|
||||
const deferredPostCommitEffects: WakeQueuePostCommitEffect[] = [];
|
||||
let cancellationReason: string | undefined;
|
||||
|
||||
return withAgentStartLock(agentId, async () => {
|
||||
@@ -1984,7 +2014,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
if (claimedRuns.length >= availableSlots) break;
|
||||
let claimed: typeof heartbeatRuns.$inferSelect | null;
|
||||
try {
|
||||
claimed = await claimQueuedRun(queuedRun, companyAgents);
|
||||
claimed = await claimQueuedRun(queuedRun, companyAgents, deferredPostCommitEffects);
|
||||
} catch (err) {
|
||||
if (isPermanentClaimRejection(err)) {
|
||||
rejectedClaims.push({ run: queuedRun, err });
|
||||
@@ -2021,8 +2051,13 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
}
|
||||
return claimedRuns;
|
||||
}).finally(async () => {
|
||||
if (cancellationReason) await cancelActiveForAgentInternal(agentId, cancellationReason);
|
||||
await cancelRejectedQueuedRuns(rejectedClaims);
|
||||
// Dispatch promoted inputs after the agent start lock is released.
|
||||
try {
|
||||
await applyWakeQueuePostCommitEffects(deferredPostCommitEffects);
|
||||
} finally {
|
||||
if (cancellationReason) await cancelActiveForAgentInternal(agentId, cancellationReason);
|
||||
await cancelRejectedQueuedRuns(rejectedClaims);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -684,7 +684,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
|
||||
|
||||
async function releaseIssueExecutionAndPromote(
|
||||
run: Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId">,
|
||||
options: { suppressImmediateRecovery?: boolean } = {},
|
||||
options: { suppressImmediateRecovery?: boolean; deferredPostCommitEffects?: WakeQueuePostCommitEffect[] } = {},
|
||||
) {
|
||||
try {
|
||||
const source = await getRun(run.id);
|
||||
@@ -696,7 +696,11 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
|
||||
// while continuation classification blocks periodic generic retries.
|
||||
suppressImmediateRecovery: options.suppressImmediateRecovery || isAiAuthenticationBlocked(source),
|
||||
});
|
||||
await applyWakeQueuePostCommitEffects(postCommitEffects);
|
||||
if (options.deferredPostCommitEffects) {
|
||||
options.deferredPostCommitEffects.push(...postCommitEffects);
|
||||
} else {
|
||||
await applyWakeQueuePostCommitEffects(postCommitEffects);
|
||||
}
|
||||
const completed = await getRun(run.id);
|
||||
const issueId = readNonEmptyString(completed?.contextSnapshot?.issueId)
|
||||
?? readNonEmptyString(completed?.contextSnapshot?.taskId) ?? completed?.nativeIssueId;
|
||||
|
||||
@@ -118,6 +118,7 @@ export interface ChatEndpoint {
|
||||
botExternalId?: string | null;
|
||||
photonAllocation?: "dedicated" | "shared";
|
||||
allowDirectMessages?: boolean;
|
||||
requireAtMention?: boolean;
|
||||
allowGroupChats?: boolean;
|
||||
allowUnlinkedPeople: boolean;
|
||||
replyMode?: "subscribed" | "mention_each_reply" | null;
|
||||
|
||||
@@ -532,6 +532,17 @@ function Settings({
|
||||
updateEndpoint.mutate({ allowDirectMessages })
|
||||
}
|
||||
/>
|
||||
{endpoint.provider === "slack" && (
|
||||
<SettingToggle
|
||||
label="Require at-mention"
|
||||
detail="Only respond to messages that @mention this bot, including thread replies and direct messages. Direct messages must also be allowed above."
|
||||
checked={endpoint.requireAtMention ?? false}
|
||||
pending={updateEndpoint.isPending}
|
||||
onChange={(requireAtMention) =>
|
||||
updateEndpoint.mutate({ requireAtMention })
|
||||
}
|
||||
/>
|
||||
)}
|
||||
{endpoint.provider === "microsoft-teams" && (
|
||||
<SettingToggle
|
||||
label="Allow group chats"
|
||||
|
||||
Reference in new issue
Block a user