Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
88 changes: 88 additions & 0 deletions docs/api-overview.md
Original file line number Diff line number Diff line change
Expand Up @@ -316,6 +316,94 @@ counts as not matched (`false`), and an `evaluator` expression that cannot be
evaluated fails the flow with `Failed to evaluate expression: ...`, including
the `ExpressionError` detail.

## Triggers

Trigger adapters (`@loushy/build-ai-agent/triggers`) wake an agent up from an
inbound webhook, a schedule or a Slack message. Wire any of them with
`listen(agent, onEvent)`, where `onEvent` runs the agent.

### Webhook authentication

`WebhookTriggerAdapter` starts an HTTP server. **Always set `auth` for a
webhook that is reachable from outside your machine**: without it, anyone who
can reach the port can run your agent (and spend your tokens). If you listen
on a non-loopback host with no `auth`, the adapter logs a one-time warning
through `options.logger`.

```ts
import { createAgent, createMockProvider } from '@loushy/build-ai-agent';
import { WebhookTriggerAdapter } from '@loushy/build-ai-agent/triggers';

const agent = createAgent({ prompt: 'You are helpful.', provider: createMockProvider() });

// HMAC of the RAW request body (GitHub / Shopify style): header `x-signature-256: sha256=<hex>`.
new WebhookTriggerAdapter({
port: 8787,
auth: { type: 'hmac', secret: process.env.WEBHOOK_SECRET ?? '' },
}).listen(agent, (input) => agent.send(input));

// Replay protection: the signed payload becomes `${timestamp}.${body}` and
// requests more than `toleranceSeconds` (default 300) old are rejected.
new WebhookTriggerAdapter({
auth: {
type: 'hmac',
secret: process.env.WEBHOOK_SECRET ?? '',
header: 'x-signature',
timestampHeader: 'x-timestamp',
toleranceSeconds: 120,
},
});

// A shared bearer token (`Authorization: Bearer <token>`).
new WebhookTriggerAdapter({ auth: { type: 'bearer', token: process.env.WEBHOOK_TOKEN ?? '' } });

// Anything else: return true to accept. `rawBody` is a Buffer of the exact bytes received.
new WebhookTriggerAdapter({
auth: { type: 'custom', verify: (req) => req.headers['x-api-key'] === process.env.API_KEY },
});
```

HMAC options: `header` (default `x-signature-256`), `algorithm` (`sha256` or
`sha1`, default `sha256`), `prefix` (default `sha256=`; `''` for a bare
digest), `timestampHeader` and `toleranceSeconds`. Signatures and bearer
tokens are compared in constant time. A request that fails authentication gets
a generic `401 {"error":"Unauthorized"}` - the response never says which check
failed - and the reason (never a secret or signature) is logged at `warn`
level. Serve webhooks over HTTPS (terminate TLS in front of the adapter) so
tokens and payloads are not sent in clear text.

### Cron schedules

`CronTriggerAdapter` takes either a fixed `intervalMs` or a real cron
expression:

```ts
import { createAgent, createMockProvider } from '@loushy/build-ai-agent';
import { CronTriggerAdapter } from '@loushy/build-ai-agent/triggers';

const agent = createAgent({ prompt: 'You are helpful.', provider: createMockProvider() });

new CronTriggerAdapter({
cron: '*/15 9-17 * * MON-FRI', // minute hour day-of-month month day-of-week
timezone: 'Europe/Paris', // IANA name; defaults to the machine's local zone
input: 'Check the support queue',
onResult: (result, error) => console.log(error ?? result?.text),
}).listen(agent, (input) => agent.send(input));
```

Supported syntax: `*`, lists (`1,15`), ranges (`1-5`), steps (`*/15`,
`10-40/10`), month names (`JAN`) and weekday names (`MON`), with `0` and `7`
both meaning Sunday, plus `@hourly`, `@daily`, `@weekly` and `@monthly`. As in
classic cron, when both day-of-month and day-of-week are restricted a day
matches if either does. An invalid expression throws a `CronExpressionError`
naming the field and showing a valid example. `parseCronExpression(expr,
timezone).nextRun(after)` is exported if you need the next fire time.

Around daylight-saving changes, a time that does not exist (spring forward) is
skipped for that day, and a time that happens twice (fall back) fires once;
an every-hour schedule keeps firing hourly. The timer is re-armed after each
run from the scheduled time (no drift, no double fire), and `stop()` clears it.

## Deployment

- `DeploymentAdapter`, `registerAdapter()`, `getAdapter()`, `listAdapters()` -
Expand Down
109 changes: 109 additions & 0 deletions src/triggers/adapters/CronTriggerAdapter.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -96,3 +96,112 @@ describe('CronTriggerAdapter', () => {
expect(adapter.reply).toBeUndefined();
});
});

describe('CronTriggerAdapter with a cron expression', () => {
const start = (options: { cron: string; timezone?: string; fireImmediately?: boolean }) => {
const onEvent = vi.fn().mockResolvedValue(fakeResult('tick'));
const onResult = vi.fn();
const adapter = new CronTriggerAdapter({ input: 'scheduled', onResult, ...options });
return { onEvent, onResult, handle: adapter.listen(noopAgent, onEvent) };
};

beforeEach(() => {
vi.useFakeTimers();
vi.setSystemTime(new Date('2026-01-01T00:00:30Z'));
});

afterEach(() => {
vi.useRealTimers();
});

it('fires at each matching minute, without drift or double-firing, with a single timer', async () => {
const { onEvent, onResult, handle } = start({ cron: '* * * * *' });
expect(vi.getTimerCount()).toBe(1);

await vi.advanceTimersByTimeAsync(29_999);
expect(onEvent).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(1);
expect(onEvent).toHaveBeenCalledTimes(1);
expect(onEvent).toHaveBeenCalledWith(
'scheduled',
expect.objectContaining({ firedAt: '2026-01-01T00:01:00.000Z', tick: 0 })
);
expect(onResult).toHaveBeenCalledWith(fakeResult('tick'), undefined, expect.anything());
expect(vi.getTimerCount()).toBe(1); // re-armed

await vi.advanceTimersByTimeAsync(10 * 60_000);
expect(onEvent).toHaveBeenCalledTimes(11);
expect(new Date().toISOString()).toBe('2026-01-01T00:11:00.000Z');
handle.stop();
});

it('honours the time zone', async () => {
const { onEvent, handle } = start({ cron: '0 9 * * *', timezone: 'Asia/Kolkata' });
await vi.advanceTimersByTimeAsync(3 * 3600_000 + 29 * 60_000); // 03:29:30Z
expect(onEvent).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(60_000);
expect(onEvent).toHaveBeenCalledTimes(1);
handle.stop();
});

it("chains timers for delays beyond setTimeout's maximum (~24.8 days) and fires once", async () => {
const { onEvent, handle } = start({ cron: '0 0 1 1 *', timezone: 'UTC' }); // next: 2027-01-01, ~365 days away
expect(vi.getTimerCount()).toBe(1);
await vi.advanceTimersByTimeAsync(364 * 86_400_000);
expect(onEvent).not.toHaveBeenCalled();
expect(vi.getTimerCount()).toBe(1);
await vi.advanceTimersByTimeAsync(2 * 86_400_000);
expect(onEvent).toHaveBeenCalledTimes(1);
expect(onEvent).toHaveBeenCalledWith(
'scheduled',
expect.objectContaining({ firedAt: '2027-01-01T00:00:00.000Z' })
);
handle.stop();
});

it('fireImmediately fires once right away in addition to the schedule', async () => {
const { onEvent, handle } = start({ cron: '* * * * *', fireImmediately: true });
expect(onEvent).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(30_000);
expect(onEvent).toHaveBeenCalledTimes(2);
handle.stop();
});

it('keeps the schedule alive when a run fails', async () => {
const onEvent = vi.fn().mockRejectedValue(new Error('boom'));
const onResult = vi.fn();
const handle = new CronTriggerAdapter({ cron: '* * * * *', input: 'x', onResult }).listen(noopAgent, onEvent);
await vi.advanceTimersByTimeAsync(90_000);
expect(onEvent).toHaveBeenCalledTimes(2);
expect(onResult).toHaveBeenCalledWith(undefined, expect.any(Error), expect.anything());
handle.stop();
});

it('stop() clears the timer and nothing fires afterwards', async () => {
const { onEvent, handle } = start({ cron: '* * * * *' });
handle.stop();
expect(vi.getTimerCount()).toBe(0);
await vi.advanceTimersByTimeAsync(10 * 60_000);
expect(onEvent).not.toHaveBeenCalled();
});

it('skips missed runs after a long suspension instead of firing a burst', async () => {
const { onEvent, handle } = start({ cron: '* * * * *' });
vi.setSystemTime(new Date('2026-01-01T05:00:10Z')); // clock jumped (e.g. laptop sleep)
await vi.advanceTimersByTimeAsync(30_000);
expect(onEvent).toHaveBeenCalledTimes(1);
handle.stop();
});

it('validates its options', () => {
const base = { input: 'x', onResult: vi.fn() };
expect(() => new CronTriggerAdapter({ ...base, cron: 'nope' })).toThrow(/expected 5 space-separated fields/);
expect(() => new CronTriggerAdapter({ ...base, cron: '* * * * *', timezone: 'Nowhere/Land' })).toThrow(
/timezone/
);
expect(() => new CronTriggerAdapter({ ...base } as never)).toThrow(/exactly one of/);
expect(() => new CronTriggerAdapter({ ...base, intervalMs: 1000, cron: '* * * * *' } as never)).toThrow(
/exactly one of/
);
});
});
152 changes: 121 additions & 31 deletions src/triggers/adapters/CronTriggerAdapter.ts
Original file line number Diff line number Diff line change
@@ -1,15 +1,10 @@
/**
* Cron trigger adapter (LOU-T5).
* Cron trigger adapter (LOU-T5, LOU-D13).
*
* Fires an agent run on a fixed interval rather than in response to an
* inbound request. Deliberately scheduled by a plain millisecond interval
* (`intervalMs`) rather than parsing real cron syntax ('* * * * *') - this
* SDK has no existing cron-expression dependency, and adding one (plus a
* scheduler loop, timezone handling, etc.) is out of scope for landing the
* TriggerAdapter interface itself. `intervalMs` covers the same "run
* periodically" need; a `CronExpressionTriggerAdapter` that layers real
* cron syntax on top of this same interface is a natural, separately
* scoped follow-up.
* Fires an agent run on a schedule rather than in response to an inbound
* request. The schedule is either a fixed interval (`intervalMs`) or a
* standard 5-field cron expression (`cron`, with an optional IANA
* `timezone`), parsed by the dependency-free parser in ../cronExpression.ts.
*
* Reply semantics: there is no caller waiting on a response the way an
* HTTP request or a Slack message is - so results are handed to the
Expand All @@ -20,24 +15,85 @@
import type { ExecutionResult } from '../../execution/AgentExecutor';
import { RunnableAgent, TriggerAdapter, TriggerContext, TriggerHandle } from '../types';

export interface CronTriggerAdapterOptions {
/** How often to fire, in milliseconds. Must be a positive integer. */
intervalMs: number;
import { CronSchedule, parseCronExpression } from '../cronExpression';

interface CronTriggerAdapterBaseOptions {
/** The input to run the agent with on each tick. */
input: string;
/** Called with the outcome of each scheduled run (success or failure). */
onResult: (result: ExecutionResult | undefined, error: unknown, context: TriggerContext) => void;
/** If true, fires once immediately in addition to the interval. Defaults to false. */
/** If true, fires once immediately in addition to the schedule. Defaults to false. */
fireImmediately?: boolean;
}

/** Fire every `intervalMs` milliseconds. */
export interface CronIntervalOptions extends CronTriggerAdapterBaseOptions {
/** How often to fire, in milliseconds. Must be a positive number. */
intervalMs: number;
cron?: never;
timezone?: never;
}

/** Fire on a cron expression. */
export interface CronExpressionOptions extends CronTriggerAdapterBaseOptions {
/**
* Standard 5-field cron expression (`minute hour day-of-month month day-of-week`)
* or `@hourly` / `@daily` / `@weekly` / `@monthly`. Supports `*`, lists, ranges,
* steps and month/weekday names; 0 and 7 are both Sunday.
*/
cron: string;
/** IANA time zone the expression is evaluated in (e.g. `'America/New_York'`). Defaults to the machine's local zone. */
timezone?: string;
intervalMs?: never;
}

/**
* Options for {@link CronTriggerAdapter}: pass either `intervalMs` or `cron`.
*
* @example
* ```ts
* new CronTriggerAdapter({
* cron: '0 9 * * MON-FRI',
* timezone: 'Europe/Paris',
* input: 'Summarise yesterday',
* onResult: (result) => console.log(result?.text),
* });
* ```
*/
export type CronTriggerAdapterOptions = CronIntervalOptions | CronExpressionOptions;

/** setTimeout stores its delay in a signed 32-bit int; anything larger fires immediately. */
const MAX_TIMEOUT_MS = 2 ** 31 - 1;

type Schedule = { kind: 'interval'; intervalMs: number } | { kind: 'cron'; schedule: CronSchedule };

function resolveSchedule(options: CronTriggerAdapterOptions): Schedule {
const hasInterval = options.intervalMs !== undefined;
if (hasInterval === (options.cron !== undefined)) {
throw new Error("CronTriggerAdapter: pass exactly one of options.intervalMs or options.cron (e.g. { cron: '*/5 * * * *' }).");
}
if (options.cron !== undefined) {
return { kind: 'cron', schedule: parseCronExpression(options.cron, options.timezone) };
}
const intervalMs = options.intervalMs as number;
if (!Number.isFinite(intervalMs) || intervalMs <= 0) {
throw new Error('CronTriggerAdapter: options.intervalMs must be a positive number');
}
return { kind: 'interval', intervalMs };
}

/** Don't keep the process alive solely because a cron trigger is listening; callers that want that hold their own handle. */
function unref(timer: NodeJS.Timeout): void {
if (typeof timer.unref === 'function') timer.unref();
}

export class CronTriggerAdapter implements TriggerAdapter {
public readonly type = 'cron';

private readonly schedule: Schedule;

constructor(private readonly options: CronTriggerAdapterOptions) {
if (!Number.isFinite(options.intervalMs) || options.intervalMs <= 0) {
throw new Error('CronTriggerAdapter: options.intervalMs must be a positive number');
}
this.schedule = resolveSchedule(options);
}

public listen(
Expand All @@ -57,19 +113,53 @@ export class CronTriggerAdapter implements TriggerAdapter {
if (this.options.fireImmediately) {
fire();
}
const timer = setInterval(fire, this.options.intervalMs);
// Don't keep the process alive solely because a cron trigger is
// listening (mirrors how e.g. Node's own http/net timers behave when
// `.unref()`-ed); a caller that wants to keep the process alive should
// hold its own reference/handle open.
if (typeof timer.unref === 'function') {
timer.unref();
}

return {
stop: () => {
clearInterval(timer);
},
};
return this.schedule.kind === 'interval'
? startInterval(this.schedule.intervalMs, fire)
: startCron(this.schedule.schedule, fire);
}
}

function startInterval(intervalMs: number, fire: () => void): TriggerHandle {
const timer = setInterval(fire, intervalMs);
unref(timer);
return { stop: () => clearInterval(timer) };
}

/**
* One timer, always set to the next run. Each run is computed from the
* previous scheduled time (cron times are absolute, so nothing drifts) and
* strictly after it (so nothing fires twice). Delays beyond setTimeout's
* ~24.8 day limit are chained.
*/
function startCron(schedule: CronSchedule, fire: () => void): TriggerHandle {
let timer: NodeJS.Timeout | undefined;
let stopped = false;

const waitUntil = (target: number) => {
const delay = Math.max(0, target - Date.now());
timer = setTimeout(() => {
if (stopped) return;
if (Date.now() < target) {
waitUntil(target);
return;
}
// Re-arm first so a throwing handler can never end the schedule. If the
// process was suspended past later runs, skip them rather than firing a burst.
arm(new Date(Math.max(target, Date.now())));
fire();
}, Math.min(delay, MAX_TIMEOUT_MS));
unref(timer);
};

const arm = (after: Date) => {
if (!stopped) waitUntil(schedule.nextRun(after).getTime());
};

arm(new Date());
return {
stop: () => {
stopped = true;
if (timer) clearTimeout(timer);
},
};
}
Loading
Loading