mirror of
https://github.com/BradGroux/veritas-kanban.git
synced 2026-08-28 02:44:59 +00:00
Add unified recurring work scheduler
Add scheduler APIs, CLI commands, settings UI, retry/event state, telemetry hooks, and documentation over scheduled deliverables and workflow schedules.
This commit is contained in:
Brad Groux
GitHub
parent
fbff5b5c40
commit
5b363303c4
19 changed files with 1940 additions and 22 deletions
|
|
@ -116,4 +116,30 @@ describe('CLI API permission preflight', () => {
|
|||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
expect(fetchMock.mock.calls[0][0]).toBe('http://vk.test/api/auth/context');
|
||||
});
|
||||
|
||||
it('requires workflow execute permission for scheduler run actions', async () => {
|
||||
const fetchMock = vi.fn().mockResolvedValue(
|
||||
jsonResponse({
|
||||
role: 'read-only',
|
||||
isLocalhost: false,
|
||||
permissions: ['workflow:read'],
|
||||
})
|
||||
);
|
||||
globalThis.fetch = fetchMock as unknown as typeof fetch;
|
||||
|
||||
const api = createGuardedApiClient('http://vk.test', 'reader-key');
|
||||
|
||||
await expect(
|
||||
api('/api/scheduler/items/workflow%3Aweekly/run', {
|
||||
method: 'POST',
|
||||
})
|
||||
).rejects.toMatchObject({
|
||||
required: ['workflow:execute'],
|
||||
path: '/api/scheduler/items/workflow%3Aweekly/run',
|
||||
method: 'POST',
|
||||
});
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
expect(fetchMock.mock.calls[0][0]).toBe('http://vk.test/api/auth/context');
|
||||
});
|
||||
});
|
||||
|
|
|
|||
156
cli/src/commands/scheduler.ts
Normal file
156
cli/src/commands/scheduler.ts
Normal file
|
|
@ -0,0 +1,156 @@
|
|||
import { Command } from 'commander';
|
||||
import chalk from 'chalk';
|
||||
import { api } from '../utils/api.js';
|
||||
import type {
|
||||
SchedulerDueRunResult,
|
||||
SchedulerItem,
|
||||
SchedulerListResponse,
|
||||
SchedulerRunResult,
|
||||
SchedulerValidationResult,
|
||||
} from '@veritas-kanban/shared';
|
||||
|
||||
export function registerSchedulerCommands(program: Command): void {
|
||||
const scheduler = program
|
||||
.command('scheduler')
|
||||
.alias('schedule')
|
||||
.description('Inspect and control recurring Veritas work');
|
||||
|
||||
scheduler
|
||||
.command('list')
|
||||
.alias('status')
|
||||
.description('List recurring work scheduler items')
|
||||
.option('--json', 'Output as JSON')
|
||||
.action(async (options) => {
|
||||
try {
|
||||
const result = await api<SchedulerListResponse>('/api/scheduler');
|
||||
if (options.json) {
|
||||
console.log(JSON.stringify(result, null, 2));
|
||||
return;
|
||||
}
|
||||
printSummary(result);
|
||||
for (const item of result.items) printItem(item);
|
||||
} catch (err) {
|
||||
printError(err);
|
||||
}
|
||||
});
|
||||
|
||||
scheduler
|
||||
.command('run <itemId>')
|
||||
.description('Run a scheduler item now')
|
||||
.option('--json', 'Output as JSON')
|
||||
.action(async (itemId, options) => {
|
||||
await runItemAction(itemId, 'run', options.json);
|
||||
});
|
||||
|
||||
scheduler
|
||||
.command('pause <itemId>')
|
||||
.description('Pause a scheduler item')
|
||||
.option('--json', 'Output as JSON')
|
||||
.action(async (itemId, options) => {
|
||||
await runItemAction(itemId, 'pause', options.json);
|
||||
});
|
||||
|
||||
scheduler
|
||||
.command('resume <itemId>')
|
||||
.description('Resume a scheduler item')
|
||||
.option('--json', 'Output as JSON')
|
||||
.action(async (itemId, options) => {
|
||||
await runItemAction(itemId, 'resume', options.json);
|
||||
});
|
||||
|
||||
scheduler
|
||||
.command('validate <itemId>')
|
||||
.description('Validate a scheduler item')
|
||||
.option('--json', 'Output as JSON')
|
||||
.action(async (itemId, options) => {
|
||||
try {
|
||||
const result = await api<SchedulerValidationResult>(
|
||||
`/api/scheduler/items/${encodeURIComponent(itemId)}/validate`,
|
||||
{ method: 'POST' }
|
||||
);
|
||||
if (options.json) {
|
||||
console.log(JSON.stringify(result, null, 2));
|
||||
return;
|
||||
}
|
||||
if (result.ok) {
|
||||
console.log(chalk.green(`Valid: ${itemId}`));
|
||||
return;
|
||||
}
|
||||
console.log(chalk.yellow(`Validation issues: ${itemId}`));
|
||||
for (const issue of result.issues) {
|
||||
console.log(` ${issue.severity}: ${issue.path} - ${issue.message}`);
|
||||
}
|
||||
process.exitCode = result.issues.some((issue) => issue.severity === 'error') ? 1 : 0;
|
||||
} catch (err) {
|
||||
printError(err);
|
||||
}
|
||||
});
|
||||
|
||||
scheduler
|
||||
.command('run-due')
|
||||
.description('Run all scheduler items due now')
|
||||
.option('--json', 'Output as JSON')
|
||||
.action(async (options) => {
|
||||
try {
|
||||
const result = await api<SchedulerDueRunResult>('/api/scheduler/due/run', {
|
||||
method: 'POST',
|
||||
});
|
||||
if (options.json) {
|
||||
console.log(JSON.stringify(result, null, 2));
|
||||
return;
|
||||
}
|
||||
console.log(
|
||||
chalk.green(
|
||||
`Checked ${result.checked}, executed ${result.executed}, skipped ${result.skipped}, failed ${result.failed}`
|
||||
)
|
||||
);
|
||||
if (result.overlapping) console.log(chalk.yellow('Due runner already active.'));
|
||||
} catch (err) {
|
||||
printError(err);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
async function runItemAction(
|
||||
itemId: string,
|
||||
action: 'run' | 'pause' | 'resume',
|
||||
json: boolean
|
||||
): Promise<void> {
|
||||
try {
|
||||
const result = await api<SchedulerRunResult>(
|
||||
`/api/scheduler/items/${encodeURIComponent(itemId)}/${action}`,
|
||||
{ method: 'POST' }
|
||||
);
|
||||
if (json) {
|
||||
console.log(JSON.stringify(result, null, 2));
|
||||
return;
|
||||
}
|
||||
console.log(chalk.green(`${action}: ${result.event.summary}`));
|
||||
if (result.event.sourceRunId) console.log(chalk.dim(`Run: ${result.event.sourceRunId}`));
|
||||
} catch (err) {
|
||||
printError(err);
|
||||
}
|
||||
}
|
||||
|
||||
function printSummary(result: SchedulerListResponse): void {
|
||||
console.log(chalk.bold('\nRecurring Work Scheduler'));
|
||||
console.log(
|
||||
chalk.dim(
|
||||
`total=${result.summary.total} enabled=${result.summary.enabled} due=${result.summary.due} failed=${result.summary.failed} blocked=${result.summary.blocked}`
|
||||
)
|
||||
);
|
||||
console.log();
|
||||
}
|
||||
|
||||
function printItem(item: SchedulerItem): void {
|
||||
const status = item.health === 'healthy' ? chalk.green(item.health) : chalk.yellow(item.health);
|
||||
console.log(`${chalk.bold(item.id)} ${status}`);
|
||||
console.log(` ${item.name}`);
|
||||
console.log(` schedule=${item.trigger.description} next=${item.nextRunAt ?? 'not set'}`);
|
||||
if (item.lastSummary) console.log(chalk.dim(` last=${item.lastSummary}`));
|
||||
}
|
||||
|
||||
function printError(err: unknown): never {
|
||||
console.error(chalk.red(`Error: ${err instanceof Error ? err.message : String(err)}`));
|
||||
process.exit(1);
|
||||
}
|
||||
|
|
@ -20,6 +20,7 @@ import { registerDoctorCommand } from './commands/doctor.js';
|
|||
import { registerSnapshotCommand } from './commands/snapshot.js';
|
||||
import { registerPromptCommands } from './commands/prompts.js';
|
||||
import { registerWorkspaceCommands } from './commands/workspaces.js';
|
||||
import { registerSchedulerCommands } from './commands/scheduler.js';
|
||||
|
||||
const program = new Command();
|
||||
const packageJson = JSON.parse(
|
||||
|
|
@ -51,5 +52,6 @@ registerDoctorCommand(program);
|
|||
registerSnapshotCommand(program);
|
||||
registerPromptCommands(program);
|
||||
registerWorkspaceCommands(program);
|
||||
registerSchedulerCommands(program);
|
||||
|
||||
program.parse();
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
# Veritas Kanban — API Reference
|
||||
|
||||
**Version**: 5.1.0
|
||||
**Last Updated**: 2026-06-18
|
||||
**Last Updated**: 2026-06-26
|
||||
**Base URL**: `http://localhost:3001/api`
|
||||
**Canonical prefix**: `/api/v1` (alias: `/api`)
|
||||
|
||||
|
|
@ -33,27 +33,28 @@
|
|||
20. [Task Comments](#task-comments)
|
||||
21. [Task Subtasks](#task-subtasks)
|
||||
22. [Task Deliverables](#task-deliverables)
|
||||
23. [Task Archive](#task-archive)
|
||||
24. [Attachments](#attachments)
|
||||
25. [Agent Permissions](#agent-permissions)
|
||||
26. [Agent Routing](#agent-routing)
|
||||
27. [Sandbox Policies](#sandbox-policies)
|
||||
28. [Shared Resources](#shared-resources)
|
||||
29. [Skill Capability Profiles](#skill-capability-profiles-apiskillscapabilities)
|
||||
30. [Skill Security Scanner](#skill-security-scanner-apiskillssecurity)
|
||||
31. [Doc Freshness](#doc-freshness)
|
||||
32. [Cost Prediction](#cost-prediction)
|
||||
33. [Error Learning](#error-learning)
|
||||
34. [Tool Policies](#tool-policies)
|
||||
35. [Watcher Continuation Policies](#watcher-continuation-policies)
|
||||
36. [Traces](#traces)
|
||||
37. [Governance Decision Traces](#governance-decision-traces-apigovernancetraces)
|
||||
38. [Audit](#audit)
|
||||
39. [Maintenance Center](#maintenance-center-apiv1maintenance)
|
||||
40. [Common Workflows](#common-workflows)
|
||||
41. [Versioning & Deprecation](#versioning--deprecation)
|
||||
42. [Rate Limits](#rate-limits)
|
||||
43. [Additional Endpoint Groups](#additional-endpoint-groups)
|
||||
23. [Recurring Work Scheduler](#recurring-work-scheduler)
|
||||
24. [Task Archive](#task-archive)
|
||||
25. [Attachments](#attachments)
|
||||
26. [Agent Permissions](#agent-permissions)
|
||||
27. [Agent Routing](#agent-routing)
|
||||
28. [Sandbox Policies](#sandbox-policies)
|
||||
29. [Shared Resources](#shared-resources)
|
||||
30. [Skill Capability Profiles](#skill-capability-profiles-apiskillscapabilities)
|
||||
31. [Skill Security Scanner](#skill-security-scanner-apiskillssecurity)
|
||||
32. [Doc Freshness](#doc-freshness)
|
||||
33. [Cost Prediction](#cost-prediction)
|
||||
34. [Error Learning](#error-learning)
|
||||
35. [Tool Policies](#tool-policies)
|
||||
36. [Watcher Continuation Policies](#watcher-continuation-policies)
|
||||
37. [Traces](#traces)
|
||||
38. [Governance Decision Traces](#governance-decision-traces-apigovernancetraces)
|
||||
39. [Audit](#audit)
|
||||
40. [Maintenance Center](#maintenance-center-apiv1maintenance)
|
||||
41. [Common Workflows](#common-workflows)
|
||||
42. [Versioning & Deprecation](#versioning--deprecation)
|
||||
43. [Rate Limits](#rate-limits)
|
||||
44. [Additional Endpoint Groups](#additional-endpoint-groups)
|
||||
|
||||
---
|
||||
|
||||
|
|
@ -1265,6 +1266,73 @@ DELETE /api/tasks/:id/deliverables/:deliverableId
|
|||
|
||||
---
|
||||
|
||||
## Recurring Work Scheduler
|
||||
|
||||
Inspect and control recurring work across scheduled deliverables and scheduled workflow definitions.
|
||||
|
||||
Mounted at `/api/scheduler`.
|
||||
|
||||
### List Scheduler Items
|
||||
|
||||
```
|
||||
GET /api/scheduler
|
||||
```
|
||||
|
||||
Returns scheduler summary counts, scheduler items, retry state, health, and recent events.
|
||||
|
||||
### Read Item
|
||||
|
||||
```
|
||||
GET /api/scheduler/items/:id
|
||||
```
|
||||
|
||||
Item IDs use the source prefix:
|
||||
|
||||
- `scheduled-deliverable:<deliverableId>`
|
||||
- `workflow:<workflowId>`
|
||||
|
||||
### Run Item Now
|
||||
|
||||
```
|
||||
POST /api/scheduler/items/:id/run
|
||||
```
|
||||
|
||||
Runs one scheduler item immediately. Deliverables execute through the scheduled deliverables runner. Workflows start a normal workflow run and return the started run ID.
|
||||
|
||||
### Pause Item
|
||||
|
||||
```
|
||||
POST /api/scheduler/items/:id/pause
|
||||
```
|
||||
|
||||
Pauses the underlying scheduled deliverable or workflow schedule.
|
||||
|
||||
### Resume Item
|
||||
|
||||
```
|
||||
POST /api/scheduler/items/:id/resume
|
||||
```
|
||||
|
||||
Re-enables the underlying scheduled deliverable or workflow schedule and clears scheduler retry delay.
|
||||
|
||||
### Validate Item
|
||||
|
||||
```
|
||||
POST /api/scheduler/items/:id/validate
|
||||
```
|
||||
|
||||
Returns validation issues. Custom cron schedules are visible and manually runnable, but automatic due-run execution is not enabled until a cron adapter is configured.
|
||||
|
||||
### Run Due Items
|
||||
|
||||
```
|
||||
POST /api/scheduler/due/run
|
||||
```
|
||||
|
||||
Runs all due standard schedules and refuses overlapping due-run passes.
|
||||
|
||||
---
|
||||
|
||||
## Task Archive
|
||||
|
||||
Archive completed tasks (by sprint or individually) and restore them. Archived tasks are removed from the active board.
|
||||
|
|
|
|||
|
|
@ -46,6 +46,7 @@ For current v5 screenshots and GIFs, see the
|
|||
- [Enforcement Gates](#enforcement-gates)
|
||||
- [Broadcast Notifications](#broadcast-notifications)
|
||||
- [Task Deliverables](#task-deliverables)
|
||||
- [Recurring Work Scheduler](#recurring-work-scheduler)
|
||||
- [Efficient Polling](#efficient-polling)
|
||||
- [Approval Delegation](#approval-delegation)
|
||||
- [Lifecycle Hooks](#task-lifecycle-hooks)
|
||||
|
|
@ -1155,6 +1156,33 @@ First-class deliverable objects attached to tasks with type and status tracking.
|
|||
|
||||
---
|
||||
|
||||
## Recurring Work Scheduler
|
||||
|
||||
Unified operator surface for scheduled deliverables and scheduled workflow definitions.
|
||||
|
||||
- **Single scheduler dashboard** — Settings exposes all recurring work with health, retry, next run, last run, and recent events
|
||||
- **Manual controls** — Run, pause, resume, and validate individual scheduler items
|
||||
- **Due runner** — `POST /api/scheduler/due/run` and `vk scheduler run-due` execute due items while refusing overlapping passes
|
||||
- **Existing service adapters** — Deliverables execute through the scheduled deliverables runner; workflows start through the workflow run service
|
||||
- **Operations telemetry** — Scheduler events emit bounded run telemetry with `agent=scheduler` for operations digest visibility
|
||||
- **Custom cron guardrail** — Cron schedules are visible and manually runnable, but automatic custom-cron due execution is deferred until a cron adapter is configured
|
||||
|
||||
→ [Full guide](features/recurring-work-scheduler.md) — API, CLI, execution model, and scheduler guardrails
|
||||
|
||||
### API Endpoints
|
||||
|
||||
| Endpoint | Method | Description |
|
||||
| ----------------------------------- | ------ | ------------------------------- |
|
||||
| `/api/scheduler` | GET | List scheduler items and events |
|
||||
| `/api/scheduler/items/:id` | GET | Read one scheduler item |
|
||||
| `/api/scheduler/items/:id/run` | POST | Run one scheduler item now |
|
||||
| `/api/scheduler/items/:id/pause` | POST | Pause one scheduler item |
|
||||
| `/api/scheduler/items/:id/resume` | POST | Resume one scheduler item |
|
||||
| `/api/scheduler/items/:id/validate` | POST | Validate one scheduler item |
|
||||
| `/api/scheduler/due/run` | POST | Run due scheduler items |
|
||||
|
||||
---
|
||||
|
||||
## Efficient Polling
|
||||
|
||||
Optimized change-detection endpoint for agents that poll instead of using WebSocket. Shipped in v2.0.
|
||||
|
|
|
|||
58
docs/features/recurring-work-scheduler.md
Normal file
58
docs/features/recurring-work-scheduler.md
Normal file
|
|
@ -0,0 +1,58 @@
|
|||
# Recurring Work Scheduler
|
||||
|
||||
The recurring work scheduler gives operators one place to inspect and control scheduled Veritas work.
|
||||
|
||||
It currently surfaces:
|
||||
|
||||
- scheduled deliverables, including operations digest deliverables
|
||||
- workflow definitions with enabled non-manual schedules
|
||||
- workflow scheduled-snapshot outputs
|
||||
|
||||
## Operator Controls
|
||||
|
||||
The scheduler is available in `Settings -> Scheduler`, through `vk scheduler`, and through `/api/scheduler`.
|
||||
|
||||
Each scheduler item exposes:
|
||||
|
||||
- health: `healthy`, `warning`, `paused`, or `blocked`
|
||||
- next and last run timestamps
|
||||
- retry attempts and next retry time
|
||||
- recent scheduler events
|
||||
- manual run, pause, resume, and validate actions
|
||||
|
||||
## Execution Model
|
||||
|
||||
Scheduled deliverables run through the existing scheduled deliverables runner. Workflow schedules run through the existing workflow run service and create a normal workflow run record.
|
||||
|
||||
The due-runner refuses overlapping scheduler passes and refuses overlapping item runs in the same server process. Failed scheduler runs record retry state with exponential backoff up to the configured retry limit. Scheduler executions also emit bounded run telemetry with `agent=scheduler` and `project=operations`, so operations digests can include scheduler activity.
|
||||
|
||||
## Custom Cron
|
||||
|
||||
Custom cron schedules are visible, manually runnable, and validated for a cron expression, but automatic cron due execution is intentionally not enabled in this first pass. Standard `daily`, `weekly`, `biweekly`, and `monthly` schedules have deterministic due calculation without adding a new production dependency.
|
||||
|
||||
## CLI
|
||||
|
||||
```bash
|
||||
vk scheduler list
|
||||
vk scheduler run-due
|
||||
vk scheduler run "scheduled-deliverable:del_ops"
|
||||
vk scheduler pause "workflow:weekly-snapshot"
|
||||
vk scheduler resume "workflow:weekly-snapshot"
|
||||
vk scheduler validate "workflow:weekly-snapshot"
|
||||
```
|
||||
|
||||
Use `--json` on any command for automation-friendly output.
|
||||
|
||||
## API
|
||||
|
||||
Mounted at `/api/scheduler`.
|
||||
|
||||
| Method | Path | Description |
|
||||
| ------ | ----------------------------------- | ------------------------------- |
|
||||
| `GET` | `/api/scheduler` | List scheduler items and events |
|
||||
| `GET` | `/api/scheduler/items/:id` | Read one scheduler item |
|
||||
| `POST` | `/api/scheduler/items/:id/run` | Run one scheduler item now |
|
||||
| `POST` | `/api/scheduler/items/:id/pause` | Pause one scheduler item |
|
||||
| `POST` | `/api/scheduler/items/:id/resume` | Resume one scheduler item |
|
||||
| `POST` | `/api/scheduler/items/:id/validate` | Validate one scheduler item |
|
||||
| `POST` | `/api/scheduler/due/run` | Run all items due now |
|
||||
196
server/src/__tests__/scheduler-service.test.ts
Normal file
196
server/src/__tests__/scheduler-service.test.ts
Normal file
|
|
@ -0,0 +1,196 @@
|
|||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import fs from 'node:fs/promises';
|
||||
import os from 'node:os';
|
||||
import path from 'node:path';
|
||||
import type { WorkflowDefinition } from '@veritas-kanban/shared';
|
||||
import { SchedulerService } from '../services/scheduler-service.js';
|
||||
import {
|
||||
ScheduledDeliverablesService,
|
||||
type Deliverable,
|
||||
} from '../services/scheduled-deliverables-service.js';
|
||||
import { WorkflowService } from '../services/workflow-service.js';
|
||||
|
||||
describe('SchedulerService', () => {
|
||||
let testRoot: string;
|
||||
let deliverablesService: ScheduledDeliverablesService;
|
||||
let workflowService: WorkflowService;
|
||||
let telemetry: { emit: ReturnType<typeof vi.fn> };
|
||||
|
||||
beforeEach(async () => {
|
||||
testRoot = await fs.mkdtemp(path.join(os.tmpdir(), 'veritas-scheduler-'));
|
||||
await fs.mkdir(path.join(testRoot, 'workflows'), { recursive: true });
|
||||
deliverablesService = new ScheduledDeliverablesService({
|
||||
dataDir: testRoot,
|
||||
storageType: 'file',
|
||||
});
|
||||
workflowService = new WorkflowService({
|
||||
workflowsDir: path.join(testRoot, 'workflows'),
|
||||
storageType: 'file',
|
||||
});
|
||||
telemetry = { emit: vi.fn(async (event) => event) };
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
deliverablesService.dispose();
|
||||
workflowService.dispose();
|
||||
await fs.rm(testRoot, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it('lists deliverable and workflow schedules with due summary', async () => {
|
||||
await seedDeliverables(testRoot, [
|
||||
scheduledDeliverable({
|
||||
id: 'del_due',
|
||||
nextRunAt: '2026-06-05T08:59:00.000Z',
|
||||
}),
|
||||
]);
|
||||
await workflowService.saveWorkflow(
|
||||
workflowDefinition({
|
||||
id: 'weekly-snapshot',
|
||||
schedule: {
|
||||
mode: 'weekly',
|
||||
enabled: true,
|
||||
startAt: '2026-06-06T09:00:00.000Z',
|
||||
timezone: 'UTC',
|
||||
},
|
||||
})
|
||||
);
|
||||
const service = schedulerService();
|
||||
|
||||
const result = await service.list(new Date('2026-06-05T09:00:00.000Z'));
|
||||
|
||||
expect(result.summary).toMatchObject({ total: 2, enabled: 2, due: 1 });
|
||||
expect(result.items.map((item) => item.id)).toEqual([
|
||||
'scheduled-deliverable:del_due',
|
||||
'workflow:weekly-snapshot',
|
||||
]);
|
||||
});
|
||||
|
||||
it('pauses and resumes scheduled deliverables through the existing service', async () => {
|
||||
await seedDeliverables(testRoot, [scheduledDeliverable({ id: 'del_ops' })]);
|
||||
const service = schedulerService();
|
||||
|
||||
const paused = await service.pause('scheduled-deliverable:del_ops');
|
||||
expect(paused.item.enabled).toBe(false);
|
||||
expect(paused.event.summary).toBe('Scheduler item paused.');
|
||||
|
||||
const resumed = await service.resume('scheduled-deliverable:del_ops');
|
||||
expect(resumed.item.enabled).toBe(true);
|
||||
expect(resumed.event.summary).toBe('Scheduler item resumed.');
|
||||
});
|
||||
|
||||
it('validates custom cron schedules without a due-run adapter', async () => {
|
||||
await seedDeliverables(testRoot, [
|
||||
scheduledDeliverable({
|
||||
id: 'del_custom',
|
||||
schedule: 'custom',
|
||||
cronExpr: '0 9 * * 1',
|
||||
scheduleDescription: 'Cron: 0 9 * * 1',
|
||||
nextRunAt: undefined,
|
||||
}),
|
||||
]);
|
||||
const service = schedulerService();
|
||||
|
||||
const result = await service.validate('scheduled-deliverable:del_custom');
|
||||
|
||||
expect(result.ok).toBe(true);
|
||||
expect(result.issues).toEqual([
|
||||
expect.objectContaining({
|
||||
severity: 'warning',
|
||||
path: 'trigger.mode',
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it('runs due deliverables and records scheduler telemetry', async () => {
|
||||
await seedDeliverables(testRoot, [
|
||||
scheduledDeliverable({
|
||||
id: 'del_due',
|
||||
tags: ['unsupported-report'],
|
||||
nextRunAt: '2026-06-05T08:59:00.000Z',
|
||||
}),
|
||||
]);
|
||||
const service = schedulerService();
|
||||
|
||||
const result = await service.runDue(new Date('2026-06-05T09:00:00.000Z'));
|
||||
|
||||
expect(result).toMatchObject({
|
||||
checked: 1,
|
||||
executed: 0,
|
||||
skipped: 1,
|
||||
failed: 0,
|
||||
overlapping: false,
|
||||
});
|
||||
expect(result.events[0]).toMatchObject({
|
||||
itemId: 'scheduled-deliverable:del_due',
|
||||
status: 'skipped',
|
||||
});
|
||||
expect(telemetry.emit).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
type: 'run.completed',
|
||||
taskId: 'scheduled-deliverable:del_due',
|
||||
agent: 'scheduler',
|
||||
project: 'operations',
|
||||
})
|
||||
);
|
||||
});
|
||||
|
||||
function schedulerService(): SchedulerService {
|
||||
return new SchedulerService({
|
||||
stateFile: path.join(testRoot, 'scheduler-state.json'),
|
||||
deliverablesService,
|
||||
workflowService,
|
||||
telemetryService: telemetry as never,
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
async function seedDeliverables(root: string, deliverables: Deliverable[]): Promise<void> {
|
||||
await fs.writeFile(path.join(root, 'scheduled-deliverables.json'), JSON.stringify(deliverables));
|
||||
await fs.writeFile(path.join(root, 'deliverable-runs.json'), '[]');
|
||||
}
|
||||
|
||||
function scheduledDeliverable(overrides: Partial<Deliverable> = {}): Deliverable {
|
||||
return {
|
||||
id: 'del_ops',
|
||||
name: 'Operations Digest',
|
||||
description: 'Generate operations digest.',
|
||||
schedule: 'daily',
|
||||
scheduleDescription: 'Every day',
|
||||
enabled: true,
|
||||
tags: ['operations-digest'],
|
||||
createdAt: '2026-06-01T09:00:00.000Z',
|
||||
lastRunAt: undefined,
|
||||
nextRunAt: '2026-06-06T09:00:00.000Z',
|
||||
totalRuns: 0,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
function workflowDefinition(overrides: Partial<WorkflowDefinition> = {}): WorkflowDefinition {
|
||||
return {
|
||||
id: 'weekly-snapshot',
|
||||
name: 'Weekly Snapshot',
|
||||
version: 1,
|
||||
description: 'Create weekly operational snapshot.',
|
||||
agents: [
|
||||
{
|
||||
id: 'writer',
|
||||
name: 'Writer',
|
||||
role: 'general',
|
||||
description: 'Writes the snapshot.',
|
||||
},
|
||||
],
|
||||
steps: [
|
||||
{
|
||||
id: 'write',
|
||||
name: 'Write snapshot',
|
||||
type: 'agent',
|
||||
agent: 'writer',
|
||||
input: 'Write snapshot.',
|
||||
},
|
||||
],
|
||||
schedule: { mode: 'weekly', enabled: true, timezone: 'UTC' },
|
||||
outputTargets: [{ type: 'scheduled-snapshot', label: 'Snapshot', required: true }],
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
56
server/src/routes/scheduler.ts
Normal file
56
server/src/routes/scheduler.ts
Normal file
|
|
@ -0,0 +1,56 @@
|
|||
import { Router, type Router as RouterType } from 'express';
|
||||
import { getSchedulerService } from '../services/scheduler-service.js';
|
||||
import { asyncHandler } from '../middleware/async-handler.js';
|
||||
|
||||
const router: RouterType = Router();
|
||||
|
||||
router.get(
|
||||
'/',
|
||||
asyncHandler(async (_req, res) => {
|
||||
res.json(await getSchedulerService().list());
|
||||
})
|
||||
);
|
||||
|
||||
router.get(
|
||||
'/items/:itemId',
|
||||
asyncHandler(async (req, res) => {
|
||||
res.json(await getSchedulerService().getItem(String(req.params.itemId)));
|
||||
})
|
||||
);
|
||||
|
||||
router.post(
|
||||
'/items/:itemId/run',
|
||||
asyncHandler(async (req, res) => {
|
||||
res.json(await getSchedulerService().runItem(String(req.params.itemId), 'manual-run'));
|
||||
})
|
||||
);
|
||||
|
||||
router.post(
|
||||
'/items/:itemId/pause',
|
||||
asyncHandler(async (req, res) => {
|
||||
res.json(await getSchedulerService().pause(String(req.params.itemId)));
|
||||
})
|
||||
);
|
||||
|
||||
router.post(
|
||||
'/items/:itemId/resume',
|
||||
asyncHandler(async (req, res) => {
|
||||
res.json(await getSchedulerService().resume(String(req.params.itemId)));
|
||||
})
|
||||
);
|
||||
|
||||
router.post(
|
||||
'/items/:itemId/validate',
|
||||
asyncHandler(async (req, res) => {
|
||||
res.json(await getSchedulerService().validate(String(req.params.itemId)));
|
||||
})
|
||||
);
|
||||
|
||||
router.post(
|
||||
'/due/run',
|
||||
asyncHandler(async (_req, res) => {
|
||||
res.json(await getSchedulerService().runDue());
|
||||
})
|
||||
);
|
||||
|
||||
export { router as schedulerRoutes };
|
||||
|
|
@ -40,6 +40,7 @@ import {
|
|||
reportAccess,
|
||||
reportRoutesAccess,
|
||||
sandboxPolicyAccess,
|
||||
schedulerAccess,
|
||||
scoringAccess,
|
||||
searchAccess,
|
||||
settingsAccess,
|
||||
|
|
@ -134,6 +135,7 @@ import { watcherPolicyRoutes } from '../watcher-policies.js';
|
|||
import sandboxPolicyRoutes from '../sandbox-policies.js';
|
||||
import { runSessionRoutes } from '../run-sessions.js';
|
||||
import { workspaceCapabilityRoutes } from '../workspace-capabilities.js';
|
||||
import { schedulerRoutes } from '../scheduler.js';
|
||||
|
||||
const v1Router: IRouter = Router();
|
||||
|
||||
|
|
@ -227,6 +229,7 @@ v1Router.use('/audit', adminAccess, auditRoutes);
|
|||
v1Router.use('/lessons', taskReadAccess, lessonsRoutes);
|
||||
v1Router.use('/delegation', delegationAccess, delegationRoutes);
|
||||
v1Router.use('/workflows', workflowAccess, workflowRoutes);
|
||||
v1Router.use('/scheduler', schedulerAccess, schedulerRoutes);
|
||||
v1Router.use('/watcher-policies', watcherPolicyAccess, watcherPolicyRoutes);
|
||||
v1Router.use('/tool-policies', policyAccess, toolPolicyRoutes);
|
||||
v1Router.use('/policies', policyAccess, policyRoutes);
|
||||
|
|
|
|||
|
|
@ -95,6 +95,12 @@ export const workflowAccess = routeAccess('workflow:read', 'workflow:write', [
|
|||
},
|
||||
]);
|
||||
|
||||
export const schedulerAccess = routeAccess('workflow:read', 'workflow:write', [
|
||||
{ methods: ['POST'], path: /^\/items\/[^/]+\/run\/?$/, permissions: 'workflow:execute' },
|
||||
{ methods: ['POST'], path: /^\/items\/[^/]+\/validate\/?$/, permissions: 'workflow:read' },
|
||||
{ methods: ['POST'], path: /^\/due\/run\/?$/, permissions: 'workflow:execute' },
|
||||
]);
|
||||
|
||||
export const watcherPolicyAccess = routeAccess('policy:read', 'policy:write', [
|
||||
{ methods: ['POST'], path: /^\/evaluate\/?$/, permissions: ['policy:read', 'agent:read'] },
|
||||
]);
|
||||
|
|
|
|||
780
server/src/services/scheduler-service.ts
Normal file
780
server/src/services/scheduler-service.ts
Normal file
|
|
@ -0,0 +1,780 @@
|
|||
import fs from 'node:fs/promises';
|
||||
import path from 'node:path';
|
||||
import { nanoid } from 'nanoid';
|
||||
import type {
|
||||
SchedulerDueRunResult,
|
||||
SchedulerEvent,
|
||||
SchedulerEventType,
|
||||
SchedulerItem,
|
||||
SchedulerItemKind,
|
||||
SchedulerListResponse,
|
||||
SchedulerRunResult,
|
||||
SchedulerRunStatus,
|
||||
SchedulerValidationIssue,
|
||||
SchedulerValidationResult,
|
||||
WorkflowDefinition,
|
||||
WorkflowSchedule,
|
||||
WorkflowScheduleMode,
|
||||
} from '@veritas-kanban/shared';
|
||||
import type { RunTelemetryEvent } from '@veritas-kanban/shared';
|
||||
import { createLogger } from '../lib/logger.js';
|
||||
import { NotFoundError, ValidationError } from '../middleware/error-handler.js';
|
||||
import { getRuntimeDir } from '../utils/paths.js';
|
||||
import {
|
||||
getScheduledDeliverablesService,
|
||||
type Deliverable,
|
||||
type DeliverableRun,
|
||||
type ScheduledDeliverablesService,
|
||||
} from './scheduled-deliverables-service.js';
|
||||
import { ScheduledDeliverablesRunner } from './scheduled-deliverables-runner-service.js';
|
||||
import { getTelemetryService, type TelemetryService } from './telemetry-service.js';
|
||||
import {
|
||||
getWorkflowAuthoringService,
|
||||
type WorkflowAuthoringService,
|
||||
} from './workflow-authoring-service.js';
|
||||
import { getWorkflowRunService, type WorkflowRunService } from './workflow-run-service.js';
|
||||
import { getWorkflowService, type WorkflowService } from './workflow-service.js';
|
||||
|
||||
const log = createLogger('scheduler');
|
||||
const STATE_VERSION = 1;
|
||||
const DEFAULT_MAX_ATTEMPTS = 3;
|
||||
const DEFAULT_BACKOFF_MINUTES = 5;
|
||||
const MAX_EVENTS = 200;
|
||||
const SCHEDULER_PROJECT = 'operations';
|
||||
const SCHEDULER_AGENT = 'scheduler';
|
||||
|
||||
interface SchedulerItemState {
|
||||
attempts?: number;
|
||||
nextAttemptAt?: string;
|
||||
lastRunAt?: string;
|
||||
nextRunAt?: string;
|
||||
lastStatus?: SchedulerRunStatus;
|
||||
lastSummary?: string;
|
||||
lastError?: string;
|
||||
sourceRunId?: string;
|
||||
}
|
||||
|
||||
interface SchedulerStateFile {
|
||||
version: typeof STATE_VERSION;
|
||||
items: Record<string, SchedulerItemState>;
|
||||
events: SchedulerEvent[];
|
||||
}
|
||||
|
||||
interface SchedulerServiceOptions {
|
||||
stateFile?: string;
|
||||
deliverablesService?: ScheduledDeliverablesService;
|
||||
workflowService?: WorkflowService;
|
||||
workflowRunService?: WorkflowRunService;
|
||||
workflowAuthoringService?: WorkflowAuthoringService;
|
||||
telemetryService?: TelemetryService;
|
||||
}
|
||||
|
||||
export class SchedulerService {
|
||||
private readonly stateFile: string;
|
||||
private readonly deliverablesService: ScheduledDeliverablesService;
|
||||
private readonly workflowService: WorkflowService;
|
||||
private readonly workflowRunService: WorkflowRunService;
|
||||
private readonly workflowAuthoringService: WorkflowAuthoringService;
|
||||
private readonly telemetryService: TelemetryService;
|
||||
private state: SchedulerStateFile | null = null;
|
||||
private runningDue = false;
|
||||
private readonly runningItems = new Set<string>();
|
||||
|
||||
constructor(options: SchedulerServiceOptions = {}) {
|
||||
this.stateFile = options.stateFile ?? path.join(getRuntimeDir(), 'scheduler-state.json');
|
||||
this.deliverablesService = options.deliverablesService ?? getScheduledDeliverablesService();
|
||||
this.workflowService = options.workflowService ?? getWorkflowService();
|
||||
this.workflowRunService = options.workflowRunService ?? getWorkflowRunService();
|
||||
this.workflowAuthoringService =
|
||||
options.workflowAuthoringService ?? getWorkflowAuthoringService();
|
||||
this.telemetryService = options.telemetryService ?? getTelemetryService();
|
||||
}
|
||||
|
||||
async list(now = new Date()): Promise<SchedulerListResponse> {
|
||||
await this.ensureLoaded();
|
||||
const items = await this.buildItems(now);
|
||||
const dueCutoff = now.getTime();
|
||||
const summary = items.reduce(
|
||||
(acc, item) => {
|
||||
acc.total++;
|
||||
if (item.enabled) acc.enabled++;
|
||||
if (!item.enabled) acc.paused++;
|
||||
if (item.health === 'blocked') acc.blocked++;
|
||||
if (item.lastStatus === 'failed') acc.failed++;
|
||||
if (isItemDue(item, dueCutoff)) acc.due++;
|
||||
return acc;
|
||||
},
|
||||
{ total: 0, enabled: 0, paused: 0, due: 0, failed: 0, blocked: 0 }
|
||||
);
|
||||
|
||||
return {
|
||||
generatedAt: now.toISOString(),
|
||||
summary,
|
||||
items,
|
||||
recentEvents: [...this.currentState().events].slice(-20).reverse(),
|
||||
};
|
||||
}
|
||||
|
||||
async getItem(itemId: string, now = new Date()): Promise<SchedulerItem> {
|
||||
const items = await this.buildItems(now);
|
||||
const item = items.find((candidate) => candidate.id === itemId);
|
||||
if (!item) throw new NotFoundError(`Scheduler item not found: ${itemId}`);
|
||||
return item;
|
||||
}
|
||||
|
||||
async validate(itemId: string, now = new Date()): Promise<SchedulerValidationResult> {
|
||||
const item = await this.getItem(itemId, now);
|
||||
const issues = this.validateItem(item);
|
||||
await this.recordEvent({
|
||||
item,
|
||||
type: 'validate',
|
||||
status: issues.some((issue) => issue.severity === 'error') ? 'failed' : 'success',
|
||||
summary:
|
||||
issues.length === 0
|
||||
? 'Scheduler item validation passed.'
|
||||
: 'Scheduler item has validation issues.',
|
||||
error: issues.find((issue) => issue.severity === 'error')?.message,
|
||||
now,
|
||||
});
|
||||
|
||||
return {
|
||||
itemId,
|
||||
ok: !issues.some((issue) => issue.severity === 'error'),
|
||||
issues,
|
||||
};
|
||||
}
|
||||
|
||||
async pause(itemId: string, now = new Date()): Promise<SchedulerRunResult> {
|
||||
const item = await this.getItem(itemId, now);
|
||||
if (!item.enabled) {
|
||||
throw new ValidationError(`Scheduler item is already paused: ${itemId}`);
|
||||
}
|
||||
|
||||
if (item.kind === 'scheduled-deliverable') {
|
||||
await this.deliverablesService.update(item.sourceId, { enabled: false });
|
||||
} else {
|
||||
await this.updateWorkflowSchedule(item.sourceId, { enabled: false });
|
||||
}
|
||||
|
||||
const state = this.currentState();
|
||||
state.items[itemId] = {
|
||||
...state.items[itemId],
|
||||
attempts: 0,
|
||||
nextAttemptAt: undefined,
|
||||
};
|
||||
const event = await this.recordEvent({
|
||||
item: { ...item, enabled: false },
|
||||
type: 'pause',
|
||||
status: 'success',
|
||||
summary: 'Scheduler item paused.',
|
||||
now,
|
||||
});
|
||||
await this.saveState();
|
||||
return { item: await this.getItem(itemId, now), event };
|
||||
}
|
||||
|
||||
async resume(itemId: string, now = new Date()): Promise<SchedulerRunResult> {
|
||||
const item = await this.getItem(itemId, now);
|
||||
if (item.enabled) {
|
||||
throw new ValidationError(`Scheduler item is already enabled: ${itemId}`);
|
||||
}
|
||||
|
||||
if (item.kind === 'scheduled-deliverable') {
|
||||
await this.deliverablesService.update(item.sourceId, { enabled: true });
|
||||
} else {
|
||||
await this.updateWorkflowSchedule(item.sourceId, { enabled: true });
|
||||
}
|
||||
|
||||
const state = this.currentState();
|
||||
state.items[itemId] = {
|
||||
...state.items[itemId],
|
||||
attempts: 0,
|
||||
nextAttemptAt: undefined,
|
||||
};
|
||||
const event = await this.recordEvent({
|
||||
item: { ...item, enabled: true },
|
||||
type: 'resume',
|
||||
status: 'success',
|
||||
summary: 'Scheduler item resumed.',
|
||||
now,
|
||||
});
|
||||
await this.saveState();
|
||||
return { item: await this.getItem(itemId, now), event };
|
||||
}
|
||||
|
||||
async runItem(
|
||||
itemId: string,
|
||||
trigger: Extract<SchedulerEventType, 'manual-run' | 'due-run'> = 'manual-run',
|
||||
now = new Date()
|
||||
): Promise<SchedulerRunResult> {
|
||||
const item = await this.getItem(itemId, now);
|
||||
if (this.runningItems.has(itemId)) {
|
||||
const event = await this.recordEvent({
|
||||
item,
|
||||
type: 'overlap',
|
||||
status: 'skipped',
|
||||
summary: 'Scheduler item is already running.',
|
||||
now,
|
||||
});
|
||||
return { item, event };
|
||||
}
|
||||
|
||||
const issues = this.validateItem(item);
|
||||
const blockingIssue = issues.find((issue) => issue.severity === 'error');
|
||||
if (blockingIssue) {
|
||||
const event = await this.recordEvent({
|
||||
item,
|
||||
type: trigger,
|
||||
status: 'failed',
|
||||
summary: 'Scheduler item failed validation before launch.',
|
||||
error: blockingIssue.message,
|
||||
now,
|
||||
});
|
||||
return { item: await this.getItem(itemId, now), event };
|
||||
}
|
||||
|
||||
this.runningItems.add(itemId);
|
||||
const startedAt = Date.now();
|
||||
try {
|
||||
const result =
|
||||
item.kind === 'scheduled-deliverable'
|
||||
? await this.runDeliverable(item, trigger, now, startedAt)
|
||||
: await this.runWorkflow(item, trigger, now, startedAt);
|
||||
await this.saveState();
|
||||
return result;
|
||||
} finally {
|
||||
this.runningItems.delete(itemId);
|
||||
}
|
||||
}
|
||||
|
||||
async runDue(now = new Date()): Promise<SchedulerDueRunResult> {
|
||||
if (this.runningDue) {
|
||||
return { checked: 0, executed: 0, skipped: 0, failed: 0, overlapping: true, events: [] };
|
||||
}
|
||||
|
||||
this.runningDue = true;
|
||||
try {
|
||||
const list = await this.list(now);
|
||||
const due = list.items.filter((item) => isItemDue(item, now.getTime()));
|
||||
const result: SchedulerDueRunResult = {
|
||||
checked: due.length,
|
||||
executed: 0,
|
||||
skipped: 0,
|
||||
failed: 0,
|
||||
overlapping: false,
|
||||
events: [],
|
||||
};
|
||||
|
||||
for (const item of due) {
|
||||
const run = await this.runItem(item.id, 'due-run', now);
|
||||
result.events.push(run.event);
|
||||
if (run.event.status === 'failed') result.failed++;
|
||||
else if (run.event.status === 'skipped') result.skipped++;
|
||||
else result.executed++;
|
||||
}
|
||||
|
||||
return result;
|
||||
} finally {
|
||||
this.runningDue = false;
|
||||
}
|
||||
}
|
||||
|
||||
private async runDeliverable(
|
||||
item: SchedulerItem,
|
||||
trigger: Extract<SchedulerEventType, 'manual-run' | 'due-run'>,
|
||||
now: Date,
|
||||
startedAt: number
|
||||
): Promise<SchedulerRunResult> {
|
||||
const result = await this.deliverablesService.get(item.sourceId);
|
||||
if (!result) throw new NotFoundError(`Scheduled deliverable not found: ${item.sourceId}`);
|
||||
|
||||
const runs: DeliverableRun[] = [];
|
||||
const runner = new ScheduledDeliverablesRunner({
|
||||
deliverablesService: {
|
||||
listDue: async () => [result.deliverable],
|
||||
recordRun: async (params) => {
|
||||
const run = await this.deliverablesService.recordRun(params);
|
||||
runs.push(run);
|
||||
return run;
|
||||
},
|
||||
},
|
||||
});
|
||||
const runnerResult = await runner.runDue(now);
|
||||
const run = runs.at(-1);
|
||||
const status = deliverableRunnerStatus(runnerResult);
|
||||
const summary =
|
||||
run?.summary ??
|
||||
(status === 'skipped'
|
||||
? 'Scheduled deliverable skipped.'
|
||||
: status === 'failed'
|
||||
? 'Scheduled deliverable failed.'
|
||||
: 'Scheduled deliverable executed.');
|
||||
const event = await this.recordEvent({
|
||||
item,
|
||||
type: trigger,
|
||||
status,
|
||||
summary,
|
||||
error: run?.error,
|
||||
sourceRunId: run?.id,
|
||||
durationMs: Date.now() - startedAt,
|
||||
now,
|
||||
nextRunAt: (await this.deliverablesService.get(item.sourceId))?.deliverable.nextRunAt,
|
||||
});
|
||||
return { item: await this.getItem(item.id, now), event };
|
||||
}
|
||||
|
||||
private async runWorkflow(
|
||||
item: SchedulerItem,
|
||||
trigger: Extract<SchedulerEventType, 'manual-run' | 'due-run'>,
|
||||
now: Date,
|
||||
startedAt: number
|
||||
): Promise<SchedulerRunResult> {
|
||||
const workflow = await this.workflowService.loadWorkflow(item.sourceId);
|
||||
if (!workflow) throw new NotFoundError(`Workflow not found: ${item.sourceId}`);
|
||||
|
||||
const dryRun = await this.workflowAuthoringService.dryRun({
|
||||
workflow,
|
||||
context: { clientMode: 'local', now: now.toISOString() },
|
||||
});
|
||||
const blocker = dryRun.messages.find((message) => message.severity === 'error');
|
||||
if (blocker) {
|
||||
const event = await this.recordEvent({
|
||||
item,
|
||||
type: trigger,
|
||||
status: 'failed',
|
||||
summary: 'Workflow schedule failed preflight validation.',
|
||||
error: blocker.message,
|
||||
durationMs: Date.now() - startedAt,
|
||||
now,
|
||||
});
|
||||
return { item: await this.getItem(item.id, now), event };
|
||||
}
|
||||
|
||||
const run = await this.workflowRunService.startRun(
|
||||
workflow.id,
|
||||
undefined,
|
||||
{
|
||||
scheduler: {
|
||||
itemId: item.id,
|
||||
trigger,
|
||||
runAt: now.toISOString(),
|
||||
},
|
||||
},
|
||||
workflow.config?.budget
|
||||
);
|
||||
const event = await this.recordEvent({
|
||||
item,
|
||||
type: trigger,
|
||||
status: 'started',
|
||||
summary: `Workflow run started: ${run.id}`,
|
||||
sourceRunId: run.id,
|
||||
durationMs: Date.now() - startedAt,
|
||||
now,
|
||||
nextRunAt: nextScheduledAt(workflow.schedule, now.toISOString()),
|
||||
});
|
||||
return { item: await this.getItem(item.id, now), event };
|
||||
}
|
||||
|
||||
private async updateWorkflowSchedule(
|
||||
workflowId: string,
|
||||
update: Partial<Pick<WorkflowSchedule, 'enabled'>>
|
||||
): Promise<void> {
|
||||
const workflow = await this.workflowService.loadWorkflow(workflowId);
|
||||
if (!workflow?.schedule) {
|
||||
throw new NotFoundError(`Scheduled workflow not found: ${workflowId}`);
|
||||
}
|
||||
workflow.schedule = { ...workflow.schedule, ...update };
|
||||
workflow.version = (workflow.version || 0) + 1;
|
||||
workflow.updatedAt = new Date().toISOString();
|
||||
await this.workflowService.saveWorkflow(workflow);
|
||||
}
|
||||
|
||||
private async buildItems(now: Date): Promise<SchedulerItem[]> {
|
||||
await this.ensureLoaded();
|
||||
const [deliverables, workflows] = await Promise.all([
|
||||
this.deliverablesService.list(),
|
||||
this.workflowService.listWorkflows(),
|
||||
]);
|
||||
|
||||
const items = [
|
||||
...deliverables.map((deliverable) => this.deliverableItem(deliverable)),
|
||||
...workflows
|
||||
.filter((workflow) => shouldExposeWorkflow(workflow))
|
||||
.map((workflow) => this.workflowItem(workflow, now)),
|
||||
];
|
||||
|
||||
return items.sort((a, b) => {
|
||||
const aNext = a.nextRunAt ? Date.parse(a.nextRunAt) : Number.POSITIVE_INFINITY;
|
||||
const bNext = b.nextRunAt ? Date.parse(b.nextRunAt) : Number.POSITIVE_INFINITY;
|
||||
if (aNext !== bNext) return aNext - bNext;
|
||||
return a.name.localeCompare(b.name);
|
||||
});
|
||||
}
|
||||
|
||||
private deliverableItem(deliverable: Deliverable): SchedulerItem {
|
||||
const id = itemId('scheduled-deliverable', deliverable.id);
|
||||
const state = this.currentState().items[id] ?? {};
|
||||
return this.decorateItem({
|
||||
id,
|
||||
kind: 'scheduled-deliverable',
|
||||
provider: 'local-server',
|
||||
sourceId: deliverable.id,
|
||||
name: deliverable.name,
|
||||
description: deliverable.description,
|
||||
enabled: deliverable.enabled,
|
||||
trigger: {
|
||||
mode: deliverable.schedule,
|
||||
description: deliverable.scheduleDescription,
|
||||
cronExpr: deliverable.cronExpr,
|
||||
customDueRunnerSupported: deliverable.schedule !== 'custom',
|
||||
},
|
||||
tags: deliverable.tags,
|
||||
nextRunAt: deliverable.nextRunAt,
|
||||
lastRunAt: deliverable.lastRunAt,
|
||||
lastStatus: state.lastStatus,
|
||||
lastSummary: state.lastSummary,
|
||||
lastError: state.lastError,
|
||||
sourceRunId: state.sourceRunId,
|
||||
health: 'healthy',
|
||||
healthSummary: 'Ready',
|
||||
retry: retryState(state),
|
||||
actions: baseActions(deliverable.enabled),
|
||||
});
|
||||
}
|
||||
|
||||
private workflowItem(workflow: WorkflowDefinition, now: Date): SchedulerItem {
|
||||
const id = itemId('workflow', workflow.id);
|
||||
const state = this.currentState().items[id] ?? {};
|
||||
const schedule = workflow.schedule as WorkflowSchedule;
|
||||
const nextRunAt = workflowNextRunAt(workflow, state, now);
|
||||
return this.decorateItem({
|
||||
id,
|
||||
kind: 'workflow',
|
||||
provider: 'local-server',
|
||||
sourceId: workflow.id,
|
||||
name: workflow.name,
|
||||
description: workflow.description,
|
||||
enabled: Boolean(schedule.enabled),
|
||||
trigger: {
|
||||
mode: schedule.mode,
|
||||
description: describeWorkflowSchedule(schedule),
|
||||
cronExpr: schedule.cronExpr,
|
||||
timezone: schedule.timezone,
|
||||
startAt: schedule.startAt,
|
||||
endAt: schedule.endAt,
|
||||
customDueRunnerSupported: schedule.mode !== 'custom',
|
||||
},
|
||||
tags: workflow.config?.telemetry_tags ?? [],
|
||||
nextRunAt,
|
||||
lastRunAt: state.lastRunAt,
|
||||
lastStatus: state.lastStatus,
|
||||
lastSummary: state.lastSummary,
|
||||
lastError: state.lastError,
|
||||
sourceRunId: state.sourceRunId,
|
||||
health: 'healthy',
|
||||
healthSummary: 'Ready',
|
||||
retry: retryState(state),
|
||||
actions: baseActions(Boolean(schedule.enabled)),
|
||||
});
|
||||
}
|
||||
|
||||
private decorateItem(item: SchedulerItem): SchedulerItem {
|
||||
const issues = this.validateItem(item);
|
||||
const retryBlocked =
|
||||
item.retry.attempts >= item.retry.maxAttempts && item.lastStatus === 'failed';
|
||||
const error = issues.find((issue) => issue.severity === 'error');
|
||||
const warning = issues.find((issue) => issue.severity === 'warning');
|
||||
|
||||
if (!item.enabled) {
|
||||
return { ...item, health: 'paused', healthSummary: 'Paused' };
|
||||
}
|
||||
if (retryBlocked || error) {
|
||||
return {
|
||||
...item,
|
||||
health: 'blocked',
|
||||
healthSummary: retryBlocked
|
||||
? 'Retry limit reached; run manually or resume to reset.'
|
||||
: (error?.message ?? 'Blocked'),
|
||||
};
|
||||
}
|
||||
if (warning) {
|
||||
return { ...item, health: 'warning', healthSummary: warning.message };
|
||||
}
|
||||
return { ...item, health: 'healthy', healthSummary: 'Ready' };
|
||||
}
|
||||
|
||||
private validateItem(item: SchedulerItem): SchedulerValidationIssue[] {
|
||||
const issues: SchedulerValidationIssue[] = [];
|
||||
if (!item.name.trim()) {
|
||||
issues.push({
|
||||
severity: 'error',
|
||||
path: 'name',
|
||||
message: 'Scheduler item is missing a name.',
|
||||
remediation: 'Name the scheduled deliverable or workflow.',
|
||||
});
|
||||
}
|
||||
if (item.enabled && item.trigger.mode === 'custom' && !item.trigger.cronExpr) {
|
||||
issues.push({
|
||||
severity: 'error',
|
||||
path: 'trigger.cronExpr',
|
||||
message: 'Custom schedule is missing a cron expression.',
|
||||
remediation: 'Add cronExpr or switch to a standard interval schedule.',
|
||||
});
|
||||
}
|
||||
if (item.enabled && !item.trigger.customDueRunnerSupported) {
|
||||
issues.push({
|
||||
severity: 'warning',
|
||||
path: 'trigger.mode',
|
||||
message:
|
||||
'Custom cron schedules are visible and manually runnable, but due-run execution is not enabled.',
|
||||
remediation: 'Add a cron adapter before relying on automatic due-run execution.',
|
||||
});
|
||||
}
|
||||
if (item.enabled && item.nextRunAt && !Number.isFinite(Date.parse(item.nextRunAt))) {
|
||||
issues.push({
|
||||
severity: 'error',
|
||||
path: 'nextRunAt',
|
||||
message: 'Next run timestamp is invalid.',
|
||||
remediation: 'Pause and resume the item to recalculate schedule state.',
|
||||
});
|
||||
}
|
||||
if (item.enabled && item.trigger.endAt && Date.parse(item.trigger.endAt) <= Date.now()) {
|
||||
issues.push({
|
||||
severity: 'warning',
|
||||
path: 'trigger.endAt',
|
||||
message: 'Schedule end time has passed.',
|
||||
remediation: 'Extend endAt or pause the schedule.',
|
||||
});
|
||||
}
|
||||
return issues;
|
||||
}
|
||||
|
||||
private async recordEvent(params: {
|
||||
item: SchedulerItem;
|
||||
type: SchedulerEventType;
|
||||
status: SchedulerRunStatus;
|
||||
summary: string;
|
||||
now: Date;
|
||||
durationMs?: number;
|
||||
error?: string;
|
||||
sourceRunId?: string;
|
||||
nextRunAt?: string;
|
||||
}): Promise<SchedulerEvent> {
|
||||
await this.ensureLoaded();
|
||||
const event: SchedulerEvent = {
|
||||
id: `sched_evt_${nanoid(10)}`,
|
||||
itemId: params.item.id,
|
||||
sourceId: params.item.sourceId,
|
||||
kind: params.item.kind,
|
||||
type: params.type,
|
||||
status: params.status,
|
||||
summary: params.summary,
|
||||
runAt: params.now.toISOString(),
|
||||
durationMs: params.durationMs,
|
||||
error: params.error,
|
||||
sourceRunId: params.sourceRunId,
|
||||
nextRunAt: params.nextRunAt,
|
||||
};
|
||||
|
||||
const state = this.currentState();
|
||||
const previous = state.items[params.item.id] ?? {};
|
||||
const failed = params.status === 'failed';
|
||||
const attempts = failed ? (previous.attempts ?? 0) + 1 : 0;
|
||||
state.items[params.item.id] = {
|
||||
...previous,
|
||||
attempts,
|
||||
nextAttemptAt:
|
||||
failed && attempts < DEFAULT_MAX_ATTEMPTS
|
||||
? retryAttemptAt(params.now, attempts)
|
||||
: undefined,
|
||||
lastRunAt: params.now.toISOString(),
|
||||
nextRunAt: params.nextRunAt ?? previous.nextRunAt,
|
||||
lastStatus: params.status,
|
||||
lastSummary: params.summary,
|
||||
lastError: params.error,
|
||||
sourceRunId: params.sourceRunId,
|
||||
};
|
||||
state.events.push(event);
|
||||
state.events = state.events.slice(-MAX_EVENTS);
|
||||
await this.saveState();
|
||||
await this.emitTelemetry(event);
|
||||
log.info(
|
||||
{ eventId: event.id, itemId: event.itemId, status: event.status },
|
||||
'Scheduler event recorded'
|
||||
);
|
||||
return event;
|
||||
}
|
||||
|
||||
private async emitTelemetry(event: SchedulerEvent): Promise<void> {
|
||||
const base = {
|
||||
taskId: event.itemId,
|
||||
project: SCHEDULER_PROJECT,
|
||||
agent: SCHEDULER_AGENT,
|
||||
attemptId: event.id,
|
||||
durationMs: event.durationMs,
|
||||
error: event.error,
|
||||
};
|
||||
if (event.status === 'failed') {
|
||||
await this.telemetryService.emit<RunTelemetryEvent>({
|
||||
...base,
|
||||
type: 'run.error',
|
||||
error: event.error ?? event.summary,
|
||||
});
|
||||
return;
|
||||
}
|
||||
await this.telemetryService.emit<RunTelemetryEvent>({
|
||||
...base,
|
||||
type: 'run.completed',
|
||||
success: event.status !== 'skipped',
|
||||
});
|
||||
}
|
||||
|
||||
private async ensureLoaded(): Promise<void> {
|
||||
if (this.state) return;
|
||||
try {
|
||||
const data = await fs.readFile(this.stateFile, 'utf-8');
|
||||
const parsed = JSON.parse(data) as SchedulerStateFile;
|
||||
this.state = {
|
||||
version: STATE_VERSION,
|
||||
items: parsed.items ?? {},
|
||||
events: Array.isArray(parsed.events) ? parsed.events.slice(-MAX_EVENTS) : [],
|
||||
};
|
||||
} catch {
|
||||
this.state = { version: STATE_VERSION, items: {}, events: [] };
|
||||
}
|
||||
}
|
||||
|
||||
private currentState(): SchedulerStateFile {
|
||||
if (!this.state) {
|
||||
throw new Error('Scheduler state has not been loaded');
|
||||
}
|
||||
return this.state;
|
||||
}
|
||||
|
||||
private async saveState(): Promise<void> {
|
||||
await fs.mkdir(path.dirname(this.stateFile), { recursive: true });
|
||||
await fs.writeFile(this.stateFile, JSON.stringify(this.state, null, 2));
|
||||
}
|
||||
}
|
||||
|
||||
function itemId(kind: SchedulerItemKind, sourceId: string): string {
|
||||
return `${kind}:${sourceId}`;
|
||||
}
|
||||
|
||||
function retryState(state: SchedulerItemState) {
|
||||
return {
|
||||
attempts: state.attempts ?? 0,
|
||||
maxAttempts: DEFAULT_MAX_ATTEMPTS,
|
||||
backoffMinutes: DEFAULT_BACKOFF_MINUTES,
|
||||
nextAttemptAt: state.nextAttemptAt,
|
||||
};
|
||||
}
|
||||
|
||||
function baseActions(enabled: boolean): SchedulerItem['actions'] {
|
||||
return {
|
||||
canRun: true,
|
||||
canPause: enabled,
|
||||
canResume: !enabled,
|
||||
canValidate: true,
|
||||
};
|
||||
}
|
||||
|
||||
function shouldExposeWorkflow(workflow: WorkflowDefinition): boolean {
|
||||
const schedule = workflow.schedule;
|
||||
if (!schedule) return false;
|
||||
return (
|
||||
schedule.mode !== 'manual' ||
|
||||
schedule.enabled ||
|
||||
workflow.outputTargets?.some((target) => target.type === 'scheduled-snapshot') === true
|
||||
);
|
||||
}
|
||||
|
||||
function workflowNextRunAt(
|
||||
workflow: WorkflowDefinition,
|
||||
state: SchedulerItemState,
|
||||
now: Date
|
||||
): string | undefined {
|
||||
const schedule = workflow.schedule;
|
||||
if (!schedule?.enabled || schedule.mode === 'manual' || schedule.mode === 'custom') {
|
||||
return state.nextRunAt;
|
||||
}
|
||||
if (state.nextRunAt) return state.nextRunAt;
|
||||
if (state.lastRunAt) return nextScheduledAt(schedule, state.lastRunAt);
|
||||
if (schedule.startAt) return schedule.startAt;
|
||||
return now.toISOString();
|
||||
}
|
||||
|
||||
function nextScheduledAt(
|
||||
schedule: WorkflowSchedule | undefined,
|
||||
baseAt: string
|
||||
): string | undefined {
|
||||
if (!schedule || schedule.mode === 'manual' || schedule.mode === 'custom') return undefined;
|
||||
return addScheduleInterval(schedule.mode, baseAt);
|
||||
}
|
||||
|
||||
function addScheduleInterval(mode: WorkflowScheduleMode, baseAt: string): string | undefined {
|
||||
const date = new Date(baseAt);
|
||||
if (!Number.isFinite(date.getTime())) return undefined;
|
||||
switch (mode) {
|
||||
case 'daily':
|
||||
date.setUTCDate(date.getUTCDate() + 1);
|
||||
break;
|
||||
case 'weekly':
|
||||
date.setUTCDate(date.getUTCDate() + 7);
|
||||
break;
|
||||
case 'biweekly':
|
||||
date.setUTCDate(date.getUTCDate() + 14);
|
||||
break;
|
||||
case 'monthly':
|
||||
date.setUTCMonth(date.getUTCMonth() + 1);
|
||||
break;
|
||||
default:
|
||||
return undefined;
|
||||
}
|
||||
return date.toISOString();
|
||||
}
|
||||
|
||||
function describeWorkflowSchedule(schedule: WorkflowSchedule): string {
|
||||
if (schedule.mode === 'custom')
|
||||
return schedule.cronExpr ? `Cron: ${schedule.cronExpr}` : 'Custom schedule';
|
||||
if (schedule.mode === 'daily') return 'Every day';
|
||||
if (schedule.mode === 'weekly') return 'Every week';
|
||||
if (schedule.mode === 'biweekly') return 'Every 2 weeks';
|
||||
if (schedule.mode === 'monthly') return 'Every month';
|
||||
return 'Manual';
|
||||
}
|
||||
|
||||
function deliverableRunnerStatus(result: {
|
||||
executed: number;
|
||||
failed: number;
|
||||
skipped: number;
|
||||
}): SchedulerRunStatus {
|
||||
if (result.failed > 0) return 'failed';
|
||||
if (result.executed > 0) return 'success';
|
||||
if (result.skipped > 0) return 'skipped';
|
||||
return 'skipped';
|
||||
}
|
||||
|
||||
function isItemDue(item: SchedulerItem, cutoff: number): boolean {
|
||||
if (!item.enabled || item.health === 'blocked' || item.health === 'paused') return false;
|
||||
if (!item.trigger.customDueRunnerSupported) return false;
|
||||
if (item.retry.nextAttemptAt && Date.parse(item.retry.nextAttemptAt) > cutoff) return false;
|
||||
if (!item.nextRunAt) return false;
|
||||
const next = Date.parse(item.nextRunAt);
|
||||
return Number.isFinite(next) && next <= cutoff;
|
||||
}
|
||||
|
||||
function retryAttemptAt(now: Date, attempts: number): string {
|
||||
const multiplier = Math.max(1, 2 ** Math.max(0, attempts - 1));
|
||||
const minutes = Math.min(DEFAULT_BACKOFF_MINUTES * multiplier, 60);
|
||||
return new Date(now.getTime() + minutes * 60_000).toISOString();
|
||||
}
|
||||
|
||||
let schedulerServiceInstance: SchedulerService | null = null;
|
||||
|
||||
export function getSchedulerService(): SchedulerService {
|
||||
if (!schedulerServiceInstance) {
|
||||
schedulerServiceInstance = new SchedulerService();
|
||||
}
|
||||
return schedulerServiceInstance;
|
||||
}
|
||||
|
|
@ -35,6 +35,7 @@ export * from './agent-budget.types.js';
|
|||
export * from './agent-profile-package.types.js';
|
||||
export * from './team-roster.types.js';
|
||||
export * from './workspace-capability.types.js';
|
||||
export * from './scheduler.types.js';
|
||||
export * from './watcher-policy.types.js';
|
||||
export * from './evidence.types.js';
|
||||
export * from './time-breakdown.types.js';
|
||||
|
|
|
|||
116
shared/src/types/scheduler.types.ts
Normal file
116
shared/src/types/scheduler.types.ts
Normal file
|
|
@ -0,0 +1,116 @@
|
|||
import type { WorkflowScheduleMode } from './workflow.js';
|
||||
|
||||
export type SchedulerDeliverableSchedule = 'daily' | 'weekly' | 'biweekly' | 'monthly' | 'custom';
|
||||
export type SchedulerItemKind = 'scheduled-deliverable' | 'workflow';
|
||||
export type SchedulerItemProvider = 'local-server';
|
||||
export type SchedulerHealth = 'healthy' | 'warning' | 'paused' | 'blocked';
|
||||
export type SchedulerRunStatus = 'success' | 'failed' | 'skipped' | 'started';
|
||||
export type SchedulerEventType =
|
||||
| 'due-run'
|
||||
| 'manual-run'
|
||||
| 'pause'
|
||||
| 'resume'
|
||||
| 'validate'
|
||||
| 'overlap';
|
||||
|
||||
export interface SchedulerTrigger {
|
||||
mode: SchedulerDeliverableSchedule | WorkflowScheduleMode;
|
||||
description: string;
|
||||
cronExpr?: string;
|
||||
timezone?: string;
|
||||
startAt?: string;
|
||||
endAt?: string;
|
||||
customDueRunnerSupported: boolean;
|
||||
}
|
||||
|
||||
export interface SchedulerRetryState {
|
||||
attempts: number;
|
||||
maxAttempts: number;
|
||||
backoffMinutes: number;
|
||||
nextAttemptAt?: string;
|
||||
}
|
||||
|
||||
export interface SchedulerItem {
|
||||
id: string;
|
||||
kind: SchedulerItemKind;
|
||||
provider: SchedulerItemProvider;
|
||||
sourceId: string;
|
||||
name: string;
|
||||
description: string;
|
||||
enabled: boolean;
|
||||
trigger: SchedulerTrigger;
|
||||
tags: string[];
|
||||
nextRunAt?: string;
|
||||
lastRunAt?: string;
|
||||
lastStatus?: SchedulerRunStatus;
|
||||
lastSummary?: string;
|
||||
lastError?: string;
|
||||
sourceRunId?: string;
|
||||
health: SchedulerHealth;
|
||||
healthSummary: string;
|
||||
retry: SchedulerRetryState;
|
||||
actions: {
|
||||
canRun: boolean;
|
||||
canPause: boolean;
|
||||
canResume: boolean;
|
||||
canValidate: boolean;
|
||||
};
|
||||
}
|
||||
|
||||
export interface SchedulerEvent {
|
||||
id: string;
|
||||
itemId: string;
|
||||
sourceId: string;
|
||||
kind: SchedulerItemKind;
|
||||
type: SchedulerEventType;
|
||||
status: SchedulerRunStatus;
|
||||
summary: string;
|
||||
runAt: string;
|
||||
durationMs?: number;
|
||||
error?: string;
|
||||
sourceRunId?: string;
|
||||
nextRunAt?: string;
|
||||
}
|
||||
|
||||
export interface SchedulerValidationIssue {
|
||||
severity: 'error' | 'warning' | 'info';
|
||||
path: string;
|
||||
message: string;
|
||||
remediation: string;
|
||||
}
|
||||
|
||||
export interface SchedulerValidationResult {
|
||||
itemId: string;
|
||||
ok: boolean;
|
||||
issues: SchedulerValidationIssue[];
|
||||
}
|
||||
|
||||
export interface SchedulerSummary {
|
||||
total: number;
|
||||
enabled: number;
|
||||
paused: number;
|
||||
due: number;
|
||||
failed: number;
|
||||
blocked: number;
|
||||
}
|
||||
|
||||
export interface SchedulerListResponse {
|
||||
generatedAt: string;
|
||||
summary: SchedulerSummary;
|
||||
items: SchedulerItem[];
|
||||
recentEvents: SchedulerEvent[];
|
||||
}
|
||||
|
||||
export interface SchedulerRunResult {
|
||||
item: SchedulerItem;
|
||||
event: SchedulerEvent;
|
||||
}
|
||||
|
||||
export interface SchedulerDueRunResult {
|
||||
checked: number;
|
||||
executed: number;
|
||||
skipped: number;
|
||||
failed: number;
|
||||
overlapping: boolean;
|
||||
events: SchedulerEvent[];
|
||||
}
|
||||
|
|
@ -257,6 +257,16 @@ const ROUTE_PERMISSIONS: RoutePermissionConfig[] = [
|
|||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
prefix: '/api/scheduler',
|
||||
read: 'workflow:read',
|
||||
write: 'workflow:write',
|
||||
overrides: [
|
||||
{ methods: ['POST'], path: /^\/items\/[^/]+\/run\/?$/, permissions: 'workflow:execute' },
|
||||
{ methods: ['POST'], path: /^\/items\/[^/]+\/validate\/?$/, permissions: 'workflow:read' },
|
||||
{ methods: ['POST'], path: /^\/due\/run\/?$/, permissions: 'workflow:execute' },
|
||||
],
|
||||
},
|
||||
{
|
||||
prefix: '/api/watcher-policies',
|
||||
read: 'policy:read',
|
||||
|
|
|
|||
|
|
@ -23,6 +23,7 @@ import {
|
|||
UserCog,
|
||||
Wrench,
|
||||
Network,
|
||||
CalendarClock,
|
||||
} from 'lucide-react';
|
||||
import { DEFAULT_FEATURE_SETTINGS } from '@veritas-kanban/shared';
|
||||
import type { ClientAuthPermission } from '@veritas-kanban/shared';
|
||||
|
|
@ -72,6 +73,9 @@ const LazyMaintenanceTab = lazy(() =>
|
|||
const LazyWorkspaceCapabilitiesTab = lazy(() =>
|
||||
import('./tabs/WorkspaceCapabilitiesTab').then((m) => ({ default: m.WorkspaceCapabilitiesTab }))
|
||||
);
|
||||
const LazySchedulerTab = lazy(() =>
|
||||
import('./tabs/SchedulerTab').then((m) => ({ default: m.SchedulerTab }))
|
||||
);
|
||||
|
||||
// ============ Tab Skeleton ============
|
||||
|
||||
|
|
@ -105,6 +109,7 @@ type TabId =
|
|||
| 'doc-freshness'
|
||||
| 'multi-user'
|
||||
| 'workspace-capabilities'
|
||||
| 'scheduler'
|
||||
| 'maintenance'
|
||||
| 'manage';
|
||||
|
||||
|
|
@ -130,6 +135,12 @@ const TABS: TabDef[] = [
|
|||
icon: Network,
|
||||
requiredPermission: 'workspace:read',
|
||||
},
|
||||
{
|
||||
id: 'scheduler',
|
||||
label: 'Scheduler',
|
||||
icon: CalendarClock,
|
||||
requiredPermission: 'workflow:read',
|
||||
},
|
||||
{ id: 'maintenance', label: 'Maintenance', icon: Wrench, requiredPermission: 'backup:read' },
|
||||
{ id: 'delegation', label: 'Delegation', icon: Plane, requiredPermission: 'agent:read' },
|
||||
{ id: 'tool-policies', label: 'Tool Policies', icon: Lock, requiredPermission: 'policy:read' },
|
||||
|
|
@ -414,6 +425,11 @@ export function SettingsDialog({ open, onOpenChange, defaultTab }: SettingsDialo
|
|||
<LazyWorkspaceCapabilitiesTab />
|
||||
</SettingsErrorBoundary>
|
||||
)}
|
||||
{activeTab === 'scheduler' && (
|
||||
<SettingsErrorBoundary tabName="Scheduler">
|
||||
<LazySchedulerTab />
|
||||
</SettingsErrorBoundary>
|
||||
)}
|
||||
{activeTab === 'delegation' && (
|
||||
<SettingsErrorBoundary tabName="Delegation">
|
||||
<LazyDelegationTab />
|
||||
|
|
|
|||
302
web/src/components/settings/tabs/SchedulerTab.tsx
Normal file
302
web/src/components/settings/tabs/SchedulerTab.tsx
Normal file
|
|
@ -0,0 +1,302 @@
|
|||
import {
|
||||
Badge,
|
||||
Button,
|
||||
Group,
|
||||
Loader,
|
||||
Paper,
|
||||
SimpleGrid,
|
||||
Stack,
|
||||
Text,
|
||||
Tooltip,
|
||||
} from '@mantine/core';
|
||||
import { CalendarClock, CheckCircle2, Pause, Play, RefreshCw, RotateCw } from 'lucide-react';
|
||||
import type { SchedulerEvent, SchedulerItem, SchedulerRunStatus } from '@veritas-kanban/shared';
|
||||
import { useIdentity } from '@/hooks/useIdentity';
|
||||
import {
|
||||
useScheduler,
|
||||
useSchedulerPause,
|
||||
useSchedulerResume,
|
||||
useSchedulerRunDue,
|
||||
useSchedulerRunItem,
|
||||
useSchedulerValidate,
|
||||
} from '@/hooks/useScheduler';
|
||||
import { useToast } from '@/hooks/useToast';
|
||||
|
||||
const EMPTY_ITEMS: SchedulerItem[] = [];
|
||||
const EMPTY_EVENTS: SchedulerEvent[] = [];
|
||||
|
||||
export function SchedulerTab() {
|
||||
const { hasPermission } = useIdentity();
|
||||
const { toast } = useToast();
|
||||
const scheduler = useScheduler();
|
||||
const runDue = useSchedulerRunDue();
|
||||
const runItem = useSchedulerRunItem();
|
||||
const pause = useSchedulerPause();
|
||||
const resume = useSchedulerResume();
|
||||
const validate = useSchedulerValidate();
|
||||
const canExecute = hasPermission('workflow:execute');
|
||||
const canWrite = hasPermission('workflow:write');
|
||||
const items = scheduler.data?.items ?? EMPTY_ITEMS;
|
||||
const events = scheduler.data?.recentEvents ?? EMPTY_EVENTS;
|
||||
|
||||
const mutate = async (action: () => Promise<unknown>, successTitle: string) => {
|
||||
try {
|
||||
await action();
|
||||
toast({ title: successTitle });
|
||||
} catch (error) {
|
||||
toast({
|
||||
title: 'Scheduler action failed',
|
||||
description: error instanceof Error ? error.message : 'Unknown error',
|
||||
variant: 'destructive',
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
if (scheduler.isLoading) {
|
||||
return (
|
||||
<Group gap="sm" className="text-muted-foreground">
|
||||
<Loader size="xs" />
|
||||
<Text size="sm">Loading scheduler...</Text>
|
||||
</Group>
|
||||
);
|
||||
}
|
||||
|
||||
return (
|
||||
<Stack gap="lg">
|
||||
<Group justify="space-between" align="center">
|
||||
<Group gap="xs">
|
||||
<CalendarClock className="h-4 w-4 text-muted-foreground" />
|
||||
<Text size="sm" fw={600}>
|
||||
Recurring Work Scheduler
|
||||
</Text>
|
||||
</Group>
|
||||
<Group gap="xs">
|
||||
<Tooltip label="Run due schedules">
|
||||
<Button
|
||||
size="xs"
|
||||
variant="light"
|
||||
color="gray"
|
||||
disabled={!canExecute || runDue.isPending}
|
||||
leftSection={<Play className="h-3.5 w-3.5" />}
|
||||
onClick={() => mutate(() => runDue.mutateAsync(), 'Due schedules checked')}
|
||||
>
|
||||
Run Due
|
||||
</Button>
|
||||
</Tooltip>
|
||||
<Tooltip label="Refresh scheduler">
|
||||
<Button
|
||||
size="xs"
|
||||
variant="subtle"
|
||||
color="gray"
|
||||
leftSection={<RefreshCw className="h-3.5 w-3.5" />}
|
||||
onClick={() => scheduler.refetch()}
|
||||
>
|
||||
Refresh
|
||||
</Button>
|
||||
</Tooltip>
|
||||
</Group>
|
||||
</Group>
|
||||
|
||||
{scheduler.data && (
|
||||
<SimpleGrid cols={{ base: 2, md: 5 }} spacing="sm">
|
||||
<SummaryStat label="Total" value={scheduler.data.summary.total} />
|
||||
<SummaryStat label="Enabled" value={scheduler.data.summary.enabled} />
|
||||
<SummaryStat label="Due" value={scheduler.data.summary.due} />
|
||||
<SummaryStat label="Failed" value={scheduler.data.summary.failed} />
|
||||
<SummaryStat label="Blocked" value={scheduler.data.summary.blocked} />
|
||||
</SimpleGrid>
|
||||
)}
|
||||
|
||||
<Stack gap="sm">
|
||||
{items.length === 0 ? (
|
||||
<Paper className="border border-dashed p-4 text-center" radius="md">
|
||||
<Text size="sm" c="dimmed">
|
||||
No recurring work is configured.
|
||||
</Text>
|
||||
</Paper>
|
||||
) : (
|
||||
items.map((item) => (
|
||||
<Paper key={item.id} className="border bg-card p-4" radius="md">
|
||||
<Stack gap="sm">
|
||||
<Group justify="space-between" align="flex-start">
|
||||
<Stack gap={2}>
|
||||
<Group gap="xs">
|
||||
<Text size="sm" fw={600}>
|
||||
{item.name}
|
||||
</Text>
|
||||
<Badge
|
||||
size="xs"
|
||||
color={item.kind === 'workflow' ? 'blue' : 'grape'}
|
||||
variant="light"
|
||||
>
|
||||
{item.kind === 'workflow' ? 'Workflow' : 'Deliverable'}
|
||||
</Badge>
|
||||
<HealthBadge item={item} />
|
||||
</Group>
|
||||
<Text size="xs" c="dimmed" lineClamp={2}>
|
||||
{item.description}
|
||||
</Text>
|
||||
</Stack>
|
||||
<Group gap="xs">
|
||||
<Button
|
||||
size="xs"
|
||||
variant="subtle"
|
||||
color="gray"
|
||||
disabled={validate.isPending}
|
||||
leftSection={<CheckCircle2 className="h-3.5 w-3.5" />}
|
||||
onClick={() =>
|
||||
mutate(() => validate.mutateAsync(item.id), 'Scheduler item validated')
|
||||
}
|
||||
>
|
||||
Validate
|
||||
</Button>
|
||||
<Button
|
||||
size="xs"
|
||||
variant="light"
|
||||
color="gray"
|
||||
disabled={!canExecute || runItem.isPending || !item.actions.canRun}
|
||||
leftSection={<Play className="h-3.5 w-3.5" />}
|
||||
onClick={() =>
|
||||
mutate(() => runItem.mutateAsync(item.id), 'Scheduler item run started')
|
||||
}
|
||||
>
|
||||
Run
|
||||
</Button>
|
||||
{item.enabled ? (
|
||||
<Button
|
||||
size="xs"
|
||||
variant="subtle"
|
||||
color="gray"
|
||||
disabled={!canWrite || pause.isPending || !item.actions.canPause}
|
||||
leftSection={<Pause className="h-3.5 w-3.5" />}
|
||||
onClick={() =>
|
||||
mutate(() => pause.mutateAsync(item.id), 'Scheduler item paused')
|
||||
}
|
||||
>
|
||||
Pause
|
||||
</Button>
|
||||
) : (
|
||||
<Button
|
||||
size="xs"
|
||||
variant="subtle"
|
||||
color="gray"
|
||||
disabled={!canWrite || resume.isPending || !item.actions.canResume}
|
||||
leftSection={<RotateCw className="h-3.5 w-3.5" />}
|
||||
onClick={() =>
|
||||
mutate(() => resume.mutateAsync(item.id), 'Scheduler item resumed')
|
||||
}
|
||||
>
|
||||
Resume
|
||||
</Button>
|
||||
)}
|
||||
</Group>
|
||||
</Group>
|
||||
<SimpleGrid cols={{ base: 1, md: 4 }} spacing="xs">
|
||||
<Meta label="Schedule" value={item.trigger.description} />
|
||||
<Meta label="Next" value={formatDate(item.nextRunAt)} />
|
||||
<Meta label="Last" value={formatDate(item.lastRunAt)} />
|
||||
<Meta label="Retry" value={`${item.retry.attempts}/${item.retry.maxAttempts}`} />
|
||||
</SimpleGrid>
|
||||
{item.lastSummary && (
|
||||
<Text size="xs" c={item.lastStatus === 'failed' ? 'red' : 'dimmed'}>
|
||||
{item.lastSummary}
|
||||
</Text>
|
||||
)}
|
||||
</Stack>
|
||||
</Paper>
|
||||
))
|
||||
)}
|
||||
</Stack>
|
||||
|
||||
{events.length > 0 && (
|
||||
<Stack gap="sm">
|
||||
<Text size="sm" fw={600}>
|
||||
Recent Events
|
||||
</Text>
|
||||
<Stack gap="xs">
|
||||
{events.slice(0, 6).map((event) => (
|
||||
<Group key={event.id} justify="space-between" className="rounded border px-3 py-2">
|
||||
<Stack gap={0}>
|
||||
<Text size="xs" fw={600}>
|
||||
{event.summary}
|
||||
</Text>
|
||||
<Text size="xs" c="dimmed">
|
||||
{event.itemId} · {formatDate(event.runAt)}
|
||||
</Text>
|
||||
</Stack>
|
||||
<StatusBadge status={event.status} />
|
||||
</Group>
|
||||
))}
|
||||
</Stack>
|
||||
</Stack>
|
||||
)}
|
||||
</Stack>
|
||||
);
|
||||
}
|
||||
|
||||
function SummaryStat({ label, value }: { label: string; value: number }) {
|
||||
return (
|
||||
<Paper className="border bg-card p-3" radius="md">
|
||||
<Text size="xs" c="dimmed">
|
||||
{label}
|
||||
</Text>
|
||||
<Text size="lg" fw={700}>
|
||||
{value}
|
||||
</Text>
|
||||
</Paper>
|
||||
);
|
||||
}
|
||||
|
||||
function Meta({ label, value }: { label: string; value: string }) {
|
||||
return (
|
||||
<Stack gap={0}>
|
||||
<Text size="xs" c="dimmed">
|
||||
{label}
|
||||
</Text>
|
||||
<Text size="xs" fw={600}>
|
||||
{value}
|
||||
</Text>
|
||||
</Stack>
|
||||
);
|
||||
}
|
||||
|
||||
function HealthBadge({ item }: { item: SchedulerItem }) {
|
||||
const color =
|
||||
item.health === 'healthy'
|
||||
? 'green'
|
||||
: item.health === 'warning'
|
||||
? 'yellow'
|
||||
: item.health === 'paused'
|
||||
? 'gray'
|
||||
: 'red';
|
||||
return (
|
||||
<Tooltip label={item.healthSummary}>
|
||||
<Badge size="xs" color={color} variant="light">
|
||||
{item.health}
|
||||
</Badge>
|
||||
</Tooltip>
|
||||
);
|
||||
}
|
||||
|
||||
function StatusBadge({ status }: { status: SchedulerRunStatus }) {
|
||||
const color =
|
||||
status === 'success'
|
||||
? 'green'
|
||||
: status === 'started'
|
||||
? 'blue'
|
||||
: status === 'skipped'
|
||||
? 'gray'
|
||||
: 'red';
|
||||
return (
|
||||
<Badge size="xs" color={color} variant="light">
|
||||
{status}
|
||||
</Badge>
|
||||
);
|
||||
}
|
||||
|
||||
function formatDate(value?: string): string {
|
||||
if (!value) return 'Not set';
|
||||
const date = new Date(value);
|
||||
if (!Number.isFinite(date.getTime())) return 'Invalid';
|
||||
return date.toLocaleString();
|
||||
}
|
||||
51
web/src/hooks/useScheduler.ts
Normal file
51
web/src/hooks/useScheduler.ts
Normal file
|
|
@ -0,0 +1,51 @@
|
|||
import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query';
|
||||
import { api } from '@/lib/api';
|
||||
|
||||
export const SCHEDULER_KEY = ['scheduler'] as const;
|
||||
|
||||
export function useScheduler() {
|
||||
return useQuery({
|
||||
queryKey: SCHEDULER_KEY,
|
||||
queryFn: api.scheduler.list,
|
||||
});
|
||||
}
|
||||
|
||||
export function useSchedulerRunDue() {
|
||||
const queryClient = useQueryClient();
|
||||
return useMutation({
|
||||
mutationFn: api.scheduler.runDue,
|
||||
onSuccess: () => queryClient.invalidateQueries({ queryKey: SCHEDULER_KEY }),
|
||||
});
|
||||
}
|
||||
|
||||
export function useSchedulerRunItem() {
|
||||
const queryClient = useQueryClient();
|
||||
return useMutation({
|
||||
mutationFn: (itemId: string) => api.scheduler.runItem(itemId),
|
||||
onSuccess: () => queryClient.invalidateQueries({ queryKey: SCHEDULER_KEY }),
|
||||
});
|
||||
}
|
||||
|
||||
export function useSchedulerPause() {
|
||||
const queryClient = useQueryClient();
|
||||
return useMutation({
|
||||
mutationFn: (itemId: string) => api.scheduler.pause(itemId),
|
||||
onSuccess: () => queryClient.invalidateQueries({ queryKey: SCHEDULER_KEY }),
|
||||
});
|
||||
}
|
||||
|
||||
export function useSchedulerResume() {
|
||||
const queryClient = useQueryClient();
|
||||
return useMutation({
|
||||
mutationFn: (itemId: string) => api.scheduler.resume(itemId),
|
||||
onSuccess: () => queryClient.invalidateQueries({ queryKey: SCHEDULER_KEY }),
|
||||
});
|
||||
}
|
||||
|
||||
export function useSchedulerValidate() {
|
||||
const queryClient = useQueryClient();
|
||||
return useMutation({
|
||||
mutationFn: (itemId: string) => api.scheduler.validate(itemId),
|
||||
onSuccess: () => queryClient.invalidateQueries({ queryKey: SCHEDULER_KEY }),
|
||||
});
|
||||
}
|
||||
|
|
@ -30,6 +30,7 @@ import { timeBreakdownsApi } from './time-breakdowns';
|
|||
import { sandboxPoliciesApi } from './sandbox-policies';
|
||||
import { runSessionsApi } from './run-sessions';
|
||||
import { workspaceCapabilitiesApi } from './workspace-capabilities';
|
||||
import { schedulerApi } from './scheduler';
|
||||
|
||||
// Assemble the full API object (matches original structure exactly)
|
||||
export const api = {
|
||||
|
|
@ -71,6 +72,7 @@ export const api = {
|
|||
sandboxPolicies: sandboxPoliciesApi,
|
||||
runSessions: runSessionsApi,
|
||||
workspaceCapabilities: workspaceCapabilitiesApi,
|
||||
scheduler: schedulerApi,
|
||||
};
|
||||
|
||||
export type {
|
||||
|
|
|
|||
41
web/src/lib/api/scheduler.ts
Normal file
41
web/src/lib/api/scheduler.ts
Normal file
|
|
@ -0,0 +1,41 @@
|
|||
import type {
|
||||
SchedulerDueRunResult,
|
||||
SchedulerListResponse,
|
||||
SchedulerRunResult,
|
||||
SchedulerValidationResult,
|
||||
} from '@veritas-kanban/shared';
|
||||
import { API_BASE, apiFetch } from './helpers';
|
||||
|
||||
function itemPath(itemId: string, action?: string): string {
|
||||
const encoded = encodeURIComponent(itemId);
|
||||
return `${API_BASE}/scheduler/items/${encoded}${action ? `/${action}` : ''}`;
|
||||
}
|
||||
|
||||
export const schedulerApi = {
|
||||
list: () => apiFetch<SchedulerListResponse>(`${API_BASE}/scheduler`),
|
||||
|
||||
runDue: () =>
|
||||
apiFetch<SchedulerDueRunResult>(`${API_BASE}/scheduler/due/run`, {
|
||||
method: 'POST',
|
||||
}),
|
||||
|
||||
runItem: (itemId: string) =>
|
||||
apiFetch<SchedulerRunResult>(itemPath(itemId, 'run'), {
|
||||
method: 'POST',
|
||||
}),
|
||||
|
||||
pause: (itemId: string) =>
|
||||
apiFetch<SchedulerRunResult>(itemPath(itemId, 'pause'), {
|
||||
method: 'POST',
|
||||
}),
|
||||
|
||||
resume: (itemId: string) =>
|
||||
apiFetch<SchedulerRunResult>(itemPath(itemId, 'resume'), {
|
||||
method: 'POST',
|
||||
}),
|
||||
|
||||
validate: (itemId: string) =>
|
||||
apiFetch<SchedulerValidationResult>(itemPath(itemId, 'validate'), {
|
||||
method: 'POST',
|
||||
}),
|
||||
};
|
||||
Loading…
Add table
Reference in a new issue