Repository Public
1/**2 * The worker process (CONTRACT §7): claims runs, executes the engine, journals steps. No listener.3 *4 * Logs are structured (pino) and carry identifiers, statuses and counts only — never prompts, model5 * output, tool input/output or keys.6 */7import { agentLogic, sweepRunTmp } from '@aos/engine';8import { configureKernel, createKernel, migrate, runWorkerLoop, type Kernel } from '@aos/kernel';9import { seedPrices } from '@aos/providers';10import { ENGINE_VERSION } from '@aos/shared';11import { pino, type Logger } from 'pino';12import type { WorkerConfig } from './config.js';1314const SWEEP_MS = 10 * 60_000;1516export interface RunningWorker {17 kernel: Kernel;18 /** Graceful stop: no new claim, in-flight runs are released (left `running` with a lease another worker takes over). */19 stop(): Promise<void>;20 /** Resolves when the loop has stopped. */21 done: Promise<void>;22}2324export function createLogger(level: string): Logger {25 return pino({26 name: 'aos-worker',27 level,28 // Belt and braces: nothing below is logged on purpose, but a stray object must not leak a secret.29 redact: { paths: ['apiKey', '*.apiKey', 'authorization', '*.authorization', 'headers', '*.headers', 'env', '*.env', 'databaseUrl', '*.databaseUrl'], censor: '[redacted]' },30 });31}3233export async function startWorker(cfg: WorkerConfig, log: Logger = createLogger(cfg.logLevel)): Promise<RunningWorker> {34 const kernel = createKernel({ databaseUrl: cfg.databaseUrl, dataDir: cfg.dataDir, logger: log.child({ component: 'kernel' }), poolMax: cfg.concurrency * 3 + 6 });35 configureKernel(kernel);36 try {37 const m = await migrate(cfg.databaseUrl);38 await seedPrices(kernel.pool);39 log.info({ migrations: m, workerId: cfg.workerId, concurrency: cfg.concurrency, providers: cfg.providers, engine: ENGINE_VERSION }, 'worker starting');40 } catch (err) {41 await kernel.close().catch(() => undefined);42 throw err;43 }44 if (!cfg.workspaceRoot) log.warn('AOS_WORKSPACE_ROOT is not set: sibling projects are not hidden from the sandboxed shell');45 for (const [name, ok] of Object.entries(cfg.providers)) {46 if (!ok) log.warn({ provider: name }, 'provider key missing: missions using this provider will end blocked');47 }4849 const sweep = (): void => {50 sweepRunTmp(kernel).then(51 (removed) => removed.length > 0 && log.info({ runs: removed.length }, 'removed temp dirs of finished runs'),52 (err: Error) => log.debug({ err: err.message }, 'temp dir sweep failed'),53 );54 };55 sweep();56 const sweeper = setInterval(sweep, SWEEP_MS);57 sweeper.unref();5859 const ac = new AbortController();60 const done = runWorkerLoop({61 kernel,62 logic: agentLogic({ kernel, workspaceRoot: cfg.workspaceRoot ?? undefined }),63 concurrency: cfg.concurrency,64 workerId: cfg.workerId,65 signal: ac.signal,66 version: ENGINE_VERSION,67 // expireWaits() runs inside the loop every 30 s (the janitor of CONTRACT §7)68 }).finally(() => clearInterval(sweeper));6970 return {71 kernel,72 done,73 stop: async () => {74 ac.abort();75 await done;76 await kernel.close().catch(() => undefined);77 },78 };79}80