dsh-im-ops/plugin-src/host/channels/slack/production.mjs

139 lines
6 KiB
JavaScript

import { unlink } from 'node:fs/promises';
import { resolve } from 'node:path';
import { SlackConfigStore } from '../../../../src/channels/slack/config-store.mjs';
import { SlackController } from '../../../../src/channels/slack/slack-controller.mjs';
import { SlackHarnessClient } from '../../../../src/channels/slack/harness-client.mjs';
import { SlackRuntime } from '../../../../src/channels/slack/slack-runtime.mjs';
import { SlackStateStore } from '../../../../src/channels/slack/state-store.mjs';
import {
BotWorkspaceStore,
createBotWorkspaceScope,
createWorkspaceAwareController,
observeBotWorkspaceRemovals,
} from '../../../../src/channels/shared/bot-workspace-store.mjs';
import { listAgentPresetCatalog } from '../../../../src/channels/shared/agent-preset.mjs';
import { createTokenConnectionSupervisor } from '../shared/connection-supervisor.mjs';
import { pluginPaths } from '../shared/production.mjs';
import { createHarnessCommandExecutor } from '../../harness-command-executor.mjs';
import { harnessConnection } from '../../harness-connection.mjs';
import { createHarnessSessionExecutors } from '../../harness-session-coordinator.mjs';
export async function createProductionController(ctx, config = {}, internals = {}) {
if (!ctx?.credentials) throw new TypeError('dsh-im slack requires ctx.credentials');
const connection = harnessConnection(ctx, config);
const ResolvedConfigStore = internals.ConfigStore ?? SlackConfigStore;
const ResolvedStateStore = internals.StateStore ?? SlackStateStore;
const ResolvedHarness = internals.HarnessClient ?? SlackHarnessClient;
const ResolvedController = internals.Controller ?? SlackController;
const ResolvedRuntime = internals.Runtime ?? SlackRuntime;
const createSupervisor = internals.createConnectionSupervisor ?? createTokenConnectionSupervisor;
const logger = typeof ctx.logger === 'function'
? ctx.logger('dsh-im:slack') : (ctx.logger ?? console);
const agentPresetCatalog = () => listAgentPresetCatalog(ctx);
const paths = pluginPaths(config, 'slack');
const configStore = await new ResolvedConfigStore(paths.config).load();
const defaultWorkspace = resolve(config.workspace ?? process.cwd());
const WorkspaceStore = internals.WorkspaceStore ?? BotWorkspaceStore;
const workspaces = internals.workspaces
?? await new WorkspaceStore(paths.workspaces, { defaultWorkspace }).load();
const configuredBots = configStore.list();
await workspaces.reconcile(configuredBots.map((bot) => bot.botId));
await Promise.all(configuredBots.map((bot) => workspaces.ensure(bot.botId, {
defaultAgentPreset: config.agentPreset,
})));
const observedConfigStore = typeof configStore.remove === 'function'
? observeBotWorkspaceRemovals(configStore, { workspaces })
: configStore;
const stateStores = new Map();
const statePath = (botId) => resolve(paths.bots, botId, 'state.json');
const stateFor = async (botId) => {
let state = stateStores.get(botId);
if (!state) {
state = await new ResolvedStateStore(statePath(botId)).load();
stateStores.set(botId, state);
}
return state;
};
const commandExecutor = createHarnessCommandExecutor(ctx, internals.commandExecutor);
const { controlExecutor, sessionMaintenanceExecutor, fileIngressExecutor } = createHarnessSessionExecutors(ctx, {
controlExecutor: internals.controlExecutor,
sessionMaintenanceExecutor: internals.sessionMaintenanceExecutor,
fileIngressExecutor: internals.fileIngressExecutor,
});
const harness = new ResolvedHarness({
...connection,
workspace: defaultWorkspace,
autostart: false,
dshBin: config.dshBin ?? 'dsh',
...(commandExecutor ? { commandExecutor } : {}),
...(controlExecutor ? { controlExecutor } : {}),
...(sessionMaintenanceExecutor ? { sessionMaintenanceExecutor } : {}),
...(fileIngressExecutor ? { fileIngressExecutor } : {}),
});
const coreController = new ResolvedController({
credentials: ctx.credentials,
configStore: observedConfigStore,
logger,
...(internals.inspectCredentials ? { inspectCredentials: internals.inspectCredentials } : {}),
createRuntime: async ({ botId, config: botConfig, botToken, appToken }) => {
const state = await stateFor(botId);
await workspaces.ensure(botId, { defaultAgentPreset: config.agentPreset });
const workspaceScope = createBotWorkspaceScope(harness, {
botId, workspaces, state, agentPresetCatalog,
});
return new ResolvedRuntime({
config: botConfig,
botToken,
appToken,
harness: workspaceScope.harness,
state: workspaceScope.state,
contextEnhancement: { botId, getSettings: () => workspaces.contextEnhancementFor(botId) },
replyTimeoutMs: config.replyTimeoutMs ?? 600_000,
connectTimeoutMs: config.connectTimeoutMs ?? 20_000,
logger: {
error: (...args) => logger.error?.(`[${botId}]`, ...args),
warn: (...args) => logger.warn?.(`[${botId}]`, ...args),
info: (...args) => logger.info?.(`[${botId}]`, ...args),
debug: (...args) => logger.debug?.(`[${botId}]`, ...args),
},
});
},
deleteState: async ({ botId }) => {
const state = stateStores.get(botId);
stateStores.delete(botId);
if (state && typeof state.remove === 'function') {
await state.remove();
} else {
try {
await unlink(statePath(botId));
} catch (error) {
if (error?.code !== 'ENOENT') throw error;
}
}
},
});
const controller = createWorkspaceAwareController(coreController, {
workspaces,
stateFor,
agentPresetCatalog,
});
const supervisor = createSupervisor({
channel: 'slack',
controller,
harness,
logger,
retryDelaysMs: config.retryDelaysMs,
healthyIntervalMs: config.healthyIntervalMs,
}).start();
return {
controller,
ready: supervisor.ready,
async close() {
await supervisor.close();
await controller.close();
harness.stopManagedProcess();
},
};
}