From c814bf739034fbfb7b0f5ae052edeaf0bed09b84 Mon Sep 17 00:00:00 2001 From: Brad Groux <3053586+BradGroux@users.noreply.github.com> Date: Sat, 25 Jul 2026 21:42:31 -0500 Subject: [PATCH] feat: add durable reflection extraction jobs (#1085) --- docs/FEATURES.md | 2 + docs/SQLITE-SCHEMA.md | 21 +- .../reflection-extraction-job-service.test.ts | 212 +++++++++++++ .../reflection-extraction-job-schemas.ts | 104 ++++++ .../reflection-extraction-job-service.ts | 296 ++++++++++++++++++ server/src/storage/file-storage.ts | 7 + server/src/storage/index.ts | 6 + server/src/storage/interfaces.ts | 31 ++ .../reflection-extraction-job-repository.ts | 183 +++++++++++ .../reflection-extraction-job-state.ts | 127 ++++++++ server/src/storage/sqlite/migrations.ts | 30 ++ .../reflection-extraction-job-repository.ts | 190 +++++++++++ server/src/storage/sqlite/sqlite-storage.ts | 3 + shared/src/types/index.ts | 1 + .../types/reflection-extraction-job.types.ts | 95 ++++++ 15 files changed, 1298 insertions(+), 10 deletions(-) create mode 100644 server/src/__tests__/reflection-extraction-job-service.test.ts create mode 100644 server/src/schemas/reflection-extraction-job-schemas.ts create mode 100644 server/src/services/reflection-extraction-job-service.ts create mode 100644 server/src/storage/reflection-extraction-job-repository.ts create mode 100644 server/src/storage/reflection-extraction-job-state.ts create mode 100644 server/src/storage/sqlite/reflection-extraction-job-repository.ts create mode 100644 shared/src/types/reflection-extraction-job.types.ts diff --git a/docs/FEATURES.md b/docs/FEATURES.md index 27f04aac..e9fb552c 100644 --- a/docs/FEATURES.md +++ b/docs/FEATURES.md @@ -602,6 +602,8 @@ Reviewed promotion queue for agent corrections, repeated mistakes, and durable l - **Task lesson promotion** — Accepted task-linked candidates append a reviewed reflection lesson to the task's lessons field - **Duplicate grouping and merge** — Similar candidates share a duplicate key and can be soft-merged into a representative while preserving audit history - **Redaction at ingestion** — Tokens, credentials, and local private paths are redacted before candidates are stored +- **Durable extraction jobs** — `reflection-extraction-job/v1` persists only source task, attempt, completion, digest, and event identities; raw conversations and unrestricted transcripts are not copied into the queue +- **Lease-safe worker foundation** — File and SQLite repositories atomically enforce global and per-workspace concurrency, stable idempotent enqueue, lease ownership and renewal, deterministic retry backoff, restart recovery, and bounded dead-lettering - **Settings UI** — Review, accept, reject, delete, and merge candidates from Settings → Reflections - **Audit trail** — Create, accept, reject, merge, and delete actions write metadata-only audit events diff --git a/docs/SQLITE-SCHEMA.md b/docs/SQLITE-SCHEMA.md index c9ef5350..af2bcf9a 100644 --- a/docs/SQLITE-SCHEMA.md +++ b/docs/SQLITE-SCHEMA.md @@ -1065,19 +1065,20 @@ SQLite tables with JSON payload columns plus query indexes. This keeps the v4 service contracts intact while preventing SQLite mode from writing operational state back to `.veritas-kanban/*.json` or telemetry NDJSON files. -| Runtime table | Stored data | -| ------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `activity_events` | Complete activity entries plus type, task, agent, and created-time columns. | -| `status_history` | Complete status transition entries plus previous/new status and task columns. | -| `telemetry_events` | Complete telemetry events plus type, task, project, token, duration, and result columns. | -| `run_events` | Complete `run-event/v1` envelopes plus ordered attempt cursor, provider identity, dedupe, and receive columns. | -| `run_supervisors` | Complete `run-supervisor/v1` snapshots plus task/attempt, state, revision, lease owner/expiry, and recovery indexes. | -| `durable_goals` | Complete `durable-goal/v1` objective state, root task/workflow identity, compare-and-set revision, blockers, continuation chain, usage, and completion-evidence requirements. | -| `admission_reservations` | Versioned capacity and execution-tree budget reservations plus task/workspace/root/provider/host scopes, root objective/node/parent indexes, lease state, revision, and idempotency evidence. | +| Runtime table | Stored data | +| ---------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `activity_events` | Complete activity entries plus type, task, agent, and created-time columns. | +| `status_history` | Complete status transition entries plus previous/new status and task columns. | +| `telemetry_events` | Complete telemetry events plus type, task, project, token, duration, and result columns. | +| `run_events` | Complete `run-event/v1` envelopes plus ordered attempt cursor, provider identity, dedupe, and receive columns. | +| `run_supervisors` | Complete `run-supervisor/v1` snapshots plus task/attempt, state, revision, lease owner/expiry, and recovery indexes. | +| `durable_goals` | Complete `durable-goal/v1` objective state, root task/workflow identity, compare-and-set revision, blockers, continuation chain, usage, and completion-evidence requirements. | +| `reflection_extraction_jobs` | Bounded `reflection-extraction-job/v1` source identities, state, revision, idempotency key, retry availability, lease owner/expiry, candidate IDs, and failure history. | +| `admission_reservations` | Versioned capacity and execution-tree budget reservations plus task/workspace/root/provider/host scopes, root objective/node/parent indexes, lease state, revision, and idempotency evidence. | `ActivityService`, `StatusHistoryService`, `TelemetryService`, `RunEventJournalService`, `RunSupervisorService`, `DurableGoalService`, and -`AdmissionControlService` select these SQLite repositories when +`ReflectionExtractionJobService`, and `AdmissionControlService` select these SQLite repositories when `VERITAS_STORAGE=sqlite`. File storage still forces the file-backed services to `storageType='file'`, so explicit file mode cannot be accidentally flipped by the environment. diff --git a/server/src/__tests__/reflection-extraction-job-service.test.ts b/server/src/__tests__/reflection-extraction-job-service.test.ts new file mode 100644 index 00000000..ab5f42dc --- /dev/null +++ b/server/src/__tests__/reflection-extraction-job-service.test.ts @@ -0,0 +1,212 @@ +import fs from 'node:fs/promises'; +import os from 'node:os'; +import path from 'node:path'; +import { afterEach, describe, expect, it } from 'vitest'; +import { ReflectionExtractionJobService } from '../services/reflection-extraction-job-service.js'; +import { FileReflectionExtractionJobRepository } from '../storage/reflection-extraction-job-repository.js'; +import type { ReflectionExtractionJobRepository } from '../storage/interfaces.js'; +import { SqliteDatabase } from '../storage/sqlite/database.js'; +import { SqliteReflectionExtractionJobRepository } from '../storage/sqlite/reflection-extraction-job-repository.js'; + +const roots: string[] = []; +const databases: SqliteDatabase[] = []; + +afterEach(async () => { + for (const database of databases.splice(0)) database.close(); + await Promise.all(roots.splice(0).map((root) => fs.rm(root, { recursive: true, force: true }))); +}); + +describe.each(['file', 'sqlite'] as const)( + 'ReflectionExtractionJobService %s repository', + (storage) => { + it('enqueues idempotently and survives a repository restart', async () => { + const harness = await createHarness(storage); + const first = await harness.service.enqueue(jobInput('workspace-a', 'completion-1')); + const duplicate = await harness.service.enqueue(jobInput('workspace-a', 'completion-1')); + + expect(first.created).toBe(true); + expect(duplicate).toEqual({ job: first.job, created: false }); + + const restarted = await harness.reopen(); + expect(await restarted.get(first.job.id)).toEqual(first.job); + + await expect( + restarted.enqueue({ + ...jobInput('workspace-a', 'completion-1'), + source: { + ...jobInput('workspace-a', 'completion-1').source, + completionDigest: 'sha256:different', + }, + }) + ).rejects.toMatchObject({ statusCode: 409, code: 'CONFLICT' }); + }); + + it('enforces global and workspace concurrency with deterministic retry backoff', async () => { + const harness = await createHarness(storage, { + maxActiveGlobal: 2, + maxActivePerWorkspace: 1, + }); + const firstA = await harness.service.enqueue(jobInput('workspace-a', 'completion-a1')); + harness.advance(1); + await harness.service.enqueue(jobInput('workspace-a', 'completion-a2')); + harness.advance(1); + await harness.service.enqueue(jobInput('workspace-b', 'completion-b1')); + harness.advance(1); + + const claimA = await harness.service.claim('worker-a'); + const claimB = await harness.service.claim('worker-b'); + const atGlobalLimit = await harness.service.claim('worker-c'); + + expect(claimA).toMatchObject({ + claimed: true, + job: { id: firstA.job.id, workspaceId: 'workspace-a', attemptCount: 1 }, + }); + expect(claimB).toMatchObject({ + claimed: true, + job: { workspaceId: 'workspace-b', attemptCount: 1 }, + }); + expect(atGlobalLimit).toEqual({ claimed: false, reason: 'global-limit' }); + if (!claimA.claimed || !claimB.claimed) throw new Error('Expected both claims.'); + + await expect( + harness.service.fail(claimA.job.id, { + expectedRevision: claimA.job.revision, + ownerId: 'other-worker', + code: 'EXTRACTION_FAILED', + summary: 'Wrong owner must not mutate the job.', + }) + ).rejects.toMatchObject({ statusCode: 409, code: 'CONFLICT' }); + + const renewedB = await harness.service.renew(claimB.job.id, { + expectedRevision: claimB.job.revision, + ownerId: 'worker-b', + }); + expect(renewedB).toMatchObject({ + state: 'leased', + revision: claimB.job.revision + 1, + lease: { ownerId: 'worker-b' }, + }); + + const completedB = await harness.service.complete(claimB.job.id, { + expectedRevision: renewedB.revision, + ownerId: 'worker-b', + candidateIds: ['reflection-candidate-b1'], + }); + expect(completedB).toMatchObject({ + state: 'completed', + candidateIds: ['reflection-candidate-b1'], + lease: undefined, + }); + + const failedA = await harness.service.fail(claimA.job.id, { + expectedRevision: claimA.job.revision, + ownerId: 'worker-a', + code: 'MODEL_TIMEOUT', + summary: 'The bounded extractor timed out.', + }); + expect(failedA).toMatchObject({ + state: 'queued', + attemptCount: 1, + failures: [{ attempt: 1, code: 'MODEL_TIMEOUT' }], + }); + expect(Date.parse(failedA.availableAt) - harness.now().getTime()).toBe(1_000); + + const next = await harness.service.claim('worker-c'); + expect(next).toMatchObject({ + claimed: true, + job: { workspaceId: 'workspace-a', attemptCount: 1 }, + }); + if (!next.claimed) throw new Error('Expected the second workspace-a job.'); + expect(next.job.id).not.toBe(failedA.id); + }); + + it('dead-letters an exhausted expired lease instead of duplicating work', async () => { + const harness = await createHarness(storage); + const enqueued = await harness.service.enqueue({ + ...jobInput('workspace-a', 'completion-expired'), + maxAttempts: 1, + }); + const claimed = await harness.service.claim('worker-a'); + expect(claimed).toMatchObject({ claimed: true, job: { id: enqueued.job.id } }); + if (!claimed.claimed) throw new Error('Expected an extraction claim.'); + + harness.advance(60_001); + const afterExpiry = await harness.service.claim('worker-b'); + expect(afterExpiry).toEqual({ claimed: false, reason: 'empty' }); + const expired = await harness.service.get(enqueued.job.id); + expect(expired).toMatchObject({ + state: 'dead-letter', + attemptCount: 1, + failures: [{ attempt: 1, code: 'LEASE_EXPIRED' }], + }); + expect(expired.lease).toBeUndefined(); + }); + } +); + +function jobInput(workspaceId: string, completionId: string) { + return { + workspaceId, + idempotencyKey: `extract:${completionId}`, + source: { + taskId: `task-${workspaceId}`, + attemptId: `attempt-${completionId}`, + completionId, + completionDigest: `sha256:${completionId}`, + runEventId: `event-${completionId}`, + }, + }; +} + +async function createHarness( + storage: 'file' | 'sqlite', + limits: { maxActiveGlobal?: number; maxActivePerWorkspace?: number } = {} +) { + const root = await fs.mkdtemp(path.join(os.tmpdir(), 'veritas-reflection-jobs-')); + roots.push(root); + let currentTime = Date.parse('2026-07-26T04:00:00.000Z'); + let repository: ReflectionExtractionJobRepository; + let reopenRepository: () => Promise; + + if (storage === 'file') { + const filePath = path.join(root, 'jobs.jsonl'); + repository = new FileReflectionExtractionJobRepository(filePath); + reopenRepository = async () => new FileReflectionExtractionJobRepository(filePath); + } else { + const databasePath = path.join(root, 'veritas.db'); + let database = new SqliteDatabase({ databasePath }); + databases.push(database); + database.open(); + repository = new SqliteReflectionExtractionJobRepository(database); + reopenRepository = async () => { + database.close(); + databases.splice(databases.indexOf(database), 1); + database = new SqliteDatabase({ databasePath }); + databases.push(database); + database.open(); + return new SqliteReflectionExtractionJobRepository(database); + }; + } + + const options = { + now: () => new Date(currentTime), + leaseDurationMs: 60_000, + retryBaseDelayMs: 1_000, + retryMaxDelayMs: 4_000, + maxActiveGlobal: limits.maxActiveGlobal ?? 4, + maxActivePerWorkspace: limits.maxActivePerWorkspace ?? 1, + }; + return { + service: new ReflectionExtractionJobService({ ...options, repository }), + advance(milliseconds: number) { + currentTime += milliseconds; + }, + now: () => new Date(currentTime), + async reopen() { + return new ReflectionExtractionJobService({ + ...options, + repository: await reopenRepository(), + }); + }, + }; +} diff --git a/server/src/schemas/reflection-extraction-job-schemas.ts b/server/src/schemas/reflection-extraction-job-schemas.ts new file mode 100644 index 00000000..57b6f30f --- /dev/null +++ b/server/src/schemas/reflection-extraction-job-schemas.ts @@ -0,0 +1,104 @@ +import { z } from 'zod'; +import { + REFLECTION_EXTRACTION_JOB_SCHEMA_VERSION, + REFLECTION_EXTRACTION_JOB_STATES, + type ReflectionExtractionJob, +} from '@veritas-kanban/shared'; + +const IdentifierSchema = z.string().trim().min(1).max(240); +const TimestampSchema = z.string().datetime(); +const BoundedSummarySchema = z.string().trim().min(1).max(2_000); + +export const ReflectionExtractionJobSchema: z.ZodType = z + .object({ + schemaVersion: z.literal(REFLECTION_EXTRACTION_JOB_SCHEMA_VERSION), + id: z.string().regex(/^reflection_job_[a-f0-9]{32}$/), + workspaceId: IdentifierSchema, + idempotencyKey: IdentifierSchema, + source: z + .object({ + taskId: IdentifierSchema, + attemptId: IdentifierSchema, + completionId: IdentifierSchema, + completionDigest: IdentifierSchema, + runEventId: IdentifierSchema.optional(), + }) + .strict(), + state: z.enum(REFLECTION_EXTRACTION_JOB_STATES), + revision: z.number().int().positive(), + attemptCount: z.number().int().nonnegative().max(20), + maxAttempts: z.number().int().positive().max(20), + availableAt: TimestampSchema, + lease: z + .object({ + ownerId: IdentifierSchema, + acquiredAt: TimestampSchema, + expiresAt: TimestampSchema, + }) + .strict() + .optional(), + candidateIds: z.array(IdentifierSchema).max(100), + failures: z + .array( + z + .object({ + attempt: z.number().int().positive().max(20), + code: IdentifierSchema, + summary: BoundedSummarySchema, + failedAt: TimestampSchema, + retryAt: TimestampSchema.optional(), + }) + .strict() + ) + .max(20), + createdAt: TimestampSchema, + updatedAt: TimestampSchema, + completedAt: TimestampSchema.optional(), + }) + .strict() + .superRefine((job, context) => { + if (job.attemptCount > job.maxAttempts) { + context.addIssue({ + code: 'custom', + path: ['attemptCount'], + message: 'Extraction job attempts cannot exceed maxAttempts.', + }); + } + if ((job.state === 'leased') !== Boolean(job.lease)) { + context.addIssue({ + code: 'custom', + path: ['lease'], + message: 'Only leased extraction jobs may retain a lease.', + }); + } + if (job.state === 'leased' && job.attemptCount === 0) { + context.addIssue({ + code: 'custom', + path: ['attemptCount'], + message: 'A leased extraction job must have at least one attempt.', + }); + } + if ((job.state === 'completed') !== Boolean(job.completedAt)) { + context.addIssue({ + code: 'custom', + path: ['completedAt'], + message: 'Only completed extraction jobs may have completedAt.', + }); + } + if (new Set(job.candidateIds).size !== job.candidateIds.length) { + context.addIssue({ + code: 'custom', + path: ['candidateIds'], + message: 'Extraction candidate IDs must be unique.', + }); + } + for (const [index, failure] of job.failures.entries()) { + if (failure.attempt > job.attemptCount) { + context.addIssue({ + code: 'custom', + path: ['failures', index, 'attempt'], + message: 'Extraction failures cannot reference a future attempt.', + }); + } + } + }); diff --git a/server/src/services/reflection-extraction-job-service.ts b/server/src/services/reflection-extraction-job-service.ts new file mode 100644 index 00000000..ea750b85 --- /dev/null +++ b/server/src/services/reflection-extraction-job-service.ts @@ -0,0 +1,296 @@ +import { createHash } from 'node:crypto'; +import type { + ReflectionExtractionJob, + ReflectionExtractionJobClaimResult, + ReflectionExtractionJobListQuery, + ReflectionExtractionJobSource, +} from '@veritas-kanban/shared'; +import { REFLECTION_EXTRACTION_JOB_SCHEMA_VERSION } from '@veritas-kanban/shared'; +import { ConflictError, NotFoundError, ValidationError } from '../middleware/error-handler.js'; +import { ReflectionExtractionJobSchema } from '../schemas/reflection-extraction-job-schemas.js'; +import { FileReflectionExtractionJobRepository } from '../storage/reflection-extraction-job-repository.js'; +import { extractionRetryDelayMs } from '../storage/reflection-extraction-job-state.js'; +import type { ReflectionExtractionJobRepository } from '../storage/interfaces.js'; +import { getStorage, getStorageTypeFromEnv } from '../storage/index.js'; + +const DEFAULT_MAX_ATTEMPTS = 5; +const DEFAULT_LEASE_DURATION_MS = 2 * 60_000; +const DEFAULT_MAX_ACTIVE_GLOBAL = 4; +const DEFAULT_MAX_ACTIVE_PER_WORKSPACE = 1; +const DEFAULT_RETRY_BASE_DELAY_MS = 5_000; +const DEFAULT_RETRY_MAX_DELAY_MS = 5 * 60_000; + +export interface EnqueueReflectionExtractionJobInput { + workspaceId: string; + idempotencyKey: string; + source: ReflectionExtractionJobSource; + maxAttempts?: number; + availableAt?: string; +} + +export interface ReflectionExtractionJobMutationInput { + expectedRevision: number; + ownerId: string; +} + +export interface CompleteReflectionExtractionJobInput extends ReflectionExtractionJobMutationInput { + candidateIds: string[]; +} + +export interface FailReflectionExtractionJobInput extends ReflectionExtractionJobMutationInput { + code: string; + summary: string; +} + +export interface ReflectionExtractionJobServiceOptions { + repository?: ReflectionExtractionJobRepository; + now?: () => Date; + leaseDurationMs?: number; + maxActiveGlobal?: number; + maxActivePerWorkspace?: number; + retryBaseDelayMs?: number; + retryMaxDelayMs?: number; +} + +let fileRepository: FileReflectionExtractionJobRepository | undefined; + +function defaultRepository(): ReflectionExtractionJobRepository { + if (getStorageTypeFromEnv() === 'sqlite') return getStorage().reflectionExtractionJobs; + fileRepository ??= new FileReflectionExtractionJobRepository(); + return fileRepository; +} + +export class ReflectionExtractionJobService { + private readonly repositoryOverride?: ReflectionExtractionJobRepository; + private readonly now: () => Date; + private readonly leaseDurationMs: number; + private readonly maxActiveGlobal: number; + private readonly maxActivePerWorkspace: number; + private readonly retryBaseDelayMs: number; + private readonly retryMaxDelayMs: number; + + constructor(options: ReflectionExtractionJobServiceOptions = {}) { + this.repositoryOverride = options.repository; + this.now = options.now ?? (() => new Date()); + this.leaseDurationMs = options.leaseDurationMs ?? DEFAULT_LEASE_DURATION_MS; + this.maxActiveGlobal = options.maxActiveGlobal ?? DEFAULT_MAX_ACTIVE_GLOBAL; + this.maxActivePerWorkspace = options.maxActivePerWorkspace ?? DEFAULT_MAX_ACTIVE_PER_WORKSPACE; + this.retryBaseDelayMs = options.retryBaseDelayMs ?? DEFAULT_RETRY_BASE_DELAY_MS; + this.retryMaxDelayMs = options.retryMaxDelayMs ?? DEFAULT_RETRY_MAX_DELAY_MS; + } + + private get repository(): ReflectionExtractionJobRepository { + return this.repositoryOverride ?? defaultRepository(); + } + + async enqueue(input: EnqueueReflectionExtractionJobInput) { + const timestamp = this.now().toISOString(); + const workspaceId = input.workspaceId.trim(); + const idempotencyKey = input.idempotencyKey.trim(); + const record = ReflectionExtractionJobSchema.parse({ + schemaVersion: REFLECTION_EXTRACTION_JOB_SCHEMA_VERSION, + id: stableJobId(workspaceId, idempotencyKey), + workspaceId, + idempotencyKey, + source: input.source, + state: 'queued', + revision: 1, + attemptCount: 0, + maxAttempts: input.maxAttempts ?? DEFAULT_MAX_ATTEMPTS, + availableAt: input.availableAt ?? timestamp, + candidateIds: [], + failures: [], + createdAt: timestamp, + updatedAt: timestamp, + }); + const result = await this.repository.enqueue(record); + if ( + !result.created && + (result.job.workspaceId !== record.workspaceId || + !sameExtractionSource(result.job.source, record.source)) + ) { + throw new ConflictError('Extraction job idempotency key references different evidence.', { + idempotencyKey, + existingJobId: result.job.id, + }); + } + return result; + } + + async get(id: string): Promise { + const job = await this.repository.get(id); + if (!job) throw new NotFoundError('Reflection extraction job not found.'); + return job; + } + + list(query: ReflectionExtractionJobListQuery = {}): Promise { + return this.repository.list(query); + } + + claim(ownerId: string, workspaceId?: string): Promise { + return this.repository.claim({ + ownerId, + now: this.now().toISOString(), + leaseDurationMs: this.leaseDurationMs, + maxActiveGlobal: this.maxActiveGlobal, + maxActivePerWorkspace: this.maxActivePerWorkspace, + retryBaseDelayMs: this.retryBaseDelayMs, + retryMaxDelayMs: this.retryMaxDelayMs, + workspaceId, + }); + } + + async renew( + id: string, + input: ReflectionExtractionJobMutationInput + ): Promise { + const current = await this.requireOwnedLease(id, input); + const timestamp = this.now().toISOString(); + const next = ReflectionExtractionJobSchema.parse({ + ...current, + revision: current.revision + 1, + lease: { + ...current.lease, + expiresAt: new Date(Date.parse(timestamp) + this.leaseDurationMs).toISOString(), + }, + updatedAt: timestamp, + }); + return this.compareAndSet(current, next); + } + + async complete( + id: string, + input: CompleteReflectionExtractionJobInput + ): Promise { + const current = await this.requireOwnedLease(id, input); + const timestamp = this.now().toISOString(); + const next = ReflectionExtractionJobSchema.parse({ + ...current, + state: 'completed', + revision: current.revision + 1, + lease: undefined, + candidateIds: input.candidateIds, + completedAt: timestamp, + updatedAt: timestamp, + }); + return this.compareAndSet(current, next); + } + + async fail( + id: string, + input: FailReflectionExtractionJobInput + ): Promise { + const current = await this.requireOwnedLease(id, input); + const timestamp = this.now().toISOString(); + const terminal = current.attemptCount >= current.maxAttempts; + const retryAt = terminal + ? undefined + : new Date( + Date.parse(timestamp) + + extractionRetryDelayMs( + current.attemptCount, + this.retryBaseDelayMs, + this.retryMaxDelayMs + ) + ).toISOString(); + const next = ReflectionExtractionJobSchema.parse({ + ...current, + state: terminal ? 'dead-letter' : 'queued', + revision: current.revision + 1, + availableAt: retryAt ?? timestamp, + lease: undefined, + failures: [ + ...current.failures, + { + attempt: current.attemptCount, + code: input.code, + summary: input.summary, + failedAt: timestamp, + retryAt, + }, + ].slice(-20), + updatedAt: timestamp, + }); + return this.compareAndSet(current, next); + } + + private async requireOwnedLease( + id: string, + input: ReflectionExtractionJobMutationInput + ): Promise { + const current = await this.get(id); + if (current.revision !== input.expectedRevision) { + throw new ConflictError('Extraction job compare-and-set revision is stale.', { + jobId: id, + expectedRevision: input.expectedRevision, + currentRevision: current.revision, + }); + } + if (current.state !== 'leased' || !current.lease) { + throw new ValidationError('Extraction job does not have an active lease.', { + jobId: id, + state: current.state, + }); + } + if (current.lease.ownerId !== input.ownerId) { + throw new ConflictError('Extraction job lease belongs to another worker.', { + jobId: id, + }); + } + if (Date.parse(current.lease.expiresAt) <= this.now().getTime()) { + throw new ConflictError('Extraction job lease has expired.', { jobId: id }); + } + return current; + } + + private async compareAndSet( + current: ReflectionExtractionJob, + next: ReflectionExtractionJob + ): Promise { + const result = await this.repository.compareAndSet({ + id: current.id, + expectedRevision: current.revision, + next, + }); + if (result.updated && result.job) return result.job; + if (result.reason === 'not-found') { + throw new NotFoundError('Reflection extraction job not found.'); + } + throw new ConflictError('Extraction job compare-and-set update was rejected.', { + jobId: current.id, + expectedRevision: current.revision, + currentRevision: result.job?.revision, + reason: result.reason, + }); + } +} + +function stableJobId(workspaceId: string, idempotencyKey: string): string { + const digest = createHash('sha256') + .update(workspaceId) + .update('\0') + .update(idempotencyKey) + .digest('hex') + .slice(0, 32); + return `reflection_job_${digest}`; +} + +function sameExtractionSource( + left: ReflectionExtractionJobSource, + right: ReflectionExtractionJobSource +): boolean { + return ( + left.taskId === right.taskId && + left.attemptId === right.attemptId && + left.completionId === right.completionId && + left.completionDigest === right.completionDigest && + left.runEventId === right.runEventId + ); +} + +let reflectionExtractionJobService: ReflectionExtractionJobService | undefined; + +export function getReflectionExtractionJobService(): ReflectionExtractionJobService { + reflectionExtractionJobService ??= new ReflectionExtractionJobService(); + return reflectionExtractionJobService; +} diff --git a/server/src/storage/file-storage.ts b/server/src/storage/file-storage.ts index 321d59b5..224cdde9 100644 --- a/server/src/storage/file-storage.ts +++ b/server/src/storage/file-storage.ts @@ -44,6 +44,7 @@ import type { PhaseTransitionRepository, RunSupervisorRepository, DurableGoalRepository, + ReflectionExtractionJobRepository, AdmissionReservationRepository, ToolControlPlaneRepository, } from './interfaces.js'; @@ -74,6 +75,7 @@ import { FileRunApprovalRepository } from './run-approval-repository.js'; import { FilePhaseTransitionRepository } from './phase-transition-repository.js'; import { FileRunSupervisorRepository } from './run-supervisor-repository.js'; import { FileDurableGoalRepository } from './durable-goal-repository.js'; +import { FileReflectionExtractionJobRepository } from './reflection-extraction-job-repository.js'; import { FileAdmissionReservationRepository } from './admission-reservation-repository.js'; import { FileToolControlPlaneRepository } from './tool-control-plane-repository.js'; @@ -497,6 +499,7 @@ export interface FileStorageOptions { phaseTransitionsPath?: string; runSupervisorsPath?: string; durableGoalsPath?: string; + reflectionExtractionJobsPath?: string; admissionReservationsPath?: string; toolControlPlanePath?: string; } @@ -515,6 +518,7 @@ export class FileStorageProvider implements StorageProvider { readonly phaseTransitions: PhaseTransitionRepository; readonly runSupervisors: RunSupervisorRepository; readonly durableGoals: DurableGoalRepository; + readonly reflectionExtractionJobs: ReflectionExtractionJobRepository; readonly admissionReservations: AdmissionReservationRepository; readonly toolControlPlane: ToolControlPlaneRepository; @@ -571,6 +575,9 @@ export class FileStorageProvider implements StorageProvider { this.phaseTransitions = new FilePhaseTransitionRepository(options.phaseTransitionsPath); this.runSupervisors = new FileRunSupervisorRepository(options.runSupervisorsPath); this.durableGoals = new FileDurableGoalRepository(options.durableGoalsPath); + this.reflectionExtractionJobs = new FileReflectionExtractionJobRepository( + options.reflectionExtractionJobsPath + ); this.admissionReservations = new FileAdmissionReservationRepository( options.admissionReservationsPath ); diff --git a/server/src/storage/index.ts b/server/src/storage/index.ts index f840a9eb..f00428a2 100644 --- a/server/src/storage/index.ts +++ b/server/src/storage/index.ts @@ -29,6 +29,7 @@ export type { PhaseTransitionRepository, RunSupervisorRepository, DurableGoalRepository, + ReflectionExtractionJobRepository, AdmissionReservationRepository, ToolControlPlaneRepository, SetupContextRepository, @@ -66,6 +67,10 @@ export { getPhaseTransitionsPath, } from './phase-transition-repository.js'; export { FileDurableGoalRepository, getDurableGoalsPath } from './durable-goal-repository.js'; +export { + FileReflectionExtractionJobRepository, + getReflectionExtractionJobsPath, +} from './reflection-extraction-job-repository.js'; export { FileAdmissionReservationRepository, getAdmissionReservationsPath, @@ -99,6 +104,7 @@ export { SqliteRunEventRepository } from './sqlite/run-event-repository.js'; export { SqliteRunApprovalRepository } from './sqlite/run-approval-repository.js'; export { SqlitePhaseTransitionRepository } from './sqlite/phase-transition-repository.js'; export { SqliteDurableGoalRepository } from './sqlite/durable-goal-repository.js'; +export { SqliteReflectionExtractionJobRepository } from './sqlite/reflection-extraction-job-repository.js'; export { SqliteAdmissionReservationRepository } from './sqlite/admission-reservation-repository.js'; export { FileToolControlPlaneRepository, diff --git a/server/src/storage/interfaces.ts b/server/src/storage/interfaces.ts index fbccbfed..26095e25 100644 --- a/server/src/storage/interfaces.ts +++ b/server/src/storage/interfaces.ts @@ -48,6 +48,13 @@ import type { DurableGoalCompareAndSetResult, DurableGoalListQuery, DurableGoalRecord, + ReflectionExtractionJob, + ReflectionExtractionJobClaimInput, + ReflectionExtractionJobClaimResult, + ReflectionExtractionJobCompareAndSetInput, + ReflectionExtractionJobCompareAndSetResult, + ReflectionExtractionJobEnqueueResult, + ReflectionExtractionJobListQuery, AdmissionReservation, AdmissionReservationClaimInput, AdmissionReservationClaimResult, @@ -457,6 +464,29 @@ export interface DurableGoalRepository { compareAndSet(input: DurableGoalCompareAndSetInput): Promise; } +// --------------------------------------------------------------------------- +// Reflection Extraction Job Repository +// --------------------------------------------------------------------------- + +export interface ReflectionExtractionJobRepository { + /** Atomically create one job or return the existing idempotent job. */ + enqueue(job: ReflectionExtractionJob): Promise; + + /** Return one materialized extraction job. */ + get(id: string): Promise; + + /** Query bounded materialized extraction jobs. */ + list(query: ReflectionExtractionJobListQuery): Promise; + + /** Atomically expire stale leases, enforce concurrency limits, and claim one eligible job. */ + claim(input: ReflectionExtractionJobClaimInput): Promise; + + /** Replace one job with a revision-guarded compare-and-set operation. */ + compareAndSet( + input: ReflectionExtractionJobCompareAndSetInput + ): Promise; +} + // --------------------------------------------------------------------------- // Durable Admission Reservation Repository // --------------------------------------------------------------------------- @@ -517,6 +547,7 @@ export interface StorageProvider { readonly phaseTransitions: PhaseTransitionRepository; readonly runSupervisors: RunSupervisorRepository; readonly durableGoals: DurableGoalRepository; + readonly reflectionExtractionJobs: ReflectionExtractionJobRepository; readonly admissionReservations: AdmissionReservationRepository; readonly toolControlPlane: ToolControlPlaneRepository; readonly setupContext?: SetupContextRepository; diff --git a/server/src/storage/reflection-extraction-job-repository.ts b/server/src/storage/reflection-extraction-job-repository.ts new file mode 100644 index 00000000..be8c1efc --- /dev/null +++ b/server/src/storage/reflection-extraction-job-repository.ts @@ -0,0 +1,183 @@ +import { constants } from 'node:fs'; +import { lstat, mkdir, open } from 'node:fs/promises'; +import path from 'node:path'; +import type { + ReflectionExtractionJob, + ReflectionExtractionJobClaimInput, + ReflectionExtractionJobClaimResult, + ReflectionExtractionJobCompareAndSetInput, + ReflectionExtractionJobCompareAndSetResult, + ReflectionExtractionJobEnqueueResult, + ReflectionExtractionJobListQuery, +} from '@veritas-kanban/shared'; +import { ReflectionExtractionJobSchema } from '../schemas/reflection-extraction-job-schemas.js'; +import { withFileLock } from '../services/file-lock.js'; +import { getRuntimeDir } from '../utils/paths.js'; +import { ensureWithinBase } from '../utils/sanitize.js'; +import type { ReflectionExtractionJobRepository } from './interfaces.js'; +import { + leaseExtractionJob, + normalizeExpiredExtractionJob, + selectExtractionJob, +} from './reflection-extraction-job-state.js'; + +const MAX_JOB_LOG_BYTES = 64 * 1024 * 1024; +const MAX_JOB_SNAPSHOTS = 50_000; + +export function getReflectionExtractionJobsPath(): string { + return path.join(getRuntimeDir(), 'reflection-extraction-jobs.jsonl'); +} + +export class FileReflectionExtractionJobRepository implements ReflectionExtractionJobRepository { + constructor(private readonly filePath = getReflectionExtractionJobsPath()) { + ensureWithinBase(path.dirname(filePath), filePath); + } + + async enqueue(job: ReflectionExtractionJob): Promise { + const parsed = ReflectionExtractionJobSchema.parse(job); + if (parsed.revision !== 1) throw new Error('New extraction jobs must start at revision 1.'); + await this.prepareParent(); + return withFileLock(this.filePath, async () => { + const snapshots = await this.readSnapshots(); + const jobs = this.materialize(snapshots); + const existing = [...jobs.values()].find( + (candidate) => candidate.idempotencyKey === parsed.idempotencyKey + ); + if (existing) return { job: existing, created: false }; + if (jobs.has(parsed.id)) throw new Error(`Extraction job ${parsed.id} already exists.`); + await this.appendSnapshots([parsed], snapshots); + return { job: parsed, created: true }; + }); + } + + async get(id: string): Promise { + return this.materialize(await this.readSnapshots()).get(id) ?? null; + } + + async list(query: ReflectionExtractionJobListQuery): Promise { + const states = query.states ? new Set(query.states) : undefined; + const limit = Math.min(Math.max(query.limit ?? 100, 1), 1_000); + return [...this.materialize(await this.readSnapshots()).values()] + .filter((job) => !query.workspaceId || job.workspaceId === query.workspaceId) + .filter((job) => !states || states.has(job.state)) + .sort( + (left, right) => + Date.parse(left.availableAt) - Date.parse(right.availableAt) || + Date.parse(left.createdAt) - Date.parse(right.createdAt) || + left.id.localeCompare(right.id) + ) + .slice(0, limit); + } + + async claim( + input: ReflectionExtractionJobClaimInput + ): Promise { + await this.prepareParent(); + return withFileLock(this.filePath, async () => { + const snapshots = await this.readSnapshots(); + const jobs = [...this.materialize(snapshots).values()]; + const normalized = jobs.map((job) => normalizeExpiredExtractionJob(job, input)); + const expired = normalized.filter((job, index) => job.revision !== jobs[index]?.revision); + const selection = selectExtractionJob(normalized, input); + if ('reason' in selection) { + if (expired.length > 0) await this.appendSnapshots(expired, snapshots); + return { claimed: false, reason: selection.reason }; + } + const claimed = leaseExtractionJob(selection.job, input); + await this.appendSnapshots([...expired, claimed], snapshots); + return { claimed: true, job: claimed }; + }); + } + + async compareAndSet( + input: ReflectionExtractionJobCompareAndSetInput + ): Promise { + await this.prepareParent(); + return withFileLock(this.filePath, async () => { + const snapshots = await this.readSnapshots(); + const current = this.materialize(snapshots).get(input.id); + if (!current) return { updated: false, reason: 'not-found' }; + if (current.revision !== input.expectedRevision) { + return { job: current, updated: false, reason: 'stale-revision' }; + } + if (input.next.revision !== input.expectedRevision + 1 || input.next.id !== input.id) { + return { job: current, updated: false, reason: 'invalid-revision' }; + } + const next = ReflectionExtractionJobSchema.parse(input.next); + await this.appendSnapshots([next], snapshots); + return { job: next, updated: true }; + }); + } + + private async prepareParent(): Promise { + const parent = path.dirname(this.filePath); + await mkdir(parent, { recursive: true, mode: 0o700 }); + const stat = await lstat(parent); + if (!stat.isDirectory() || stat.isSymbolicLink()) { + throw new Error('Extraction job directory is not a private regular directory.'); + } + } + + private async readSnapshots(): Promise { + let handle: Awaited> | undefined; + try { + handle = await open(this.filePath, constants.O_RDONLY | constants.O_NOFOLLOW); + const stat = await handle.stat(); + if (!stat.isFile() || stat.size > MAX_JOB_LOG_BYTES) { + throw new Error('Extraction job log is not a bounded regular file.'); + } + const content = await handle.readFile({ encoding: 'utf8' }); + if (!content.trim()) return []; + const lines = content.split(/\r?\n/).filter(Boolean); + if (lines.length > MAX_JOB_SNAPSHOTS) { + throw new Error('Extraction job log reached its bounded snapshot limit.'); + } + return lines.map((line) => ReflectionExtractionJobSchema.parse(JSON.parse(line))); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return []; + if ((error as NodeJS.ErrnoException).code === 'ELOOP') { + throw new Error('Extraction job log is not a bounded regular file.', { cause: error }); + } + throw error; + } finally { + await handle?.close(); + } + } + + private materialize(snapshots: ReflectionExtractionJob[]): Map { + const byId = new Map(); + for (const snapshot of snapshots) { + const current = byId.get(snapshot.id); + if (!current || snapshot.revision > current.revision) byId.set(snapshot.id, snapshot); + } + return byId; + } + + private async appendSnapshots( + next: ReflectionExtractionJob[], + existing: ReflectionExtractionJob[] + ): Promise { + if (existing.length + next.length > MAX_JOB_SNAPSHOTS) { + throw new Error('Extraction job log reached its bounded snapshot limit.'); + } + const content = next.map((job) => `${JSON.stringify(job)}\n`).join(''); + const existingBytes = existing.reduce( + (total, job) => total + Buffer.byteLength(JSON.stringify(job), 'utf8') + 1, + 0 + ); + if (existingBytes + Buffer.byteLength(content, 'utf8') > MAX_JOB_LOG_BYTES) { + throw new Error('Extraction job log reached its bounded byte limit.'); + } + const handle = await open( + this.filePath, + constants.O_APPEND | constants.O_CREAT | constants.O_WRONLY | constants.O_NOFOLLOW, + 0o600 + ); + try { + await handle.write(content, undefined, 'utf8'); + await handle.sync(); + } finally { + await handle.close(); + } + } +} diff --git a/server/src/storage/reflection-extraction-job-state.ts b/server/src/storage/reflection-extraction-job-state.ts new file mode 100644 index 00000000..fb1a7147 --- /dev/null +++ b/server/src/storage/reflection-extraction-job-state.ts @@ -0,0 +1,127 @@ +import type { + ReflectionExtractionJob, + ReflectionExtractionJobClaimInput, + ReflectionExtractionJobClaimResult, +} from '@veritas-kanban/shared'; +import { ReflectionExtractionJobSchema } from '../schemas/reflection-extraction-job-schemas.js'; + +export function normalizeExpiredExtractionJob( + job: ReflectionExtractionJob, + input: ReflectionExtractionJobClaimInput +): ReflectionExtractionJob { + if ( + job.state !== 'leased' || + !job.lease || + Date.parse(job.lease.expiresAt) > Date.parse(input.now) + ) { + return job; + } + const terminal = job.attemptCount >= job.maxAttempts; + const retryAt = terminal + ? undefined + : new Date( + Date.parse(input.now) + + extractionRetryDelayMs(job.attemptCount, input.retryBaseDelayMs, input.retryMaxDelayMs) + ).toISOString(); + return ReflectionExtractionJobSchema.parse({ + ...job, + state: terminal ? 'dead-letter' : 'queued', + revision: job.revision + 1, + availableAt: retryAt ?? input.now, + lease: undefined, + failures: [ + ...job.failures, + { + attempt: job.attemptCount, + code: 'LEASE_EXPIRED', + summary: 'The extraction worker lease expired before completion.', + failedAt: input.now, + retryAt, + }, + ].slice(-20), + updatedAt: input.now, + }); +} + +export function selectExtractionJob( + jobs: ReflectionExtractionJob[], + input: ReflectionExtractionJobClaimInput +): + | { job: ReflectionExtractionJob } + | { reason: Extract['reason'] } { + validateClaimInput(input); + const active = jobs.filter( + (job) => + job.state === 'leased' && job.lease && Date.parse(job.lease.expiresAt) > Date.parse(input.now) + ); + if (active.length >= input.maxActiveGlobal) return { reason: 'global-limit' }; + + const candidates = jobs + .filter((job) => job.state === 'queued') + .filter((job) => job.attemptCount < job.maxAttempts) + .filter((job) => Date.parse(job.availableAt) <= Date.parse(input.now)) + .filter((job) => !input.workspaceId || job.workspaceId === input.workspaceId) + .sort( + (left, right) => + Date.parse(left.availableAt) - Date.parse(right.availableAt) || + Date.parse(left.createdAt) - Date.parse(right.createdAt) || + left.id.localeCompare(right.id) + ); + if (candidates.length === 0) return { reason: 'empty' }; + + const activeByWorkspace = new Map(); + for (const job of active) { + activeByWorkspace.set(job.workspaceId, (activeByWorkspace.get(job.workspaceId) ?? 0) + 1); + } + const job = candidates.find( + (candidate) => (activeByWorkspace.get(candidate.workspaceId) ?? 0) < input.maxActivePerWorkspace + ); + return job ? { job } : { reason: 'workspace-limit' }; +} + +export function leaseExtractionJob( + job: ReflectionExtractionJob, + input: ReflectionExtractionJobClaimInput +): ReflectionExtractionJob { + return ReflectionExtractionJobSchema.parse({ + ...job, + state: 'leased', + revision: job.revision + 1, + attemptCount: job.attemptCount + 1, + lease: { + ownerId: input.ownerId, + acquiredAt: input.now, + expiresAt: new Date(Date.parse(input.now) + input.leaseDurationMs).toISOString(), + }, + updatedAt: input.now, + }); +} + +export function extractionRetryDelayMs( + attempt: number, + baseDelayMs: number, + maxDelayMs: number +): number { + const exponent = Math.max(0, attempt - 1); + return Math.min(maxDelayMs, baseDelayMs * 2 ** Math.min(exponent, 30)); +} + +function validateClaimInput(input: ReflectionExtractionJobClaimInput): void { + if (!input.ownerId.trim() || !Number.isFinite(Date.parse(input.now))) { + throw new Error('Extraction claim requires a valid owner and timestamp.'); + } + for (const [field, value] of Object.entries({ + leaseDurationMs: input.leaseDurationMs, + maxActiveGlobal: input.maxActiveGlobal, + maxActivePerWorkspace: input.maxActivePerWorkspace, + retryBaseDelayMs: input.retryBaseDelayMs, + retryMaxDelayMs: input.retryMaxDelayMs, + })) { + if (!Number.isInteger(value) || value <= 0) { + throw new Error(`Extraction claim ${field} must be a positive integer.`); + } + } + if (input.retryMaxDelayMs < input.retryBaseDelayMs) { + throw new Error('Extraction retry maximum must be at least the base delay.'); + } +} diff --git a/server/src/storage/sqlite/migrations.ts b/server/src/storage/sqlite/migrations.ts index 441c78ec..186e5c77 100644 --- a/server/src/storage/sqlite/migrations.ts +++ b/server/src/storage/sqlite/migrations.ts @@ -1417,6 +1417,36 @@ export const SQLITE_BASE_MIGRATIONS: readonly SqliteMigration[] = [ ON durable_goals(root_workflow_id, updated_at DESC); `, }, + { + version: 28, + name: '0028_reflection_extraction_jobs', + up: ` + CREATE TABLE reflection_extraction_jobs ( + id TEXT PRIMARY KEY, + workspace_id TEXT NOT NULL, + state TEXT NOT NULL CHECK ( + state IN ('queued', 'leased', 'completed', 'dead-letter') + ), + revision INTEGER NOT NULL CHECK (revision > 0), + idempotency_key TEXT NOT NULL UNIQUE, + available_at TEXT NOT NULL, + lease_owner_id TEXT, + lease_expires_at TEXT, + job_json TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + + CREATE INDEX idx_reflection_extraction_jobs_available + ON reflection_extraction_jobs(state, available_at, created_at, id); + + CREATE INDEX idx_reflection_extraction_jobs_workspace + ON reflection_extraction_jobs(workspace_id, state, available_at); + + CREATE INDEX idx_reflection_extraction_jobs_lease + ON reflection_extraction_jobs(state, lease_expires_at); + `, + }, ]; export function sortedMigrations(migrations: readonly SqliteMigration[]): SqliteMigration[] { diff --git a/server/src/storage/sqlite/reflection-extraction-job-repository.ts b/server/src/storage/sqlite/reflection-extraction-job-repository.ts new file mode 100644 index 00000000..538f6353 --- /dev/null +++ b/server/src/storage/sqlite/reflection-extraction-job-repository.ts @@ -0,0 +1,190 @@ +import type { + ReflectionExtractionJob, + ReflectionExtractionJobClaimInput, + ReflectionExtractionJobClaimResult, + ReflectionExtractionJobCompareAndSetInput, + ReflectionExtractionJobCompareAndSetResult, + ReflectionExtractionJobEnqueueResult, + ReflectionExtractionJobListQuery, +} from '@veritas-kanban/shared'; +import { ReflectionExtractionJobSchema } from '../../schemas/reflection-extraction-job-schemas.js'; +import type { ReflectionExtractionJobRepository } from '../interfaces.js'; +import { + leaseExtractionJob, + normalizeExpiredExtractionJob, + selectExtractionJob, +} from '../reflection-extraction-job-state.js'; +import type { SqliteDatabase } from './database.js'; + +interface JobRow { + job_json: string; +} + +export class SqliteReflectionExtractionJobRepository implements ReflectionExtractionJobRepository { + constructor(private readonly database: SqliteDatabase) {} + + async enqueue(job: ReflectionExtractionJob): Promise { + const parsed = ReflectionExtractionJobSchema.parse(job); + if (parsed.revision !== 1) throw new Error('New extraction jobs must start at revision 1.'); + const connection = this.database.getConnection(); + connection.exec('BEGIN IMMEDIATE'); + try { + const existing = connection + .prepare('SELECT job_json FROM reflection_extraction_jobs WHERE idempotency_key = ?') + .get(parsed.idempotencyKey) as JobRow | undefined; + if (existing) { + connection.exec('COMMIT'); + return { + job: ReflectionExtractionJobSchema.parse(JSON.parse(existing.job_json)), + created: false, + }; + } + connection + .prepare( + `INSERT INTO reflection_extraction_jobs ( + id, workspace_id, state, revision, idempotency_key, available_at, + lease_owner_id, lease_expires_at, job_json, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` + ) + .run( + parsed.id, + parsed.workspaceId, + parsed.state, + parsed.revision, + parsed.idempotencyKey, + parsed.availableAt, + null, + null, + JSON.stringify(parsed), + parsed.createdAt, + parsed.updatedAt + ); + connection.exec('COMMIT'); + return { job: parsed, created: true }; + } catch (error) { + connection.exec('ROLLBACK'); + throw error; + } + } + + async get(id: string): Promise { + const row = this.database + .getConnection() + .prepare('SELECT job_json FROM reflection_extraction_jobs WHERE id = ?') + .get(id) as JobRow | undefined; + return row ? ReflectionExtractionJobSchema.parse(JSON.parse(row.job_json)) : null; + } + + async list(query: ReflectionExtractionJobListQuery): Promise { + const clauses: string[] = []; + const parameters: Array = []; + if (query.workspaceId) { + clauses.push('workspace_id = ?'); + parameters.push(query.workspaceId); + } + if (query.states?.length) { + clauses.push(`state IN (${query.states.map(() => '?').join(', ')})`); + parameters.push(...query.states); + } + parameters.push(Math.min(Math.max(query.limit ?? 100, 1), 1_000)); + const rows = this.database + .getConnection() + .prepare( + `SELECT job_json + FROM reflection_extraction_jobs + ${clauses.length > 0 ? `WHERE ${clauses.join(' AND ')}` : ''} + ORDER BY available_at, created_at, id + LIMIT ?` + ) + .all(...parameters) as unknown as JobRow[]; + return rows.map((row) => ReflectionExtractionJobSchema.parse(JSON.parse(row.job_json))); + } + + async claim( + input: ReflectionExtractionJobClaimInput + ): Promise { + const connection = this.database.getConnection(); + connection.exec('BEGIN IMMEDIATE'); + try { + const rows = connection + .prepare('SELECT job_json FROM reflection_extraction_jobs') + .all() as unknown as JobRow[]; + const jobs = rows.map((row) => ReflectionExtractionJobSchema.parse(JSON.parse(row.job_json))); + const normalized = jobs.map((job) => normalizeExpiredExtractionJob(job, input)); + for (const [index, job] of normalized.entries()) { + if (job.revision !== jobs[index]?.revision) this.updateJob(connection, job); + } + const selection = selectExtractionJob(normalized, input); + if ('reason' in selection) { + connection.exec('COMMIT'); + return { claimed: false, reason: selection.reason }; + } + const claimed = leaseExtractionJob(selection.job, input); + this.updateJob(connection, claimed); + connection.exec('COMMIT'); + return { claimed: true, job: claimed }; + } catch (error) { + connection.exec('ROLLBACK'); + throw error; + } + } + + async compareAndSet( + input: ReflectionExtractionJobCompareAndSetInput + ): Promise { + const connection = this.database.getConnection(); + connection.exec('BEGIN IMMEDIATE'); + try { + const row = connection + .prepare('SELECT job_json FROM reflection_extraction_jobs WHERE id = ?') + .get(input.id) as JobRow | undefined; + if (!row) { + connection.exec('COMMIT'); + return { updated: false, reason: 'not-found' }; + } + const current = ReflectionExtractionJobSchema.parse(JSON.parse(row.job_json)); + if (current.revision !== input.expectedRevision) { + connection.exec('COMMIT'); + return { job: current, updated: false, reason: 'stale-revision' }; + } + if (input.next.revision !== input.expectedRevision + 1 || input.next.id !== input.id) { + connection.exec('COMMIT'); + return { job: current, updated: false, reason: 'invalid-revision' }; + } + const next = ReflectionExtractionJobSchema.parse(input.next); + const result = this.updateJob(connection, next, input.expectedRevision); + if (result !== 1) throw new Error('Extraction job compare-and-set changed unexpectedly.'); + connection.exec('COMMIT'); + return { job: next, updated: true }; + } catch (error) { + connection.exec('ROLLBACK'); + throw error; + } + } + + private updateJob( + connection: ReturnType, + job: ReflectionExtractionJob, + expectedRevision?: number + ): number { + const result = connection + .prepare( + `UPDATE reflection_extraction_jobs + SET state = ?, revision = ?, available_at = ?, lease_owner_id = ?, + lease_expires_at = ?, job_json = ?, updated_at = ? + WHERE id = ?${expectedRevision === undefined ? '' : ' AND revision = ?'}` + ) + .run( + job.state, + job.revision, + job.availableAt, + job.lease?.ownerId ?? null, + job.lease?.expiresAt ?? null, + JSON.stringify(job), + job.updatedAt, + job.id, + ...(expectedRevision === undefined ? [] : [expectedRevision]) + ); + return Number(result.changes); + } +} diff --git a/server/src/storage/sqlite/sqlite-storage.ts b/server/src/storage/sqlite/sqlite-storage.ts index 06aa08b2..d4a474f2 100644 --- a/server/src/storage/sqlite/sqlite-storage.ts +++ b/server/src/storage/sqlite/sqlite-storage.ts @@ -16,6 +16,7 @@ import { SqliteRunApprovalRepository } from './run-approval-repository.js'; import { SqlitePhaseTransitionRepository } from './phase-transition-repository.js'; import { SqliteRunSupervisorRepository } from './run-supervisor-repository.js'; import { SqliteDurableGoalRepository } from './durable-goal-repository.js'; +import { SqliteReflectionExtractionJobRepository } from './reflection-extraction-job-repository.js'; import { SqliteAdmissionReservationRepository } from './admission-reservation-repository.js'; import { SqliteToolControlPlaneRepository } from './tool-control-plane-repository.js'; import { createDefaultConfig, normalizeAppConfig } from '../../services/config-service.js'; @@ -41,6 +42,7 @@ export class SqliteStorageProvider implements StorageProvider { readonly phaseTransitions: SqlitePhaseTransitionRepository; readonly runSupervisors: SqliteRunSupervisorRepository; readonly durableGoals: SqliteDurableGoalRepository; + readonly reflectionExtractionJobs: SqliteReflectionExtractionJobRepository; readonly admissionReservations: SqliteAdmissionReservationRepository; readonly toolControlPlane: SqliteToolControlPlaneRepository; @@ -67,6 +69,7 @@ export class SqliteStorageProvider implements StorageProvider { this.phaseTransitions = new SqlitePhaseTransitionRepository(this.sqlite); this.runSupervisors = new SqliteRunSupervisorRepository(this.sqlite); this.durableGoals = new SqliteDurableGoalRepository(this.sqlite); + this.reflectionExtractionJobs = new SqliteReflectionExtractionJobRepository(this.sqlite); this.admissionReservations = new SqliteAdmissionReservationRepository(this.sqlite); this.toolControlPlane = new SqliteToolControlPlaneRepository(this.sqlite); } diff --git a/shared/src/types/index.ts b/shared/src/types/index.ts index 517f9e35..2b25d173 100644 --- a/shared/src/types/index.ts +++ b/shared/src/types/index.ts @@ -10,6 +10,7 @@ export * from './chat.types.js'; export * from './communication-adapter.types.js'; export * from './ceremony.types.js'; export * from './reflection.types.js'; +export * from './reflection-extraction-job.types.js'; export * from './external-tracker.types.js'; export * from './transition-hooks.types.js'; export * from './delegation.types.js'; diff --git a/shared/src/types/reflection-extraction-job.types.ts b/shared/src/types/reflection-extraction-job.types.ts new file mode 100644 index 00000000..f3913a7c --- /dev/null +++ b/shared/src/types/reflection-extraction-job.types.ts @@ -0,0 +1,95 @@ +export const REFLECTION_EXTRACTION_JOB_SCHEMA_VERSION = 'reflection-extraction-job/v1' as const; + +export const REFLECTION_EXTRACTION_JOB_STATES = [ + 'queued', + 'leased', + 'completed', + 'dead-letter', +] as const; + +export type ReflectionExtractionJobState = (typeof REFLECTION_EXTRACTION_JOB_STATES)[number]; + +export interface ReflectionExtractionJobSource { + taskId: string; + attemptId: string; + completionId: string; + completionDigest: string; + runEventId?: string; +} + +export interface ReflectionExtractionJobLease { + ownerId: string; + acquiredAt: string; + expiresAt: string; +} + +export interface ReflectionExtractionJobFailure { + attempt: number; + code: string; + summary: string; + failedAt: string; + retryAt?: string; +} + +export interface ReflectionExtractionJob { + schemaVersion: typeof REFLECTION_EXTRACTION_JOB_SCHEMA_VERSION; + id: string; + workspaceId: string; + idempotencyKey: string; + source: ReflectionExtractionJobSource; + state: ReflectionExtractionJobState; + revision: number; + attemptCount: number; + maxAttempts: number; + availableAt: string; + lease?: ReflectionExtractionJobLease; + candidateIds: string[]; + failures: ReflectionExtractionJobFailure[]; + createdAt: string; + updatedAt: string; + completedAt?: string; +} + +export interface ReflectionExtractionJobListQuery { + workspaceId?: string; + states?: ReflectionExtractionJobState[]; + limit?: number; +} + +export interface ReflectionExtractionJobClaimInput { + ownerId: string; + now: string; + leaseDurationMs: number; + maxActiveGlobal: number; + maxActivePerWorkspace: number; + retryBaseDelayMs: number; + retryMaxDelayMs: number; + workspaceId?: string; +} + +export type ReflectionExtractionJobClaimResult = + | { + claimed: true; + job: ReflectionExtractionJob; + } + | { + claimed: false; + reason: 'empty' | 'global-limit' | 'workspace-limit'; + }; + +export interface ReflectionExtractionJobCompareAndSetInput { + id: string; + expectedRevision: number; + next: ReflectionExtractionJob; +} + +export interface ReflectionExtractionJobCompareAndSetResult { + job?: ReflectionExtractionJob; + updated: boolean; + reason?: 'not-found' | 'stale-revision' | 'invalid-revision'; +} + +export interface ReflectionExtractionJobEnqueueResult { + job: ReflectionExtractionJob; + created: boolean; +}