diff --git a/README.md b/README.md index 6eaf2e3..70daaca 100644 --- a/README.md +++ b/README.md @@ -127,6 +127,13 @@ Slack alert via `FACTORY_CANARY_SLACK_WEBHOOK` on failure. See `scripts/com.agentrelay.factory-canary.plist.example` for an every-6h launchd template. +### Slack questions + +Set `slack.channel` to the Slack channel name, channel ID, or mounted channel +directory. For example, `factory`, `C1234567890`, and +`C1234567890__factory` are all accepted when the channel is present under the +relayfile Slack mount. + ## Tell it what to work on Two ways to hand the factory an issue — both are just labeling/titling, nothing diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 7577389..907c66b 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -6875,6 +6875,39 @@ describe('FactoryLoop', () => { expect(mount.confirmedPaths.filter((path) => path.includes('/replies/'))).toEqual([]) }) + it('resolves a configured Slack channel name to the mounted channel directory', async () => { + const channelDir = 'C0FACTORY__factory-e2e' + const mount = new ConfirmRecordingSlackMountClient({ + [issuePath(35)]: issueFile(35), + [`/slack/channels/${channelDir}/meta.json`]: {}, + }) + const fleet = new FakeFleetClient() + const factory = createFactory(config({ slack: slackConfig('factory-e2e') }), { + mount, + fleet, + triage: new StaticTriage(), + }) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(35), issueFile(35)))) + + const slackRoots = mount.writes.filter((write) => isSlackRootWritePath(write.path)) + expect(slackRoots).toHaveLength(1) + expect(slackRoots[0]?.path).toMatch(new RegExp(`^/slack/channels/${channelDir}/messages/`)) + expect(fleet.messages[0]?.text).toContain(`/slack/channels/${channelDir}/messages/${mount.threadTs.replace(/\./g, '_')}/replies/question.json`) + + emitSlackReply(mount, slackReplyFixturePath(channelDir, mount.threadTs, 'human-answer-35'), 'human-answer-35', { + text: 'Use the mounted channel directory.', + user: 'U123', + user_is_bot: false, + }) + await flush() + await flush() + + expect(slackAnswerInputs(fleet)).toEqual([ + { name: 'ar-35-impl-pear', data: '\nHuman reply in the Slack thread:\nUse the mounted channel directory.\n\r' }, + ]) + }) + it('routes a mid-task agent question to the Slack dispatch thread and returns the human answer via sendInput', async () => { const mount = new ConfirmRecordingSlackMountClient({ [issuePath(36)]: issueFile(36) }) const fleet = new FakeFleetClient() diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 78f6192..5442df0 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -53,7 +53,7 @@ import type { TriageDecision, TriageEngine, } from '../types' -import { MountGithubRead, MountLinearWriteback, MountSlackWriteback } from '../writeback' +import { MountGithubRead, MountLinearWriteback, MountSlackWriteback, slackChannelAliases, slackChannelSegment } from '../writeback' import { asRecord, parseJsonContent, stableHash, wrappedPayload } from '../writeback/shared' import { type InFlightIssue, issueKey, type TrackedAgent } from './batch-tracker' import { findAgentProcessByName, readProcessIdentity, type AgentProcessFinder } from './process-identity' @@ -194,6 +194,8 @@ export class FactoryLoop implements Factory { readonly #resumeInFlight = new Map>() readonly #slackWatchers = new Map() readonly #slackWatcherStarts = new Map>() + #resolvedSlackChannelDir?: string + #slackChannelDirRefresh?: Promise // Agents we've already logged an ambiguous-PID-lookup warning for, so the // reaper doesn't spam the same benign "ambiguous process lookup" line on every // poll (a joined/cloud agent has no local PID to resolve — expected). @@ -2833,12 +2835,15 @@ export class FactoryLoop implements Factory { if (this.#slack && this.#config.slack && !await this.#shouldSkipSlackWriteback('merge-done-thread')) { try { - const root = await this.#slack.postThread({ - channel: this.#config.slack.channel, - text: `${issue.key}: PR merged; Linear state set to Done.`, - }) - await this.#slack.reply(root.threadId, `${issue.key}: Linear state set to Done.`) - this.#recordSlackWritebackSuccess('merge-done-thread') + const channel = await this.#slackChannelDir() + if (channel) { + const root = await this.#slack.postThread({ + channel, + text: `${issue.key}: PR merged; Linear state set to Done.`, + }) + await this.#slack.reply(root.threadId, `${issue.key}: Linear state set to Done.`) + this.#recordSlackWritebackSuccess('merge-done-thread') + } } catch (error) { this.#markSlackWritebackFailure('merge-done-thread', error) } @@ -3084,21 +3089,24 @@ export class FactoryLoop implements Factory { if (this.#slack && this.#config.slack && !await this.#shouldSkipSlackWriteback('completion-thread')) { try { - const merged = opts.completionReason === 'pr-merged' - const completionText = merged - ? `${record.issue.key}: PR merged; Linear state set to ${statusLabel}.` - : `${record.issue.key}: factory agents completed${humanReview ? '; awaiting human review' : ''}.\nStatus: ${statusLabel}\nMerge policy: ${this.#config.mergePolicy}` - const stateText = merged - ? `${record.issue.key}: PR merged; Linear state set to ${statusLabel}.` - : humanReview - ? `${record.issue.key}: awaiting human review; Linear state set to ${statusLabel}.` - : `${record.issue.key}: Linear state set to ${statusLabel}.` - const root = await this.#slack.postThread({ - channel: this.#config.slack.channel, - text: completionText, - }) - await this.#slack.reply(root.threadId, stateText) - this.#recordSlackWritebackSuccess('completion-thread') + const channel = await this.#slackChannelDir() + if (channel) { + const merged = opts.completionReason === 'pr-merged' + const completionText = merged + ? `${record.issue.key}: PR merged; Linear state set to ${statusLabel}.` + : `${record.issue.key}: factory agents completed${humanReview ? '; awaiting human review' : ''}.\nStatus: ${statusLabel}\nMerge policy: ${this.#config.mergePolicy}` + const stateText = merged + ? `${record.issue.key}: PR merged; Linear state set to ${statusLabel}.` + : humanReview + ? `${record.issue.key}: awaiting human review; Linear state set to ${statusLabel}.` + : `${record.issue.key}: Linear state set to ${statusLabel}.` + const root = await this.#slack.postThread({ + channel, + text: completionText, + }) + await this.#slack.reply(root.threadId, stateText) + this.#recordSlackWritebackSuccess('completion-thread') + } } catch (error) { this.#markSlackWritebackFailure('completion-thread', error) } @@ -3362,7 +3370,7 @@ export class FactoryLoop implements Factory { } const root = await this.#slack.postThread({ - channel: this.#config.slack.channel, + channel: await this.#slackChannelDir() ?? this.#config.slack.channel, text: [ `${record.issue.key}: factory agents dispatched.`, `State: ${result.stateId ?? 'dispatching'}`, @@ -3427,7 +3435,7 @@ export class FactoryLoop implements Factory { const issue = await this.#readIssue(decision.issue.path) const root = await this.#slack.postThread({ - channel: this.#config.slack.channel, + channel: await this.#slackChannelDir() ?? this.#config.slack.channel, text: [ `${decision.issue.key}: factory triage escalation for ${issue?.title ?? decision.issue.key}`, `Reason: ${reason}`, @@ -3449,7 +3457,7 @@ export class FactoryLoop implements Factory { return } - const channelDir = this.#config.slack.channel + const channelDir = await this.#slackChannelDir() ?? this.#config.slack.channel const messagesPrefix = slackChannelMessagesPrefix(channelDir) const preExistingPaths = new Set() const seenReplies = new Set() @@ -3687,14 +3695,86 @@ export class FactoryLoop implements Factory { return resolve(process.cwd(), '.integrations') } + async #slackChannelDir(): Promise { + if (!this.#config.slack) { + return undefined + } + if (this.#resolvedSlackChannelDir) { + return this.#resolvedSlackChannelDir + } + if (this.#slackChannelDirRefresh) { + return this.#slackChannelDirRefresh + } + + this.#slackChannelDirRefresh = this.#resolveSlackChannelDir() + .finally(() => { + this.#slackChannelDirRefresh = undefined + }) + return this.#slackChannelDirRefresh + } + + async #resolveSlackChannelDir(): Promise { + const configured = this.#config.slack?.channel.trim() + if (!configured) { + return undefined + } + + const configuredSegment = slackChannelSegment(configured) + const configuredAliases = slackChannelAliases(configured) + if (configuredAliases.size === 0) { + return undefined + } + + let paths: string[] + try { + paths = await this.#mount.listTree('/slack/channels') + } catch (error) { + this.#logger.warn?.('[factory] unable to resolve Slack channel from mount; using configured channel value', { + channel: configured, + error, + }) + this.#resolvedSlackChannelDir = configuredSegment + return this.#resolvedSlackChannelDir + } + + const channelDirs = [...new Set(paths + .map((path) => path.match(/^\/slack\/channels\/([^/]+)/u)?.[1]) + .filter((channelDir): channelDir is string => Boolean(channelDir)))] + .sort() + const matches = channelDirs.filter((channelDir) => { + const aliases = slackChannelAliases(channelDir) + return [...aliases].some((alias) => configuredAliases.has(alias)) + }) + + if (matches.length === 0) { + this.#logger.warn?.('[factory] Slack channel was not found in mount; using configured channel value', { + channel: configured, + }) + this.#resolvedSlackChannelDir = configuredSegment + return this.#resolvedSlackChannelDir + } + + const exact = matches.find((channelDir) => slackChannelSegment(channelDir).toLowerCase() === configuredSegment.toLowerCase()) + this.#resolvedSlackChannelDir = exact ?? matches[0] + if (matches.length > 1 && !exact) { + this.#logger.warn?.('[factory] Slack channel name matched multiple mount directories; using first match', { + channel: configured, + selected: this.#resolvedSlackChannelDir, + matches, + }) + } + return this.#resolvedSlackChannelDir + } + async #slackDispatchThreadFor(record: InFlightIssue): Promise<{ channel: string; threadId: string; mountRoot: string } | undefined> { if (!this.#config.slack) { return undefined } const threadId = await this.#state.getSlackThread(this.#workspaceId, issueKey(record.issue)) + const channel = await this.#slackChannelDir() ?? this.#config.slack.channel return threadId - ? { channel: this.#config.slack.channel, threadId, mountRoot: this.#integrationsMountRoot() } + ? { channel, threadId, mountRoot: this.#integrationsMountRoot() } : undefined } diff --git a/src/writeback/index.ts b/src/writeback/index.ts index 2c3091a..83d80f6 100644 --- a/src/writeback/index.ts +++ b/src/writeback/index.ts @@ -10,6 +10,8 @@ export type { } from './linear' export { MountSlackWriteback, + slackChannelAliases, + slackChannelSegment, } from './slack' export type { MountSlackWritebackConfig, diff --git a/src/writeback/slack.ts b/src/writeback/slack.ts index 3e79e6f..e452068 100644 --- a/src/writeback/slack.ts +++ b/src/writeback/slack.ts @@ -16,14 +16,14 @@ interface ThreadRef { const channelIdFromDir = (channelDir: string): string => channelDir.split('__')[0] ?? channelDir -const channelSegment = (value: string): string => { +export const slackChannelSegment = (value: string): string => { const trimmed = value.trim().replace(/^#/u, '') const match = trimmed.match(/(?:^|\/)slack\/channels\/([^/]+)/u) return match?.[1] ?? trimmed.split('/')[0] ?? trimmed } -const channelAliases = (value?: string): Set => { - const segment = typeof value === 'string' ? channelSegment(value) : '' +export const slackChannelAliases = (value?: string): Set => { + const segment = typeof value === 'string' ? slackChannelSegment(value) : '' if (!segment) return new Set() const aliases = new Set() @@ -45,8 +45,8 @@ const assertSlackChannelAllowed = ( targetChannel: string | undefined, context: string, ): void => { - const configured = channelAliases(configuredChannel) - const target = channelAliases(targetChannel) + const configured = slackChannelAliases(configuredChannel) + const target = slackChannelAliases(targetChannel) if (configured.size === 0) { throw new Error(`Refusing Slack writeback for ${context}: configured factory-e2e channel is required`) }