feat(automation): accept a per-kernel scheduledWorkPolicy in the engine and both schedule triggers

.changeset/19834-per-kernel-scheduled-work-policy.md

@@ -0,0 +1,23 @@
+---
+'@objectstack/service-automation': minor
+'@objectstack/trigger-schedule': minor
+---
+
+A kernel can now carry its own scheduled-work policy. `AutomationEngineOptions`, `AutomationServicePluginOptions` (forwarded to the engine), `ScheduleTriggerPlugin` and `TimeRelativeTriggerPlugin` accept an optional `scheduledWorkPolicy`: a `ScheduledWorkPolicy` value, or a resolver called at each bind. When it is present, the engine's bind gate and each trigger's own gate read it instead of the deployment resolver. When it is absent, they call the zero-argument `resolveScheduledWorkPolicy()` exactly as before.
+
+It exists for a host that runs several kernels of different plans in one process. `OS_AUTOMATION_SCHEDULED_WORK_ENABLED` is one reading for the whole process, so such a host could not turn scheduled work off for one kernel and leave it on for the kernel beside it:
+
+```ts
+const policy = { enabled: false, posture: 'single', requiresActingOrganization: false, runOwnership: 'unscoped' } as const;
+kernel.use(new AutomationServicePlugin({ scheduledWorkPolicy: policy }));
+kernel.use(new ScheduleTriggerPlugin({ scheduledWorkPolicy: policy }));
+kernel.use(new TimeRelativeTriggerPlugin({ scheduledWorkPolicy: policy }));
+```
+
+Give all three the same policy. The engine gates first, and each trigger keeps its own gate for hosts that drive it without the engine. If a trigger reads a different answer from its engine, the engine's audit reports that trigger's refusal as a binding failure. A hand-built value must keep the resolver's invariant, `requiresActingOrganization === (enabled && runOwnership === 'declared')`.
+
+A time-triggered flow that the per-kernel policy leaves unarmed is reported the same way as one the deployment leaves unarmed. `getTriggerBindingAudit()` and the `getFlowRuntimeStates()` row both give `SCHEDULED_WORK_DISABLED_REASON`, never a binding failure and never "add `requires: ['triggers']`". `ScheduleTrigger` and `TimeRelativeTrigger` also accept the same option in a new trailing constructor argument. The types `ScheduleTriggerPluginOptions`, `TimeRelativeTriggerPluginOptions`, `ScheduledWorkTriggerOptions` and `ScheduledWorkPolicySource` are exported from `@objectstack/trigger-schedule`.
+
+Nothing changes for a host that passes no policy. The deployment default keeps its meaning and its spelling, and `objectstack serve` is unchanged. This change only adds options, so there is nothing to migrate.
+
+Clause-②: no

packages/services/service-automation/src/engine.ts

@@ -28,6 +28,7 @@ import { resolveFlowTriggerKind, resolveScheduleOrganization } from '@objectstac
 import {
     resolveScheduledWorkPolicy,
     SCHEDULED_WORK_DISABLED_REASON,
+    type ScheduledWorkPolicy,
 } from '@objectstack/types';
 import { predicateSlotRefusal, resolveFlowNodeExpressions, structuralConditionRefusal } from '@objectstack/spec/automation';
 // [#15137] The `value`-role half of the ledger. Both halves of "is this envelope
@@ -851,6 +852,31 @@ export interface AutomationEngineOptions {
      * See {@link RunSummaryLogLevel}.
      */
     runSummaryLog?: RunSummaryLogLevel;
+    /**
+     * [#19834] THIS kernel's scheduled-work policy — a value, or a resolver
+     * called at each bind. Absent (the default), the engine reads the
+     * zero-argument deployment resolver `resolveScheduledWorkPolicy()` exactly
+     * as before, so `OS_AUTOMATION_SCHEDULED_WORK_ENABLED` keeps its meaning.
+     *
+     * For a host that runs several kernels in one process whose plans differ
+     * (one scheduled work OFF, its sibling ON): the per-kernel answer has
+     * nowhere else to live, because the deployment resolver reads one
+     * process-wide environment. A time-triggered flow this policy leaves
+     * unarmed is reported exactly as a deployment-disabled one —
+     * `SCHEDULED_WORK_DISABLED_REASON` on the binding audit and the status row.
+     *
+     * ⚠️ Hand the SAME policy to `ScheduleTriggerPlugin` and
+     * `TimeRelativeTriggerPlugin` of the same kernel: each trigger keeps its own
+     * gate (for a host that drives it without this engine), and a trigger
+     * reading a different answer from this engine refuses what the engine let
+     * through, which the audit then reports as a binding failure.
+     *
+     * ⚠️ A hand-built value owes the resolver's invariant:
+     * `requiresActingOrganization === enabled && runOwnership === 'declared'`.
+     * Spreading the deployment's reading and overriding only `enabled` breaks
+     * it under an `isolated` posture.
+     */
+    scheduledWorkPolicy?: ScheduledWorkPolicy | (() => ScheduledWorkPolicy);
 }
 
 /**
@@ -2376,6 +2402,8 @@ export class AutomationEngine implements IAutomationService {
     private readonly maxLogSize: number;
     /** Level for the per-run summary line (#4354). See {@link RunSummaryLogLevel}. */
     private readonly runSummaryLog: RunSummaryLogLevel;
+    /** [#19834] Per-kernel policy override; see {@link AutomationEngineOptions.scheduledWorkPolicy}. */
+    private readonly scheduledWorkPolicy: AutomationEngineOptions['scheduledWorkPolicy'];
     private logger: Logger;
     /**
      * Runs paused at a node, keyed by runId (ADR-0019). Process-local copy of
@@ -2493,6 +2521,18 @@ export class AutomationEngine implements IAutomationService {
         this.store = store;
         this.maxLogSize = options?.maxLogSize ?? DEFAULT_MAX_EXECUTION_LOG_SIZE;
         this.runSummaryLog = options?.runSummaryLog ?? 'info';
+        this.scheduledWorkPolicy = options?.scheduledWorkPolicy;
+    }
+
+    /**
+     * [#19834] The scheduled-work policy for THIS engine, read at each bind:
+     * the per-kernel option when one was given, else the zero-argument
+     * deployment resolver, unchanged.
+     */
+    private readScheduledWorkPolicy(): ScheduledWorkPolicy {
+        const source = this.scheduledWorkPolicy;
+        if (source === undefined) return resolveScheduledWorkPolicy();
+        return typeof source === 'function' ? source() : source;
     }
 
     /**
@@ -3511,12 +3551,14 @@ export class AutomationEngine implements IAutomationService {
         // ⛔ The POLICY is not cached — the resolver reads `process.env` live,
         // so a host that rebinds after changing the environment (the CLI's
         // `--fresh` harness, a test flipping the switch between kernels in one
-        // process) sees the value current at the bind. What IS recorded is the
+        // process) sees the value current at the bind. [#19834] A per-kernel
+        // `scheduledWorkPolicy` option replaces that read for this engine only;
+        // a resolver given there is likewise called at each bind. What IS recorded is the
         // REFUSAL, on {@link policyDisabledFlows}, and that is the difference
         // between the two: see its own docblock for why the audit must read
         // what happened rather than re-derive it from an environment that may
         // have moved since.
-        if (isTimeTriggeredKind(resolved.triggerType) && !resolveScheduledWorkPolicy().enabled) {
+        if (isTimeTriggeredKind(resolved.triggerType) && !this.readScheduledWorkPolicy().enabled) {
             if (!this.policyDisabledFlows.has(flowName)) {
                 this.policyDisabledFlows.add(flowName);
                 // Said once per flow while it stays refused, at `info`, for the

packages/services/service-automation/src/per-kernel-scheduled-work-policy.test.ts

@@ -0,0 +1,258 @@
+// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
+
+/**
+ * [#19834] A per-kernel `ScheduledWorkPolicy` on the engine — the engine half.
+ *
+ * `resolveScheduledWorkPolicy()` reads ONE process-wide environment, so a host
+ * running several kernels of different plans in one Node process had nowhere
+ * to say "this kernel: scheduled work OFF". `AutomationEngineOptions` (and
+ * `AutomationServicePluginOptions`, which forwards it) now carry an optional
+ * `scheduledWorkPolicy` — a value or a resolver — read in place of the
+ * deployment resolver when present.
+ *
+ * What is pinned here, all against the engine SOURCE (this package aliases
+ * nothing it tests):
+ *
+ *   1. two engines in one process disagree exactly as their options say, in
+ *      BOTH deployment states — the option outranks the environment both ways;
+ *   2. a policy-unarmed flow is reported exactly as a deployment-unarmed one
+ *      (the policy sentence, ⛔ never "add requires: ['triggers']");
+ *   3. a resolver is called at each bind, not once;
+ *   4. the zero-argument path is unchanged: no option ⇒ the environment.
+ *
+ * The acceptance case with the REAL schedule triggers lives in
+ * `@objectstack/trigger-schedule`'s `per-kernel-scheduled-work-policy.test.ts`.
+ */
+
+import { describe, it, expect, afterEach } from 'vitest';
+import { LiteKernel } from '@objectstack/core';
+import type { Plugin, PluginContext } from '@objectstack/core';
+import { AutomationEngine } from './engine.js';
+import { AutomationServicePlugin } from './plugin.js';
+import {
+    SCHEDULED_WORK_ENV,
+    SCHEDULED_WORK_DISABLED_REASON,
+    type ScheduledWorkPolicy,
+} from '@objectstack/types';
+
+const POSTURE_ENV = 'OS_TENANCY_POSTURE';
+
+const OFF: ScheduledWorkPolicy = {
+    enabled: false,
+    posture: 'single',
+    requiresActingOrganization: false,
+    runOwnership: 'unscoped',
+};
+const ON: ScheduledWorkPolicy = { ...OFF, enabled: true };
+
+function silentLogger() {
+    const l: any = { info() {}, warn() {}, error() {}, debug() {} };
+    l.child = () => l;
+    return l;
+}
+
+function scheduleFlow(name: string) {
+    return {
+        name,
+        label: name,
+        type: 'schedule' as const,
+        status: 'active',
+        nodes: [
+            {
+                id: 'start',
+                type: 'start' as const,
+                label: 'Start',
+                config: { schedule: { type: 'cron', expression: '0 8 * * *' } },
+            },
+            { id: 'end', type: 'end' as const, label: 'End' },
+        ],
+        edges: [{ id: 'e1', source: 'start', target: 'end' }],
+    };
+}
+
+/** A `schedule` trigger that records every binding it is asked to arm. */
+function recordingTrigger() {
+    const started: string[] = [];
+    return {
+        started,
+        trigger: {
+            type: 'schedule',
+            start(binding: { flowName: string }) {
+                started.push(binding.flowName);
+            },
+            stop() {},
+        },
+    };
+}
+
+/** Engine + trigger + one schedule flow; returns what the trigger was asked to arm. */
+function bindOne(options?: ConstructorParameters<typeof AutomationEngine>[2]) {
+    const engine = new AutomationEngine(silentLogger(), undefined, options);
+    const rec = recordingTrigger();
+    engine.registerTrigger(rec.trigger);
+    engine.registerFlow('digest', scheduleFlow('digest'));
+    return { engine, started: rec.started };
+}
+
+function withDeployment(value: string | undefined) {
+    if (value === undefined) delete process.env[SCHEDULED_WORK_ENV];
+    else process.env[SCHEDULED_WORK_ENV] = value;
+    delete process.env[POSTURE_ENV];
+}
+
+describe('AutomationEngine — a per-kernel scheduled-work policy (#19834)', () => {
+    const PRIOR_SWITCH = process.env[SCHEDULED_WORK_ENV];
+    const PRIOR_POSTURE = process.env[POSTURE_ENV];
+    afterEach(() => {
+        if (PRIOR_SWITCH === undefined) delete process.env[SCHEDULED_WORK_ENV];
+        else process.env[SCHEDULED_WORK_ENV] = PRIOR_SWITCH;
+        if (PRIOR_POSTURE === undefined) delete process.env[POSTURE_ENV];
+        else process.env[POSTURE_ENV] = PRIOR_POSTURE;
+    });
+
+    for (const [label, deployment] of [
+        ['deployment OFF (unset)', undefined],
+        ['deployment ON', 'true'],
+    ] as const) {
+        it(`${label}: two engines in one process — OFF arms nothing, its ON sibling arms`, () => {
+            withDeployment(deployment);
+            const off = bindOne({ scheduledWorkPolicy: OFF });
+            const on = bindOne({ scheduledWorkPolicy: ON });
+
+            expect(off.started, 'the OFF kernel called its trigger').toEqual([]);
+            expect(on.started, 'the ON kernel did not arm its flow').toEqual(['digest']);
+
+            // The refusal is structured exactly as a deployment refusal is.
+            const audit = off.engine.getTriggerBindingAudit();
+            expect(audit.map((a) => a.flowName)).toEqual(['digest']);
+            expect(audit[0].reason).toBe(SCHEDULED_WORK_DISABLED_REASON);
+            expect(audit[0].reason).not.toMatch(/requires: \['triggers'\]/);
+            expect(audit[0].reason).not.toMatch(/binding failed/);
+            const row = off.engine.getFlowRuntimeStates().find((s) => s.name === 'digest');
+            expect(row).toMatchObject({ enabled: true, bound: false, triggerType: 'schedule' });
+            expect(row?.reason).toBe(SCHEDULED_WORK_DISABLED_REASON);
+
+            expect(on.engine.getTriggerBindingAudit()).toEqual([]);
+        });
+    }
+
+    it('a policy-unarmed flow with NO trigger registered still names the policy, never the missing trigger', () => {
+        // The ordering #17396 set for the deployment switch holds for the
+        // per-kernel one: with the policy off, registering the trigger would
+        // change nothing, so "add requires: ['triggers']" is the wrong remedy.
+        withDeployment('true');
+        const engine = new AutomationEngine(silentLogger(), undefined, { scheduledWorkPolicy: OFF });
+        engine.registerFlow('digest', scheduleFlow('digest'));
+        const audit = engine.getTriggerBindingAudit();
+        expect(audit).toHaveLength(1);
+        expect(audit[0].reason).toBe(SCHEDULED_WORK_DISABLED_REASON);
+        expect(audit[0].reason).not.toMatch(/requires: \['triggers'\]/);
+    });
+
+    it('a resolver is called at each bind, not once at construction', () => {
+        withDeployment(undefined);
+        let current: ScheduledWorkPolicy = OFF;
+        let calls = 0;
+        const engine = new AutomationEngine(silentLogger(), undefined, {
+            scheduledWorkPolicy: () => {
+                calls += 1;
+                return current;
+            },
+        });
+        expect(calls, 'nothing is read before a bind').toBe(0);
+        const rec = recordingTrigger();
+        engine.registerTrigger(rec.trigger);
+        engine.registerFlow('a', scheduleFlow('a'));
+        expect(rec.started).toEqual([]);
+
+        current = ON;
+        engine.registerFlow('b', scheduleFlow('b'));
+        expect(rec.started).toEqual(['b']);
+        expect(calls).toBeGreaterThanOrEqual(2);
+    });
+
+    it('ZERO-ARGUMENT PATH UNCHANGED: no option ⇒ the deployment environment decides', () => {
+        withDeployment(undefined);
+        expect(bindOne().started, 'unset switch must still refuse').toEqual([]);
+        expect(bindOne({}).started, 'an options bag without the key must still refuse').toEqual([]);
+        expect(bindOne({ scheduledWorkPolicy: undefined }).started).toEqual([]);
+
+        withDeployment('true');
+        expect(bindOne().started, 'switch on must still arm').toEqual(['digest']);
+        expect(bindOne({ maxLogSize: 10 }).started).toEqual(['digest']);
+    });
+});
+
+// ─── the plugin forwards it — two kernels, one process ──────────────
+
+/** The `objectql` seam the boot flow-pull reads (see flow-node-type-audit.test.ts). */
+function fakeObjectqlPlugin(flows: unknown[]): Plugin {
+    return {
+        name: 'fake-objectql',
+        version: '1.0.0',
+        async init(ctx: PluginContext) {
+            (ctx as unknown as { registerService(n: string, s: unknown): void }).registerService('objectql', {
+                registry: {
+                    listItems: (type: string) => (type === 'flow' ? flows : []),
+                    getObject: () => undefined,
+                },
+            });
+        },
+    };
+}
+
+/** Registers a recording `schedule` trigger on `kernel:ready`, as the real trigger plugin does. */
+function recordingTriggerPlugin(started: string[]): Plugin {
+    return {
+        name: 'recording-schedule-trigger',
+        version: '1.0.0',
+        async init() {},
+        async start(ctx: PluginContext) {
+            ctx.hook('kernel:ready', async () => {
+                ctx.getService<AutomationEngine>('automation').registerTrigger({
+                    type: 'schedule',
+                    start(binding: { flowName: string }) {
+                        started.push(binding.flowName);
+                    },
+                    stop() {},
+                });
+            });
+        },
+    };
+}
+
+async function bootKernel(scheduledWorkPolicy: ScheduledWorkPolicy | undefined) {
+    const started: string[] = [];
+    const kernel = new LiteKernel();
+    kernel.use(fakeObjectqlPlugin([scheduleFlow('digest')]));
+    kernel.use(new AutomationServicePlugin({ suspendedRunStore: 'memory', scheduledWorkPolicy }));
+    kernel.use(recordingTriggerPlugin(started));
+    await kernel.bootstrap();
+    return { kernel, started, engine: kernel.getService<AutomationEngine>('automation') };
+}
+
+describe('AutomationServicePlugin — forwards the per-kernel policy (#19834)', () => {
+    const PRIOR_SWITCH = process.env[SCHEDULED_WORK_ENV];
+    afterEach(() => {
+        if (PRIOR_SWITCH === undefined) delete process.env[SCHEDULED_WORK_ENV];
+        else process.env[SCHEDULED_WORK_ENV] = PRIOR_SWITCH;
+    });
+
+    it('two kernels in one process: OFF arms nothing, its ON sibling arms, under a deployment that is ON', async () => {
+        withDeployment('true');
+        const off = await bootKernel(OFF);
+        const on = await bootKernel(ON);
+        const control = await bootKernel(undefined);
+        try {
+            expect(off.started).toEqual([]);
+            expect(off.engine.getTriggerBindingAudit().map((a) => a.reason)).toEqual([SCHEDULED_WORK_DISABLED_REASON]);
+            expect(on.started).toEqual(['digest']);
+            // Control: the kernel with no option follows the deployment (ON).
+            expect(control.started).toEqual(['digest']);
+        } finally {
+            await off.kernel.shutdown();
+            await on.kernel.shutdown();
+            await control.kernel.shutdown();
+        }
+    });
+});

packages/services/service-automation/src/plugin.ts

@@ -13,7 +13,7 @@ import type {
 import { isConnectorUpstreamUnavailable, RetryConfigSchema } from '@objectstack/spec/integration';
 import { stripReadDecorations } from '@objectstack/spec/kernel';
 import { AutomationEngine } from './engine.js';
-import type { RunSummaryLogLevel } from './engine.js';
+import type { AutomationEngineOptions, RunSummaryLogLevel } from './engine.js';
 import { describeThrownForLog, thrownMessageText } from './thrown-cause-diagnostics.js';
 import { resolveFlowPrecedence, renderFlowContender } from './flow-precedence.js';
 import { installBuiltinNodes, rearmSuspendedWaitTimers } from './builtin/index.js';
@@ -127,6 +127,13 @@ export interface AutomationServicePluginOptions {
      * — only its default-level narration is.
      */
     runSummaryLog?: RunSummaryLogLevel;
+    /**
+     * [#19834] This kernel's scheduled-work policy, forwarded to the engine
+     * unchanged — see {@link AutomationEngineOptions.scheduledWorkPolicy}.
+     * Absent, the deployment default (`OS_AUTOMATION_SCHEDULED_WORK_ENABLED`)
+     * applies exactly as before.
+     */
+    scheduledWorkPolicy?: AutomationEngineOptions['scheduledWorkPolicy'];
     /**
      * Per-flow cap on terminal run-history rows, enforced at write time (the
      * "or 100 runs/flow, whichever first" half of the #2585 retention
@@ -595,6 +602,7 @@ export class AutomationServicePlugin implements Plugin {
         this.engine = new AutomationEngine(ctx.logger, undefined, {
             maxLogSize: this.options.maxLogSize,
             runSummaryLog: this.options.runSummaryLog,
+            scheduledWorkPolicy: this.options.scheduledWorkPolicy,
         });
 
         // Register as global service — other plugins access via ctx.getService('automation')