feat: Saga Manager (Sprint 4-6)

Motor generic pentru procese cu mai multi pasi care nu incap intr-o
tranzactie. Garantia nu e atomicitatea, ci compensarea in ordine inversa.

- saga-registry: definitii versionate in cod, backoff exponential plafonat
- saga-runner: corelator outbox->instanta, step runner, timeout, retry,
  compensare; revendicare cu FOR UPDATE SKIP LOCKED ca doua procese sa nu
  avanseze aceeasi saga simultan
- idempotenta la pornire prin index unic (saga, versiune, trigger_event),
  nu prin SELECT-apoi-INSERT care ar avea race
- compensarea are propriul retry: o compensare esuata lasa sistemul mai rau
  decat esecul original
- monitor /v1/sagas cu instante blocate si retry manual care NU sare pasi
- prima saga reala: imbogatire research brief, fara pas AI (un apel AI
  automat per brief ar schimba profilul de cost)

drizzle/0007_saga_manager.sql

@@ -0,0 +1,61 @@
+-- Saga Manager (Sprint 4-6). Coordoneaza procese cu mai multi pasi care nu
+-- incap intr-o tranzactie. Garantia nu e atomicitatea, ci compensarea in
+-- ordine inversa a pasilor deja executati.
+
+DO $$ BEGIN
+  CREATE TYPE "saga_status" AS ENUM
+    ('RUNNING','COMPLETED','COMPENSATING','COMPENSATED','FAILED','TIMED_OUT');
+EXCEPTION WHEN duplicate_object THEN NULL; END $$;
+
+DO $$ BEGIN
+  CREATE TYPE "saga_step_status" AS ENUM
+    ('PENDING','RUNNING','COMPLETED','FAILED','COMPENSATED','SKIPPED');
+EXCEPTION WHEN duplicate_object THEN NULL; END $$;
+
+CREATE TABLE IF NOT EXISTS "saga_instances" (
+  "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
+  "tenant_id" uuid NOT NULL,
+  "workspace_id" uuid,
+  "saga_name" text NOT NULL,
+  "saga_version" integer NOT NULL,
+  "status" "saga_status" DEFAULT 'RUNNING' NOT NULL,
+  "current_step" integer DEFAULT 0 NOT NULL,
+  "trigger_event_id" uuid NOT NULL,
+  "correlation_id" uuid,
+  "context" jsonb DEFAULT '{}'::jsonb NOT NULL,
+  "attempt" integer DEFAULT 0 NOT NULL,
+  "next_attempt_at" timestamp DEFAULT now() NOT NULL,
+  "timeout_at" timestamp,
+  "last_error" text,
+  "locked_until" timestamp,
+  "created_at" timestamp DEFAULT now() NOT NULL,
+  "updated_at" timestamp DEFAULT now() NOT NULL,
+  "completed_at" timestamp
+);
+
+-- Un eveniment porneste o singura instanta dintr-o saga data. Aceasta
+-- constrangere e ce face pornirea idempotenta, nu o verificare in cod.
+CREATE UNIQUE INDEX IF NOT EXISTS "saga_trigger_uq"
+  ON "saga_instances" ("saga_name","saga_version","trigger_event_id");
+CREATE INDEX IF NOT EXISTS "saga_due_idx"
+  ON "saga_instances" ("status","next_attempt_at");
+CREATE INDEX IF NOT EXISTS "saga_tenant_idx"
+  ON "saga_instances" ("tenant_id","created_at");
+
+CREATE TABLE IF NOT EXISTS "saga_steps" (
+  "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
+  "saga_instance_id" uuid NOT NULL,
+  "step_name" text NOT NULL,
+  "sequence" integer NOT NULL,
+  "status" "saga_step_status" DEFAULT 'PENDING' NOT NULL,
+  "attempts" integer DEFAULT 0 NOT NULL,
+  "output" jsonb,
+  "last_error" text,
+  "started_at" timestamp,
+  "completed_at" timestamp,
+  "compensated_at" timestamp
+);
+
+-- Un pas apare o singura data per instanta: baza idempotentei la avans.
+CREATE UNIQUE INDEX IF NOT EXISTS "saga_step_uq"
+  ON "saga_steps" ("saga_instance_id","sequence");

drizzle/meta/_journal.json

@@ -50,6 +50,13 @@
       "when": 1785328533134,
       "tag": "0006_notifications_engine",
       "breakpoints": true
+    },
+    {
+      "idx": 7,
+      "version": "7",
+      "when": 1785331800000,
+      "tag": "0007_saga_manager",
+      "breakpoints": true
     }
   ]
 }

src/app.module.ts

@@ -16,6 +16,7 @@ import { TenantsModule } from './tenants/tenants.module';
 import { NavigationModule } from './navigation/navigation.module';
 import { ProjectionsModule } from './projections/projections.module';
 import { NotificationsModule } from './notifications/notifications.module';
+import { SagasModule } from './sagas/sagas.module';
 import { IntelligenceModule } from './intelligence/intelligence.module';
 import { SegmentsModule } from './segments/segments.module';
 import { ResearchBriefsModule } from './research-briefs/research-briefs.module';
@@ -50,6 +51,7 @@ function parseRedisConnection(redisUrl: string | undefined) {
     NavigationModule,
     ProjectionsModule,
     NotificationsModule,
+    SagasModule,
     OrganizationsModule,
     TasksModule,
     IntelligenceModule,

src/db/schema.ts

@@ -471,6 +471,92 @@ export const notificationProcessedEvents = pgTable(
   (table) => [uniqueIndex('notification_processed_uq').on(table.ruleId, table.eventId)],
 );
 
+// --- Saga Manager (spec Sprint 4-6) ---------------------------------------
+// O saga coordoneaza un proces cu mai multi pasi care NU incap intr-o singura
+// tranzactie (apeluri externe, latenta, pasi care pot esua independent).
+// Garantia nu e atomicitatea, ci: ori toti pasii reusesc, ori cei deja executati
+// sunt compensati in ordine inversa.
+
+export const sagaStatus = pgEnum('saga_status', [
+  'RUNNING',
+  'COMPLETED',
+  'COMPENSATING',
+  'COMPENSATED',
+  'FAILED',
+  'TIMED_OUT',
+]);
+
+export const sagaStepStatus = pgEnum('saga_step_status', [
+  'PENDING',
+  'RUNNING',
+  'COMPLETED',
+  'FAILED',
+  'COMPENSATED',
+  'SKIPPED',
+]);
+
+export const sagaInstances = pgTable(
+  'saga_instances',
+  {
+    id: uuid('id').defaultRandom().primaryKey(),
+    tenantId: uuid('tenant_id').notNull(),
+    workspaceId: uuid('workspace_id'),
+    sagaName: text('saga_name').notNull(),
+    sagaVersion: integer('saga_version').notNull(),
+    status: sagaStatus('status').notNull().default('RUNNING'),
+    /** Indexul pasului curent in definitie. */
+    currentStep: integer('current_step').notNull().default(0),
+    /** Evenimentul care a pornit saga -- baza idempotentei la start. */
+    triggerEventId: uuid('trigger_event_id').notNull(),
+    correlationId: uuid('correlation_id'),
+    /** Starea acumulata, citita si scrisa de pasi. */
+    context: jsonb('context').notNull().default({}),
+    attempt: integer('attempt').notNull().default(0),
+    /** Cand poate fi reluata: backoff exponential dupa esec. */
+    nextAttemptAt: timestamp('next_attempt_at').defaultNow().notNull(),
+    /** Termen absolut pentru intreaga saga, nu doar pentru pasul curent. */
+    timeoutAt: timestamp('timeout_at'),
+    lastError: text('last_error'),
+    /**
+     * Lease de executie. Doua procese runner nu trebuie sa avanseze aceeasi
+     * saga simultan; randul se ia cu FOR UPDATE SKIP LOCKED, iar lockedUntil
+     * elibereaza saga daca procesul care o tinea a murit.
+     */
+    lockedUntil: timestamp('locked_until'),
+    createdAt: timestamp('created_at').defaultNow().notNull(),
+    updatedAt: timestamp('updated_at').defaultNow().notNull(),
+    completedAt: timestamp('completed_at'),
+  },
+  (table) => [
+    // Un eveniment porneste o singura instanta dintr-o saga data.
+    uniqueIndex('saga_trigger_uq').on(table.sagaName, table.sagaVersion, table.triggerEventId),
+    index('saga_due_idx').on(table.status, table.nextAttemptAt),
+    index('saga_tenant_idx').on(table.tenantId, table.createdAt),
+  ],
+);
+
+export const sagaSteps = pgTable(
+  'saga_steps',
+  {
+    id: uuid('id').defaultRandom().primaryKey(),
+    sagaInstanceId: uuid('saga_instance_id').notNull(),
+    stepName: text('step_name').notNull(),
+    /** Pozitia in definitie; determina si ordinea inversa a compensarii. */
+    sequence: integer('sequence').notNull(),
+    status: sagaStepStatus('status').notNull().default('PENDING'),
+    attempts: integer('attempts').notNull().default(0),
+    output: jsonb('output'),
+    lastError: text('last_error'),
+    startedAt: timestamp('started_at'),
+    completedAt: timestamp('completed_at'),
+    compensatedAt: timestamp('compensated_at'),
+  },
+  (table) => [
+    // Un pas apare o singura data per instanta: baza idempotentei la avans.
+    uniqueIndex('saga_step_uq').on(table.sagaInstanceId, table.sequence),
+  ],
+);
+
 // Append-oriented (spec sectiunea 29): se scrie, nu se modifica si nu se sterge.
 export const auditLog = pgTable(
   'audit_log',

src/sagas/research-brief-enrichment.saga.ts

@@ -0,0 +1,162 @@
+import { Logger } from '@nestjs/common';
+import { eq } from 'drizzle-orm';
+import { db } from '../db/client';
+import { outboxEvents, researchBriefs } from '../db/schema';
+import type { IntelligenceService } from '../intelligence/intelligence.service';
+import { registerSaga, type SagaDefinition } from './saga-registry';
+
+const logger = new Logger('ResearchBriefEnrichmentSaga');
+
+/** Marcaj in sources care arata ca imbogatirea a rulat. Baza idempotentei. */
+const ENRICHMENT_LABEL = 'intelligence-api (enrichment)';
+
+interface SourceEntry {
+  label: string;
+  url?: string;
+}
+
+/**
+ * Imbogatirea unui research brief dupa creare.
+ *
+ * De ce saga si nu un simplu handler: sunt trei pasi din care primul apeleaza
+ * un serviciu extern care poate fi lent sau picat. Daca pasul 2 esueaza dupa ce
+ * pasul 1 a scris deja in brief, un handler obisnuit ar lasa brieful marcat ca
+ * imbogatit fara sa fie. Saga anuleaza scrierea.
+ *
+ * NU contine pas de AI in mod deliberat: un apel AI declansat automat la fiecare
+ * brief creat ar schimba profilul de cost. Draftul AI ramane pe actiunea
+ * manuala existenta (POST .../draft).
+ */
+export function buildResearchBriefEnrichmentSaga(
+  intelligence: IntelligenceService,
+): SagaDefinition {
+  return {
+    name: 'research_brief_enrichment',
+    version: 1,
+    triggerEvent: 'research_brief.created',
+    timeoutMinutes: 30,
+
+    shouldStart(event) {
+      // Fara organizationId nu avem ce imbogati; nu pornim o saga care ar
+      // esua garantat la primul pas.
+      return typeof event.payload.organizationId === 'string' && !!event.payload.organizationId;
+    },
+
+    steps: [
+      {
+        name: 'fetch_company_intelligence',
+        // Pasul care are cea mai mare sansa de esec tranzitoriu.
+        maxAttempts: 4,
+        async run(_ctx, event) {
+          const organizationId = String(event.payload.organizationId);
+          const company = await intelligence.getCompany(organizationId);
+          return {
+            context: {
+              company: {
+                name: company.organization_name ?? null,
+                domain: company.normalized_domain ?? null,
+                industries: company.industries ?? [],
+                hqCity: company.hq_city ?? null,
+                hqCountry: company.hq_country ?? null,
+              },
+            },
+            output: { organizationId, resolved: !!company.organization_name },
+          };
+        },
+        // Citire pura: nu are efecte de anulat.
+      },
+
+      {
+        name: 'attach_provenance',
+        async run(ctx, event) {
+          const briefId = await resolveBriefId(event.id);
+          if (!briefId) throw new Error('Briefu-l declansator nu mai exista');
+
+          const brief = await db.query.researchBriefs.findFirst({
+            where: eq(researchBriefs.id, briefId),
+          });
+          if (!brief) throw new Error(`Brief ${briefId} nu a fost gasit`);
+
+          const sources = (brief.sources ?? []) as SourceEntry[];
+          // Idempotent: a doua rulare nu adauga a doua oara aceeasi sursa.
+          if (sources.some((s) => s.label === ENRICHMENT_LABEL)) {
+            return { context: { briefId }, output: { alreadyAttached: true } };
+          }
+
+          const company = (ctx.company ?? {}) as Record<string, unknown>;
+          await db
+            .update(researchBriefs)
+            .set({
+              sources: [
+                ...sources,
+                {
+                  label: ENRICHMENT_LABEL,
+                  url: 'https://intelligence-api.boardmind.dev',
+                  resolvedName: company.name ?? null,
+                  resolvedDomain: company.domain ?? null,
+                },
+              ],
+              organizationName: (company.name as string) ?? brief.organizationName,
+              updatedAt: new Date(),
+            })
+            .where(eq(researchBriefs.id, briefId));
+
+          return { context: { briefId }, output: { attached: true } };
+        },
+        async compensate(ctx) {
+          const briefId = ctx.briefId as string | undefined;
+          if (!briefId) return;
+          const brief = await db.query.researchBriefs.findFirst({
+            where: eq(researchBriefs.id, briefId),
+          });
+          // Briefu-l poate fi sters intre timp: compensam ce gasim, nu presupunem.
+          if (!brief) return;
+
+          const sources = (brief.sources ?? []) as SourceEntry[];
+          await db
+            .update(researchBriefs)
+            .set({
+              sources: sources.filter((s) => s.label !== ENRICHMENT_LABEL),
+              updatedAt: new Date(),
+            })
+            .where(eq(researchBriefs.id, briefId));
+          logger.log(`Provenance retrasa de pe briefu-l ${briefId}`);
+        },
+      },
+
+      {
+        name: 'emit_enriched_event',
+        async run(ctx, event) {
+          const briefId = ctx.briefId as string | undefined;
+          if (!briefId) throw new Error('Lipseste briefId din context');
+
+          // Evenimentul e consumat de Notifications Engine si de proiectii;
+          // saga nu trimite ea notificari, ci anunta faptul.
+          await db.insert(outboxEvents).values({
+            tenantId: event.tenantId,
+            workspaceId: event.workspaceId,
+            eventType: 'research_brief.enriched',
+            aggregateType: 'research_brief',
+            subjectId: briefId,
+            correlationId: event.correlationId,
+            causationId: event.id,
+            payload: { briefId, company: ctx.company ?? null },
+          });
+          return { output: { emitted: true } };
+        },
+      },
+    ],
+  };
+}
+
+/** subject_id al evenimentului declansator = id-ul briefului. */
+async function resolveBriefId(triggerEventId: string): Promise<string | null> {
+  const row = await db.query.outboxEvents.findFirst({
+    where: eq(outboxEvents.id, triggerEventId),
+  });
+  return row?.subjectId ?? null;
+}
+
+export function registerResearchBriefEnrichment(intelligence: IntelligenceService): void {
+  registerSaga(buildResearchBriefEnrichmentSaga(intelligence));
+}

src/sagas/saga-registry.spec.ts

@@ -0,0 +1,75 @@
+import {
+  backoffDelayMs,
+  findSaga,
+  maxAttemptsFor,
+  registerSaga,
+  resetRegistry,
+  sagasForEvent,
+  triggerEventTypes,
+  type SagaDefinition,
+} from './saga-registry';
+
+const stubSaga = (over: Partial<SagaDefinition> = {}): SagaDefinition => ({
+  name: 'test_saga',
+  version: 1,
+  triggerEvent: 'thing.created',
+  timeoutMinutes: 10,
+  steps: [{ name: 'step_one', run: async () => {} }],
+  ...over,
+});
+
+describe('saga registry', () => {
+  beforeEach(() => resetRegistry());
+
+  it('gaseste sagas dupa evenimentul declansator', () => {
+    registerSaga(stubSaga());
+    expect(sagasForEvent('thing.created')).toHaveLength(1);
+    expect(sagasForEvent('altceva.created')).toEqual([]);
+  });
+
+  it('refuza inregistrarea aceleiasi sagas de doua ori pe aceeasi versiune', () => {
+    // Doua definitii cu acelasi nume+versiune ar porni doua instante per
+    // eveniment; e o eroare de programare, deci esueaza tare la bootstrap.
+    registerSaga(stubSaga());
+    expect(() => registerSaga(stubSaga())).toThrow(/deja inregistrata/);
+  });
+
+  it('permite versiuni diferite ale aceleiasi sagas sa coexiste', () => {
+    registerSaga(stubSaga({ version: 1 }));
+    registerSaga(stubSaga({ version: 2 }));
+    expect(findSaga('test_saga', 1)?.version).toBe(1);
+    expect(findSaga('test_saga', 2)?.version).toBe(2);
+  });
+
+  it('deduplica tipurile de evenimente la care e abonat macar o saga', () => {
+    registerSaga(stubSaga({ name: 'a' }));
+    registerSaga(stubSaga({ name: 'b' }));
+    expect(triggerEventTypes()).toEqual(['thing.created']);
+  });
+
+  describe('backoff', () => {
+    it('creste exponential intre incercari', () => {
+      expect(backoffDelayMs(1)).toBe(30_000);
+      expect(backoffDelayMs(2)).toBe(60_000);
+      expect(backoffDelayMs(3)).toBe(120_000);
+    });
+
+    it('se plafoneaza, ca reincercarile tarzii sa nu ajunga peste zile', () => {
+      expect(backoffDelayMs(50)).toBe(30 * 60_000);
+    });
+
+    it('nu intoarce delay negativ pentru incercarea zero', () => {
+      expect(backoffDelayMs(0)).toBeGreaterThan(0);
+    });
+  });
+
+  describe('maxAttempts', () => {
+    it('are un default cand pasul nu specifica', () => {
+      expect(maxAttemptsFor({ name: 's', run: async () => {} })).toBe(3);
+    });
+
+    it('respecta valoarea explicita a pasului', () => {
+      expect(maxAttemptsFor({ name: 's', run: async () => {}, maxAttempts: 7 })).toBe(7);
+    });
+  });
+});

src/sagas/saga-registry.ts

@@ -0,0 +1,108 @@
+/**
+ * Definitiile de saga stau in cod, versionate, ca la notification-rules si
+ * projection-registry. Motivul e acelasi: un proces cu mai multi pasi e logica
+ * de business care merita review si teste, nu configuratie editabila la runtime.
+ *
+ * Contractul unui pas:
+ *  - run() trebuie sa fie IDEMPOTENT. Poate fi apelat de mai multe ori pentru
+ *    acelasi pas (retry dupa timeout de retea, restart de proces). Daca a scris
+ *    deja ceva, a doua rulare trebuie sa observe asta si sa nu dubleze.
+ *  - compensate() anuleaza efectul lui run(). Trebuie sa fie idempotent SI
+ *    tolerant la faptul ca run() poate sa fi esuat la jumatate -- compenseaza
+ *    ce gaseste, nu presupune ca totul a fost creat.
+ */
+
+export interface SagaEvent {
+  id: string;
+  tenantId: string;
+  workspaceId: string | null;
+  eventType: string;
+  actorId: string | null;
+  subjectId: string | null;
+  correlationId: string | null;
+  payload: Record<string, unknown>;
+}
+
+export type SagaContext = Record<string, unknown>;
+
+export interface SagaStepResult {
+  /** Se uneste in contextul sagai si e vizibil pasilor urmatori. */
+  context?: SagaContext;
+  output?: Record<string, unknown>;
+}
+
+export interface SagaStepDefinition {
+  name: string;
+  run(ctx: SagaContext, event: SagaEvent): Promise<SagaStepResult | void>;
+  /**
+   * Lipsa lui compensate inseamna ca pasul nu are efecte de anulat (o citire,
+   * de exemplu). Nu inseamna "nu stim cum sa compensam".
+   */
+  compensate?(ctx: SagaContext, event: SagaEvent): Promise<void>;
+  /** Cate incercari inainte de a declara pasul esuat. */
+  maxAttempts?: number;
+}
+
+export interface SagaDefinition {
+  name: string;
+  version: number;
+  /** Ce eveniment porneste saga. */
+  triggerEvent: string;
+  /** Termen absolut pentru toata saga; depasirea duce la compensare. */
+  timeoutMinutes: number;
+  steps: SagaStepDefinition[];
+  /**
+   * Filtru optional: chiar daca evenimentul se potriveste, saga poate decide
+   * ca nu o priveste (ex. lipseste un camp din payload).
+   */
+  shouldStart?(event: SagaEvent): boolean;
+}
+
+const DEFAULT_MAX_ATTEMPTS = 3;
+const BASE_BACKOFF_MS = 30_000;
+const MAX_BACKOFF_MS = 30 * 60_000;
+
+export function maxAttemptsFor(step: SagaStepDefinition): number {
+  return step.maxAttempts ?? DEFAULT_MAX_ATTEMPTS;
+}
+
+/**
+ * Backoff exponential plafonat. Fara plafon, a 10-a incercare ar fi programata
+ * peste zile, ceea ce in practica inseamna "niciodata".
+ */
+export function backoffDelayMs(attempt: number): number {
+  return Math.min(BASE_BACKOFF_MS * 2 ** Math.max(0, attempt - 1), MAX_BACKOFF_MS);
+}
+
+const REGISTRY: SagaDefinition[] = [];
+
+export function registerSaga(definition: SagaDefinition): void {
+  const duplicate = REGISTRY.find(
+    (s) => s.name === definition.name && s.version === definition.version,
+  );
+  if (duplicate) {
+    throw new Error(`Saga ${definition.name}@v${definition.version} este deja inregistrata`);
+  }
+  REGISTRY.push(definition);
+}
+
+export function allSagas(): SagaDefinition[] {
+  return [...REGISTRY];
+}
+
+export function sagasForEvent(eventType: string): SagaDefinition[] {
+  return REGISTRY.filter((s) => s.triggerEvent === eventType);
+}
+
+export function findSaga(name: string, version: number): SagaDefinition | undefined {
+  return REGISTRY.find((s) => s.name === name && s.version === version);
+}
+
+export function triggerEventTypes(): string[] {
+  return [...new Set(REGISTRY.map((s) => s.triggerEvent))];
+}
+
+/** Doar pentru teste: goleste registrul intre cazuri. */
+export function resetRegistry(): void {
+  REGISTRY.length = 0;
+}

src/sagas/saga-runner.service.ts

@@ -0,0 +1,485 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { Cron, CronExpression } from '@nestjs/schedule';
+import { and, asc, desc, eq, gt, inArray, sql } from 'drizzle-orm';
+import { db } from '../db/client';
+import { outboxEvents, sagaInstances, sagaSteps } from '../db/schema';
+import {
+  allSagas,
+  backoffDelayMs,
+  findSaga,
+  maxAttemptsFor,
+  sagasForEvent,
+  triggerEventTypes,
+  type SagaContext,
+  type SagaDefinition,
+  type SagaEvent,
+} from './saga-registry';
+
+const BATCH_SIZE = 50;
+const CORRELATOR_BATCH = 200;
+const SAFETY_LAG_MS = 60_000;
+/** Cat timp o instanta revendicata ramane a acestui proces. */
+const LEASE_MS = 5 * 60_000;
+
+type SagaInstanceRow = typeof sagaInstances.$inferSelect;
+
+@Injectable()
+export class SagaRunnerService {
+  private readonly logger = new Logger(SagaRunnerService.name);
+  private lastSeenAt: Date | null = null;
+
+  // --- Corelator: eveniment din outbox -> instanta noua de saga -------------
+
+  @Cron(CronExpression.EVERY_10_SECONDS)
+  async correlate(): Promise<void> {
+    try {
+      await this.correlateBatch();
+    } catch (error) {
+      this.logger.error('Saga correlator failed', error as Error);
+    }
+  }
+
+  async correlateBatch(): Promise<number> {
+    const eventTypes = triggerEventTypes();
+    if (eventTypes.length === 0) return 0;
+
+    const since = this.lastSeenAt
+      ? new Date(this.lastSeenAt.getTime() - SAFETY_LAG_MS)
+      : new Date(Date.now() - 24 * 60 * 60 * 1000);
+
+    const events = await db
+      .select()
+      .from(outboxEvents)
+      .where(and(gt(outboxEvents.createdAt, since), inArray(outboxEvents.eventType, eventTypes)))
+      .orderBy(asc(outboxEvents.createdAt))
+      .limit(CORRELATOR_BATCH);
+
+    let started = 0;
+    for (const row of events) {
+      const event = this.toSagaEvent(row);
+      for (const definition of sagasForEvent(event.eventType)) {
+        if (definition.shouldStart && !definition.shouldStart(event)) continue;
+        started += await this.startInstance(definition, event);
+      }
+      this.lastSeenAt = row.createdAt;
+    }
+
+    if (started > 0) this.logger.log(`Started ${started} saga instance(s)`);
+    return started;
+  }
+
+  private async startInstance(definition: SagaDefinition, event: SagaEvent): Promise<number> {
+    // Idempotenta la pornire vine din indexul unic (saga, versiune, eveniment),
+    // nu dintr-un SELECT-apoi-INSERT care ar avea race intre doua procese.
+    const inserted = await db
+      .insert(sagaInstances)
+      .values({
+        tenantId: event.tenantId,
+        workspaceId: event.workspaceId,
+        sagaName: definition.name,
+        sagaVersion: definition.version,
+        triggerEventId: event.id,
+        correlationId: event.correlationId,
+        context: { event: event.payload },
+        timeoutAt: new Date(Date.now() + definition.timeoutMinutes * 60_000),
+      })
+      .onConflictDoNothing()
+      .returning({ id: sagaInstances.id });
+    return inserted.length;
+  }
+
+  // --- Step runner ---------------------------------------------------------
+
+  @Cron(CronExpression.EVERY_10_SECONDS)
+  async advance(): Promise<void> {
+    try {
+      await this.advanceBatch();
+    } catch (error) {
+      this.logger.error('Saga runner failed', error as Error);
+    }
+  }
+
+  async advanceBatch(): Promise<number> {
+    const claimed = await this.claimDueInstances(BATCH_SIZE);
+    let processed = 0;
+    for (const instance of claimed) {
+      try {
+        await this.processInstance(instance);
+      } catch (error) {
+        this.logger.error(`Saga ${instance.sagaName}/${instance.id} crashed`, error as Error);
+        await this.releaseWithError(instance, error);
+      }
+      processed += 1;
+    }
+    return processed;
+  }
+
+  /**
+   * Revendica instantele scadente intr-un singur UPDATE atomic.
+   *
+   * FOR UPDATE SKIP LOCKED e ce impiedica doua procese runner sa avanseze
+   * aceeasi saga simultan: al doilea proces sare peste randurile blocate in loc
+   * sa astepte. lockedUntil e plasa de siguranta pentru cazul in care procesul
+   * care detinea lease-ul moare fara sa il elibereze.
+   */
+  private async claimDueInstances(limit: number): Promise<SagaInstanceRow[]> {
+    const leaseUntil = new Date(Date.now() + LEASE_MS);
+    const result = await db.execute(sql`
+      UPDATE saga_instances SET locked_until = ${leaseUntil}, updated_at = now()
+      WHERE id IN (
+        SELECT id FROM saga_instances
+        WHERE status IN ('RUNNING','COMPENSATING')
+          AND next_attempt_at <= now()
+          AND (locked_until IS NULL OR locked_until < now())
+        ORDER BY next_attempt_at ASC
+        LIMIT ${limit}
+        FOR UPDATE SKIP LOCKED
+      )
+      RETURNING *;
+    `);
+    return (result.rows as unknown[]).map((row) => this.mapInstance(row as Record<string, unknown>));
+  }
+
+  private async processInstance(instance: SagaInstanceRow): Promise<void> {
+    const definition = findSaga(instance.sagaName, instance.sagaVersion);
+    if (!definition) {
+      // Definitia a disparut sau a fost re-versionata sub o instanta activa.
+      // Nu ghicim ce ar fi trebuit sa faca: o marcam si o lasam vizibila.
+      await this.fail(instance, `Definitia ${instance.sagaName}@v${instance.sagaVersion} lipseste`);
+      return;
+    }
+
+    if (instance.status === 'COMPENSATING') {
+      await this.compensateNext(instance, definition);
+      return;
+    }
+
+    // Timeoutul se verifica inainte de a mai executa un pas: o saga expirata
+    // nu trebuie sa mai produca efecte noi, ci sa le anuleze pe cele vechi.
+    if (instance.timeoutAt && instance.timeoutAt < new Date()) {
+      this.logger.warn(`Saga ${instance.sagaName}/${instance.id} a expirat; compensez`);
+      await this.beginCompensation(instance, 'Saga a depasit timeout-ul');
+      return;
+    }
+
+    const step = definition.steps[instance.currentStep];
+    if (!step) {
+      await this.complete(instance);
+      return;
+    }
+
+    const event = this.eventFromContext(instance);
+    const context = (instance.context ?? {}) as SagaContext;
+    const attempts = instance.attempt + 1;
+
+    await this.upsertStep(instance, step.name, instance.currentStep, 'RUNNING', attempts);
+
+    try {
+      const result = await step.run(context, event);
+      const merged = { ...context, ...(result?.context ?? {}) };
+
+      await db.transaction(async (tx) => {
+        await tx
+          .update(sagaSteps)
+          .set({
+            status: 'COMPLETED',
+            output: result?.output ?? null,
+            completedAt: new Date(),
+            lastError: null,
+          })
+          .where(
+            and(
+              eq(sagaSteps.sagaInstanceId, instance.id),
+              eq(sagaSteps.sequence, instance.currentStep),
+            ),
+          );
+
+        await tx
+          .update(sagaInstances)
+          .set({
+            currentStep: instance.currentStep + 1,
+            context: merged,
+            attempt: 0,
+            lastError: null,
+            lockedUntil: null,
+            nextAttemptAt: new Date(),
+            updatedAt: new Date(),
+          })
+          .where(eq(sagaInstances.id, instance.id));
+      });
+    } catch (error) {
+      await this.handleStepFailure(instance, definition, step.name, attempts, error);
+    }
+  }
+
+  private async handleStepFailure(
+    instance: SagaInstanceRow,
+    definition: SagaDefinition,
+    stepName: string,
+    attempts: number,
+    error: unknown,
+  ): Promise<void> {
+    const message = error instanceof Error ? error.message : String(error);
+    const step = definition.steps[instance.currentStep];
+    const exhausted = attempts >= maxAttemptsFor(step);
+
+    await db
+      .update(sagaSteps)
+      .set({ status: exhausted ? 'FAILED' : 'PENDING', lastError: message })
+      .where(
+        and(eq(sagaSteps.sagaInstanceId, instance.id), eq(sagaSteps.sequence, instance.currentStep)),
+      );
+
+    if (!exhausted) {
+      const retryAt = new Date(Date.now() + backoffDelayMs(attempts));
+      this.logger.warn(
+        `Saga ${definition.name}/${instance.id} pas "${stepName}" a esuat (${attempts}/${maxAttemptsFor(step)}); reincerc la ${retryAt.toISOString()}`,
+      );
+      await db
+        .update(sagaInstances)
+        .set({
+          attempt: attempts,
+          lastError: message,
+          nextAttemptAt: retryAt,
+          lockedUntil: null,
+          updatedAt: new Date(),
+        })
+        .where(eq(sagaInstances.id, instance.id));
+      return;
+    }
+
+    this.logger.error(
+      `Saga ${definition.name}/${instance.id} pas "${stepName}" epuizat dupa ${attempts} incercari; compensez`,
+    );
+    await this.beginCompensation(instance, `Pasul "${stepName}" a esuat: ${message}`);
+  }
+
+  // --- Compensare ----------------------------------------------------------
+
+  private async beginCompensation(instance: SagaInstanceRow, reason: string): Promise<void> {
+    await db
+      .update(sagaInstances)
+      .set({
+        status: 'COMPENSATING',
+        lastError: reason,
+        attempt: 0,
+        nextAttemptAt: new Date(),
+        lockedUntil: null,
+        updatedAt: new Date(),
+      })
+      .where(eq(sagaInstances.id, instance.id));
+  }
+
+  /**
+   * Compenseaza un singur pas per rulare, in ordine inversa, si lasa instanta
+   * pentru urmatorul tur. Compensarea are propriul retry: o compensare esuata
+   * lasa sistemul intr-o stare mai proasta decat esecul original, deci nu o
+   * abandonam dupa prima incercare.
+   */
+  private async compensateNext(
+    instance: SagaInstanceRow,
+    definition: SagaDefinition,
+  ): Promise<void> {
+    const [pending] = await db
+      .select()
+      .from(sagaSteps)
+      .where(and(eq(sagaSteps.sagaInstanceId, instance.id), eq(sagaSteps.status, 'COMPLETED')))
+      .orderBy(desc(sagaSteps.sequence))
+      .limit(1);
+
+    if (!pending) {
+      await db
+        .update(sagaInstances)
+        .set({
+          status: 'COMPENSATED',
+          completedAt: new Date(),
+          lockedUntil: null,
+          updatedAt: new Date(),
+        })
+        .where(eq(sagaInstances.id, instance.id));
+      this.logger.log(`Saga ${definition.name}/${instance.id} compensata complet`);
+      return;
+    }
+
+    const step = definition.steps[pending.sequence];
+    const context = (instance.context ?? {}) as SagaContext;
+    const event = this.eventFromContext(instance);
+    const attempts = instance.attempt + 1;
+
+    try {
+      // Un pas fara compensate() nu are efecte de anulat; il marcam si mergem
+      // mai departe, nu il tratam ca esec.
+      if (step?.compensate) {
+        await step.compensate(context, event);
+      }
+      await db
+        .update(sagaSteps)
+        .set({ status: 'COMPENSATED', compensatedAt: new Date() })
+        .where(eq(sagaSteps.id, pending.id));
+      await db
+        .update(sagaInstances)
+        .set({ attempt: 0, lockedUntil: null, nextAttemptAt: new Date(), updatedAt: new Date() })
+        .where(eq(sagaInstances.id, instance.id));
+    } catch (error) {
+      const message = error instanceof Error ? error.message : String(error);
+      const exhausted = attempts >= maxAttemptsFor(step ?? { name: pending.stepName, run: async () => {} });
+
+      if (exhausted) {
+        // Compensare imposibila: nu pretindem ca s-a curatat. FAILED e o stare
+        // care cere interventie umana, si de asta e vizibila in monitor.
+        this.logger.error(
+          `Saga ${definition.name}/${instance.id}: compensarea pasului "${pending.stepName}" a esuat definitiv`,
+        );
+        await this.fail(instance, `Compensare esuata la "${pending.stepName}": ${message}`);
+        return;
+      }
+
+      await db
+        .update(sagaSteps)
+        .set({ lastError: message })
+        .where(eq(sagaSteps.id, pending.id));
+      await db
+        .update(sagaInstances)
+        .set({
+          attempt: attempts,
+          lastError: message,
+          nextAttemptAt: new Date(Date.now() + backoffDelayMs(attempts)),
+          lockedUntil: null,
+          updatedAt: new Date(),
+        })
+        .where(eq(sagaInstances.id, instance.id));
+    }
+  }
+
+  // --- Tranzitii terminale -------------------------------------------------
+
+  private async complete(instance: SagaInstanceRow): Promise<void> {
+    await db
+      .update(sagaInstances)
+      .set({
+        status: 'COMPLETED',
+        completedAt: new Date(),
+        lockedUntil: null,
+        lastError: null,
+        updatedAt: new Date(),
+      })
+      .where(eq(sagaInstances.id, instance.id));
+  }
+
+  private async fail(instance: SagaInstanceRow, reason: string): Promise<void> {
+    await db
+      .update(sagaInstances)
+      .set({
+        status: 'FAILED',
+        lastError: reason,
+        completedAt: new Date(),
+        lockedUntil: null,
+        updatedAt: new Date(),
+      })
+      .where(eq(sagaInstances.id, instance.id));
+  }
+
+  private async releaseWithError(instance: SagaInstanceRow, error: unknown): Promise<void> {
+    const message = error instanceof Error ? error.message : String(error);
+    await db
+      .update(sagaInstances)
+      .set({
+        lastError: message,
+        lockedUntil: null,
+        nextAttemptAt: new Date(Date.now() + backoffDelayMs(instance.attempt + 1)),
+        updatedAt: new Date(),
+      })
+      .where(eq(sagaInstances.id, instance.id));
+  }
+
+  // --- Ajutoare ------------------------------------------------------------
+
+  private async upsertStep(
+    instance: SagaInstanceRow,
+    stepName: string,
+    sequence: number,
+    status: 'RUNNING',
+    attempts: number,
+  ): Promise<void> {
+    await db
+      .insert(sagaSteps)
+      .values({
+        sagaInstanceId: instance.id,
+        stepName,
+        sequence,
+        status,
+        attempts,
+        startedAt: new Date(),
+      })
+      .onConflictDoUpdate({
+        target: [sagaSteps.sagaInstanceId, sagaSteps.sequence],
+        set: { status, attempts, startedAt: new Date() },
+      });
+  }
+
+  private toSagaEvent(row: typeof outboxEvents.$inferSelect): SagaEvent {
+    return {
+      id: row.id,
+      tenantId: row.tenantId,
+      workspaceId: row.workspaceId,
+      eventType: row.eventType,
+      actorId: row.actorId,
+      subjectId: row.subjectId,
+      correlationId: row.correlationId,
+      payload: (row.payload ?? {}) as Record<string, unknown>,
+    };
+  }
+
+  /**
+   * Reconstituie evenimentul declansator din contextul persistat. Nu recitim
+   * outbox_events: evenimentul original poate fi arhivat, iar saga trebuie sa
+   * ramana rulabila si dupa aceea.
+   */
+  private eventFromContext(instance: SagaInstanceRow): SagaEvent {
+    const context = (instance.context ?? {}) as SagaContext;
+    return {
+      id: instance.triggerEventId,
+      tenantId: instance.tenantId,
+      workspaceId: instance.workspaceId,
+      eventType: '',
+      actorId: null,
+      subjectId: null,
+      correlationId: instance.correlationId,
+      payload: (context.event ?? {}) as Record<string, unknown>,
+    };
+  }
+
+  private mapInstance(row: Record<string, unknown>): SagaInstanceRow {
+    const date = (v: unknown) => (v ? new Date(v as string) : null);
+    return {
+      id: row.id as string,
+      tenantId: row.tenant_id as string,
+      workspaceId: (row.workspace_id as string) ?? null,
+      sagaName: row.saga_name as string,
+      sagaVersion: Number(row.saga_version),
+      status: row.status as SagaInstanceRow['status'],
+      currentStep: Number(row.current_step),
+      triggerEventId: row.trigger_event_id as string,
+      correlationId: (row.correlation_id as string) ?? null,
+      context: (row.context ?? {}) as SagaContext,
+      attempt: Number(row.attempt),
+      nextAttemptAt: date(row.next_attempt_at) as Date,
+      timeoutAt: date(row.timeout_at),
+      lastError: (row.last_error as string) ?? null,
+      lockedUntil: date(row.locked_until),
+      createdAt: date(row.created_at) as Date,
+      updatedAt: date(row.updated_at) as Date,
+      completedAt: date(row.completed_at),
+    };
+  }
+
+  /** Expus pentru monitor: cate definitii sunt incarcate. */
+  registeredSagas(): { name: string; version: number; triggerEvent: string; steps: number }[] {
+    return allSagas().map((s) => ({
+      name: s.name,
+      version: s.version,
+      triggerEvent: s.triggerEvent,
+      steps: s.steps.length,
+    }));
+  }
+}

src/sagas/sagas.controller.ts

@@ -0,0 +1,40 @@
+import { Controller, Get, Param, ParseUUIDPipe, Post, Query } from '@nestjs/common';
+import { CurrentSession } from '../auth/session.decorator';
+import type { AuthenticatedSession } from '../auth/tenant.guard';
+import { SagasService } from './sagas.service';
+
+const STATUSES = [
+  'RUNNING',
+  'COMPLETED',
+  'COMPENSATING',
+  'COMPENSATED',
+  'FAILED',
+  'TIMED_OUT',
+] as const;
+
+@Controller('sagas')
+export class SagasController {
+  constructor(private readonly sagas: SagasService) {}
+
+  @Get()
+  overview(@CurrentSession() session: AuthenticatedSession) {
+    return this.sagas.overview(session);
+  }
+
+  @Get('instances')
+  list(@CurrentSession() session: AuthenticatedSession, @Query('status') status?: string) {
+    // Allowlist, nu interpolare: statusul ajunge intr-o comparatie SQL.
+    const safe = STATUSES.includes(status as (typeof STATUSES)[number]) ? status : undefined;
+    return this.sagas.list(session, safe);
+  }
+
+  @Get('instances/:id')
+  detail(@CurrentSession() session: AuthenticatedSession, @Param('id', ParseUUIDPipe) id: string) {
+    return this.sagas.detail(session, id);
+  }
+
+  @Post('instances/:id/retry')
+  retry(@CurrentSession() session: AuthenticatedSession, @Param('id', ParseUUIDPipe) id: string) {
+    return this.sagas.retry(session, id);
+  }
+}

src/sagas/sagas.module.ts

@@ -0,0 +1,26 @@
+import { Module, OnModuleInit } from '@nestjs/common';
+import { IntelligenceModule } from '../intelligence/intelligence.module';
+import { IntelligenceService } from '../intelligence/intelligence.service';
+import { registerResearchBriefEnrichment } from './research-brief-enrichment.saga';
+import { SagaRunnerService } from './saga-runner.service';
+import { SagasController } from './sagas.controller';
+import { SagasService } from './sagas.service';
+
+@Module({
+  imports: [IntelligenceModule],
+  controllers: [SagasController],
+  providers: [SagaRunnerService, SagasService],
+  exports: [SagaRunnerService],
+})
+export class SagasModule implements OnModuleInit {
+  constructor(private readonly intelligence: IntelligenceService) {}
+
+  /**
+   * Sagas se inregistreaza la bootstrap, nu la import: definitia are nevoie de
+   * servicii injectate, iar registrul trebuie sa fie complet inainte ca primul
+   * tur de corelator sa ruleze.
+   */
+  onModuleInit(): void {
+    registerResearchBriefEnrichment(this.intelligence);
+  }
+}

src/sagas/sagas.service.ts

@@ -0,0 +1,144 @@
+import { Injectable, NotFoundException } from '@nestjs/common';
+import { and, asc, count, desc, eq, lt, sql } from 'drizzle-orm';
+import { db } from '../db/client';
+import { sagaInstances, sagaSteps } from '../db/schema';
+import type { AuthenticatedSession } from '../auth/tenant.guard';
+import { allSagas } from './saga-registry';
+
+/** O saga blocata mai mult de atat merita atentie umana. */
+const STUCK_AFTER_MINUTES = 30;
+
+@Injectable()
+export class SagasService {
+  /**
+   * Monitorul raspunde la o singura intrebare: ce nu merge?
+   * De aceea numara pe status si scoate separat instantele blocate, in loc sa
+   * intoarca o lista plata pe care ar trebui sa o citeasca cineva manual.
+   */
+  async overview(session: AuthenticatedSession) {
+    const stuckBefore = new Date(Date.now() - STUCK_AFTER_MINUTES * 60_000);
+
+    const [byStatus, stuck] = await Promise.all([
+      db
+        .select({ status: sagaInstances.status, n: count() })
+        .from(sagaInstances)
+        .where(eq(sagaInstances.tenantId, session.tenantId))
+        .groupBy(sagaInstances.status),
+      db
+        .select()
+        .from(sagaInstances)
+        .where(
+          and(
+            eq(sagaInstances.tenantId, session.tenantId),
+            sql`${sagaInstances.status} IN ('RUNNING','COMPENSATING')`,
+            lt(sagaInstances.updatedAt, stuckBefore),
+          ),
+        )
+        .orderBy(asc(sagaInstances.updatedAt))
+        .limit(20),
+    ]);
+
+    return {
+      definitions: allSagas().map((s) => ({
+        name: s.name,
+        version: s.version,
+        triggerEvent: s.triggerEvent,
+        steps: s.steps.map((step) => step.name),
+        timeoutMinutes: s.timeoutMinutes,
+      })),
+      counts: Object.fromEntries(byStatus.map((r) => [r.status, r.n])),
+      stuck: stuck.map((row) => this.summarize(row)),
+      stuckThresholdMinutes: STUCK_AFTER_MINUTES,
+    };
+  }
+
+  async list(session: AuthenticatedSession, status?: string, limit = 50) {
+    const where = status
+      ? and(eq(sagaInstances.tenantId, session.tenantId), sql`${sagaInstances.status} = ${status}`)
+      : eq(sagaInstances.tenantId, session.tenantId);
+
+    const rows = await db
+      .select()
+      .from(sagaInstances)
+      .where(where)
+      .orderBy(desc(sagaInstances.createdAt))
+      .limit(limit);
+    return { instances: rows.map((row) => this.summarize(row)) };
+  }
+
+  /** Detaliul include pasii: fara ei nu se poate spune unde s-a oprit. */
+  async detail(session: AuthenticatedSession, id: string) {
+    const instance = await db.query.sagaInstances.findFirst({
+      where: and(eq(sagaInstances.id, id), eq(sagaInstances.tenantId, session.tenantId)),
+    });
+    if (!instance) throw new NotFoundException('Saga instance not found');
+
+    const steps = await db
+      .select()
+      .from(sagaSteps)
+      .where(eq(sagaSteps.sagaInstanceId, instance.id))
+      .orderBy(asc(sagaSteps.sequence));
+
+    return {
+      ...this.summarize(instance),
+      context: instance.context,
+      steps: steps.map((step) => ({
+        sequence: step.sequence,
+        name: step.stepName,
+        status: step.status,
+        attempts: step.attempts,
+        output: step.output,
+        lastError: step.lastError,
+        startedAt: step.startedAt,
+        completedAt: step.completedAt,
+        compensatedAt: step.compensatedAt,
+      })),
+    };
+  }
+
+  /**
+   * Reia o saga blocata: reseteaza backoff-ul si elibereaza lease-ul.
+   * NU schimba statusul si nu sare peste pasi -- reincercarea trebuie sa treaca
+   * prin aceeasi logica, altfel operatorul ar putea "repara" o saga sarind
+   * exact pasul care esua.
+   */
+  async retry(session: AuthenticatedSession, id: string) {
+    const instance = await db.query.sagaInstances.findFirst({
+      where: and(eq(sagaInstances.id, id), eq(sagaInstances.tenantId, session.tenantId)),
+    });
+    if (!instance) throw new NotFoundException('Saga instance not found');
+    if (instance.status === 'COMPLETED' || instance.status === 'COMPENSATED') {
+      return { retried: false, reason: 'Saga s-a incheiat deja' };
+    }
+
+    await db
+      .update(sagaInstances)
+      .set({
+        status: instance.status === 'FAILED' ? 'RUNNING' : instance.status,
+        attempt: 0,
+        nextAttemptAt: new Date(),
+        lockedUntil: null,
+        updatedAt: new Date(),
+      })
+      .where(eq(sagaInstances.id, instance.id));
+    return { retried: true };
+  }
+
+  private summarize(row: typeof sagaInstances.$inferSelect) {
+    return {
+      id: row.id,
+      sagaName: row.sagaName,
+      sagaVersion: row.sagaVersion,
+      status: row.status,
+      currentStep: row.currentStep,
+      attempt: row.attempt,
+      correlationId: row.correlationId,
+      lastError: row.lastError,
+      nextAttemptAt: row.nextAttemptAt,
+      timeoutAt: row.timeoutAt,
+      createdAt: row.createdAt,
+      updatedAt: row.updatedAt,
+      completedAt: row.completedAt,
+    };
+  }
+}