Skip to content

Commit 51296e7

Browse files
steipeteDarlingtonDeveloper
steipete
and
DarlingtonDeveloper
committed
feat(slack): land thread-ownership from @DarlingtonDeveloper (#15775)
Land PR #15775 by @DarlingtonDeveloper: - add thread-ownership plugin and Slack message_sending hook wiring - include regression tests and changelog update Co-authored-by: Mike <[email protected]>
1 parent 874ff70 commit 51296e7

6 files changed

Lines changed: 513 additions & 2 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ Docs: https://docs.openclaw.ai
99
- Skills: remove duplicate `local-places` Google Places skill/proxy and keep `goplaces` as the single supported Google Places path.
1010
- Discord: send voice messages with waveform previews from local audio files (including silent delivery). (#7253) Thanks @nyanjou.
1111
- Discord: add configurable presence status/activity/type/url (custom status defaults to activity text). (#10855) Thanks @h0tp-ftw.
12+
- Slack/Plugins: add thread-ownership outbound gating via `message_sending` hooks, including @-mention bypass tracking and Slack outbound hook wiring for cancel/modify behavior. (#15775) Thanks @DarlingtonDeveloper.
1213

1314
### Fixes
1415

‎extensions/thread-ownership/index.test.ts‎

Lines changed: 180 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,180 @@
1+
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
2+
import register from "./index.js";
3+
4+
describe("thread-ownership plugin", () => {
5+
const hooks: Record<string, Function> = {};
6+
const api = {
7+
pluginConfig: {},
8+
config: {
9+
agents: {
10+
list: [{ id: "test-agent", default: true, identity: { name: "TestBot" } }],
11+
},
12+
},
13+
id: "thread-ownership",
14+
name: "Thread Ownership",
15+
logger: { info: vi.fn(), warn: vi.fn(), debug: vi.fn() },
16+
on: vi.fn((hookName: string, handler: Function) => {
17+
hooks[hookName] = handler;
18+
}),
19+
};
20+
21+
let originalFetch: typeof globalThis.fetch;
22+
23+
beforeEach(() => {
24+
vi.clearAllMocks();
25+
for (const key of Object.keys(hooks)) delete hooks[key];
26+
27+
process.env.SLACK_FORWARDER_URL = "http://localhost:8750";
28+
process.env.SLACK_BOT_USER_ID = "U999";
29+
30+
originalFetch = globalThis.fetch;
31+
globalThis.fetch = vi.fn();
32+
});
33+
34+
afterEach(() => {
35+
globalThis.fetch = originalFetch;
36+
delete process.env.SLACK_FORWARDER_URL;
37+
delete process.env.SLACK_BOT_USER_ID;
38+
vi.restoreAllMocks();
39+
});
40+
41+
it("registers message_received and message_sending hooks", () => {
42+
register(api as any);
43+
44+
expect(api.on).toHaveBeenCalledTimes(2);
45+
expect(api.on).toHaveBeenCalledWith("message_received", expect.any(Function));
46+
expect(api.on).toHaveBeenCalledWith("message_sending", expect.any(Function));
47+
});
48+
49+
describe("message_sending", () => {
50+
beforeEach(() => {
51+
register(api as any);
52+
});
53+
54+
it("allows non-slack channels", async () => {
55+
const result = await hooks.message_sending(
56+
{ content: "hello", metadata: { threadTs: "1234.5678", channelId: "C123" }, to: "C123" },
57+
{ channelId: "discord", conversationId: "C123" },
58+
);
59+
60+
expect(result).toBeUndefined();
61+
expect(globalThis.fetch).not.toHaveBeenCalled();
62+
});
63+
64+
it("allows top-level messages (no threadTs)", async () => {
65+
const result = await hooks.message_sending(
66+
{ content: "hello", metadata: {}, to: "C123" },
67+
{ channelId: "slack", conversationId: "C123" },
68+
);
69+
70+
expect(result).toBeUndefined();
71+
expect(globalThis.fetch).not.toHaveBeenCalled();
72+
});
73+
74+
it("claims ownership successfully", async () => {
75+
vi.mocked(globalThis.fetch).mockResolvedValue(
76+
new Response(JSON.stringify({ owner: "test-agent" }), { status: 200 }),
77+
);
78+
79+
const result = await hooks.message_sending(
80+
{ content: "hello", metadata: { threadTs: "1234.5678", channelId: "C123" }, to: "C123" },
81+
{ channelId: "slack", conversationId: "C123" },
82+
);
83+
84+
expect(result).toBeUndefined();
85+
expect(globalThis.fetch).toHaveBeenCalledWith(
86+
"http://localhost:8750/api/v1/ownership/C123/1234.5678",
87+
expect.objectContaining({
88+
method: "POST",
89+
body: JSON.stringify({ agent_id: "test-agent" }),
90+
}),
91+
);
92+
});
93+
94+
it("cancels when thread owned by another agent", async () => {
95+
vi.mocked(globalThis.fetch).mockResolvedValue(
96+
new Response(JSON.stringify({ owner: "other-agent" }), { status: 409 }),
97+
);
98+
99+
const result = await hooks.message_sending(
100+
{ content: "hello", metadata: { threadTs: "1234.5678", channelId: "C123" }, to: "C123" },
101+
{ channelId: "slack", conversationId: "C123" },
102+
);
103+
104+
expect(result).toEqual({ cancel: true });
105+
expect(api.logger.info).toHaveBeenCalledWith(expect.stringContaining("cancelled send"));
106+
});
107+
108+
it("fails open on network error", async () => {
109+
vi.mocked(globalThis.fetch).mockRejectedValue(new Error("ECONNREFUSED"));
110+
111+
const result = await hooks.message_sending(
112+
{ content: "hello", metadata: { threadTs: "1234.5678", channelId: "C123" }, to: "C123" },
113+
{ channelId: "slack", conversationId: "C123" },
114+
);
115+
116+
expect(result).toBeUndefined();
117+
expect(api.logger.warn).toHaveBeenCalledWith(
118+
expect.stringContaining("ownership check failed"),
119+
);
120+
});
121+
});
122+
123+
describe("message_received @-mention tracking", () => {
124+
beforeEach(() => {
125+
register(api as any);
126+
});
127+
128+
it("tracks @-mentions and skips ownership check for mentioned threads", async () => {
129+
// Simulate receiving a message that @-mentions the agent.
130+
await hooks.message_received(
131+
{ content: "Hey @TestBot help me", metadata: { threadTs: "9999.0001", channelId: "C456" } },
132+
{ channelId: "slack", conversationId: "C456" },
133+
);
134+
135+
// Now send in the same thread -- should skip the ownership HTTP call.
136+
const result = await hooks.message_sending(
137+
{ content: "Sure!", metadata: { threadTs: "9999.0001", channelId: "C456" }, to: "C456" },
138+
{ channelId: "slack", conversationId: "C456" },
139+
);
140+
141+
expect(result).toBeUndefined();
142+
expect(globalThis.fetch).not.toHaveBeenCalled();
143+
});
144+
145+
it("ignores @-mentions on non-slack channels", async () => {
146+
// Use a unique thread key so module-level state from other tests doesn't interfere.
147+
await hooks.message_received(
148+
{ content: "Hey @TestBot", metadata: { threadTs: "7777.0001", channelId: "C999" } },
149+
{ channelId: "discord", conversationId: "C999" },
150+
);
151+
152+
// The mention should not have been tracked, so sending should still call fetch.
153+
vi.mocked(globalThis.fetch).mockResolvedValue(
154+
new Response(JSON.stringify({ owner: "test-agent" }), { status: 200 }),
155+
);
156+
157+
await hooks.message_sending(
158+
{ content: "Sure!", metadata: { threadTs: "7777.0001", channelId: "C999" }, to: "C999" },
159+
{ channelId: "slack", conversationId: "C999" },
160+
);
161+
162+
expect(globalThis.fetch).toHaveBeenCalled();
163+
});
164+
165+
it("tracks bot user ID mentions via <@U999> syntax", async () => {
166+
await hooks.message_received(
167+
{ content: "Hey <@U999> help", metadata: { threadTs: "8888.0001", channelId: "C789" } },
168+
{ channelId: "slack", conversationId: "C789" },
169+
);
170+
171+
const result = await hooks.message_sending(
172+
{ content: "On it!", metadata: { threadTs: "8888.0001", channelId: "C789" }, to: "C789" },
173+
{ channelId: "slack", conversationId: "C789" },
174+
);
175+
176+
expect(result).toBeUndefined();
177+
expect(globalThis.fetch).not.toHaveBeenCalled();
178+
});
179+
});
180+
});

‎extensions/thread-ownership/index.ts‎

Lines changed: 133 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,133 @@
1+
import type { OpenClawConfig, OpenClawPluginApi } from "openclaw/plugin-sdk";
2+
3+
type ThreadOwnershipConfig = {
4+
forwarderUrl?: string;
5+
abTestChannels?: string[];
6+
};
7+
8+
type AgentEntry = NonNullable<NonNullable<OpenClawConfig["agents"]>["list"]>[number];
9+
10+
// In-memory set of {channel}:{thread} keys where this agent was @-mentioned.
11+
// Entries expire after 5 minutes.
12+
const mentionedThreads = new Map<string, number>();
13+
const MENTION_TTL_MS = 5 * 60 * 1000;
14+
15+
function cleanExpiredMentions(): void {
16+
const now = Date.now();
17+
for (const [key, ts] of mentionedThreads) {
18+
if (now - ts > MENTION_TTL_MS) {
19+
mentionedThreads.delete(key);
20+
}
21+
}
22+
}
23+
24+
function resolveOwnershipAgent(config: OpenClawConfig): { id: string; name: string } {
25+
const list = Array.isArray(config.agents?.list)
26+
? config.agents.list.filter((entry): entry is AgentEntry =>
27+
Boolean(entry && typeof entry === "object"),
28+
)
29+
: [];
30+
const selected = list.find((entry) => entry.default === true) ?? list[0];
31+
32+
const id =
33+
typeof selected?.id === "string" && selected.id.trim() ? selected.id.trim() : "unknown";
34+
const identityName =
35+
typeof selected?.identity?.name === "string" ? selected.identity.name.trim() : "";
36+
const fallbackName = typeof selected?.name === "string" ? selected.name.trim() : "";
37+
const name = identityName || fallbackName;
38+
39+
return { id, name };
40+
}
41+
42+
export default function register(api: OpenClawPluginApi) {
43+
const pluginCfg = (api.pluginConfig ?? {}) as ThreadOwnershipConfig;
44+
const forwarderUrl = (
45+
pluginCfg.forwarderUrl ??
46+
process.env.SLACK_FORWARDER_URL ??
47+
"http://slack-forwarder:8750"
48+
).replace(/\/$/, "");
49+
50+
const abTestChannels = new Set(
51+
pluginCfg.abTestChannels ??
52+
process.env.THREAD_OWNERSHIP_CHANNELS?.split(",").filter(Boolean) ??
53+
[],
54+
);
55+
56+
const { id: agentId, name: agentName } = resolveOwnershipAgent(api.config);
57+
const botUserId = process.env.SLACK_BOT_USER_ID ?? "";
58+
59+
// ---------------------------------------------------------------------------
60+
// message_received: track @-mentions so the agent can reply even if it
61+
// doesn't own the thread.
62+
// ---------------------------------------------------------------------------
63+
api.on("message_received", async (event, ctx) => {
64+
if (ctx.channelId !== "slack") return;
65+
66+
const text = event.content ?? "";
67+
const threadTs = (event.metadata?.threadTs as string) ?? "";
68+
const channelId = (event.metadata?.channelId as string) ?? ctx.conversationId ?? "";
69+
70+
if (!threadTs || !channelId) return;
71+
72+
// Check if this agent was @-mentioned.
73+
const mentioned =
74+
(agentName && text.includes(`@${agentName}`)) ||
75+
(botUserId && text.includes(`<@${botUserId}>`));
76+
77+
if (mentioned) {
78+
cleanExpiredMentions();
79+
mentionedThreads.set(`${channelId}:${threadTs}`, Date.now());
80+
}
81+
});
82+
83+
// ---------------------------------------------------------------------------
84+
// message_sending: check thread ownership before sending to Slack.
85+
// Returns { cancel: true } if another agent owns the thread.
86+
// ---------------------------------------------------------------------------
87+
api.on("message_sending", async (event, ctx) => {
88+
if (ctx.channelId !== "slack") return;
89+
90+
const threadTs = (event.metadata?.threadTs as string) ?? "";
91+
const channelId = (event.metadata?.channelId as string) ?? event.to;
92+
93+
// Top-level messages (no thread) are always allowed.
94+
if (!threadTs) return;
95+
96+
// Only enforce in A/B test channels (if set is empty, skip entirely).
97+
if (abTestChannels.size > 0 && !abTestChannels.has(channelId)) return;
98+
99+
// If this agent was @-mentioned in this thread recently, skip ownership check.
100+
cleanExpiredMentions();
101+
if (mentionedThreads.has(`${channelId}:${threadTs}`)) return;
102+
103+
// Try to claim ownership via the forwarder HTTP API.
104+
try {
105+
const resp = await fetch(`${forwarderUrl}/api/v1/ownership/${channelId}/${threadTs}`, {
106+
method: "POST",
107+
headers: { "Content-Type": "application/json" },
108+
body: JSON.stringify({ agent_id: agentId }),
109+
signal: AbortSignal.timeout(3000),
110+
});
111+
112+
if (resp.ok) {
113+
// We own it (or just claimed it), proceed.
114+
return;
115+
}
116+
117+
if (resp.status === 409) {
118+
// Another agent owns this thread — cancel the send.
119+
const body = (await resp.json()) as { owner?: string };
120+
api.logger.info?.(
121+
`thread-ownership: cancelled send to ${channelId}:${threadTs} — owned by ${body.owner}`,
122+
);
123+
return { cancel: true };
124+
}
125+
126+
// Unexpected status — fail open.
127+
api.logger.warn?.(`thread-ownership: unexpected status ${resp.status}, allowing send`);
128+
} catch (err) {
129+
// Network error — fail open.
130+
api.logger.warn?.(`thread-ownership: ownership check failed (${String(err)}), allowing send`);
131+
}
132+
});
133+
}

‎extensions/thread-ownership/openclaw.plugin.json‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
{
2+
"id": "thread-ownership",
3+
"name": "Thread Ownership",
4+
"description": "Prevents multiple agents from responding in the same Slack thread. Uses HTTP calls to the slack-forwarder ownership API.",
5+
"configSchema": {
6+
"type": "object",
7+
"additionalProperties": false,
8+
"properties": {
9+
"forwarderUrl": {
10+
"type": "string"
11+
},
12+
"abTestChannels": {
13+
"type": "array",
14+
"items": { "type": "string" }
15+
}
16+
}
17+
},
18+
"uiHints": {
19+
"forwarderUrl": {
20+
"label": "Forwarder URL",
21+
"help": "Base URL of the slack-forwarder ownership API (default: http://slack-forwarder:8750)"
22+
},
23+
"abTestChannels": {
24+
"label": "A/B Test Channels",
25+
"help": "Slack channel IDs where thread ownership is enforced"
26+
}
27+
}
28+
}

0 commit comments

Comments
 (0)