scriptoriumadmin/veritas-kanban
1
0
Fork 0
mirror of https://github.com/BradGroux/veritas-kanban.git synced 2026-08-28 02:44:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

refactor: move workflow runs behind storage (#1207)

Browse source
This commit is contained in:
Brad Groux 2026-08-23 19:09:57 -05:00 • committed by GitHub
parent a38d42c005
commit 392e65d551
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 443 additions and 170 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
docs/architecture/service-filesystem-boundary.json
View file

@ -1,6 +1,6 @@
{
"schemaVersion": 1,
"maximumEntries": 37,
"maximumEntries": 36,
"entries": [
{
"path": "server/src/services/agent-health-service.ts",
@ -217,12 +217,6 @@
"category": "authoritative-persistence",
"owner": "#1187",
"rationale": "Operational evidence storage migration is tracked in issue #1187."
},
{
"path": "server/src/services/workflow-run-service.ts",
"category": "authoritative-persistence",
"owner": "#1186",
"rationale": "Coordination-state storage migration is tracked in issue #1186."
}
]
}

1
docs/testing/critical-path-coverage.json
View file

@ -95,6 +95,7 @@
"src/__tests__/work-product-schemas.test.ts",
"src/__tests__/workflow-definition-repository.test.ts",
"src/__tests__/workflow-execution-file-repository.test.ts",
"src/__tests__/workflow-run-repository.test.ts",
"src/__tests__/workflow-run-service.test.ts",
"src/__tests__/workflow-service.test.ts",
"src/__tests__/workflow-step-executor-codex.test.ts",

133
server/src/__tests__/workflow-run-repository.test.ts Normal file
View file

@ -0,0 +1,133 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { lstat, mkdir, mkdtemp, readFile, rm, symlink, writeFile } from 'node:fs/promises';
import path from 'node:path';
import type { WorkflowDefinition, WorkflowRun } from '../types/workflow.js';
import { FileWorkflowRunRepository } from '../storage/workflow-run-repository.js';
vi.mock('node:fs/promises', async (importOriginal) => {
const actual = await importOriginal<typeof import('node:fs/promises')>();
return { ...actual, lstat: vi.fn(actual.lstat) };
});
function run(id: string, revision = 1, startedAt = '2026-08-23T00:00:00.000Z'): WorkflowRun {
return {
id,
workflowId: 'workflow-1',
workflowVersion: 1,
status: 'running',
context: {},
steps: [],
startedAt,
revision,
};
}
describe('FileWorkflowRunRepository', () => {
let root: string;
let runsDir: string;
let repository: FileWorkflowRunRepository;
beforeEach(async () => {
root = await mkdtemp(path.join(process.cwd(), '.veritas-workflow-runs-'));
runsDir = path.join(root, 'runs');
repository = new FileWorkflowRunRepository(runsDir);
});
afterEach(async () => {
await rm(root, { recursive: true, force: true });
});
it('enforces revisions and preserves one concurrent winner', async () => {
await expect(repository.get('run_one')).resolves.toBeNull();
await expect(repository.save(run('run_one'), 0)).resolves.toBe(true);
const results = await Promise.all([
repository.save(run('run_one', 2), 1),
repository.save({ ...run('run_one', 2), status: 'completed' }, 1),
]);
expect(results.filter(Boolean)).toHaveLength(1);
await expect(repository.save(run('run_one', 3), 1)).resolves.toBe(false);
await expect(repository.get('run_one')).resolves.toMatchObject({ revision: 2 });
});
it('lists filtered runs and writes immutable workflow snapshots', async () => {
await repository.save(run('run_old', 1, '2026-08-22T00:00:00.000Z'), 0);
await repository.save(run('run_new', 1, '2026-08-23T00:00:00.000Z'), 0);
await expect(repository.list({ workflowId: 'workflow-1' })).resolves.toMatchObject([
{ id: 'run_new' },
{ id: 'run_old' },
]);
await expect(repository.listMetadata({ status: 'missing' })).resolves.toEqual([]);
await repository.saveWorkflowSnapshot('run_new', {
id: 'workflow-1',
name: 'Workflow',
version: 1,
steps: [],
} as WorkflowDefinition);
await expect(
readFile(path.join(runsDir, 'run_new', 'workflow.yml'), 'utf8')
).resolves.toContain('workflow-1');
});
it('rejects symbolic links, changed files, and non-file state', async () => {
const runDir = path.join(runsDir, 'run_unsafe');
await mkdir(runDir, { recursive: true });
const runPath = path.join(runDir, 'run.json');
const target = path.join(root, 'outside.json');
await writeFile(target, JSON.stringify(run('run_unsafe')), 'utf8');
await symlink(target, runPath);
await expect(repository.get('run_unsafe')).rejects.toThrow(/symbolic link/i);
await rm(runPath);
await writeFile(runPath, JSON.stringify(run('run_unsafe')), 'utf8');
const actual = await vi.importActual<typeof import('node:fs/promises')>('node:fs/promises');
vi.mocked(lstat).mockImplementation(async (filePath) => {
const stats = await actual.lstat(filePath);
if (path.resolve(String(filePath)) !== runPath) return stats;
return Object.assign(Object.create(Object.getPrototypeOf(stats)), stats, {
ino: stats.ino + 1,
});
});
await expect(repository.get('run_unsafe')).rejects.toThrow(/changed file/i);
vi.mocked(lstat).mockImplementation(actual.lstat);
await rm(runPath);
await mkdir(runPath);
await expect(repository.get('run_unsafe')).rejects.toThrow(/bounded regular file/i);
});
it('fails closed for oversized content and unsafe run directories', async () => {
await expect(
repository.save(
{
...run('run_oversized'),
context: { payload: 'x'.repeat(16 * 1024 * 1024) },
},
0
)
).rejects.toThrow(/16 MiB storage limit/);
await expect(
repository.saveWorkflowSnapshot('run_snapshot', {
id: 'workflow-1',
name: 'x'.repeat(4 * 1024 * 1024),
version: 1,
steps: [],
} as WorkflowDefinition)
).rejects.toThrow(/4 MiB storage limit/);
await mkdir(path.join(runsDir, 'run_..invalid'), { recursive: true });
await expect(repository.list()).resolves.toEqual([]);
await rm(runsDir, { recursive: true });
await writeFile(runsDir, 'not a directory', 'utf8');
await expect(repository.list()).rejects.toThrow();
await expect(repository.get('run_regular')).rejects.toThrow(/regular directory/i);
await rm(runsDir);
const targetDir = path.join(root, 'linked-runs');
await mkdir(targetDir);
await symlink(targetDir, runsDir);
await expect(repository.list()).rejects.toThrow(/regular directory/i);
await expect(repository.save(run('run_linked'), 0)).rejects.toThrow(/regular directory/i);
});
});

68
server/src/__tests__/workflow-run-service.test.ts
View file

@ -665,6 +665,74 @@ describe('WorkflowRunService', () => {
]);
});
it('validates human-gate approval and rejection state before mutation', async () => {
const repository = (service as any).repository;
const persist = async (id: string, overrides: Record<string, any> = {}) => {
await repository.save(
{
id,
workflowId: 'wf-1',
workflowVersion: 3,
status: 'blocked',
context: {},
steps: [],
startedAt: '2026-08-23T00:00:00.000Z',
revision: 1,
...overrides,
},
0
);
};
await expect(
service.approveGateStep('run_1111111111_missing', 'gate-1', 'operator')
).rejects.toThrow(/not found/);
await persist('run_1111111111_running', { status: 'running' });
await expect(
service.approveGateStep('run_1111111111_running', 'gate-1', 'operator')
).rejects.toThrow(/is not blocked/);
await persist('run_1111111111_nogate');
await expect(
service.approveGateStep('run_1111111111_nogate', 'gate-1', 'operator')
).rejects.toThrow(/not blocked at a human gate/);
await persist('run_1111111111_wronggate', {
context: { _gateBlock: { stepId: 'gate-2' } },
});
await expect(
service.approveGateStep('run_1111111111_wronggate', 'gate-1', 'operator')
).rejects.toThrow(/blocked at gate/);
await persist('run_1111111111_nostep', {
context: { _gateBlock: { stepId: 'gate-1' } },
});
await expect(
service.approveGateStep('run_1111111111_nostep', 'gate-1', 'operator')
).rejects.toThrow(/Step gate-1 not found/);
await expect(
service.rejectGateStep('run_1111111111_rejectmissing', 'gate-1', 'operator')
).rejects.toThrow(/not found/);
await persist('run_1111111111_rejectrunning', { status: 'running' });
await expect(
service.rejectGateStep('run_1111111111_rejectrunning', 'gate-1', 'operator')
).rejects.toThrow(/is not blocked/);
await persist('run_1111111111_rejectnogate');
await expect(
service.rejectGateStep('run_1111111111_rejectnogate', 'gate-1', 'operator')
).rejects.toThrow(/not blocked at a human gate/);
await persist('run_1111111111_rejectwrong', {
context: { _gateBlock: { stepId: 'gate-2' } },
});
await expect(
service.rejectGateStep('run_1111111111_rejectwrong', 'gate-1', 'operator')
).rejects.toThrow(/blocked at gate/);
await persist('run_1111111111_rejectnostep', {
context: { _gateBlock: { stepId: 'gate-1' } },
});
await expect(
service.rejectGateStep('run_1111111111_rejectnostep', 'gate-1', 'operator')
).rejects.toThrow(/Step gate-1 not found/);
});
it('rejects invalid ids, missing workflows, invalid metadata reads, and incompatible agent escalation', async () => {
await expect(service.getRun('../bad')).rejects.toThrow(/illegal path characters/);
await expect(service.getRun('run_invalid')).rejects.toThrow(/format is invalid/);

176
server/src/services/workflow-run-service.ts
View file

@ -3,8 +3,6 @@
* Phase 1: Core Engine (sequential steps, basic retry logic)
*/
import fs from 'fs/promises';
import path from 'path';
import { createHash } from 'node:crypto';
import { nanoid } from 'nanoid';
import {
@ -45,14 +43,16 @@ import { broadcastWorkflowStatus } from './broadcast-service.js';
import { getTaskService } from './task-service.js';
import { SqliteDatabase, type SqliteConnectionOptions } from '../storage/sqlite/database.js';
import { SqliteWorkflowRunRepository } from '../storage/sqlite/workflow-repositories.js';
import {
FileWorkflowRunRepository,
type WorkflowRunRepository,
} from '../storage/workflow-run-repository.js';
import { getConfigService } from './config-service.js';
import { getAgentBudgetService } from './agent-budget-service.js';
import { getGovernanceTraceService } from './governance-trace-service.js';
import { ConflictError, NotFoundError, ValidationError } from '../middleware/error-handler.js';
import { RunRecoveryPolicyService } from './run-recovery-policy-service.js';
import { getAgentRoutingService } from './agent-routing-service.js';
import { atomicWriteFile } from '../storage/fs-helpers.js';
import { withFileLock } from './file-lock.js';
import {
AdmissionControlService,
getAdmissionControlService,
@ -118,7 +118,7 @@ export class WorkflowRunService {
private runsDir: string;
private workflowService: ReturnType<typeof getWorkflowService>;
private stepExecutor: WorkflowStepExecutor;
private readonly repository: SqliteWorkflowRunRepository | null = null;
private readonly repository: WorkflowRunRepository;
private readonly sqliteDatabase: SqliteDatabase | null = null;
private readonly ownsSqliteDatabase: boolean = false;
private readonly runRecoveryPolicy: RunRecoveryPolicyService;
@ -147,15 +147,9 @@ export class WorkflowRunService {
this.ownsSqliteDatabase = !resolvedOptions.sqliteDatabase;
this.sqliteDatabase.open();
this.repository = new SqliteWorkflowRunRepository(this.sqliteDatabase);
} else {
this.repository = new FileWorkflowRunRepository(this.runsDir);
}
if (!this.repository) {
this.ensureDirectories();
}
}
private async ensureDirectories(): Promise<void> {
await fs.mkdir(this.runsDir, { recursive: true });
}
private normalizeRunId(runId: string): string {
@ -2322,19 +2316,7 @@ export class WorkflowRunService {
*/
async getRun(runId: string): Promise<WorkflowRun | null> {
const safeRunId = this.normalizeRunId(runId);
if (this.repository) {
return this.repository.get(safeRunId);
}
const runPath = path.join(this.runsDir, safeRunId, 'run.json');
try {
const content = await fs.readFile(runPath, 'utf-8');
return JSON.parse(content) as WorkflowRun;
} catch (err: unknown) {
if (err && typeof err === 'object' && 'code' in err && err.code === 'ENOENT') return null;
throw err;
}
return this.repository.get(safeRunId);
}
/**
@ -2345,45 +2327,7 @@ export class WorkflowRunService {
workflowId?: string;
status?: string;
}): Promise<WorkflowRun[]> {
if (this.repository) {
return this.repository.list(filters);
}
const runDirs = await fs.readdir(this.runsDir).catch(() => []);
const runs: WorkflowRun[] = [];
for (const dir of runDirs) {
if (!dir.startsWith('run_')) continue;
let run: WorkflowRun | null;
try {
run = await this.getRun(dir);
} catch (err) {
if (err instanceof ValidationError) {
log.warn({ runDir: dir }, 'Skipping run directory with invalid ID');
continue;
}
throw err;
}
if (!run) continue;
// Apply filters
if (filters?.taskId && run.taskId !== filters.taskId) continue;
if (filters?.workflowId && run.workflowId !== filters.workflowId) continue;
if (filters?.status && run.status !== filters.status) continue;
runs.push(run);
}
// Sort by startedAt descending
runs.sort(
(a, b) =>
new Date(b.startedAt).getTime() - new Date(a.startedAt).getTime() ||
b.id.localeCompare(a.id)
);
return runs;
return this.repository.list(filters);
}
/**
@ -2409,64 +2353,7 @@ export class WorkflowRunService {
>
>
> {
if (this.repository) {
const metadata = this.repository.listMetadata(filters);
log.info({ count: metadata.length }, 'Listed run metadata');
return metadata;
}
const runDirs = await fs.readdir(this.runsDir).catch(() => []);
const metadata: Array<
Pick<
WorkflowRun,
| 'id'
| 'workflowId'
| 'workflowVersion'
| 'taskId'
| 'status'
| 'startedAt'
| 'completedAt'
| 'error'
>
> = [];
for (const dir of runDirs) {
if (!dir.startsWith('run_')) continue;
const runPath = path.join(this.runsDir, dir, 'run.json');
try {
const content = await fs.readFile(runPath, 'utf-8');
const run = JSON.parse(content) as WorkflowRun;
// Apply filters
if (filters?.taskId && run.taskId !== filters.taskId) continue;
if (filters?.workflowId && run.workflowId !== filters.workflowId) continue;
if (filters?.status && run.status !== filters.status) continue;
metadata.push({
id: run.id,
workflowId: run.workflowId,
workflowVersion: run.workflowVersion,
taskId: run.taskId,
status: run.status,
startedAt: run.startedAt,
completedAt: run.completedAt,
error: run.error,
});
} catch (err: unknown) {
log.warn({ runDir: dir, err }, 'Failed to read run metadata');
continue;
}
}
// Sort by startedAt descending
metadata.sort(
(a, b) =>
new Date(b.startedAt).getTime() - new Date(a.startedAt).getTime() ||
b.id.localeCompare(a.id)
);
const metadata = await this.repository.listMetadata(filters);
log.info({ count: metadata.length }, 'Listed run metadata');
return metadata;
}
@ -2803,36 +2690,9 @@ export class WorkflowRunService {
lastCheckpoint: new Date().toISOString(),
};
if (this.repository) {
if (!this.repository.save(nextRun, expectedRevision)) {
throw new WorkflowRunChangedError(run.id, expectedRevision);
}
Object.assign(run, nextRun);
return;
if (!(await this.repository.save(nextRun, expectedRevision))) {
throw new WorkflowRunChangedError(run.id, expectedRevision);
}
const runDir = path.join(this.runsDir, run.id);
await fs.mkdir(runDir, { recursive: true });
const runPath = path.join(runDir, 'run.json');
await withFileLock(runPath, async () => {
let current: WorkflowRun | null = null;
try {
current = JSON.parse(await fs.readFile(runPath, 'utf-8')) as WorkflowRun;
} catch (error) {
if (!error || typeof error !== 'object' || !('code' in error) || error.code !== 'ENOENT') {
throw error;
}
}
const currentRevision = current?.revision ?? 0;
if (
(current && currentRevision !== expectedRevision) ||
(!current && expectedRevision !== 0)
) {
throw new WorkflowRunChangedError(run.id, expectedRevision, current?.revision);
}
await atomicWriteFile(runPath, JSON.stringify(nextRun, null, 2));
});
Object.assign(run, nextRun);
}
@ -2840,17 +2700,7 @@ export class WorkflowRunService {
* Snapshot workflow YAML into run directory (for version immutability)
*/
private async snapshotWorkflow(runId: string, workflow: WorkflowDefinition): Promise<void> {
if (this.repository) {
this.repository.saveWorkflowSnapshot(runId, workflow);
return;
}
const runDir = path.join(this.runsDir, runId);
await fs.mkdir(runDir, { recursive: true });
const snapshotPath = path.join(runDir, 'workflow.yml');
const yaml = await import('yaml');
await fs.writeFile(snapshotPath, yaml.stringify(workflow), 'utf-8');
await this.repository.saveWorkflowSnapshot(runId, workflow);
}
dispose(): void {

6
server/src/storage/index.ts
View file

@ -64,6 +64,12 @@ export {
FileWorkflowExecutionFileRepository,
type WorkflowExecutionFileRepository,
} from './workflow-execution-file-repository.js';
export {
FileWorkflowRunRepository,
type WorkflowRunFilters,
type WorkflowRunMetadata,
type WorkflowRunRepository,
} from './workflow-run-repository.js';
export { FileScheduledDeliverablesStore } from './scheduled-deliverables-repository.js';
export { FileBroadcastRepository } from './broadcast-repository.js';
export {

221
server/src/storage/workflow-run-repository.ts Normal file
View file

@ -0,0 +1,221 @@
import { constants } from 'node:fs';
import { lstat, mkdir, open, readdir } from 'node:fs/promises';
import path from 'node:path';
import yaml from 'yaml';
import type { WorkflowDefinition, WorkflowRun } from '../types/workflow.js';
import { withFileLock } from '../services/file-lock.js';
import { ensureWithinBase, validatePathSegment } from '../utils/sanitize.js';
import { atomicWriteFile } from './fs-helpers.js';
const MAX_WORKFLOW_RUN_BYTES = 16 * 1024 * 1024;
const MAX_WORKFLOW_SNAPSHOT_BYTES = 4 * 1024 * 1024;
export type WorkflowRunFilters = { taskId?: string; workflowId?: string; status?: string };
export type WorkflowRunMetadata = Pick<
WorkflowRun,
| 'id'
| 'workflowId'
| 'workflowVersion'
| 'taskId'
| 'status'
| 'startedAt'
| 'completedAt'
| 'error'
>;
type MaybePromise<T> = T | Promise<T>;
export interface WorkflowRunRepository {
get(runId: string): MaybePromise<WorkflowRun | null>;
list(filters?: WorkflowRunFilters): MaybePromise<WorkflowRun[]>;
listMetadata(filters?: WorkflowRunFilters): MaybePromise<WorkflowRunMetadata[]>;
save(run: WorkflowRun, expectedRevision?: number): MaybePromise<boolean>;
saveWorkflowSnapshot(runId: string, workflow: WorkflowDefinition): MaybePromise<void>;
}
export class FileWorkflowRunRepository implements WorkflowRunRepository {
private readonly runsDir: string;
constructor(runsDir: string) {
this.runsDir = path.resolve(runsDir);
}
async get(runId: string): Promise<WorkflowRun | null> {
const runDir = this.runDir(runId);
if (!(await this.regularDirectoryExists(this.runsDir))) return null;
if (!(await this.regularDirectoryExists(runDir))) return null;
return this.readRun(this.runPath(runId));
}
async list(filters: WorkflowRunFilters = {}): Promise<WorkflowRun[]> {
const runs: WorkflowRun[] = [];
for (const runId of await this.listRunIds()) {
const run = await this.get(runId);
if (!run || !matchesFilters(run, filters)) continue;
runs.push(run);
}
return runs.sort(compareRuns);
}
async listMetadata(filters: WorkflowRunFilters = {}): Promise<WorkflowRunMetadata[]> {
const metadata: WorkflowRunMetadata[] = [];
for (const runId of await this.listRunIds()) {
try {
const run = await this.get(runId);
if (!run || !matchesFilters(run, filters)) continue;
metadata.push({
id: run.id,
workflowId: run.workflowId,
workflowVersion: run.workflowVersion,
taskId: run.taskId,
status: run.status,
startedAt: run.startedAt,
completedAt: run.completedAt,
error: run.error,
});
} catch {
continue;
}
}
return metadata.sort(
(left, right) =>
new Date(right.startedAt).getTime() - new Date(left.startedAt).getTime() ||
right.id.localeCompare(left.id)
);
}
async save(run: WorkflowRun, expectedRevision = 0): Promise<boolean> {
const runDir = this.runDir(run.id);
const runPath = this.runPath(run.id);
await this.prepareDirectory(this.runsDir);
await this.prepareDirectory(runDir);
return withFileLock(runPath, async () => {
const current = await this.readRun(runPath);
const currentRevision = current?.revision ?? 0;
if (
(current && currentRevision !== expectedRevision) ||
(!current && expectedRevision !== 0)
) {
return false;
}
const content = JSON.stringify(run, null, 2);
if (Buffer.byteLength(content, 'utf8') > MAX_WORKFLOW_RUN_BYTES) {
throw new Error('Workflow run exceeds the 16 MiB storage limit');
}
await atomicWriteFile(runPath, content, 'utf8');
return true;
});
}
async saveWorkflowSnapshot(runId: string, workflow: WorkflowDefinition): Promise<void> {
const runDir = this.runDir(runId);
const snapshotPath = ensureWithinBase(runDir, path.join(runDir, 'workflow.yml'));
await this.prepareDirectory(this.runsDir);
await this.prepareDirectory(runDir);
const content = yaml.stringify(workflow);
if (Buffer.byteLength(content, 'utf8') > MAX_WORKFLOW_SNAPSHOT_BYTES) {
throw new Error('Workflow snapshot exceeds the 4 MiB storage limit');
}
await withFileLock(snapshotPath, () => atomicWriteFile(snapshotPath, content, 'utf8'));
}
private async listRunIds(): Promise<string[]> {
let entries;
try {
entries = await readdir(this.runsDir, { withFileTypes: true });
} catch (error) {
if ((error as NodeJS.ErrnoException).code === 'ENOENT') return [];
throw error;
}
const stats = await lstat(this.runsDir);
if (!stats.isDirectory() || stats.isSymbolicLink()) {
throw new Error('Workflow runs path must use a regular directory');
}
return entries
.filter(
(entry) => entry.isDirectory() && !entry.isSymbolicLink() && entry.name.startsWith('run_')
)
.map((entry) => entry.name)
.filter((runId) => {
try {
validatePathSegment(runId);
return true;
} catch {
return false;
}
});
}
private runDir(runId: string): string {
return ensureWithinBase(this.runsDir, path.join(this.runsDir, validatePathSegment(runId)));
}
private runPath(runId: string): string {
const runDir = this.runDir(runId);
return ensureWithinBase(runDir, path.join(runDir, 'run.json'));
}
private async prepareDirectory(directory: string): Promise<void> {
await mkdir(directory, { recursive: true, mode: 0o700 });
const stats = await lstat(directory);
if (!stats.isDirectory() || stats.isSymbolicLink()) {
throw new Error('Workflow runs path must use a regular directory');
}
}
private async regularDirectoryExists(directory: string): Promise<boolean> {
try {
const stats = await lstat(directory);
if (!stats.isDirectory() || stats.isSymbolicLink()) {
throw new Error('Workflow runs path must use a regular directory');
}
return true;
} catch (error) {
if ((error as NodeJS.ErrnoException).code === 'ENOENT') return false;
throw error;
}
}
private async readRun(runPath: string): Promise<WorkflowRun | null> {
let handle: Awaited<ReturnType<typeof open>> | undefined;
try {
handle = await open(runPath, constants.O_RDONLY | (constants.O_NOFOLLOW ?? 0));
const [pathStats, stats] = await Promise.all([lstat(runPath), handle.stat()]);
if (
pathStats.isSymbolicLink() ||
pathStats.dev !== stats.dev ||
pathStats.ino !== stats.ino
) {
throw new Error('Workflow run must not use a symbolic link or changed file');
}
if (!stats.isFile() || stats.size > MAX_WORKFLOW_RUN_BYTES) {
throw new Error('Workflow run must use a bounded regular file');
}
return JSON.parse(await handle.readFile({ encoding: 'utf8' })) as WorkflowRun;
} catch (error) {
const errorCode = (error as NodeJS.ErrnoException).code;
if (errorCode === 'ENOENT') return null;
if (errorCode === 'ELOOP') {
throw new Error('Workflow run must not use a symbolic link', { cause: error });
}
throw error;
} finally {
await handle?.close();
}
}
}
function matchesFilters(run: WorkflowRun, filters: WorkflowRunFilters): boolean {
return (
(!filters.taskId || run.taskId === filters.taskId) &&
(!filters.workflowId || run.workflowId === filters.workflowId) &&
(!filters.status || run.status === filters.status)
);
}
function compareRuns(left: WorkflowRun, right: WorkflowRun): number {
return (
new Date(right.startedAt).getTime() - new Date(left.startedAt).getTime() ||
right.id.localeCompare(left.id)
);
}