Merge pull request '[E10-3b] orchestrator: dynamic supergraph manager — recompose без рестарта' (#75) from feat/E10-3b-dynamic-supergraph into feat/E10-5-on-chain-watcher

Story 10.3b — custom SupergraphManager, который на каждый poll-tick
re-читает subgraph registry и если список изменился — пересобирает
supergraph и вызывает gateway update(newSdl).

Раньше Apollo Gateway получал статический список subgraph'ов на
bootstrap'е через IntrospectAndCompose — добавление нового
расширения (Story 10.4 install pipeline) требовало рестарта контейнера.
Теперь watcher install pipeline → registry upsert → manager на
следующий tick видит новую запись → recompose → gateway swap'ает
supergraph in-memory. Без рестарта.

Идемпотентность через fingerprint (`name:url` sorted join):
- список не изменился → composer НЕ зовётся, ничего не делаем;
- список изменился → composer.compose() → update().

Скрытая ловушка: fingerprint обновляем только ПОСЛЕ успешного
compose'а — иначе exception в composer'е навсегда залипает в
устаревшем состоянии (следующий tick думает что состояние актуально).
Покрыто continuity-тестом.

7/7 unit-тестов покрывают: initial, idempotent tick, add/remove/url-change,
continuity на ошибке composer'а, cleanup. Используют jest fake timers.

Интеграция с Apollo Gateway (gateway.module.ts) — отдельный follow-up
(`feat/E10-3b-apollo-integration`), требует @apollo/composition impl
для SupergraphComposerPort.

components/orchestrator/src/app.module.ts

@@ -2,6 +2,7 @@ import { Module } from '@nestjs/common';
 import { TypeOrmModule } from '@nestjs/typeorm';
 import { GatewayModule } from './gateway/gateway.module';
 import { OrchestratorModule } from './orchestrator/orchestrator.module';
+import { WatcherModule } from './watcher/watcher.module';
 import { loadAppConfig } from './config/app-config';
 import { SubgraphRegistryEntity } from './gateway/subgraph-registry.entity';
 
@@ -20,6 +21,7 @@ const cfg = loadAppConfig();
     }),
     GatewayModule,
     OrchestratorModule,
+    WatcherModule,
   ],
 })
 export class AppModule {}

components/orchestrator/src/gateway/supergraph-manager.spec.ts

@@ -0,0 +1,183 @@
+/**
+ * @fileoverview Юнит-тесты dynamic supergraph manager'а (Story 10.3b).
+ *
+ * Покрытие:
+ *  1. initial compose: возвращает SDL, composer вызывается ровно один раз;
+ *  2. tick без изменений в registry → composer НЕ вызывается, update НЕ вызывается;
+ *  3. tick с новой subgraph-записью в registry → composer вызывается, update вызывается с новым SDL;
+ *  4. tick с удалённой subgraph-записью → recompose;
+ *  5. tick с изменённым URL у subgraph'а (рестарт сервиса на другом порту) → recompose;
+ *  6. ошибка в composer'е на tick'е НЕ ломает следующий tick (continuity);
+ *  7. cleanup() останавливает таймер — composer перестаёт вызываться.
+ *
+ * Используем jest fake timers для управления интервалом.
+ */
+import {
+  createDynamicSupergraphManager,
+  SupergraphComposerPort,
+  SupergraphRegistryReader,
+} from './supergraph-manager';
+import type { SubgraphDescriptor } from './subgraph-registry.service';
+
+class FakeRegistry implements SupergraphRegistryReader {
+  current: SubgraphDescriptor[] = [];
+  calls = 0;
+  async listForCompose(): Promise<SubgraphDescriptor[]> {
+    this.calls += 1;
+    return this.current.slice();
+  }
+}
+
+class FakeComposer implements SupergraphComposerPort {
+  calls: Array<ReadonlyArray<SubgraphDescriptor>> = [];
+  results: string[] = [];
+  error?: Error;
+  async compose(subgraphs: ReadonlyArray<SubgraphDescriptor>): Promise<string> {
+    this.calls.push(subgraphs.slice());
+    if (this.error) throw this.error;
+    const sdl = `# supergraph v${this.calls.length}\n${subgraphs.map((s) => s.name).join(',')}`;
+    this.results.push(sdl);
+    return sdl;
+  }
+}
+
+const flushMicrotasks = async (): Promise<void> => {
+  await Promise.resolve();
+  await Promise.resolve();
+  await Promise.resolve();
+};
+
+const POLL_MS = 1000;
+
+describe('createDynamicSupergraphManager', () => {
+  beforeEach(() => {
+    jest.useFakeTimers();
+  });
+  afterEach(() => {
+    jest.useRealTimers();
+  });
+
+  it('initial compose: возвращает SDL, composer вызывается ровно один раз', async () => {
+    const registry = new FakeRegistry();
+    registry.current = [{ name: 'core', url: 'http://core:3000/graphql' }];
+    const composer = new FakeComposer();
+    const updates: string[] = [];
+    const lifecycle = await createDynamicSupergraphManager(
+      { registry, composer, pollIntervalMs: POLL_MS },
+      (sdl) => updates.push(sdl),
+    );
+    expect(composer.calls.length).toBe(1);
+    expect(lifecycle.initialSdl).toBe('# supergraph v1\ncore');
+    expect(updates).toEqual([]);
+    await lifecycle.cleanup();
+  });
+
+  it('tick без изменений → composer НЕ вызывается, update НЕ вызывается', async () => {
+    const registry = new FakeRegistry();
+    registry.current = [{ name: 'core', url: 'http://core:3000/graphql' }];
+    const composer = new FakeComposer();
+    const updates: string[] = [];
+    const lifecycle = await createDynamicSupergraphManager(
+      { registry, composer, pollIntervalMs: POLL_MS },
+      (sdl) => updates.push(sdl),
+    );
+    expect(composer.calls.length).toBe(1);
+    jest.advanceTimersByTime(POLL_MS);
+    await flushMicrotasks();
+    expect(composer.calls.length).toBe(1); // не вызван повторно
+    expect(updates).toEqual([]);
+    await lifecycle.cleanup();
+  });
+
+  it('tick с новой subgraph-записью → composer вызывается, update получает новый SDL', async () => {
+    const registry = new FakeRegistry();
+    registry.current = [{ name: 'core', url: 'http://core:3000/graphql' }];
+    const composer = new FakeComposer();
+    const updates: string[] = [];
+    const lifecycle = await createDynamicSupergraphManager(
+      { registry, composer, pollIntervalMs: POLL_MS },
+      (sdl) => updates.push(sdl),
+    );
+    registry.current.push({ name: 'chatcoop', url: 'http://chatcoop:3000/graphql' });
+    jest.advanceTimersByTime(POLL_MS);
+    await flushMicrotasks();
+    expect(composer.calls.length).toBe(2);
+    expect(updates).toEqual(['# supergraph v2\ncore,chatcoop']);
+    await lifecycle.cleanup();
+  });
+
+  it('tick с удалённой subgraph-записью → recompose', async () => {
+    const registry = new FakeRegistry();
+    registry.current = [
+      { name: 'core', url: 'http://core:3000/graphql' },
+      { name: 'chatcoop', url: 'http://chatcoop:3000/graphql' },
+    ];
+    const composer = new FakeComposer();
+    const updates: string[] = [];
+    const lifecycle = await createDynamicSupergraphManager(
+      { registry, composer, pollIntervalMs: POLL_MS },
+      (sdl) => updates.push(sdl),
+    );
+    registry.current = [{ name: 'core', url: 'http://core:3000/graphql' }];
+    jest.advanceTimersByTime(POLL_MS);
+    await flushMicrotasks();
+    expect(composer.calls.length).toBe(2);
+    expect(updates).toEqual(['# supergraph v2\ncore']);
+    await lifecycle.cleanup();
+  });
+
+  it('tick с изменённым URL у subgraph\'а → recompose', async () => {
+    const registry = new FakeRegistry();
+    registry.current = [{ name: 'chatcoop', url: 'http://chatcoop:3000/graphql' }];
+    const composer = new FakeComposer();
+    const updates: string[] = [];
+    const lifecycle = await createDynamicSupergraphManager(
+      { registry, composer, pollIntervalMs: POLL_MS },
+      (sdl) => updates.push(sdl),
+    );
+    registry.current = [{ name: 'chatcoop', url: 'http://chatcoop-v2:3000/graphql' }];
+    jest.advanceTimersByTime(POLL_MS);
+    await flushMicrotasks();
+    expect(composer.calls.length).toBe(2);
+    expect(updates.length).toBe(1);
+    await lifecycle.cleanup();
+  });
+
+  it('ошибка в composer\'е НЕ ломает следующий tick (continuity)', async () => {
+    const registry = new FakeRegistry();
+    registry.current = [{ name: 'core', url: 'http://core:3000/graphql' }];
+    const composer = new FakeComposer();
+    const updates: string[] = [];
+    const lifecycle = await createDynamicSupergraphManager(
+      { registry, composer, pollIntervalMs: POLL_MS },
+      (sdl) => updates.push(sdl),
+    );
+    // tick 1 — добавили subgraph, composer бросает
+    registry.current.push({ name: 'chatcoop', url: 'http://chatcoop:3000/graphql' });
+    composer.error = new Error('introspection failed');
+    jest.advanceTimersByTime(POLL_MS);
+    await flushMicrotasks();
+    expect(updates).toEqual([]);
+    // tick 2 — composer оправился
+    composer.error = undefined;
+    jest.advanceTimersByTime(POLL_MS);
+    await flushMicrotasks();
+    expect(updates.length).toBe(1);
+    await lifecycle.cleanup();
+  });
+
+  it('cleanup() останавливает таймер — composer перестаёт вызываться', async () => {
+    const registry = new FakeRegistry();
+    registry.current = [{ name: 'core', url: 'http://core:3000/graphql' }];
+    const composer = new FakeComposer();
+    const lifecycle = await createDynamicSupergraphManager(
+      { registry, composer, pollIntervalMs: POLL_MS },
+      () => undefined,
+    );
+    await lifecycle.cleanup();
+    registry.current.push({ name: 'chatcoop', url: 'http://chatcoop:3000/graphql' });
+    jest.advanceTimersByTime(POLL_MS * 5);
+    await flushMicrotasks();
+    expect(composer.calls.length).toBe(1); // только initial
+  });
+});

components/orchestrator/src/gateway/supergraph-manager.ts

@@ -0,0 +1,122 @@
+/**
+ * @fileoverview Custom SupergraphManager — Story 10.3b.
+ *
+ * Apollo Gateway по умолчанию через `IntrospectAndCompose` принимает
+ * СТАТИЧЕСКИЙ список subgraph'ов на bootstrap'е. Polling-режим
+ * подхватывает изменения СХЕМЫ у уже известных subgraph'ов, но НЕ
+ * замечает появление НОВЫХ записей в registry — для них требуется
+ * рестарт контейнера.
+ *
+ * Story 10.4 install pipeline пишет новые subgraph'ы в registry без
+ * рестарта; чтобы gateway их видел, нужен SupergraphManager, который
+ * на каждый tick re-читает registry, и если список (по `(name,url)`)
+ * изменился — пересобирает supergraph и вызывает `update(newSdl)`.
+ *
+ * Apollo Gateway это поддерживает: `gateway: { supergraphSdl: fn }`,
+ * где fn возвращает `{ supergraphSdl, cleanup }` и получает callback
+ * `update`. См. https://www.apollographql.com/docs/federation/v2/api/apollo-gateway/#supergraphsdl
+ *
+ * Этот файл — переиспользуемый менеджер, отделённый от Apollo Gateway
+ * API (его hook-fn в `gateway.module.ts` собирает менеджер и
+ * передаёт в gateway). Сам менеджер не знает про gateway — он
+ * работает с двумя портами: regsitry (читать список) и
+ * supergraphComposer (собрать SDL из subgraph descriptors).
+ */
+import { Logger } from '@nestjs/common';
+import type { SubgraphDescriptor } from './subgraph-registry.service';
+
+/**
+ * Порт композитора supergraph'а. Реальный impl делает introspection
+ * subgraph URL'ов и compose через `@apollo/composition`. Тестовый
+ * возвращает заранее заданный SDL.
+ */
+export interface SupergraphComposerPort {
+  compose(subgraphs: ReadonlyArray<SubgraphDescriptor>): Promise<string>;
+}
+
+/**
+ * Минимальный API регистра, который нужен менеджеру.
+ * Преднамеренно ужe, чем весь `SubgraphRegistryService` — менеджер
+ * вообще не должен знать про писать/изменять записи.
+ */
+export interface SupergraphRegistryReader {
+  listForCompose(): Promise<SubgraphDescriptor[]>;
+}
+
+export interface SupergraphManagerOptions {
+  composer: SupergraphComposerPort;
+  registry: SupergraphRegistryReader;
+  /** Интервал опроса registry. Должен совпадать с polling'ом IntrospectAndCompose. */
+  pollIntervalMs: number;
+}
+
+export interface SupergraphManagerLifecycle {
+  /** Начальный SDL для отдачи gateway'ю на bootstrap'е. */
+  initialSdl: string;
+  /** Освободить таймер — gateway вызовет при shutdown'е. */
+  cleanup(): Promise<void>;
+}
+
+/**
+ * Создаёт менеджер supergraph'а с динамическим refresh'ем по registry.
+ *
+ *  1. Делает первый compose(registry.listForCompose()) → возвращает
+ *     `initialSdl` и сохраняет «текущее состояние» (отпечаток списка).
+ *  2. Запускает setInterval, который на каждый tick читает registry;
+ *     если отпечаток списка отличается от текущего — recompose и
+ *     вызов `update(newSdl)`.
+ *  3. `cleanup()` останавливает таймер.
+ *
+ * Идемпотентность: если registry вернул тот же список (по name+url) —
+ * compose НЕ вызывается, экономим CPU и сеть (introspection).
+ */
+export async function createDynamicSupergraphManager(
+  opts: SupergraphManagerOptions,
+  update: (sdl: string) => void,
+): Promise<SupergraphManagerLifecycle> {
+  const logger = new Logger('DynamicSupergraphManager');
+  let lastFingerprint = '';
+
+  const fingerprintOf = (subgraphs: ReadonlyArray<SubgraphDescriptor>): string =>
+    subgraphs
+      .slice()
+      .sort((a, b) => a.name.localeCompare(b.name))
+      .map((s) => `${s.name}:${s.url}`)
+      .join('|');
+
+  const composeOnce = async (): Promise<string> => {
+    const subgraphs = await opts.registry.listForCompose();
+    const fp = fingerprintOf(subgraphs);
+    lastFingerprint = fp;
+    return opts.composer.compose(subgraphs);
+  };
+
+  const initialSdl = await composeOnce();
+
+  const timer = setInterval(() => {
+    void (async () => {
+      try {
+        const subgraphs = await opts.registry.listForCompose();
+        const fp = fingerprintOf(subgraphs);
+        if (fp === lastFingerprint) return;
+        logger.log(`supergraph registry changed (was ${lastFingerprint.split('|').length} subgraphs, now ${fp.split('|').length}) — recompose`);
+        // ВАЖНО: fingerprint обновляем только ПОСЛЕ успешного compose.
+        // Иначе при exception в composer'е следующий tick посчитает,
+        // что состояние уже актуально (fp === lastFingerprint) и пропустит
+        // retry — supergraph навсегда останется в устаревшем состоянии.
+        const sdl = await opts.composer.compose(subgraphs);
+        lastFingerprint = fp;
+        update(sdl);
+      } catch (e) {
+        logger.error(`recompose tick failed: ${e instanceof Error ? e.message : String(e)}`);
+      }
+    })();
+  }, opts.pollIntervalMs);
+
+  return {
+    initialSdl,
+    cleanup: async () => {
+      clearInterval(timer);
+    },
+  };
+}

components/orchestrator/src/watcher/on-chain-watcher.service.spec.ts

@@ -0,0 +1,258 @@
+/**
+ * @fileoverview Юнит-тесты on-chain watcher'а.
+ *
+ * Покрытие (Story 10.5):
+ *  - release-published scope='all' → install зовётся;
+ *  - release-published scope='cooperatives' с нашим coopname → install;
+ *  - release-published scope='cooperatives' без нашего coopname → ignore;
+ *  - release-published scope='empty' → ignore;
+ *  - release-published metadata=null → install не зовётся (skip + warn);
+ *  - subscription-activated для нашего coop → install зовётся;
+ *  - subscription-activated для чужого coop → ignore;
+ *  - release-withdrawn → registry.deactivate;
+ *  - subscription-expired для нашего coop → registry.deactivate;
+ *  - subscription-expired для чужого coop → ignore.
+ */
+import {
+  AppsContractEvent,
+  AppsContractEventStreamPort,
+  ReleaseInstallSpec,
+  ReleaseMetadataPort,
+} from './ports';
+import {
+  OnChainWatcherService,
+  OnChainWatcherConfig,
+} from './on-chain-watcher.service';
+import {
+  InstallOrchestratorService,
+  InstallExtensionInput,
+  InstallOutcome,
+} from '../orchestrator/install-orchestrator.service';
+import { SubgraphRegistryService } from '../gateway/subgraph-registry.service';
+
+class FakeEventStream implements AppsContractEventStreamPort {
+  handler?: (e: AppsContractEvent) => Promise<void>;
+  async subscribe(handler: (e: AppsContractEvent) => Promise<void>): Promise<{ unsubscribe(): void }> {
+    this.handler = handler;
+    return { unsubscribe: () => undefined };
+  }
+}
+
+class FakeMetadata implements ReleaseMetadataPort {
+  byKey = new Map<string, ReleaseInstallSpec>();
+  calls: Array<{ packageId: string; version: string }> = [];
+  setSpec(packageId: string, version: string, spec: ReleaseInstallSpec): void {
+    this.byKey.set(`${packageId}@${version}`, spec);
+  }
+  async fetchInstallSpec(opts: { packageId: string; version: string }): Promise<ReleaseInstallSpec | null> {
+    this.calls.push(opts);
+    return this.byKey.get(`${opts.packageId}@${opts.version}`) ?? null;
+  }
+}
+
+class FakeOrchestrator {
+  calls: Array<InstallExtensionInput> = [];
+  outcome: InstallOutcome = { status: 'applied', packageId: '', healthAfterMs: 1 };
+  async install(input: InstallExtensionInput): Promise<InstallOutcome> {
+    this.calls.push(input);
+    return { ...this.outcome, packageId: input.packageId };
+  }
+}
+
+class FakeRegistryService {
+  deactivated: string[] = [];
+  async deactivate(packageId: string): Promise<void> {
+    this.deactivated.push(packageId);
+  }
+}
+
+const buildHarness = (
+  cfg: Partial<OnChainWatcherConfig> = {},
+): {
+  watcher: OnChainWatcherService;
+  stream: FakeEventStream;
+  metadata: FakeMetadata;
+  orchestrator: FakeOrchestrator;
+  registry: FakeRegistryService;
+} => {
+  const stream = new FakeEventStream();
+  const metadata = new FakeMetadata();
+  const orchestrator = new FakeOrchestrator();
+  const registry = new FakeRegistryService();
+  const watcher = new OnChainWatcherService(
+    stream,
+    metadata,
+    { coopname: 'voskhod', ...cfg },
+    orchestrator as unknown as InstallOrchestratorService,
+    registry as unknown as SubgraphRegistryService,
+  );
+  return { watcher, stream, metadata, orchestrator, registry };
+};
+
+const PACKAGE_ID = '@coopenomics/chatcoop';
+const VERSION = '1.0.0';
+const SPEC: ReleaseInstallSpec = {
+  url: 'http://chatcoop:3000/graphql',
+  imageRef: 'reg.local/chatcoop:1.0.0',
+};
+
+describe('OnChainWatcherService', () => {
+  describe('release-published', () => {
+    it("scope='all' → install зовётся", async () => {
+      const h = buildHarness();
+      h.metadata.setSpec(PACKAGE_ID, VERSION, SPEC);
+      await h.watcher.handle({
+        kind: 'release-published',
+        packageId: PACKAGE_ID,
+        version: VERSION,
+        scopeType: 'all',
+        blockNum: 100,
+      });
+      expect(h.orchestrator.calls).toEqual([
+        {
+          packageId: PACKAGE_ID,
+          version: VERSION,
+          url: SPEC.url,
+          imageRef: SPEC.imageRef,
+          composeService: undefined,
+          composeFile: undefined,
+          cooperativeJwt: undefined,
+        },
+      ]);
+    });
+
+    it("scope='cooperatives' с нашим coopname → install", async () => {
+      const h = buildHarness();
+      h.metadata.setSpec(PACKAGE_ID, VERSION, SPEC);
+      await h.watcher.handle({
+        kind: 'release-published',
+        packageId: PACKAGE_ID,
+        version: VERSION,
+        scopeType: 'cooperatives',
+        scopeCoopnames: ['voskhod', 'alpha'],
+        blockNum: 101,
+      });
+      expect(h.orchestrator.calls.length).toBe(1);
+    });
+
+    it("scope='cooperatives' без нашего coopname → ignore", async () => {
+      const h = buildHarness();
+      h.metadata.setSpec(PACKAGE_ID, VERSION, SPEC);
+      await h.watcher.handle({
+        kind: 'release-published',
+        packageId: PACKAGE_ID,
+        version: VERSION,
+        scopeType: 'cooperatives',
+        scopeCoopnames: ['alpha', 'beta'],
+        blockNum: 102,
+      });
+      expect(h.orchestrator.calls).toEqual([]);
+      expect(h.metadata.calls).toEqual([]);
+    });
+
+    it("scope='empty' → ignore", async () => {
+      const h = buildHarness();
+      await h.watcher.handle({
+        kind: 'release-published',
+        packageId: PACKAGE_ID,
+        version: VERSION,
+        scopeType: 'empty',
+        blockNum: 103,
+      });
+      expect(h.orchestrator.calls).toEqual([]);
+    });
+
+    it('metadata=null → install не зовётся (skip + warn)', async () => {
+      const h = buildHarness();
+      // ничего не setSpec — fetchInstallSpec вернёт null
+      await h.watcher.handle({
+        kind: 'release-published',
+        packageId: PACKAGE_ID,
+        version: VERSION,
+        scopeType: 'all',
+        blockNum: 104,
+      });
+      expect(h.metadata.calls).toEqual([{ packageId: PACKAGE_ID, version: VERSION }]);
+      expect(h.orchestrator.calls).toEqual([]);
+    });
+
+    it('cooperativeJwt из cfg пробрасывается в install pipeline', async () => {
+      const h = buildHarness({ cooperativeJwt: 'jwt-of-voskhod' });
+      h.metadata.setSpec(PACKAGE_ID, VERSION, SPEC);
+      await h.watcher.handle({
+        kind: 'release-published',
+        packageId: PACKAGE_ID,
+        version: VERSION,
+        scopeType: 'all',
+        blockNum: 105,
+      });
+      expect(h.orchestrator.calls[0]?.cooperativeJwt).toBe('jwt-of-voskhod');
+    });
+  });
+
+  describe('subscription-activated', () => {
+    it('для нашего coop → install зовётся с version="active"', async () => {
+      const h = buildHarness();
+      h.metadata.setSpec(PACKAGE_ID, '', SPEC);
+      await h.watcher.handle({
+        kind: 'subscription-activated',
+        coopname: 'voskhod',
+        packageId: PACKAGE_ID,
+        expiresAtUnix: 1_800_000_000,
+        blockNum: 200,
+      });
+      expect(h.orchestrator.calls.length).toBe(1);
+      expect(h.orchestrator.calls[0]?.version).toBe('active');
+    });
+
+    it('для чужого coop → ignore', async () => {
+      const h = buildHarness();
+      await h.watcher.handle({
+        kind: 'subscription-activated',
+        coopname: 'alpha',
+        packageId: PACKAGE_ID,
+        expiresAtUnix: 1_800_000_000,
+        blockNum: 201,
+      });
+      expect(h.orchestrator.calls).toEqual([]);
+      expect(h.metadata.calls).toEqual([]);
+    });
+  });
+
+  describe('release-withdrawn', () => {
+    it('→ registry.deactivate', async () => {
+      const h = buildHarness();
+      await h.watcher.handle({
+        kind: 'release-withdrawn',
+        packageId: PACKAGE_ID,
+        version: VERSION,
+        blockNum: 300,
+      });
+      expect(h.registry.deactivated).toEqual([PACKAGE_ID]);
+    });
+  });
+
+  describe('subscription-expired', () => {
+    it('для нашего coop → registry.deactivate', async () => {
+      const h = buildHarness();
+      await h.watcher.handle({
+        kind: 'subscription-expired',
+        coopname: 'voskhod',
+        packageId: PACKAGE_ID,
+        blockNum: 400,
+      });
+      expect(h.registry.deactivated).toEqual([PACKAGE_ID]);
+    });
+
+    it('для чужого coop → ignore', async () => {
+      const h = buildHarness();
+      await h.watcher.handle({
+        kind: 'subscription-expired',
+        coopname: 'alpha',
+        packageId: PACKAGE_ID,
+        blockNum: 401,
+      });
+      expect(h.registry.deactivated).toEqual([]);
+    });
+  });
+});

components/orchestrator/src/watcher/on-chain-watcher.service.ts

@@ -0,0 +1,184 @@
+/**
+ * @fileoverview On-chain watcher (Story 10.5). Слушает события
+ * apps-contract в Коопеномиксе и автоматически синхронизирует
+ * состояние установленных расширений с тем, что в блокчейне.
+ *
+ * Source-of-truth — chain. Если на orchestrator-ноде установлено
+ * что-то, чего нет в chain'е (или есть, но release отозван), оно
+ * uninstall'ится. Если в chain'е появилось новое — install'ится.
+ *
+ * Логика:
+ *  - `release-published`: проверить, виден ли релиз нашему кооперативу
+ *    через scope (all / subnets / cooperatives) и есть ли у нас
+ *    активная подписка → install pipeline.
+ *  - `subscription-activated`: дотянуть active-релиз из apps-catalog'а
+ *    → install pipeline.
+ *  - `release-withdrawn`: если этот пакет был активен — uninstall.
+ *  - `subscription-expired`: uninstall расширения.
+ *
+ * Все три сценария проходят через {@link InstallOrchestratorService}
+ * → одно место решения «как ставить»; watcher только триггерит.
+ *
+ * Идемпотентность: install pipeline сам upsert'ит registry — повторный
+ * вызов с тем же `(packageId, version)` no-op. Uninstall переводит
+ * запись в active=false; повторный uninstall тоже no-op.
+ */
+import { Inject, Injectable, Logger, OnApplicationBootstrap, OnApplicationShutdown } from '@nestjs/common';
+import {
+  APPS_CONTRACT_EVENT_STREAM,
+  AppsContractEvent,
+  AppsContractEventStreamPort,
+  RELEASE_METADATA_PORT,
+  ReleaseMetadataPort,
+} from './ports';
+import { InstallOrchestratorService } from '../orchestrator/install-orchestrator.service';
+import { SubgraphRegistryService } from '../gateway/subgraph-registry.service';
+
+export interface OnChainWatcherConfig {
+  /** Имя нашего кооператива — нужно для фильтрации scope=cooperatives. */
+  coopname: string;
+  /**
+   * JWT кооператива для install pipeline'а (нужен docker pull через
+   * CA-auth OCI token). Если не задан — install будет fail'ить на
+   * шаге oci-token; полезно для frontend-only пакетов.
+   */
+  cooperativeJwt?: string;
+}
+
+@Injectable()
+export class OnChainWatcherService implements OnApplicationBootstrap, OnApplicationShutdown {
+  private readonly logger = new Logger(OnChainWatcherService.name);
+  private subscription?: { unsubscribe(): void };
+
+  constructor(
+    @Inject(APPS_CONTRACT_EVENT_STREAM) private readonly stream: AppsContractEventStreamPort,
+    @Inject(RELEASE_METADATA_PORT) private readonly metadata: ReleaseMetadataPort,
+    @Inject('ON_CHAIN_WATCHER_CONFIG') private readonly cfg: OnChainWatcherConfig,
+    private readonly orchestrator: InstallOrchestratorService,
+    private readonly registry: SubgraphRegistryService,
+  ) {}
+
+  async onApplicationBootstrap(): Promise<void> {
+    this.subscription = await this.stream.subscribe((e) => this.handle(e));
+    this.logger.log(`on-chain watcher started (coopname=${this.cfg.coopname})`);
+  }
+
+  onApplicationShutdown(): void {
+    this.subscription?.unsubscribe();
+  }
+
+  /**
+   * @internal exposed для юнит-тестов — гонять без bootstrap'а Nest'а.
+   */
+  async handle(e: AppsContractEvent): Promise<void> {
+    switch (e.kind) {
+      case 'release-published':
+        return this.handleReleasePublished(e);
+      case 'subscription-activated':
+        return this.handleSubscriptionActivated(e);
+      case 'release-withdrawn':
+        return this.handleReleaseWithdrawn(e);
+      case 'subscription-expired':
+        return this.handleSubscriptionExpired(e);
+    }
+  }
+
+  private async handleReleasePublished(
+    e: Extract<AppsContractEvent, { kind: 'release-published' }>,
+  ): Promise<void> {
+    if (!this.isReleaseVisibleToCoop(e)) {
+      this.logger.debug(
+        `release-published ${e.packageId}@${e.version} вне scope нашего coopname=${this.cfg.coopname} → ignore`,
+      );
+      return;
+    }
+    const spec = await this.metadata.fetchInstallSpec({
+      packageId: e.packageId,
+      version: e.version,
+    });
+    if (spec === null) {
+      this.logger.warn(
+        `release-published ${e.packageId}@${e.version}: metadata not found, install skipped`,
+      );
+      return;
+    }
+    await this.callInstall(e.packageId, e.version, spec);
+  }
+
+  private async handleSubscriptionActivated(
+    e: Extract<AppsContractEvent, { kind: 'subscription-activated' }>,
+  ): Promise<void> {
+    if (e.coopname !== this.cfg.coopname) return;
+    // На subscription-activated нам не приходит конкретная версия —
+    // нужно сходить в apps-catalog за active-релизом. fetchInstallSpec
+    // с пустой version интерпретируется реальным impl'ом как «дай active».
+    const spec = await this.metadata.fetchInstallSpec({
+      packageId: e.packageId,
+      version: '',
+    });
+    if (spec === null) {
+      this.logger.warn(`subscription-activated ${e.packageId}: active release not found`);
+      return;
+    }
+    // Так как реальной версии в event'е нет, оставим version пустой —
+    // реальный fetchInstallSpec impl должен резолвить и возвращать
+    // активную версию через расширение spec'а. На текущем shape'е
+    // используем packageId как метку записи — registry хранит
+    // packageId как primary key, версия пишется отдельным полем.
+    await this.callInstall(e.packageId, 'active', spec);
+  }
+
+  private async handleReleaseWithdrawn(
+    e: Extract<AppsContractEvent, { kind: 'release-withdrawn' }>,
+  ): Promise<void> {
+    await this.registry.deactivate(e.packageId);
+    this.logger.log(`release-withdrawn ${e.packageId}@${e.version} → deactivated`);
+  }
+
+  private async handleSubscriptionExpired(
+    e: Extract<AppsContractEvent, { kind: 'subscription-expired' }>,
+  ): Promise<void> {
+    if (e.coopname !== this.cfg.coopname) return;
+    await this.registry.deactivate(e.packageId);
+    this.logger.log(`subscription-expired ${e.packageId} → deactivated`);
+  }
+
+  private isReleaseVisibleToCoop(
+    e: Extract<AppsContractEvent, { kind: 'release-published' }>,
+  ): boolean {
+    switch (e.scopeType) {
+      case 'all':
+        return true;
+      case 'cooperatives':
+        return (e.scopeCoopnames ?? []).includes(this.cfg.coopname);
+      case 'subnets':
+        // Subnet-фильтрация требует знания нашего subnet_id — отдельный
+        // конфиг; пока считаем что watcher на этой ноде subnet-aware
+        // на уровне stream'а (он не должен присылать чужие subnets).
+        return true;
+      case 'empty':
+        return false;
+    }
+  }
+
+  private async callInstall(
+    packageId: string,
+    version: string,
+    spec: { url: string; imageRef?: string; composeService?: string; composeFile?: string },
+  ): Promise<void> {
+    const result = await this.orchestrator.install({
+      packageId,
+      version,
+      url: spec.url,
+      imageRef: spec.imageRef,
+      composeService: spec.composeService,
+      composeFile: spec.composeFile,
+      cooperativeJwt: this.cfg.cooperativeJwt,
+    });
+    if (result.status === 'failed') {
+      this.logger.error(
+        `install via watcher failed: ${packageId}@${version} reason=${result.reason} ${result.error}`,
+      );
+    }
+  }
+}

components/orchestrator/src/watcher/ports.ts

@@ -0,0 +1,108 @@
+/**
+ * @fileoverview Порты on-chain watcher'а (Story 10.5).
+ *
+ * Watcher слушает события apps-contract в Коопеномиксе и автоматически
+ * инициирует install/uninstall расширений для своего кооператива.
+ * Source-of-truth для «что должно быть установлено» — блокчейн.
+ *
+ * Сервис {@link OnChainWatcherService} зависит только от этих
+ * портов; реальные impl (RPC polling / parser2 client / etc.) лежат
+ * рядом и подключаются в `WatcherModule`. Тесты подменяют порты
+ * ин-мемори фейками — без реальной chain-ноды.
+ */
+
+/**
+ * События apps-contract, релевантные orchestrator'у.
+ *
+ * Discriminated по `kind` — компилятор требует обработать каждый
+ * вариант в switch'е.
+ *
+ *  - `release-published` (`apps::setrelease`) — кооперативу-оператору
+ *    был выкачан новый релиз. Watcher проверяет scope (видна ли версия
+ *    нашему кооперативу) и зовёт install pipeline.
+ *  - `subscription-activated` (`apps::regsub` или `apps::extendsub`) —
+ *    наш кооператив подписался на пакет (или продлил). Нужно
+ *    дотянуть active-релиз и install'ить.
+ *  - `release-withdrawn` (`apps::withdraw`) — релиз отозван.
+ *    Если он сейчас активен — uninstall.
+ *  - `subscription-expired` (`apps::expsub`) — подписка кончилась.
+ *    Uninstall расширения.
+ */
+export type AppsContractEvent =
+  | {
+      kind: 'release-published';
+      packageId: string;
+      version: string;
+      /** scope-тип релиза — см. apps-contract; `'cooperatives'` → дополнительная фильтрация в watcher. */
+      scopeType: 'all' | 'subnets' | 'cooperatives' | 'empty';
+      /** Если `scopeType='cooperatives'` — явный whitelist coopname'ов; иначе пусто. */
+      scopeCoopnames?: ReadonlyArray<string>;
+      blockNum: number;
+    }
+  | {
+      kind: 'subscription-activated';
+      coopname: string;
+      packageId: string;
+      expiresAtUnix: number;
+      blockNum: number;
+    }
+  | {
+      kind: 'release-withdrawn';
+      packageId: string;
+      version: string;
+      blockNum: number;
+    }
+  | {
+      kind: 'subscription-expired';
+      coopname: string;
+      packageId: string;
+      blockNum: number;
+    };
+
+/**
+ * Порт потока событий apps-contract. Реальный impl читает либо
+ * parser2 Redis stream (`ce:parser2:<chain_id>:events` отфильтрованный
+ * по contract=apps), либо poll'ит chain RPC `/v1/history/get_actions`.
+ *
+ * Контракт subscribe'а:
+ *  - handler вызывается ровно один раз на каждое уникальное событие;
+ *  - порядок — по блокам (старые first);
+ *  - при ошибке в handler'е событие НЕ pop'ается из стрима до retry'я;
+ *  - возвращаемый `unsubscribe` — синхронный, останавливает poll.
+ */
+export interface AppsContractEventStreamPort {
+  subscribe(handler: (e: AppsContractEvent) => Promise<void>): Promise<{ unsubscribe(): void }>;
+}
+
+export const APPS_CONTRACT_EVENT_STREAM = Symbol('AppsContractEventStreamPort');
+
+/**
+ * Порт получения метаданных релиза из apps-catalog'а.
+ *
+ * On-chain event `setrelease` приносит только `(packageId, version, scopeType)`.
+ * Чтобы реально запустить install pipeline, нужно знать `imageRef`,
+ * `composeService`, `subgraphUrl` — они лежат в `coopenomics`-секции
+ * package manifest'а в apps-catalog (Story 10.7a/b).
+ *
+ * Реальный impl бьёт по `GET /v1/package/:scope/:name/metadata` ca-auth'а
+ * (Story 9.4.c) или по `/v1/registry/...` напрямую.
+ */
+export interface ReleaseMetadataPort {
+  fetchInstallSpec(opts: {
+    packageId: string;
+    version: string;
+  }): Promise<ReleaseInstallSpec | null>;
+}
+
+export interface ReleaseInstallSpec {
+  /** Внутренний URL subgraph'а — для healthcheck'а и записи в registry. */
+  url: string;
+  /** OCI-ссылка образа в Nexus; если нет — frontend-only пакет, install не нужен. */
+  imageRef?: string;
+  /** Имя сервиса в docker-compose файле orchestrator'а. */
+  composeService?: string;
+  /** Путь к docker-compose файлу. */
+  composeFile?: string;
+}
+
+export const RELEASE_METADATA_PORT = Symbol('ReleaseMetadataPort');

components/orchestrator/src/watcher/watcher.module.ts

@@ -0,0 +1,58 @@
+/**
+ * @fileoverview WatcherModule — DI-проводка on-chain watcher'а (Story 10.5).
+ *
+ * На текущий момент impl'ы портов поставляются заглушками
+ * `NoopEventStream` / `NoopReleaseMetadata` — они не делают ничего.
+ * Это сделано, чтобы AppModule не падал при boot'е, пока реальные
+ * impl'ы (parser2 client / HTTP к apps-catalog) не приехали в
+ * отдельных follow-up'ах.
+ *
+ * Реальная подключка чита chain'а — отдельный PR
+ * (`feat/E10-5b-chain-rpc-impl`). Этот модуль фиксирует контракт.
+ */
+import { Module } from '@nestjs/common';
+import { GatewayModule } from '../gateway/gateway.module';
+import { OrchestratorModule } from '../orchestrator/orchestrator.module';
+import { OnChainWatcherService, OnChainWatcherConfig } from './on-chain-watcher.service';
+import {
+  APPS_CONTRACT_EVENT_STREAM,
+  AppsContractEvent,
+  AppsContractEventStreamPort,
+  RELEASE_METADATA_PORT,
+  ReleaseInstallSpec,
+  ReleaseMetadataPort,
+} from './ports';
+import { loadAppConfig } from '../config/app-config';
+
+class NoopEventStream implements AppsContractEventStreamPort {
+  async subscribe(_handler: (e: AppsContractEvent) => Promise<void>): Promise<{ unsubscribe(): void }> {
+    return { unsubscribe: () => undefined };
+  }
+}
+
+class NoopReleaseMetadata implements ReleaseMetadataPort {
+  async fetchInstallSpec(_opts: { packageId: string; version: string }): Promise<ReleaseInstallSpec | null> {
+    return null;
+  }
+}
+
+@Module({
+  imports: [GatewayModule, OrchestratorModule],
+  providers: [
+    OnChainWatcherService,
+    { provide: APPS_CONTRACT_EVENT_STREAM, useClass: NoopEventStream },
+    { provide: RELEASE_METADATA_PORT, useClass: NoopReleaseMetadata },
+    {
+      provide: 'ON_CHAIN_WATCHER_CONFIG',
+      useFactory: (): OnChainWatcherConfig => {
+        const cfg = loadAppConfig();
+        return {
+          coopname: cfg.coopname,
+          cooperativeJwt: process.env.COOPERATIVE_JWT,
+        };
+      },
+    },
+  ],
+  exports: [OnChainWatcherService],
+})
+export class WatcherModule {}