feat(p11): temps réel complet (watcher checkout principal + topic settings)

P11-A — branche du checkout principal modifiée hors Arboretum : - fs-watcher: pin « permanent » non évinçable (pinRepo/unpinRepo, repoPins dans evictIfNeeded) - worktree-manager: armMainCheckoutWatchers() + arm/désarm sur addRepo/découverte/removeRepo/hidden - index.ts: armMainCheckoutWatchers() dans runDaemon → git checkout CLI sur le principal → worktree_update <500ms sans watch client P11-B — réglages en temps réel : - protocole additif: type SettingsBroadcast (source unique, réutilisé par SettingsResponse), topic 'settings', message settings_update (validés parseClientMessage + gateway) - core/settings-bus.ts (EventEmitter) ; routes/settings émet après PATCH ; gateway relaie aux abonnés 'settings' - web: ws-client subscribeSettings + routage ; store settings applySettings/startRealtime ; AppShell abonne globalement ; SettingsView re-sync des drafts scalaires sans écraser une saisie en cours - tests: protocol (topic settings) + fs-watcher (pinRepo non évincé) ; acceptance-p11.mjs (checkout principal <500ms + settings_update)
2026-06-27 14:08:44 +02:00
parent 92670a796a
commit e8d10b7ec0
16 changed files with 304 additions and 38 deletions
--- a/packages/server/scripts/acceptance-p11.mjs
+++ b/packages/server/scripts/acceptance-p11.mjs
@@ -0,0 +1,117 @@
+#!/usr/bin/env node
+// Acceptation P11 (sans navigateur) : temps réel complet. Vrai daemon + vrai repo git tmp + client WS.
+// Couvre : (1) `git checkout` en CLI sur le CHECKOUT PRINCIPAL → worktree_update (nouvelle branche)
+// poussé en temps réel (< 500 ms) sans qu'aucun client n'ait `watch`é ce worktree ; (2) PATCH /settings
+// → settings_update reçu par un client abonné au topic 'settings'.
+import { spawn, execFileSync } from 'node:child_process';
+import { mkdtempSync, rmSync, writeFileSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join, dirname } from 'node:path';
+import { fileURLToPath } from 'node:url';
+import { createRequire } from 'node:module';
+
+const require = createRequire(import.meta.url);
+const WebSocket = require('ws');
+
+const PORT = 7554;
+const ORIGIN = `http://127.0.0.1:${PORT}`;
+const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
+const serverDir = join(dirname(fileURLToPath(import.meta.url)), '..');
+const results = [];
+const check = (name, ok, detail = '') => {
+  results.push({ name, ok, detail });
+  console.log(`${ok ? '✅' : '❌'} ${name}${detail ? ` — ${detail}` : ''}`);
+};
+
+const tmp = mkdtempSync(join(tmpdir(), 'arb-accept-p11-'));
+const repo = join(tmp, 'repo');
+execFileSync('mkdir', ['-p', repo]);
+const git = (...args) => execFileSync('git', args, { cwd: repo, stdio: 'pipe' });
+git('init', '-b', 'main');
+git('config', 'user.email', 'test@arboretum.dev');
+git('config', 'user.name', 'Test');
+writeFileSync(join(repo, 'README.md'), '# demo\n');
+git('add', '-A');
+git('commit', '-m', 'init');
+
+const srv = spawn(
+  'node',
+  [join(serverDir, 'dist', 'index.js'), '--port', String(PORT), '--db', join(tmp, 'a.db'), '--claude-home', join(tmp, 'claude'), '--no-discover'],
+  { env: { ...process.env, ARBORETUM_LOG: 'warn' }, stdio: ['ignore', 'pipe', 'pipe'] },
+);
+let srvOut = '';
+srv.stdout.on('data', (d) => (srvOut += d));
+srv.stderr.on('data', (d) => (srvOut += d));
+
+function wsClient(cookie) {
+  const ws = new WebSocket(`ws://127.0.0.1:${PORT}/ws`, { headers: { Origin: ORIGIN, Cookie: cookie } });
+  const state = { msgs: [] };
+  ws.on('message', (data, isBinary) => {
+    if (!isBinary) state.msgs.push(JSON.parse(String(data)));
+  });
+  const waitMsg = async (pred, timeout = 6000) => {
+    const t0 = Date.now();
+    while (Date.now() - t0 < timeout) {
+      const m = state.msgs.find(pred);
+      if (m) return m;
+      await sleep(40);
+    }
+    return null;
+  };
+  return { ws, state, waitMsg, send: (m) => ws.send(JSON.stringify(m)) };
+}
+
+const j = (path, method, cookie, body) =>
+  fetch(`${ORIGIN}${path}`, {
+    method,
+    headers: { Origin: ORIGIN, Cookie: cookie, ...(body ? { 'Content-Type': 'application/json' } : {}) },
+    ...(body ? { body: JSON.stringify(body) } : {}),
+  });
+
+try {
+  await sleep(1500);
+  const token = /arb_[0-9a-f]+/.exec(srvOut)?.[0];
+  check('boot + token bootstrap', !!token);
+  const login = await fetch(`${ORIGIN}/api/v1/auth/login`, {
+    method: 'POST', headers: { 'Content-Type': 'application/json', Origin: ORIGIN }, body: JSON.stringify({ token }),
+  });
+  const cookie = login.headers.get('set-cookie')?.split(';')[0] ?? '';
+  check('login → cookie', login.status === 200);
+
+  const c = wsClient(cookie);
+  await new Promise((res, rej) => (c.ws.on('open', res), c.ws.on('error', rej)));
+  c.send({ type: 'hello', protocol: 1 });
+  await c.waitMsg((m) => m.type === 'hello_ok');
+  c.send({ type: 'sub', topics: ['worktrees', 'settings'] });
+
+  // ---- (1) checkout du PRINCIPAL en CLI → worktree_update sans watch client ----
+  const addRepo = await j('/api/v1/repos', 'POST', cookie, { path: repo });
+  check('POST /repos → 201', addRepo.status === 201);
+  await sleep(1000); // laisse chokidar finir le scan initial du watcher permanent du principal
+  c.state.msgs.length = 0;
+  const t0 = Date.now();
+  git('checkout', '-b', 'feature'); // changement de branche du checkout principal, hors Arboretum
+  const branchMsg = await c.waitMsg((m) => m.type === 'worktree_update' && m.worktree?.isMain && m.worktree?.branch === 'feature', 5000);
+  const dt = branchMsg ? Date.now() - t0 : -1;
+  check('checkout principal en CLI → worktree_update (sans watch client)', !!branchMsg, branchMsg ? `${dt}ms` : 'timeout');
+  check('latence temps réel < 500 ms', branchMsg !== null && dt >= 0 && dt < 500, `${dt}ms`);
+
+  // ---- (2) PATCH /settings → settings_update reçu par l'abonné 'settings' ----
+  c.state.msgs.length = 0;
+  const patch = await j('/api/v1/settings', 'PATCH', cookie, { retentionDays: 7 });
+  check('PATCH /settings → 200', patch.status === 200);
+  const settingsMsg = await c.waitMsg((m) => m.type === 'settings_update' && m.settings?.retentionDays === 7, 3000);
+  check('settings_update reçu avec le nouvel état', !!settingsMsg);
+
+  c.ws.close();
+} catch (err) {
+  check('exception', false, String(err));
+} finally {
+  srv.kill('SIGTERM');
+  await sleep(1500);
+  check('arrêt propre du daemon (SIGTERM)', srv.exitCode === 0 || srv.signalCode === null || srv.exitCode === null);
+  rmSync(tmp, { recursive: true, force: true });
+  const failed = results.filter((r) => !r.ok);
+  console.log(failed.length === 0 ? '\nACCEPTANCE P11: ALL GREEN' : `\nACCEPTANCE P11: ${failed.length} FAILURE(S)`);
+  process.exit(failed.length === 0 ? 0 : 1);
+}
--- a/packages/server/src/app.ts
+++ b/packages/server/src/app.ts
@@ -11,6 +11,7 @@ import { AuthService, LoginRateLimiter, type AuthContext } from './auth/service.
 import { PtyManager } from './core/pty-manager.js';
 import { DiscoveryService } from './core/discovery-service.js';
 import { SessionArchiveService } from './core/session-archive.js';
+import { SettingsBus } from './core/settings-bus.js';
 import { WorktreeManager } from './core/worktree-manager.js';
 import { RepoDiscoveryService } from './core/repo-discovery.js';
 import { GroupManager } from './core/group-manager.js';
@@ -80,6 +81,7 @@ export interface AppBundle {
  groups: GroupManager;
  push: PushService;
  fsWatcher: FsWatcherService;
+  settingsBus: SettingsBus;
 }

 export function buildApp(config: Config, db: Db, serverVersion: string): AppBundle {
@@ -105,6 +107,8 @@ export function buildApp(config: Config, db: Db, serverVersion: string): AppBund
  // Démarré dans runDaemon() (jamais ici) → le scan ne tourne pas pendant les tests qui appellent buildApp.
  const repoDiscovery = new RepoDiscoveryService(db, worktrees);
  const groups = new GroupManager(db);
+  // P11 — bus de diffusion des réglages : PATCH /settings émet, la gateway relaie au topic 'settings'.
+  const settingsBus = new SettingsBus();

  // En-têtes de sécurité sur toute réponse + no-store sur les réponses sensibles (API/WS).
  // onSend DOIT retourner le payload (sinon Fastify vide la réponse).
@@ -189,14 +193,14 @@ export function buildApp(config: Config, db: Db, serverVersion: string): AppBund
  registerGitRoutes(app, worktrees, db);
  registerFileRoutes(app, worktrees, db);
  registerPushRoutes(app, push, db);
-  registerSettingsRoutes(app, db, config, serverVersion, push);
+  registerSettingsRoutes(app, db, config, serverVersion, push, settingsBus);
  registerFsRoutes(app);
  registerAuditRoutes(app, db);
  registerDataRoutes(app, db, auth);
  // La route websocket doit être déclarée APRÈS le chargement du plugin (contexte
  // encapsulé) — sinon le handler reçoit la signature REST (request, reply).
  void app.register(async (scoped) => {
-    registerWsGateway(scoped, manager, discovery, sessionArchive, worktrees, groups, serverVersion);
+    registerWsGateway(scoped, manager, discovery, sessionArchive, worktrees, groups, settingsBus, serverVersion);
  });

  // SPA buildée embarquée dans le paquet npm (public/) — absente en dev (vite dev sert le front)
@@ -211,5 +215,5 @@ export function buildApp(config: Config, db: Db, serverVersion: string): AppBund
    });
  }

-  return { app, auth, manager, discovery, sessionArchive, repoDiscovery, worktrees, groups, push, fsWatcher };
+  return { app, auth, manager, discovery, sessionArchive, repoDiscovery, worktrees, groups, push, fsWatcher, settingsBus };
 }
--- a/packages/server/src/core/fs-watcher.ts
+++ b/packages/server/src/core/fs-watcher.ts
@@ -23,6 +23,8 @@ interface WatchEntry {
  refCount: number;
  /** nombre de sessions vivantes épinglant ce worktree. */
  sessionPins: number;
+  /** épingle « permanente » (checkout principal d'un repo enregistré) — jamais évincée (P11). */
+  repoPins: number;
  lastUsed: number;
  debounce: NodeJS.Timeout | null;
  /** résolue quand chokidar a fini son scan initial (les events deviennent fiables). */
@@ -91,6 +93,24 @@ export class FsWatcherService extends EventEmitter<FsWatcherEvents> {
    e.lastUsed = Date.now();
  }

+  /**
+   * Épingle en PERMANENCE le checkout principal d'un repo enregistré (P11) : ainsi un `git checkout`
+   * en CLI sur le principal est détecté et rediffusé sans qu'aucun client n'ait « regardé » ce
+   * worktree. Idempotent (un seul pin par repo+path conservé). Jamais évincé par la LRU.
+   */
+  pinRepo(repoId: string, path: string): void {
+    const e = this.ensure(repoId, path);
+    e.repoPins = 1; // idempotent : on ne cumule pas (un seul checkout principal par repo)
+    e.lastUsed = Date.now();
+  }
+
+  unpinRepo(repoId: string, path: string): void {
+    const e = this.entries.get(this.key(repoId, path));
+    if (!e) return;
+    e.repoPins = 0;
+    e.lastUsed = Date.now();
+  }
+
  /** Nombre de watchers actifs (test/diagnostic). */
  size(): number {
    return this.entries.size;
@@ -120,7 +140,7 @@ export class FsWatcherService extends EventEmitter<FsWatcherEvents> {
    let resolveReady: () => void = () => {};
    const ready = new Promise<void>((r) => (resolveReady = r));
    watcher.once('ready', () => resolveReady());
-    const entry: WatchEntry = { repoId, path: abs, watcher, refCount: 0, sessionPins: 0, lastUsed: Date.now(), debounce: null, ready };
+    const entry: WatchEntry = { repoId, path: abs, watcher, refCount: 0, sessionPins: 0, repoPins: 0, lastUsed: Date.now(), debounce: null, ready };
    const onChange = (): void => this.schedule(entry);
    watcher.on('add', onChange).on('change', onChange).on('unlink', onChange).on('addDir', onChange).on('unlinkDir', onChange);
    this.entries.set(key, entry);
@@ -151,7 +171,7 @@ export class FsWatcherService extends EventEmitter<FsWatcherEvents> {
  private evictIfNeeded(): void {
    if (this.entries.size <= this.maxWatchers) return;
    const idle = [...this.entries.entries()]
-      .filter(([, e]) => e.refCount === 0 && e.sessionPins === 0)
+      .filter(([, e]) => e.refCount === 0 && e.sessionPins === 0 && e.repoPins === 0)
      .sort((a, b) => a[1].lastUsed - b[1].lastUsed);
    for (const [key, e] of idle) {
      if (this.entries.size <= this.maxWatchers) break;
--- a/packages/server/src/core/settings-bus.ts
+++ b/packages/server/src/core/settings-bus.ts
@@ -0,0 +1,11 @@
+// Bus d'événements des réglages (P11) : la route PATCH /settings émet le snapshot non sensible
+// après une mise à jour réussie ; la gateway WS le relaie aux abonnés du topic 'settings'. Découple
+// les routes de la gateway (pas de dépendance directe), comme les EventEmitter des managers.
+import { EventEmitter } from 'node:events';
+import type { SettingsBroadcast } from '@arboretum/shared';
+
+export interface SettingsBusEvents {
+  settings_update: [SettingsBroadcast];
+}
+
+export class SettingsBus extends EventEmitter<SettingsBusEvents> {}
--- a/packages/server/src/core/worktree-manager.ts
+++ b/packages/server/src/core/worktree-manager.ts
@@ -202,9 +202,24 @@ export class WorktreeManager extends EventEmitter<WorktreeManagerEvents> {
    }
    const summary = await this.rowToSummary(row);
    this.emit('repo_update', summary);
+    this.armRepoWatcher(row); // P11 : temps réel du checkout principal dès l'enregistrement
    return summary;
  }

+  /**
+   * P11 — arme un watcher PERMANENT sur le checkout principal de chaque repo visible : un
+   * `git checkout`/`switch` en CLI sur le principal est ainsi rediffusé en temps réel sans qu'un
+   * client ne l'ait « regardé ». Appelé depuis runDaemon (jamais buildApp → pas d'effet en tests purs).
+   */
+  armMainCheckoutWatchers(): void {
+    const rows = this.db.prepare('SELECT id, path FROM repos WHERE hidden = 0').all() as Array<{ id: string; path: string }>;
+    for (const r of rows) this.armRepoWatcher(r);
+  }
+
+  private armRepoWatcher(row: { id: string; path: string }): void {
+    this.fsWatcher?.pinRepo(row.id, resolve(row.path));
+  }
+
  async updateRepo(id: string, patch: { label?: string; postCreateHooks?: PostCreateHook[]; preTrust?: boolean; hidden?: boolean }): Promise<RepoSummary> {
    const row = this.getRepoRow(id);
    if (!row) throw httpError(404, 'NOT_FOUND', 'No repo with this id');
@@ -217,12 +232,19 @@ export class WorktreeManager extends EventEmitter<WorktreeManagerEvents> {
      .run(row.label, row.post_create_hooks, row.pre_trust, row.hidden, id);
    const summary = await this.rowToSummary(row);
    this.emit('repo_update', summary);
+    // P11 : masqué → on libère le watcher permanent du principal ; ré-affiché → on le réarme.
+    if (patch.hidden !== undefined) {
+      if (row.hidden === 1) this.fsWatcher?.unpinRepo(id, resolve(row.path));
+      else this.armRepoWatcher(row);
+    }
    return summary;
  }

  removeRepo(id: string): boolean {
+    const row = this.getRepoRow(id);
    const res = this.db.prepare('DELETE FROM repos WHERE id = ?').run(id);
    if (res.changes === 0) return false;
+    if (row) this.fsWatcher?.unpinRepo(id, resolve(row.path)); // P11 : libère le watcher permanent
    this.factsCache.delete(id);
    this.emit('repo_removed', id);
    return true;
@@ -270,6 +292,7 @@ export class WorktreeManager extends EventEmitter<WorktreeManagerEvents> {
      if (res.changes === 1) {
        added++;
        this.emit('repo_update', await this.rowToSummary(row)); // nouveaux uniquement
+        this.armRepoWatcher(row); // P11 : temps réel du checkout principal du repo découvert
      }
    }
    return { scanned: paths.length, added, durationMs: Date.now() - t0, truncated };
--- a/packages/server/src/index.ts
+++ b/packages/server/src/index.ts
@@ -30,12 +30,13 @@ function applyClaudeHomeOverride(config: Config, db: Db): void {
 export async function runDaemon(config: Config): Promise<void> {
  const db = openDb(config.dbPath);
  applyClaudeHomeOverride(config, db); // override claude_home (réglage UI) avant de câbler les services
-  const { app, auth, manager, discovery, sessionArchive, repoDiscovery, fsWatcher } = buildApp(config, db, pkg.version);
+  const { app, auth, manager, discovery, sessionArchive, repoDiscovery, worktrees, fsWatcher } = buildApp(config, db, pkg.version);

  const bootstrapToken = auth.ensureBootstrapToken();
  await app.listen({ port: config.port, host: config.bind });
  discovery.start(); // scan initial + rafraîchissement périodique des sessions découvertes
  sessionArchive.start(); // archivage auto des sessions terminées (rétention configurable)
+  worktrees.armMainCheckoutWatchers(); // P11 : temps réel du checkout principal de chaque repo
  if (config.autoDiscover) repoDiscovery.start(); // découverte auto des repos : scan au boot + re-scan périodique

  const url = `http://${config.bind === '0.0.0.0' ? '127.0.0.1' : config.bind}:${config.port}`;
--- a/packages/server/src/routes/settings.ts
+++ b/packages/server/src/routes/settings.ts
@@ -6,6 +6,7 @@ import type { ServerInfo, SettingsResponse, UpdateSettingsRequest } from '@arbor
 import type { Config } from '../config.js';
 import { type Db, setSetting } from '../db/index.js';
 import type { PushService } from '../core/push-service.js';
+import type { SettingsBus } from '../core/settings-bus.js';
 import { recordAudit } from '../core/audit-log.js';
 import { diagnoseClaudeBin } from '../core/claude-launcher.js';
 import {
@@ -40,6 +41,7 @@ export function registerSettingsRoutes(
  config: Config,
  serverVersion: string,
  push: PushService,
+  settingsBus: SettingsBus,
 ): void {
  const serverInfo = (): ServerInfo => ({
    version: serverVersion,
@@ -116,6 +118,9 @@ export function registerSettingsRoutes(
      action: 'settings.update',
      details: { keys: Object.keys(body) },
    });
-    return reply.send(snapshot());
+    const snap = snapshot();
+    // P11 — diffuse le nouvel état non sensible à tous les clients abonnés au topic 'settings'.
+    settingsBus.emit('settings_update', snap.settings);
+    return reply.send(snap);
  });
 }
--- a/packages/server/src/ws/gateway.ts
+++ b/packages/server/src/ws/gateway.ts
@@ -9,6 +9,7 @@ import {
  type RepoSummary,
  type ServerMessage,
  type SessionSummary,
+  type SettingsBroadcast,
  type WorktreeSummary,
 } from '@arboretum/shared';
 import type { ClientBinding, PtyManager } from '../core/pty-manager.js';
@@ -16,6 +17,7 @@ import type { DiscoveryService } from '../core/discovery-service.js';
 import type { SessionArchiveService } from '../core/session-archive.js';
 import type { WorktreeManager } from '../core/worktree-manager.js';
 import type { GroupManager } from '../core/group-manager.js';
+import type { SettingsBus } from '../core/settings-bus.js';

 const HEARTBEAT_MS = 30_000;

@@ -31,6 +33,7 @@ export function registerWsGateway(
  sessionArchive: SessionArchiveService,
  worktrees: WorktreeManager,
  groups: GroupManager,
+  settingsBus: SettingsBus,
  serverVersion: string,
 ): void {
  app.get('/ws', { websocket: true }, (socket: WebSocket, req) => {
@@ -41,6 +44,7 @@ export function registerWsGateway(
    let subscribedSessions = false;
    let subscribedWorktrees = false;
    let subscribedGroups = false;
+    let subscribedSettings = false;
    let alive = true;
    // Worktrees « regardés » par CETTE connexion (clé repoId\0path) → push ciblé de worktree_changes.
    const watched = new Set<string>();
@@ -85,6 +89,10 @@ export function registerWsGateway(
    const onGroupRemoved = (groupId: string): void => {
      if (subscribedGroups) send({ type: 'group_removed', groupId });
    };
+    // P11 — un réglage a changé : relayé aux connexions abonnées au topic 'settings'.
+    const onSettingsUpdate = (settings: SettingsBroadcast): void => {
+      if (subscribedSettings) send({ type: 'settings_update', settings });
+    };
    // P7 — détail d'un worktree modifié : poussé UNIQUEMENT si cette connexion le regarde.
    const onWorktreeChanges = (e: { repoId: string; path: string }): void => {
      if (watched.has(watchKey(e.repoId, e.path))) send({ type: 'worktree_changes', ...e });
@@ -100,6 +108,7 @@ export function registerWsGateway(
    worktrees.on('worktree_changes', onWorktreeChanges);
    groups.on('group_update', onGroupUpdate);
    groups.on('group_removed', onGroupRemoved);
+    settingsBus.on('settings_update', onSettingsUpdate);

    const heartbeat = setInterval(() => {
      if (!alive) {
@@ -139,6 +148,7 @@ export function registerWsGateway(
          subscribedSessions = msg.topics.includes('sessions');
          subscribedWorktrees = msg.topics.includes('worktrees');
          subscribedGroups = msg.topics.includes('groups');
+          subscribedSettings = msg.topics.includes('settings');
          return;
        }
        case 'watch': {
@@ -250,6 +260,7 @@ export function registerWsGateway(
      worktrees.off('worktree_changes', onWorktreeChanges);
      groups.off('group_update', onGroupUpdate);
      groups.off('group_removed', onGroupRemoved);
+      settingsBus.off('settings_update', onSettingsUpdate);
      for (const [, st] of channels) manager.detach(st.sessionId, st.binding);
      channels.clear();
      // Libère le refcount du watcher pour chaque worktree regardé par cette connexion.
--- a/packages/server/test/fs-watcher.test.ts
+++ b/packages/server/test/fs-watcher.test.ts
@@ -100,6 +100,21 @@ describe('FsWatcherService', () => {
    expect(s.isWatching('b', b)).toBe(false);
  });

+  it('pinRepo (checkout principal, P11) : jamais évincé par la LRU', () => {
+    const s = new FsWatcherService({ maxWatchers: 1 });
+    services.push(s);
+    const a = makeTmpRepo();
+    const b = makeTmpRepo();
+    s.pinRepo('a', a); // épingle permanente
+    s.watch('b', b);
+    s.unwatch('b', b);
+    expect(s.isWatching('a', a)).toBe(true); // toujours là malgré le dépassement du plafond
+    s.unpinRepo('a', a);
+    const c = makeTmpRepo();
+    s.watch('c', c); // 'a' n'est plus épinglé et idle → évinçable
+    expect(s.isWatching('a', a)).toBe(false);
+  });
+
  it('closeAll libère tout', async () => {
    const repo = makeTmpRepo();
    const s = new FsWatcherService();