403 lines
17 KiB
TypeScript
403 lines
17 KiB
TypeScript
/**
|
||
* worker-bootstrap.ts — entry point for the default "worker" mode of AAO.
|
||
*
|
||
* Phase 1 of the AAO Gateway split (2026-05-18) extracted the original
|
||
* `src/index.ts` startup logic into this function so `src/main.ts` can
|
||
* dispatch on `AAO_MODE` without two parallel entry points. The previous
|
||
* `src/index.ts` is kept as a thin compatibility shim that just calls
|
||
* `start()` here so `node dist/index.js` continues to work for anyone
|
||
* who has wired that path into their service manager.
|
||
*
|
||
* Worker mode behaviour is unchanged: load config → init DB → start
|
||
* WorkerManager + Scheduler → mount Express API. Anything that used to
|
||
* live in `index.ts main()` now lives in `start()`.
|
||
*/
|
||
import { Repository, BrowserSessionRepo } from './db/repository.js';
|
||
import { startCoreServer } from './bridge/server.js';
|
||
import { resolveListenPort } from './server/config.js';
|
||
import { runMigrations } from './db/migrate.js';
|
||
import { logger } from './logger.js';
|
||
import { accessSync, existsSync, mkdirSync, constants } from 'fs';
|
||
import { dirname, resolve, join } from 'path';
|
||
import { OpenAICompatClient, type Message, type ToolDef, type LLMEvent } from './llm/openai-compat.js';
|
||
import { runPromptCoach, PROMPT_COACH_TOOL_SCHEMA } from './engine/prompt-coach.js';
|
||
import { readUserAgentsMd } from './user-folder/paths.js';
|
||
import { readMemoryIndex } from './user-folder/memory.js';
|
||
import { setLlmUsageRecorder } from './llm/usage-recorder.js';
|
||
import { llmRoutingKey } from './llm/routing-key.js';
|
||
import { resolveMaxStreamMs } from './config.js';
|
||
import { ConfigManager } from './config-manager.js';
|
||
import { WorkerManager } from './worker-manager.js';
|
||
import { classifyPiece } from './engine/piece-classifier.js';
|
||
import { PieceCatalog } from './engine/piece-catalog.js';
|
||
import { Scheduler } from './scheduler.js';
|
||
import { buildTitleFallback, buildTitlePrompt } from './title-generation.js';
|
||
import { initMcpKeyFromFile } from './mcp/crypto.js';
|
||
import { registerShutdownHook, installSignalHandlers } from './bridge/shutdown.js';
|
||
import { acquireWorkerLock } from './instance-lock.js';
|
||
import { checkBwrapAvailable } from './engine/tools/sandbox.js';
|
||
import { SkillCatalog } from './engine/skills.js';
|
||
import { VapidKeyStore } from './vapid-store.js';
|
||
import { PushService } from './push-service.js';
|
||
|
||
function runPreflight(configPath: string, dbPath: string): void {
|
||
const errors: string[] = [];
|
||
|
||
if (!existsSync(configPath)) {
|
||
logger.info(`[preflight] config file not found: ${configPath}; continuing with defaults`);
|
||
}
|
||
|
||
logger.info('[preflight] running in local-only mode');
|
||
|
||
const dbDir = dirname(dbPath);
|
||
try {
|
||
mkdirSync(dbDir, { recursive: true });
|
||
accessSync(dbDir, constants.W_OK);
|
||
} catch (e) {
|
||
errors.push(`DB directory is not writable: ${dbDir} (${String(e)})`);
|
||
}
|
||
|
||
const schemaCandidates = [
|
||
resolve(process.cwd(), 'dist/db/schema.sql'),
|
||
resolve(process.cwd(), 'src/db/schema.sql'),
|
||
];
|
||
const hasSchema = schemaCandidates.some(p => existsSync(p));
|
||
if (!hasSchema) {
|
||
errors.push(`schema.sql not found. expected one of: ${schemaCandidates.join(', ')}`);
|
||
}
|
||
|
||
if (errors.length > 0) {
|
||
throw new Error(`Preflight failed:\n- ${errors.join('\n- ')}`);
|
||
}
|
||
}
|
||
|
||
export interface StartWorkerOptions {
|
||
/** Override config.yaml path. Defaults to 'config.yaml' (cwd-relative). */
|
||
configPath?: string;
|
||
}
|
||
|
||
export async function start(opts: StartWorkerOptions = {}): Promise<void> {
|
||
logger.info('maestro starting (mode=worker)...');
|
||
|
||
const configPath = opts.configPath ?? 'config.yaml';
|
||
const configManager = new ConfigManager(configPath);
|
||
const config = configManager.getConfig();
|
||
|
||
const dbPath = process.env['DB_PATH'] ?? './data/maestro.db';
|
||
runPreflight(configPath, dbPath);
|
||
|
||
// Single-writer guard: refuse to start if another worker process on this
|
||
// host already owns this DB. Prevents two schedulers from double-claiming
|
||
// and double-executing jobs (see instance-lock.ts).
|
||
const workerLock = acquireWorkerLock(dbPath);
|
||
registerShutdownHook('worker-instance-lock', async () => workerLock.release());
|
||
|
||
const requiresBwrap =
|
||
config.safety?.bashUnrestricted || config.safety?.bashSandbox === 'always';
|
||
if (requiresBwrap) {
|
||
const bwrapCheck = await checkBwrapAvailable();
|
||
if (!bwrapCheck.ok) {
|
||
throw new Error(
|
||
`Bash sandboxing requires bwrap but it is not available: ${bwrapCheck.reason}\n` +
|
||
`Enable user namespaces in the container/host, or set safety.bash_sandbox: auto.`
|
||
);
|
||
}
|
||
logger.info('[startup] bash sandbox enabled — bwrap verified');
|
||
} else if ((config.safety?.bashSandbox ?? 'auto') === 'auto') {
|
||
const bwrapCheck = await checkBwrapAvailable();
|
||
if (!bwrapCheck.ok) {
|
||
logger.warn(
|
||
'[security] bwrap unavailable — Bash falls back to hardened whitelist ' +
|
||
'(env-scrubbed, but NO filesystem/network isolation). A task can read host ' +
|
||
'config secrets and other tenants\' data. Set bash_sandbox: always (requires ' +
|
||
'bwrap) for multi-user isolation.'
|
||
);
|
||
}
|
||
} else if (config.safety?.bashSandbox === 'off') {
|
||
// `off` is an explicit opt-out: env is still scrubbed, but there is no
|
||
// filesystem/network isolation, so a task's Bash can reach host config
|
||
// secrets and the shared DB. Warn so a multi-user operator notices.
|
||
logger.warn(
|
||
'[security] bash_sandbox=off — the Bash tool runs without filesystem/network ' +
|
||
'isolation (env is still scrubbed). Acceptable for single-user/dev only; set ' +
|
||
'bash_sandbox: always (requires bwrap) for multi-user deployments.'
|
||
);
|
||
}
|
||
|
||
const repo = new Repository(dbPath);
|
||
runMigrations(repo.getDb());
|
||
|
||
// Install the process-global LLM usage recorder so every OpenAICompatClient
|
||
// completion (agent loop, title, classify, reflection — gateway + direct)
|
||
// lands in the per-user hour-grain ledger. Hour grain lets the dashboard
|
||
// re-bucket usage into the viewer's local calendar day. Best-effort: the
|
||
// recorder helper swallows write errors so a DB hiccup never kills a stream.
|
||
// Spec: docs/superpowers/specs/2026-06-11-usage-dashboard-v2-design.md
|
||
setLlmUsageRecorder((event) => {
|
||
repo.incrementLlmUsageHourly({
|
||
userId: event.userId,
|
||
source: event.source,
|
||
model: event.model,
|
||
route: event.route,
|
||
tokensIn: event.tokensIn,
|
||
tokensOut: event.tokensOut,
|
||
requests: 1,
|
||
});
|
||
});
|
||
// 起動時に孤立ジョブを回復
|
||
await repo.recoverOrphanedJobs();
|
||
|
||
// Worker 起動（WorkerManager に委譲）
|
||
const workerManager = new WorkerManager(repo, configManager);
|
||
workerManager.start();
|
||
|
||
// Listen port precedence: PORT env (incl. .env loaded by scripts/server.sh)
|
||
// > config.yaml server.port > default 9876. UI/config changes apply on the
|
||
// next restart (no live re-bind), same as the TLS settings.
|
||
const port = resolveListenPort(process.env['PORT'], config.server?.port, (m) => logger.warn(m));
|
||
|
||
// タイトル自動生成関数を作成（roles に 'title' を持つ worker を優先、なければ最初の worker）
|
||
const titleWorker =
|
||
config.provider.workers.find(w => w.enabled !== false && w.roles?.includes('title')) ??
|
||
config.provider.workers[0];
|
||
let titleClient: OpenAICompatClient | null = null;
|
||
let generateTitle: ((body: string, ownerId?: string) => Promise<string>) | undefined;
|
||
|
||
if (titleWorker) {
|
||
const titleModel = titleWorker.model ?? config.provider.model;
|
||
// Gateway (proxy) mode routes by role, not model name. Title generation
|
||
// is cheap, so send a tier the title worker serves (first of auto/fast/
|
||
// quality), defaulting to 'auto'. Direct mode sends the configured model.
|
||
const titleTier =
|
||
titleWorker.roles?.find(r => r === 'auto' || r === 'fast' || r === 'quality') ?? 'auto';
|
||
const titleRoutingKey = llmRoutingKey({
|
||
isGateway: titleWorker.proxy === true,
|
||
role: titleTier,
|
||
resolveDirectModel: () => titleModel,
|
||
});
|
||
logger.info(
|
||
`Config: title generation worker=${titleWorker.id} key=${titleRoutingKey ?? '<none>'}` +
|
||
(titleWorker.proxy === true ? ' (gateway:role)' : ''),
|
||
);
|
||
titleClient = new OpenAICompatClient(
|
||
titleWorker.endpoint,
|
||
titleRoutingKey,
|
||
titleWorker.apiKey,
|
||
config.provider.retry,
|
||
(config.provider.timeoutMinutes ?? 10) * 60 * 1000,
|
||
undefined,
|
||
undefined,
|
||
undefined,
|
||
// proxy mode so title/classification usage is recorded as
|
||
// source='gateway' with the backendId route (not mislabeled 'direct').
|
||
{ proxy: titleWorker.proxy === true, maxStreamMs: resolveMaxStreamMs(config.provider) },
|
||
);
|
||
|
||
generateTitle = async (body: string, ownerId?: string): Promise<string> => {
|
||
const fallback = buildTitleFallback(body);
|
||
const prompt = buildTitlePrompt(body);
|
||
if (!prompt) return fallback;
|
||
let title = '';
|
||
try {
|
||
for await (const event of titleClient!.chat([{ role: 'user', content: prompt }], undefined, undefined, { userId: ownerId })) {
|
||
if (event.type === 'text') title += event.text;
|
||
if (event.type === 'error') return fallback;
|
||
if (event.type === 'done') break;
|
||
}
|
||
title = title.replace(/<think>[\s\S]*?<\/think>/g, '').trim();
|
||
title = title.split('\n')[0]?.trim() ?? '';
|
||
return title.slice(0, 60) || fallback;
|
||
} catch {
|
||
return fallback;
|
||
}
|
||
};
|
||
}
|
||
|
||
const piecesDir = resolve(process.cwd(), 'pieces');
|
||
const customPiecesDir = config.customPiecesDir ? resolve(config.customPiecesDir) : undefined;
|
||
logger.info(`[startup] piecesDir=${piecesDir} customPiecesDir=${customPiecesDir ?? 'none'}`);
|
||
|
||
// カスタムディレクトリが設定されていれば作成
|
||
if (customPiecesDir) {
|
||
mkdirSync(customPiecesDir, { recursive: true });
|
||
}
|
||
|
||
// Per-user piece catalog: built-ins loaded once, per-user overrides with 60s TTL cache.
|
||
// The fallback userId 'local' is used when no authenticated user is present (local mode).
|
||
const userFolderRoot = config.userFolderRoot ?? './data/users';
|
||
const pieceCatalog = new PieceCatalog(piecesDir, userFolderRoot);
|
||
logger.info(`[startup] PieceCatalog initialised builtinDir=${piecesDir} dataDir=${userFolderRoot}`);
|
||
|
||
const skillsDir = resolve(process.cwd(), 'data', 'skills');
|
||
mkdirSync(skillsDir, { recursive: true });
|
||
const skillCatalog = new SkillCatalog(skillsDir, userFolderRoot);
|
||
workerManager.setSkillCatalog(skillCatalog);
|
||
logger.info(`[startup] SkillCatalog initialised systemDir=${skillsDir} userRoot=${userFolderRoot}`);
|
||
|
||
// V2 Web Push notifications.
|
||
// Spec: docs/superpowers/specs/2026-05-28-browser-notifications-v2-webpush.md.
|
||
// When `notifications.push.enabled` is false (the default), we still mount
|
||
// the API routes so the UI can see a clear 503 / "管理者により無効化"
|
||
// signal, but workers do not fire any pushes.
|
||
let pushService: PushService | null = null;
|
||
let vapidStore: VapidKeyStore | null = null;
|
||
const pushCfg = config.notifications?.push;
|
||
if (pushCfg?.enabled) {
|
||
const currentPath =
|
||
pushCfg.vapidCurrentPath ?? join(process.cwd(), 'data/secrets/vapid.json');
|
||
const historyDir =
|
||
pushCfg.vapidHistoryDir ?? join(process.cwd(), 'data/secrets/vapid-history');
|
||
const subject = pushCfg.vapidSubject ?? 'https://maestro.example.com/';
|
||
vapidStore = new VapidKeyStore(currentPath, historyDir);
|
||
vapidStore.loadOrGenerate(subject);
|
||
pushService = new PushService(repo, vapidStore, {
|
||
queueConcurrency: pushCfg.queueConcurrency,
|
||
perSendTimeoutMs: pushCfg.perSendTimeoutMs,
|
||
payloadMaxBytes: pushCfg.payloadMaxBytes,
|
||
});
|
||
workerManager.setPushService(pushService);
|
||
logger.info(`[startup] Web Push V2 enabled keyId=${vapidStore.getCurrent().keyId}`);
|
||
} else {
|
||
logger.info('[startup] Web Push V2 disabled (notifications.push.enabled=false)');
|
||
}
|
||
|
||
const selectPiece = titleClient
|
||
? async (body: string, fileNames: string[], userId?: string): Promise<string> => {
|
||
const pieces = pieceCatalog.getForUser(userId ?? 'local');
|
||
const result = await classifyPiece(titleClient!, body, pieces, fileNames, undefined, userId ?? 'local');
|
||
return result ?? 'chat';
|
||
}
|
||
: undefined;
|
||
|
||
// On-demand prompt coach (create-dialog draft evaluation). Reuses the cheap
|
||
// title client and forces the submit_evaluation tool, mirroring reflection.
|
||
const evaluatePrompt = titleClient
|
||
? (r: { instruction: string; piece?: string; userId: string; signal?: AbortSignal }) =>
|
||
runPromptCoach(
|
||
{
|
||
userFolderRoot,
|
||
pieceCatalog,
|
||
skillCatalog,
|
||
readMemory: readMemoryIndex,
|
||
readAgents: readUserAgentsMd,
|
||
callLlm: async ({ system, user, userId, signal }) => {
|
||
const messages: Message[] = [
|
||
{ role: 'system', content: system },
|
||
{ role: 'user', content: user },
|
||
];
|
||
let input: Record<string, unknown> | null = null;
|
||
let errorMsg: string | null = null;
|
||
for await (const event of titleClient!.chat(
|
||
messages,
|
||
[PROMPT_COACH_TOOL_SCHEMA as unknown as ToolDef],
|
||
signal,
|
||
{ userId },
|
||
{
|
||
temperature: 0.2,
|
||
toolChoice: { type: 'function', function: { name: 'submit_evaluation' } },
|
||
},
|
||
) as AsyncGenerator<LLMEvent>) {
|
||
if (event.type === 'tool_use' && event.name === 'submit_evaluation' && input === null) {
|
||
input = event.input;
|
||
} else if (event.type === 'error') {
|
||
errorMsg = event.error;
|
||
}
|
||
}
|
||
if (errorMsg !== null) throw new Error(`prompt-coach LLM ${errorMsg}`);
|
||
if (input === null) throw new Error('prompt-coach LLM returned no submit_evaluation tool_call');
|
||
return input;
|
||
},
|
||
},
|
||
r,
|
||
)
|
||
: undefined;
|
||
|
||
// スケジューラ起動 (selectPiece 注入で 'auto' を実 piece に解決)
|
||
// task_kind='script' をサポートするため、sessRepo / masterKeyPath / userFolderRoot も渡す。
|
||
const sessRepoForScheduler = new BrowserSessionRepo(repo.getDb());
|
||
const scheduler = new Scheduler(repo, config.worktreeDir, {
|
||
selectPiece,
|
||
sessRepo: sessRepoForScheduler,
|
||
masterKeyPath: config.secrets?.masterKeyPath ?? './data/secrets/master.key',
|
||
userFolderRoot: config.userFolderRoot ?? './data/users',
|
||
});
|
||
scheduler.start();
|
||
|
||
// Auto-init MCP encryption key from file (generates on first boot if missing).
|
||
// Must run before startCoreServer so isKeyConfigured() is truthy when routes register.
|
||
try {
|
||
initMcpKeyFromFile(config.secrets?.mcpKeyPath ?? './data/secrets/mcp.key');
|
||
logger.info('[startup] MCP encryption key loaded from file');
|
||
} catch (e) {
|
||
logger.warn(`[startup] MCP encryption key init failed: ${e} — MCP client features will be disabled`);
|
||
}
|
||
|
||
startCoreServer({
|
||
repo,
|
||
worktreeDir: config.worktreeDir,
|
||
configuredRepos: [],
|
||
generateTitle,
|
||
selectPiece,
|
||
evaluatePrompt,
|
||
configManager,
|
||
piecesDir,
|
||
customPiecesDir,
|
||
scheduler,
|
||
authConfig: config.auth,
|
||
brandingDir: join(dirname(dbPath), 'branding'),
|
||
workerManager,
|
||
skillCatalog,
|
||
pushService,
|
||
vapidStore,
|
||
}, port);
|
||
|
||
// Graceful shutdown — see installWorkerShutdownHooks() docstring.
|
||
installWorkerShutdownHooks({
|
||
scheduler,
|
||
workerManager,
|
||
repo,
|
||
});
|
||
|
||
logger.info('maestro ready (mode=worker)');
|
||
}
|
||
|
||
/**
|
||
* Shutdown-hook wiring for worker mode. Extracted so unit tests can
|
||
* verify the registry calls without spinning up the entire boot path.
|
||
*
|
||
* Registers hooks with the shared `shutdown` registry instead of
|
||
* stacking per-subsystem signal listeners. The gateway path uses the
|
||
* same registry, so SIGTERM behaviour is now consistent across modes
|
||
* and future subsystems can `registerShutdownHook` without hitting
|
||
* Node's default MaxListeners cap.
|
||
*
|
||
* The registry runs hooks concurrently (Promise.allSettled), so any
|
||
* hook with internal ordering requirements collapses its sequence
|
||
* into a single closure. Worker drain → repo close is one such case:
|
||
* closing the DB before WorkerManager.stop() finishes risks
|
||
* SQLITE_BUSY on the in-flight job write, so we serialise them inside
|
||
* one hook.
|
||
*/
|
||
export interface WorkerShutdownDeps {
|
||
scheduler: { stop(): void };
|
||
/**
|
||
* WorkerManager.stop() returns { drained, requeued } as of the AAO Gateway
|
||
* refactor, but this shutdown hook discards the result. Widen to
|
||
* `Promise<unknown>` so the structural type accepts any stop() return shape
|
||
* without recreating a coupling to WorkerManager's concrete type here.
|
||
*/
|
||
workerManager: { stop(): Promise<unknown> };
|
||
repo: { close(): void };
|
||
}
|
||
|
||
export function installWorkerShutdownHooks(deps: WorkerShutdownDeps): void {
|
||
registerShutdownHook('worker-scheduler', async () => {
|
||
deps.scheduler.stop();
|
||
});
|
||
registerShutdownHook('worker-manager+repo', async () => {
|
||
await deps.workerManager.stop();
|
||
deps.repo.close();
|
||
});
|
||
installSignalHandlers();
|
||
}
|