feat(sdk): duties — periodic work that exactly one process on a machine or workspace runs

apps/e2e/demo-e2e-bundle-size/e2e/browser-bundle-size.e2e.spec.ts

@@ -24,8 +24,8 @@ import {
 } from './helpers/browser-bundle';
 
 const BUDGETS: Array<{ entry: string; label: string; maxBytes: number }> = [
-  { entry: 'create-tool.ts', label: 'create() + one tool', maxBytes: 1_323_819 },
-  { entry: 'create-webmcp-tool.ts', label: 'create() + WebMcpPlugin + one tool', maxBytes: 1_330_696 },
+  { entry: 'create-tool.ts', label: 'create() + one tool', maxBytes: 1_325_128 },
+  { entry: 'create-webmcp-tool.ts', label: 'create() + WebMcpPlugin + one tool', maxBytes: 1_332_004 },
 ];
 
 const VITE_INITIAL_CHUNK_MAX_BYTES = 960_000;
@@ -46,6 +46,7 @@ const FEATURE_MODULES: Record<string, RegExp> = {
   ),
   channels: sdkModule('channel/(?!channel\\.utils)'),
   'distributed transport': sdkModule('ha/|transport/(bus|relay|event-stores)/'),
+  duties: sdkModule('duty/'),
   elicitation: sdkModule('elicitation/(store|flows)/|elicitation/send-elicitation-result\\.tool'),
   health: sdkModule('health/'),
   'http transport': sdkModule(

apps/e2e/demo-e2e-dts-consumer/e2e/dts-consumer.e2e.spec.ts

@@ -43,6 +43,7 @@ const fixtureFiles = [
   'hook-ordering-must-compile.ts',
   'hook-ordering-must-fail.ts',
   'live-surface-must-compile.ts',
+  'duties-must-compile.ts',
   'offload-must-compile.ts',
 ].map((fileName) => path.join(fixtureDir, fileName));
 
@@ -204,6 +205,10 @@ describe('@frontmcp/sdk declarations in a strict consumer project', () => {
     expect(describeEach(inFixtureFile('live-surface-must-compile.ts'))).toEqual([]);
   });
 
+  it('types @Duty, its metadata and run context, and createContainer({ runDuties, info })', () => {
+    expect(describeEach(inFixtureFile('duties-must-compile.ts'))).toEqual([]);
+  });
+
   it('types this.offload() from the offloaded function, @Resource({ worker }) and the offload option', () => {
     expect(describeEach(inFixtureFile('offload-must-compile.ts'))).toEqual([]);
   });

apps/e2e/demo-e2e-dts-consumer/fixture/duties-must-compile.ts

@@ -0,0 +1,49 @@
+import 'reflect-metadata';
+
+import {
+  App,
+  createContainer,
+  Duty,
+  FrontMcp,
+  type DutyInterface,
+  type DutyMetadata,
+  type DutyRunContext,
+} from '@frontmcp/sdk';
+
+@Duty({ every: '10m', scope: 'machine' })
+export class PruneCacheDuty implements DutyInterface {
+  async run({ signal }: DutyRunContext): Promise<void> {
+    if (signal.aborted) return;
+  }
+}
+
+const nightly: DutyMetadata = { name: 'reindex', every: 86_400_000, scope: 'workspace', leaseMs: 30_000 };
+
+@Duty(nightly)
+export class ReindexDuty {
+  run(): void {
+    return undefined;
+  }
+}
+
+@App({ name: 'maintenance', providers: [PruneCacheDuty, ReindexDuty] })
+class MaintenanceApp {}
+
+@FrontMcp({ info: { name: 'duties', version: '1.0.0' }, apps: [MaintenanceApp] })
+export class DutiesServer {}
+
+// @ts-expect-error a duty needs a run() method
+@Duty({ every: '1m', scope: 'machine' })
+export class DutyWithoutRun {}
+
+export class InvalidDuties {
+  // @ts-expect-error scope is 'machine' or 'workspace'
+  static readonly scope: DutyMetadata = { every: '1m', scope: 'galaxy' };
+  // @ts-expect-error every is milliseconds or a duration string
+  static readonly every: DutyMetadata = { every: true, scope: 'machine' };
+}
+
+export async function runDutiesInAScript(): Promise<void> {
+  const container = await createContainer({ apps: [MaintenanceApp], runDuties: true, info: { name: 'duties' } });
+  await container.dispose();
+}

apps/e2e/demo-e2e-dts-consumer/fixture/tsconfig.json

@@ -30,6 +30,7 @@
     "call-recorder-must-compile.ts",
     "hook-ordering-must-compile.ts",
     "hook-ordering-must-fail.ts",
-    "live-surface-must-compile.ts"
+    "live-surface-must-compile.ts",
+    "duties-must-compile.ts"
   ]
 }

apps/e2e/demo-e2e-local-runtime/e2e/duties.e2e.spec.ts

@@ -0,0 +1,98 @@
+/**
+ * E2E: five servers on one machine share a `@Duty`'s lease. One of them runs it each interval; when it is killed,
+ * another takes over once the lease runs out, and when it stops gracefully it hands the lease over at once.
+ */
+import * as fs from 'node:fs';
+import * as os from 'node:os';
+import * as path from 'node:path';
+
+import { killAll, spawnWorker, type Worker } from './helpers/workers';
+
+const SERVERS = 5;
+const EVERY_MS = 500;
+const LEASE_MS = 12_000;
+
+interface Run {
+  pid: number;
+  at: number;
+}
+
+function runsOf(workers: Worker<unknown>[]): Run[] {
+  return workers
+    .flatMap(({ lines }) => lines.filter((line) => line.startsWith('ran ')))
+    .map((line) => {
+      const [, pid, at] = line.split(' ');
+      return { pid: Number(pid), at: Number(at) };
+    })
+    .sort((a, b) => a.at - b.at);
+}
+
+function workerOf(workers: Worker<unknown>[], pid: number): Worker<unknown> {
+  const found = workers.find(({ child }) => child.pid === pid);
+  if (!found) throw new Error(`no worker with pid ${pid}`);
+  return found;
+}
+
+async function waitUntil(condition: () => boolean, timeoutMs: number, what: string): Promise<void> {
+  const deadline = Date.now() + timeoutMs;
+  while (!condition()) {
+    if (Date.now() > deadline) throw new Error(`timed out waiting for ${what}`);
+    await new Promise((resolve) => setTimeout(resolve, 50));
+  }
+}
+
+/** One process ran every run, and no two runs came closer than half an interval. */
+function expectOneHolderOncePerInterval(runs: Run[]): void {
+  expect(new Set(runs.map(({ pid }) => pid)).size).toBe(1);
+  for (let index = 1; index < runs.length; index++) {
+    expect(runs[index].at - runs[index - 1].at).toBeGreaterThanOrEqual(EVERY_MS / 2);
+  }
+}
+
+describe('@Duty across processes', () => {
+  let home: string;
+  let started: Worker<unknown>[] = [];
+
+  beforeEach(() => {
+    home = fs.mkdtempSync(path.join(os.tmpdir(), 'local-runtime-duties-'));
+  });
+
+  afterEach(() => {
+    killAll(started);
+    fs.rmSync(home, { recursive: true, force: true });
+  });
+
+  it('runs once per interval in one of five servers, and another takes over when the holder dies or stops', async () => {
+    const workers = Array.from({ length: SERVERS }, () =>
+      spawnWorker('duty-server.ts', { FRONTMCP_HOME: home, EVERY_MS: String(EVERY_MS), LEASE_MS: String(LEASE_MS) }),
+    );
+    started = workers;
+    await Promise.all(workers.map((worker) => worker.waitFor('ready')));
+
+    await waitUntil(() => runsOf(workers).length >= 6, 30_000, 'six runs');
+    const beforeKill = runsOf(workers);
+    expectOneHolderOncePerInterval(beforeKill);
+    if (process.platform !== 'win32') expect(fs.statSync(path.join(home, 'duties.sqlite')).mode & 0o777).toBe(0o600);
+
+    const first = workerOf(workers, beforeKill[0].pid);
+    const killedAt = Date.now();
+    first.child.kill('SIGKILL');
+    const survivors = workers.filter((worker) => worker !== first);
+    await waitUntil(() => runsOf(survivors).length >= 4, LEASE_MS * 3, 'a takeover after the holder was killed');
+    const afterKill = runsOf(survivors);
+    expect(afterKill[0].at - killedAt).toBeLessThan(LEASE_MS * 2);
+    expectOneHolderOncePerInterval(afterKill);
+
+    const second = workerOf(survivors, afterKill[0].pid);
+    const terminatedAt = Date.now();
+    second.child.kill('SIGTERM');
+    await expect(second.result).resolves.toEqual({ stopped: true });
+    const lastRunOfSecond = runsOf([second]).at(-1)?.at ?? 0;
+    const rest = survivors.filter((worker) => worker !== second);
+    await waitUntil(() => runsOf(rest).length >= 2, LEASE_MS * 3, 'a takeover after the holder stopped');
+    const afterStop = runsOf(rest);
+    expect(afterStop[0].at - terminatedAt).toBeLessThan((LEASE_MS * 2) / 3);
+    expect(afterStop[0].at - lastRunOfSecond).toBeGreaterThanOrEqual(EVERY_MS / 2);
+    expectOneHolderOncePerInterval(afterStop);
+  });
+});

apps/e2e/demo-e2e-local-runtime/e2e/helpers/workers.ts

@@ -6,6 +6,8 @@ const projectRoot = join(__dirname, '..', '..');
 
 export interface Worker<Result> {
   readonly child: ChildProcess;
+  /** Every complete line the worker printed so far. */
+  readonly lines: readonly string[];
   /** Resolves when the worker prints `line`; rejects if it exits first. */
   waitFor(line: string): Promise<void>;
   send(line: string): void;
@@ -53,6 +55,7 @@ export function spawnWorker<Result>(script: string, env: Record<string, string>
 
   return {
     child,
+    lines,
     result,
     send: (line) => child.stdin?.write(`${line}\n`),
     waitFor: (line) =>

apps/e2e/demo-e2e-local-runtime/fixture/duty-server.ts

@@ -0,0 +1,37 @@
+/** A server with one machine duty that prints `ran <pid> <ms>` on each run; SIGTERM shuts it down gracefully. */
+import 'reflect-metadata';
+
+import { App, Duty, FrontMcpInstance, LogLevel } from '@frontmcp/sdk';
+
+import { report, requiredEnv, runWorker, signal } from './worker-protocol';
+
+@Duty({
+  name: 'tick',
+  every: Number(requiredEnv('EVERY_MS')),
+  scope: 'machine',
+  leaseMs: Number(requiredEnv('LEASE_MS')),
+})
+class TickDuty {
+  run(): void {
+    signal(`ran ${process.pid} ${Date.now()}`);
+  }
+}
+
+@App({ name: 'duties', providers: [TickDuty] })
+class DutiesApp {}
+
+runWorker(async () => {
+  const server = await FrontMcpInstance.createDirect({
+    info: { name: 'duty-e2e', version: '1.0.0' },
+    apps: [DutiesApp],
+    logging: { level: LogLevel.Off, enableConsole: false },
+  });
+  const keepAlive = setInterval(() => undefined, 60_000);
+  const terminated = new Promise<void>((resolve) => process.once('SIGTERM', () => resolve()));
+  signal('ready');
+
+  await terminated;
+  await server.dispose();
+  clearInterval(keepAlive);
+  report({ stopped: true });
+});

libs/sdk/src/common/decorators/duty.decorator.ts

@@ -0,0 +1,29 @@
+import 'reflect-metadata';
+
+import { InvalidDecoratorMetadataError } from '../../errors/decorator.errors';
+import { type DutyInterface } from '../interfaces/duty.interface';
+import { defineDutyMetadata, dutyMetadataSchema, type DutyMetadata } from '../metadata/duty.metadata';
+import { FrontMcpProviderTokens } from '../tokens';
+
+type DutyClass = abstract new (...args: never[]) => DutyInterface;
+
+/**
+ * Marks a provider class as a duty: once the server serves, `run()` is called every `every` in exactly one process
+ * of the machine or project (`scope`), the one that holds the duty's lease. Register it in `providers`.
+ */
+export function Duty(metadata: DutyMetadata): (target: DutyClass) => void {
+  return (target) => {
+    const result = dutyMetadataSchema.safeParse(metadata);
+    if (!result.success) {
+      const [issue] = result.error.issues;
+      throw new InvalidDecoratorMetadataError('Duty', issue.path.join('.'), issue.message);
+    }
+    if (typeof target.prototype?.run !== 'function') {
+      throw new InvalidDecoratorMetadataError('Duty', 'run', `${target.name} has no run() method`);
+    }
+    const { name = target.name, every: everyMs, scope, leaseMs } = result.data;
+    const defaultLeaseMs = Math.min(Math.max(3 * everyMs, 3_000), 60_000);
+    Reflect.defineMetadata(FrontMcpProviderTokens.type, true, target);
+    defineDutyMetadata(target, { name, everyMs, scope, leaseMs: leaseMs ?? defaultLeaseMs });
+  };
+}

libs/sdk/src/common/decorators/index.ts

@@ -15,3 +15,4 @@ export * from './skill.decorator';
 export * from './job.decorator';
 export * from './workflow.decorator';
 export * from './channel.decorator';
+export * from './duty.decorator';

libs/sdk/src/common/interfaces/duty.interface.ts

@@ -0,0 +1,10 @@
+/** What a duty's `run()` receives. */
+export interface DutyRunContext {
+  /** Aborted when this process loses the duty's lease or shuts down: stop the run early when it fires. */
+  readonly signal: AbortSignal;
+}
+
+/** A `@Duty` class: `run()` is called every interval in the one process that holds the duty's lease. */
+export interface DutyInterface {
+  run(context: DutyRunContext): void | Promise<void>;
+}

libs/sdk/src/common/interfaces/index.ts

@@ -37,5 +37,6 @@ export * from './skill.interface';
 export * from './job.interface';
 export * from './workflow.interface';
 export * from './channel.interface';
+export * from './duty.interface';
 
 export * from './internal';

libs/sdk/src/common/metadata/duty.metadata.ts

@@ -0,0 +1,67 @@
+import { z } from '@frontmcp/lazy-zod';
+
+/** Where one process at a time runs a duty: one per machine, or one per project root (the nearest package.json). */
+export type DutyScope = 'machine' | 'workspace';
+
+export interface DutyMetadata {
+  /** The lease's name within the server. Set it when a build renames classes. @default the class name */
+  name?: string;
+  /** How often the duty runs: milliseconds, or a duration such as `'500ms'`, `'30s'`, `'5m'`, `'1h'` or `'1d'`. */
+  every: number | string;
+  scope: DutyScope;
+  /**
+   * How long the lease outlives a holder that stopped renewing it (killed, paused, hung), so how soon another process
+   * takes the duty over. Renewed every third of it. @default three intervals, between 3000 and 60000
+   */
+  leaseMs?: number;
+}
+
+export interface ResolvedDutyMetadata {
+  name: string;
+  everyMs: number;
+  scope: DutyScope;
+  leaseMs: number;
+}
+
+const DUTY_METADATA = Symbol.for('frontmcp:duty');
+const MAX_TIMER_MS = 2_147_483_647;
+const UNIT_MS: Record<string, number> = { ms: 1, s: 1_000, m: 60_000, h: 3_600_000, d: 86_400_000 };
+const EVERY_FORMAT = 'every must be milliseconds or a duration such as "30s", "5m", "1h"';
+
+function toMilliseconds(every: number | string): number {
+  if (typeof every === 'number') return every;
+  const match = /^(\d+)(ms|s|m|h|d)$/.exec(every.trim());
+  return match ? Number(match[1]) * UNIT_MS[match[2]] : Number.NaN;
+}
+
+function wholeMilliseconds(field: string) {
+  return z
+    .number({ message: field === 'every' ? EVERY_FORMAT : `${field} must be a number of milliseconds` })
+    .int(`${field} must be a whole number of milliseconds`)
+    .max(MAX_TIMER_MS, `${field} must be at most ${MAX_TIMER_MS} ms (about 24.8 days)`);
+}
+
+export const dutyMetadataSchema = z
+  .object({
+    name: z.string({ message: 'name must be a string' }).min(1, 'name must not be empty').optional(),
+    every: z
+      .union([z.number(), z.string()], { message: EVERY_FORMAT })
+      .transform(toMilliseconds)
+      .pipe(wholeMilliseconds('every').positive('every must be greater than 0')),
+    scope: z.enum(['machine', 'workspace'], { message: "scope must be 'machine' or 'workspace'" }),
+    leaseMs: wholeMilliseconds('leaseMs')
+      .min(1_000, 'leaseMs must be at least 1000 (stores keep a TTL in whole seconds)')
+      .optional(),
+  })
+  .strict();
+
+/** @internal Records what `@Duty` resolved on the class. */
+export function defineDutyMetadata(target: object, metadata: ResolvedDutyMetadata): void {
+  Reflect.defineMetadata(DUTY_METADATA, metadata, target);
+}
+
+/** The `@Duty` metadata of an instance's class, when it has some. */
+export function dutyMetadataOf(value: unknown): ResolvedDutyMetadata | undefined {
+  if (typeof value !== 'object' || value === null) return undefined;
+  return Reflect.getMetadata(DUTY_METADATA, value.constructor);
+}

libs/sdk/src/common/metadata/index.ts

@@ -25,3 +25,4 @@ export * from './workflow.metadata';
 export * from './channel.metadata';
 export * from './app-filter.metadata';
 export * from './remote-primitive.metadata';
+export * from './duty.metadata';

libs/sdk/src/common/types/options/sqlite/resolve.ts

@@ -68,7 +68,8 @@ function sanitizeForFs(name: string | undefined): string {
   return name.replace(/[^a-zA-Z0-9._-]/g, '-').replace(/^[-.]+/, '');
 }
 
-function findProjectRoot(start: string): string | undefined {
+/** The nearest directory above `start`, itself included, that holds a package.json. */
+export function findProjectRoot(start: string): string | undefined {
   let dir = start;
   // Cap the walk at 32 levels to bound worst-case (sandbox / symlink loops).
   // `@frontmcp/utils.readFileSync` is the sync FS escape hatch per the