feat(draco): version du service d'agents visible dans le panneau Spark, mode sidebar (7.83.1) — fusion #64

ACC-5.1 (#24) — app/routers/mcp_server.py : catalogue §7.2 complété
(list_projects, get_project, read_page tronqué proprement,
list_project_tasks), search_pages en plein texte, liens profonds +
kind sur chaque résultat, scopes du jeton appliqués (écriture
refusée -32002 sans scope write) et journalisation §7.4 de chaque
tools/call dans api_audit_log (clés d'arguments seulement).

ACC-2.2 (#20) — pont MCP Flowdeck → ObsiGate, côté client :
- mcp_client : serveurs à session (capture/réémission de
  Mcp-Session-Id, handshake avant tools/call), décodage SSE
  multi-messages par id de réponse, classification des outils à la
  découverte (readOnlyHint sinon heuristique ; lecture par défaut).
- preset connecteur « obsigate » (URL dérivée dev/prod), kinds MCP
  factorisés (connectors.MCP_KINDS), option dans le panneau agent.
- garde-fou n° 2 : permission_manager.is_write_tool couvre les
  outils MCP écrivants → portes politique d'approbation
  (agent_engine) et rôles (assert_can) ; McpTool bloque tout
  apply_* distant (patron §7.5).

Tests : test_fusion_mcp_server 18, nouveau test_fusion_pont_obsigate
7 (faux serveur ObsiGate à session stricte via le vrai _rpc).
Preuve réelle T0-T5 : attend la clé ObsiGate dédiée (1 vault) et
Q-A/Q-B côté Bruno.

CHANGELOG.md

@@ -1,5 +1,167 @@
 # Changelog - FlowDeck
 
+## v7.83.1 (2026-10-26) — Panneau Spark : version du service d'agents visible en mode sidebar
+
+- **Panneau Spark** (`static/js/draco-panneau.js`) : le pied du panneau
+  affiche désormais la ligne discrète et permanente « Service d'agents
+  vX », comme la console pleine page. La version est lue **en direct**
+  dans la santé du service (`GET /healthz` de l'origine que le panneau
+  connaît déjà), à l'ouverture du panneau, avec cache de session —
+  jamais une valeur codée en dur ; service injoignable = rien
+  d'affiché, aucune erreur visible. Carte fusion #64.
+
+## v7.83.0 (2026-10-26) — Fusion : serveur MCP complété (ACC-5.1) et pont MCP vers ObsiGate (ACC-2.2)
+
+- **ACC-5.1 — serveur MCP Flowdeck complété** (`app/routers/mcp_server.py`,
+  spec d'accès centralisé §7.2/§7.4). Le catalogue de lecture gagne
+  `list_projects`, `get_project` (métadonnées + schéma + titres seuls),
+  `read_page` (page classique ou ligne de collection, contenu tronqué
+  proprement à 20 000 caractères) et `list_project_tasks` (tâches d'une
+  base projet, tous assignés). `search_pages` devient une vraie
+  recherche **plein texte** (titre + contenu des pages, titre +
+  propriétés des lignes de collection), et chaque résultat porte son
+  **lien profond** (`url`) et son `kind` — contrat figé : un projet
+  Flowdeck est une collection. §7.4 tenu : les **scopes du jeton**
+  sont appliqués (un jeton `read` se voit refuser les outils
+  d'écriture, erreur -32002) et **chaque appel `tools/call` est
+  journalisé** dans `api_audit_log` (outil, jeton, statut — jamais les
+  valeurs d'arguments). `list_meetings`/`get_meeting` restent hors
+  catalogue tant que la décision Meetings A/B n'est pas tranchée.
+- **ACC-2.2 — pont MCP Flowdeck → ObsiGate, côté client**
+  (`app/services/mcp_client.py`, plan « étape 1 : lectures seules »).
+  Le client MCP gère désormais les serveurs **à session** (Streamable
+  HTTP d'ObsiGate) : capture et réémission de l'en-tête
+  `Mcp-Session-Id`, handshake complet avant `tools/call`, et décodage
+  SSE **multi-messages** (le message de réponse est sélectionné, pas
+  le premier `data:`). Nouveau preset de connecteur **« obsigate »**
+  (URL dérivée de la configuration dev/prod, clé dédiée au pont créée
+  dans ObsiGate → Configurations → Clés API & MCP).
+- **Garde-fou n° 2 du plan du pont** : les outils MCP sont **classés**
+  à la découverte (annotation `readOnlyHint` du serveur, sinon
+  heuristique de nom ; lecture par défaut). Un outil MCP écrivant
+  (`propose_*`, `create_*`…) passe désormais par les mêmes portes que
+  les outils internes — politique d'approbation (`agent_policies`) et
+  rôles (`PermissionManager`, via le nouvel `is_write_tool`) — là où
+  il passait avant pour une lecture. Et un outil distant **`apply_*`
+  est bloqué côté client** : une écriture croisée ne s'exécute jamais
+  par un simple appel d'outil ; elle suit le patron §7.5 (propose →
+  approbation dans l'application propriétaire).
+- **Preuve réelle T0–T5** : le code est testé contre un faux serveur
+  ObsiGate à session stricte (`tests/test_fusion_pont_obsigate.py`,
+  7 tests ; `tests/test_fusion_mcp_server.py` porté à 18 tests). La
+  démonstration contre l'ObsiGate réel attend encore deux gestes de
+  Bruno : créer la clé API ObsiGate dédiée au pont, **limitée à un
+  seul vault**, et trancher **Q-A/Q-B** (fusion-trois-apps #1).
+
+## v7.82.4 (2026-10-25) — Résolution dev/prod par l'hôte de la requête : le mode console du panneau revit en dev
+
+- **Bug corrigé (dev)** : sur flowdeck-dev, le mode console du
+  panneau Spark restait une page morte — prouvé en direct avant
+  correctif : la 7.82.3 tournait bien (`GET /api/health` →
+  `7.82.3`), mais sa CSP servait `frame-src 'self'
+  https://agent.dracodev.net` (et `connect-src` idem), pointant
+  l'agent **prod, pas encore installée**, et la source de l'iframe
+  console (`DRACO_PANNEAU_CONFIG.agent_url`, rendue par
+  `base.html`) valait la même origine morte. Deux trous laissés par
+  la v7.82.3 : (1) `agent_url()` n'avait **aucune dérivation** —
+  même avec `AGENT_EVENEMENTS_URL` en `-dev`, l'agent lui-même
+  retombait sur le nom officiel ; (2) **aucune** résolution ne
+  lisait l'hôte de la requête — un `.env` dev sans
+  `DRACO_AGENT_URL` ni `AGENT_EVENEMENTS_URL` retombait
+  entièrement sur la prod.
+- **Cure** (`app/services/draco_urls.py`, `app/middleware/security.py`) :
+  pour tout ce qui est **rendu à une requête** (liens de la barre /
+  `DRACO_URLS` servi, source de l'iframe du panneau, origine agent
+  des CSP `frame-src`/`connect-src`, boutons et redirections SSO —
+  émettrice via `DRACO_SSO_URL` résolu, callback =
+  `OAUTH_REDIRECT_URI` dont l'origine est sa propre origine
+  résolue), l'ordre de résolution devient : 1) variable explicite
+  de l'installation, prioritaire et inchangée ; 2) **hôte public de
+  la requête** (posé par le middleware CSP dans un `ContextVar`,
+  même patron que le nonce CSP ; `X-Forwarded-Host` sinon `Host`) —
+  motif fermé `^(og|flowdeck|tiktak|kanbante|agent)(<SUFFIXE>)\.dracodev\.net$`,
+  suffixe optionnel (`-dev`, vide en prod) appliqué à toutes les
+  sœurs **et à l'agent** ; jamais de suffixe inventé depuis un hôte
+  arbitraire, hors motif (LAN, IP) on passe à la suite ; 3)
+  dérivation par les adresses de l'agent connues en configuration,
+  **élargie** (URL d'événements **ou** base du service — notamment
+  `agent_url()` se dérive désormais de l'URL d'événements
+  explicite) — aussi la résolution des contextes **sans requête**,
+  inchangée sinon ; 4) défaut officiel. L'iframe du panneau et la
+  CSP de la même réponse consomment la **même** origine résolue :
+  les deux lisent `agent_url()` pendant la même requête.
+- Tests : `tests/test_draco_urls_devprod.py` +12 (25/25 au total) —
+  requête `Host: flowdeck-dev.dracodev.net` et **aucune variable
+  d'environnement** → barre, iframe, CSP et SSO rendus entièrement
+  en `-dev` ; Host prod → noms officiels ; Host LAN → repli
+  config/défauts ; explicite gagne sur l'hôte ; motif fermé
+  (hôtes arbitraires → aucune dérivation) ; agent dérivé du seul
+  `AGENT_EVENEMENTS_URL` ; contextes sans requête inchangés.
+  Suite complète relue : **4 failed / 26 errors, identiques sur
+  arbre propre** (1307 passed en propre ; familles préexistantes
+  MCP/realtime/my_tasks/gallery/sync, sans rapport avec ce
+  correctif) — 1319 passed avec le correctif. Ruff vert.
+
+## v7.82.3 (2026-10-25) — URLs sœurs dérivées de l'environnement de l'agent
+
+- **Bug corrigé (dev)** : les URL sœurs servies par `draco_urls()`
+  (menu inter-apps, SSO, preset TikTak, références croisées)
+  restaient aux noms **prod** quand le `.env` ne déclarait pas les
+  quatre URL une à une — seul l'agent était en `-dev`, via
+  `AGENT_EVENEMENTS_URL`. Le menu inter-apps de la dev envoyait
+  donc vers la prod.
+- **Dérivation d'environnement** (`app/services/draco_urls.py`,
+  résolution uniforme des quatre sœurs — ObsiGate, Flowdeck,
+  TikTak, Kanbante) : 1) URL explicite prioritaire, contrat
+  inchangé (`DRACO_SSO_URL`, `TIKTAK_URL`, `KANBANTE_URL`,
+  `FLOWDECK_URL`/`OAUTH_REDIRECT_URI`…) ; 2) sinon, si l'hôte de
+  l'agent tel que l'app le connaît déjà (`AGENT_EVENEMENTS_URL`,
+  via `agent_evenements_url()`) est de la forme
+  `agent<SUFFIXE>.dracodev.net` (SUFFIXE éventuellement vide,
+  typiquement `-dev`), l'URL sœur est dérivée :
+  `https://<slug><SUFFIXE>.dracodev.net` (slugs `og`, `flowdeck`,
+  `tiktak`, `kanbante`) ; hôte hors motif (autre domaine, LAN,
+  IP) = **aucune dérivation**, jamais un lien deviné ; 3) sinon
+  défaut officiel, comme avant. L'URL **Gitea ne change pas** ;
+  les gabarits et le JS de la barre, qui consomment `draco_urls()`,
+  sont inchangés.
+- Tests : `tests/test_draco_urls_devprod.py` +4 — seul
+  `AGENT_EVENEMENTS_URL=…agent-dev…` posé → sœurs en `-dev` ;
+  explicite gagne sur la dérivation ; hôtes d'agent hors motif
+  → défauts officiels ; agent officiel (suffixe vide) → sœurs
+  officielles. Non-régression « aucun env » déjà couverte.
+
+## v7.82.2 (2026-10-25) — CI test entièrement verte : pages-supports de templates exclues des listes
+
+Cause racine commune à 7 des 8 échecs restants : la refonte templates
+(`fc85481`, 2026-10-10) a introduit les **pages-supports** — une page
+interne par template système (99 sur une base fraîche), créée par
+`TemplateService._create_holder` avec `search_excluded=1` et la consigne
+« masquée, jamais listée ». Aucune liste ne respectait cette consigne :
+datées `updated_at = now`, elles occupaient toutes les places des
+listes triées par récence (LIMIT 50/20/8), devant les vraies pages.
+
+- **Bibliothèque** (`app/routers/library.py`) : filtre `NO_HOLDER`
+  (`COALESCE(p.search_excluded, 0) = 0`) ajouté aux onglets meetings,
+  recents, favorites, shared, published, private et workspace.
+- **Sidebar Recents** (`dashboard/_common.py::_recent_sidebar_pages`)
+  et **projets builtin** (`dashboard/workspace.py`) : même exclusion —
+  les pages-supports n'y sont plus ni des « fichiers récents » ni des
+  « projets ».
+- **Mentions agent** (`routers/agent.py::list_mentions`) : les trois
+  scans de pages excluent les pages-supports, jamais mentionnables.
+- **Flake `test_v55` (og parser)** : `mock_http`
+  (`tests/test_agent_web_tools.py`) répare désormais aussi la référence
+  `shared_client` figée par `og_fetcher` quand il est importé pendant
+  la fenêtre patchée (même classe de fuite que celle déjà traitée pour
+  `url_fetch`) — le handler d'un test `fetch_url` répondait sinon aux
+  fetchs OG des tests suivants du même worker xdist.
+- **`test_editor_front_end_wired`** : depuis `fc85481`, le picker de
+  `base.html` passe par le registre unifié (`/api/templates`) et le
+  câblage `page-templates` vit dans l'éditeur
+  (`static/js/page_editor_scripts.js`) ; le marqueur est vérifié dans
+  le fichier qui le porte. Aucune assertion supprimée ni affaiblie.
+
 ## v7.82.1 (2026-10-25) — CI : lint ruff vert et littéral de version resynchronisé
 
 - **Lint (CI run 2304, job 7058)** : les 16 erreurs `ruff check

VERSION

@@ -1 +1 @@
-7.82.1
+7.83.1

app/main.py

@@ -216,7 +216,7 @@ def _lire_version() -> str:
 
 
 # Littéral de version synchronisé avec le fichier VERSION :
-# version="7.82.1"
+# version="7.83.1"
 # Patron historique des bumps (voir scripts/bump_version.py, qui cherche
 # ce littéral) et contrôle de tests/test_pwa_offline.py. La version
 # runtime reste lue dynamiquement depuis VERSION par _lire_version()

app/middleware/security.py

@@ -134,8 +134,9 @@ class ContentSecurityPolicyMiddleware(BaseHTTPMiddleware):
         # service d'agents commun (premier parti, même programme) —
         # seul hôte externe nommé, rien d'autre ne s'ouvre.
         # v7.82.0 (dev/prod) : cet hôte n'est plus figé — {agent} est
-        # l'origine résolue par la config (DRACO_AGENT_URL ; défaut =
-        # nom officiel, un dev déclare son agent-dev dans le .env).
+        # l'origine résolue par draco_urls (DRACO_AGENT_URL explicite,
+        # sinon hôte de la requête — v7.82.4 —, sinon dérivation par
+        # l'agent connu en config ; défaut = nom officiel).
         # Même origine pour frame-src : la console du service est
         # encadrée en iframe par le panneau (surimpression v3).
         "connect-src 'self' {agent} ws://{host} wss://{host}; "
@@ -152,6 +153,17 @@ class ContentSecurityPolicyMiddleware(BaseHTTPMiddleware):
         # tâche créée maintenant → le contexte (donc le nonce) y est copié,
         # exactement ce que les templates liront via `csp_nonce()`.
         CSP_NONCE.set(nonce)
+        # v7.82.4 : même patron pour l'hôte public de la requête —
+        # `draco_urls` le lit pour résoudre l'environnement (dev/prod)
+        # de tout ce qui est rendu à cette requête (barre, iframe du
+        # panneau, SSO), et la CSP ci-dessous consomme la MÊME origine
+        # agent résolue. X-Forwarded-Host prime, comme
+        # `get_redirect_uri()` (app/routers/auth.py).
+        from app.services import draco_urls
+
+        draco_urls.definir_hote_requete(
+            request.headers.get("x-forwarded-host", "")
+            or request.headers.get("host", ""))
         response = await call_next(request)
         # Only set CSP on HTML responses
         content_type = response.headers.get("content-type", "")
@@ -160,8 +172,6 @@ class ContentSecurityPolicyMiddleware(BaseHTTPMiddleware):
             # on retire quand même tout caractère hors base URL par sécurité.
             host = re.sub(r"[^0-9A-Za-z.\-:\[\]]", "",
                           request.headers.get("host", ""))
-            from app.services import draco_urls
-
             agent = re.sub(r"[^0-9A-Za-z.\-:/]", "",
                            draco_urls.origine(draco_urls.agent_url()))
             response.headers[self.CSP_HEADER] = self.CSP_VALUE.format(

app/routers/agent.py

@@ -838,6 +838,8 @@ def list_mentions(request: Request, q: str = ""):
     Les entrées identiques (même type + même titre, ex. plusieurs pages « Sans
     titre ») sont **dédupliquées** pour ne pas afficher le même fichier plusieurs fois.
     """
+    # Les scans de pages excluent les pages-supports de templates
+    # (search_excluded=1 — cf. library.NO_HOLDER) : jamais mentionnables.
     query = (q or "").strip()[:60]
     like = f"%{query}%"
     limit = 8
@@ -873,7 +875,7 @@ def list_mentions(request: Request, q: str = ""):
         if ws:
             sql_ws = (
                 "SELECT p.id, p.title FROM pages p "
-                "WHERE p.workspace_id=? AND p.deleted_at IS NULL AND (? = '' OR p.title LIKE ?) "
+                "WHERE p.workspace_id=? AND p.deleted_at IS NULL AND COALESCE(p.search_excluded, 0) = 0 AND (? = '' OR p.title LIKE ?) "
                 "ORDER BY p.updated_at DESC LIMIT ?"
             )
             files = [
@@ -922,7 +924,7 @@ def list_mentions(request: Request, q: str = ""):
             for r in conn.execute(
                 "SELECT p.id, p.title, w.name AS ws_name "
                 "FROM pages p LEFT JOIN workspaces w ON w.id=p.workspace_id "
-                "WHERE p.deleted_at IS NULL AND (p.workspace_id IS ? OR p.workspace_id IS NULL) "
+                "WHERE p.deleted_at IS NULL AND COALESCE(p.search_excluded, 0) = 0 AND (p.workspace_id IS ? OR p.workspace_id IS NULL) "
                 "AND (? = '' OR p.title LIKE ?) ORDER BY p.updated_at DESC LIMIT ?",
                 (ws, query, like, limit)
             ).fetchall():
@@ -939,7 +941,7 @@ def list_mentions(request: Request, q: str = ""):
             for r in conn.execute(
                 "SELECT p.id, p.title, w.name AS ws_name FROM pages p "
                 "LEFT JOIN workspaces w ON w.id=p.workspace_id "
-                "WHERE p.deleted_at IS NULL AND (? = '' OR p.title LIKE ?) "
+                "WHERE p.deleted_at IS NULL AND COALESCE(p.search_excluded, 0) = 0 AND (? = '' OR p.title LIKE ?) "
                 "ORDER BY p.updated_at DESC LIMIT ?", (query, like, limit)
             ).fetchall():
                 docs.append({"type": "document", "icon": "📄", "id": r["id"],

app/routers/auth.py

@@ -33,12 +33,14 @@ def get_redirect_uri(request: Request) -> str:
     return f"{scheme}://{host}/auth/callback"
 
 def _sso_login_url() -> str:
-    """URL du bouton « Login with ObsiGate (SSO) », résolue par la config.
+    """URL du bouton « Login with ObsiGate (SSO) ».
 
-    Émettrice = DRACO_SSO_URL (défaut : nom officiel) ; retour =
-    URL publique de cette instance (FLOWDECK_URL / OAUTH_REDIRECT_URI /
-    APP_BASE_URL, défaut : nom officiel) + ``/auth/draco``. Aucun nom
-    public figé : un dev déclare ses noms en ``-dev`` dans le .env.
+    Émettrice = ``obsigate_url()`` (DRACO_SSO_URL explicite, sinon
+    l'environnement de la requête — v7.82.4) ; retour = URL publique
+    de cette instance résolue de la même façon (FLOWDECK_URL /
+    OAUTH_REDIRECT_URI explicites, sinon hôte de la requête, le
+    callback = sa propre origine résolue) + ``/auth/draco``. Aucun
+    nom public figé.
     """
     from urllib.parse import quote
 

app/routers/dashboard/_common.py

@@ -107,6 +107,7 @@ def _recent_sidebar_pages(uid: int, limit: int = 8) -> list:
                    FROM pages p
                    LEFT JOIN workspaces w ON w.id = p.workspace_id
                    WHERE p.deleted_at IS NULL AND p.parent_section != 'Trash'
+                     AND COALESCE(p.search_excluded, 0) = 0
                      AND (w.owner_id = ? OR p.workspace_id IS NULL)
                      AND REPLACE(COALESCE(p.content,''), ' ', '')
                          NOT LIKE '%"type":"meeting"%'

app/routers/dashboard/workspace.py

@@ -222,7 +222,8 @@ async def list_workspace_projects(request: Request):
     builtin = []
     with get_conn() as conn:
         rows = conn.execute(
-            "SELECT id, title FROM pages WHERE parent_id IS NULL AND collection_row_id IS NULL ORDER BY updated_at DESC LIMIT 20"
+            "SELECT id, title FROM pages WHERE parent_id IS NULL AND collection_row_id IS NULL "
+            "AND COALESCE(search_excluded, 0) = 0 ORDER BY updated_at DESC LIMIT 20"
         ).fetchall()
         # A23 : une seule agrégation GROUP BY au lieu d'un COUNT par ligne.
         counts = {}

app/routers/library.py

@@ -210,6 +210,12 @@ BASE_SELECT = (
     "p.parent_id, p.share_mode, p.parent_section, p.page_icon"
 )
 
+# Pages-supports de templates (fd_templates._create_holder) : pages internes
+# marquées search_excluded=1 — « masquées, jamais listées ». Elles ne doivent
+# apparaître dans AUCUN onglet de la bibliothèque (une base fraîche en sème 99,
+# qui noieraient sinon recents/private/workspace sous leur updated_at = now).
+NO_HOLDER = "COALESCE(p.search_excluded, 0) = 0"
+
 
 # Lignes contenant un bloc AI Meeting Notes (formats JSON navigateur/Python).
 MEETING_BLOCK_LIKE = "REPLACE(COALESCE(p.content,''), ' ', '') LIKE '%\"type\":\"meeting\"%'"
@@ -240,7 +246,7 @@ def library_meetings(
         " ORDER BY mt.id DESC LIMIT 1) AS meeting_summary "
         "FROM pages p LEFT JOIN workspaces w ON w.id = p.workspace_id "
         "WHERE p.deleted_at IS NULL AND (w.owner_id = ? OR p.workspace_id IS NULL) "
-        f"AND {MEETING_BLOCK_LIKE}"
+        f"AND {NO_HOLDER} AND {MEETING_BLOCK_LIKE}"
     )
     params: list = [uid]
     query, params = _apply_source_filter(query, params, source_type)
@@ -275,7 +281,7 @@ def library_recents(
         source_type = "all"
 
     uid = _get_user_id(request)
-    query = f"{BASE_SELECT} FROM pages p WHERE p.parent_section != 'Trash' AND p.deleted_at IS NULL AND {NO_MEETING_BLOCK}"
+    query = f"{BASE_SELECT} FROM pages p WHERE p.parent_section != 'Trash' AND p.deleted_at IS NULL AND {NO_HOLDER} AND {NO_MEETING_BLOCK}"
     params: list = []
 
     # Hierarchical view: show only root-level pages at the top
@@ -313,6 +319,7 @@ def library_favorites(
         "p.parent_id, p.share_mode, p.parent_section, 1 as favorited "
         "FROM favorites f JOIN pages p ON p.id = f.page_id "
         "WHERE f.user_id = ? AND p.parent_section != 'Trash' AND p.deleted_at IS NULL "
+        f"AND {NO_HOLDER}"
     )
     params: list = [uid]
     if tree:
@@ -379,7 +386,7 @@ def library_shared(
 
     query = (
         f"{BASE_SELECT} FROM pages p "
-        f"WHERE ({' OR '.join(conds)}) AND p.parent_section != 'Trash' AND p.deleted_at IS NULL"
+        f"WHERE ({' OR '.join(conds)}) AND p.parent_section != 'Trash' AND p.deleted_at IS NULL AND {NO_HOLDER}"
     )
     if tree:
         query += " AND p.parent_id IS NULL"
@@ -410,7 +417,7 @@ def library_published(
 
     uid = _get_user_id(request)
     # v7.69.6 : les deux drapeaux (lignes historiques issues d'un seul chemin).
-    query = f"{BASE_SELECT} FROM pages p WHERE (p.is_published = 1 OR p.published = 1) AND p.parent_section != 'Trash' AND p.deleted_at IS NULL"
+    query = f"{BASE_SELECT} FROM pages p WHERE (p.is_published = 1 OR p.published = 1) AND p.parent_section != 'Trash' AND p.deleted_at IS NULL AND {NO_HOLDER}"
     params: list = []
     if tree:
         query += " AND p.parent_id IS NULL"
@@ -435,7 +442,7 @@ def library_private(
         source_type = "all"
 
     uid = _get_user_id(request)
-    query = f"{BASE_SELECT} FROM pages p WHERE p.parent_section = 'Private' AND p.deleted_at IS NULL AND {NO_MEETING_BLOCK}"
+    query = f"{BASE_SELECT} FROM pages p WHERE p.parent_section = 'Private' AND p.deleted_at IS NULL AND {NO_HOLDER} AND {NO_MEETING_BLOCK}"
     params: list = []
     if tree:
         query += " AND p.parent_id IS NULL"
@@ -501,7 +508,7 @@ def library_workspace(
         source_type = "all"
 
     uid = _get_user_id(request)
-    query = f"{BASE_SELECT} FROM pages p WHERE (p.workspace != '' OR p.workspace_id IS NOT NULL) AND p.parent_section != 'Trash' AND p.deleted_at IS NULL AND {NO_MEETING_BLOCK}"
+    query = f"{BASE_SELECT} FROM pages p WHERE (p.workspace != '' OR p.workspace_id IS NOT NULL) AND p.parent_section != 'Trash' AND p.deleted_at IS NULL AND {NO_HOLDER} AND {NO_MEETING_BLOCK}"
     params: list = []
 
     if tree:

app/routers/mcp_server.py

@@ -18,6 +18,21 @@ Doctrine du programme : Flowdeck
 référence les tâches, il ne les copie pas — aucune tâche TikTak ne
 transite par ici. Écritures limitées à la création/mise à jour de
 pages et à la création de collections ; aucune suppression.
+
+**ACC-5.1 (architecture v2.0, §7.2)** — le catalogue de lecture cible
+est couvert : ``list_projects`` / ``get_project`` (un « projet » du
+contrat de liens profonds est une collection : ``/collections/<id>``),
+``search_pages`` (plein texte : titres **et** contenus, pages
+classiques et lignes de collections), ``read_page`` (contenu tronqué
+proprement) et ``list_project_tasks`` (collections ``is_task``). Les
+outils réunions restent absents (décision Meetings A/B ouverte).
+Chaque résultat d'objet porte son **URL profonde** (forme figée par
+``contrats/liens-profonds.md``), comme le §7.2 l'exige.
+
+**§7.4** — chaque ``tools/call`` est journalisé dans
+``api_audit_log`` (jeton, outil, horodatage) et la **portée du
+jeton** est appliquée : sans le scope ``write``, les trois outils
+d'écriture sont refusés (« jeton en lecture seule »).
 """
 from __future__ import annotations
 
@@ -74,6 +89,41 @@ OUTILS = [
     {"name": "list_repos",
      "description": "Liste le registre des dépôts suivis par Flowdeck (builtin, Gitea, GitHub).",
      "inputSchema": {"type": "object", "properties": {}}},
+    {"name": "list_projects",
+     "description": "Liste les PROJETS Flowdeck du périmètre du jeton "
+                    "(catalogue §7.2) : au sens du contrat de liens "
+                    "profonds, un projet est une collection — id, titre, "
+                    "état (is_task, nombre de lignes) et URL profonde "
+                    "/collections/<id>.",
+     "inputSchema": {"type": "object", "properties": {}}},
+    {"name": "get_project",
+     "description": "Détail d'un projet (collection) SANS le contenu "
+                    "complet de ses pages : métadonnées, schéma des "
+                    "propriétés et titres de ses lignes, URL profonde.",
+     "inputSchema": {"type": "object",
+                     "properties": {"id": {"type": "integer"}},
+                     "required": ["id"]}},
+    {"name": "read_page",
+     "description": "Lit le contenu d'une page (§7.2), tronqué "
+                    "proprement à 20 000 caractères (drapeau "
+                    "« truncated »). « kind » départage les deux "
+                    "familles d'identifiants : « page » (page "
+                    "classique d'un workspace, URL /pages/<id>) ou "
+                    "« collection_page » (ligne d'une collection, "
+                    "contenu porté par sa page liée). Sans « kind », "
+                    "la page classique est essayée d'abord.",
+     "inputSchema": {"type": "object",
+                     "properties": {"id": {"type": "integer"},
+                                    "kind": {"type": "string",
+                                             "enum": ["page", "collection_page"]}},
+                     "required": ["id"]}},
+    {"name": "list_project_tasks",
+     "description": "Liste les tâches des projets « base de tâches » "
+                    "(collections is_task) du périmètre du jeton : "
+                    "titre, statut, échéance, assignés, URL profonde "
+                    "du projet. « project_id » restreint à un projet.",
+     "inputSchema": {"type": "object",
+                     "properties": {"project_id": {"type": "integer"}}}},
     {"name": "create_page",
      "description": "Crée une page dans une collection Flowdeck.",
      "inputSchema": {"type": "object",
@@ -99,6 +149,38 @@ OUTILS = [
                      "required": ["name"]}},
 ]
 
+#: Outils d'écriture — refusés aux jetons sans le scope ``write`` (§7.4).
+OUTILS_ECRITURE = {"create_page", "update_page", "create_collection"}
+
+#: Borne de lecture d'un contenu de page (§7.2 : « tronqué proprement »).
+CONTENU_MAX = 20000
+
+
+def _url_projet(collection_id: int) -> str:
+    """URL profonde d'un projet (contrat figé : /collections/<id>)."""
+    from app.services import draco_urls
+
+    return f"{draco_urls.flowdeck_url()}/collections/{collection_id}"
+
+
+def _url_page(page_id: int) -> str:
+    """URL profonde d'une page classique (contrat figé : /pages/<id>)."""
+    from app.services import draco_urls
+
+    return f"{draco_urls.flowdeck_url()}/pages/{page_id}"
+
+
+def _tronquer(texte: str) -> tuple[str, bool]:
+    """Troncature propre : jamais au milieu d'un mot si on peut l'éviter."""
+    texte = texte or ""
+    if len(texte) <= CONTENU_MAX:
+        return texte, False
+    coupe = texte[:CONTENU_MAX]
+    dernier_espace = coupe.rfind(" ", CONTENU_MAX - 200)
+    if dernier_espace > 0:
+        coupe = coupe[:dernier_espace]
+    return coupe, True
+
 
 def _props(brut):
     """property_values_json → dict quand c'est du JSON, sinon inchangé."""
@@ -112,8 +194,10 @@ def _props(brut):
 
 def _page_resume(row) -> dict:
     return {"id": row["id"], "collection_id": row["collection_id"],
+            "kind": "collection_page",
             "title": row["title"], "icon": row["icon"],
             "properties": _props(row["property_values_json"]),
+            "url": _url_projet(row["collection_id"]),
             "updated_at": row["updated_at"]}
 
 
@@ -169,6 +253,7 @@ def _appel_outil(nom: str, args: dict, user_id: int):
             return [{"id": r["id"], "name": r["name"],
                      "description": r["description"], "icon": r["icon"],
                      "is_task": bool(r["is_task"]) if "is_task" in r.keys() else False,
+                     "url": _url_projet(r["id"]),
                      "page_count": r["page_count"]}
                     for r in rows
                     if r["eff_ws"] is None or r["eff_ws"] in accessible]
@@ -182,7 +267,9 @@ def _appel_outil(nom: str, args: dict, user_id: int):
                 "ORDER BY position, id", (coll["id"],)).fetchall()
             return {"collection": {"id": coll["id"], "name": coll["name"],
                                     "description": coll["description"],
-                                    "icon": coll["icon"]},
+                                    "icon": coll["icon"],
+                                    "is_task": bool(coll["is_task"]) if "is_task" in coll.keys() else False,
+                                    "url": _url_projet(coll["id"])},
                     "pages": [_page_resume(p) for p in pages]}
         if nom == "get_page":
             p = conn.execute("SELECT * FROM collection_pages WHERE id = ?",
@@ -195,19 +282,181 @@ def _appel_outil(nom: str, args: dict, user_id: int):
                     res[cle] = p[cle]
             return res
         if nom == "search_pages":
-            q = f"%{str(args.get('query', '')).strip()}%"
-            if q == "%%":
+            # §7.2 : recherche plein texte — titres ET contenus, pages
+            # classiques comme lignes de collections, URL profonde et
+            # famille (« kind ») sur chaque résultat pour que read_page
+            # puisse être appelé sans ambiguïté d'identifiant.
+            terme = str(args.get("query", "")).strip()
+            if not terme:
                 return []
+            q = f"%{terme}%"
+            resultats: list[dict] = []
             rows = conn.execute(
-                "SELECT * FROM collection_pages WHERE title LIKE ? "
-                "ORDER BY updated_at DESC LIMIT 25", (q,)).fetchall()
-            return [_page_resume(p) for p in rows
-                    if collection_visible(p["collection_id"])]
+                "SELECT * FROM collection_pages "
+                "WHERE title LIKE ? OR property_values_json LIKE ? "
+                "ORDER BY updated_at DESC LIMIT 25", (q, q)).fetchall()
+            for p in rows:
+                if collection_visible(p["collection_id"]):
+                    resultats.append(_page_resume(p))
+            pages = conn.execute(
+                "SELECT id, title, workspace_id, updated_at FROM pages "
+                "WHERE deleted_at IS NULL AND (title LIKE ? OR content LIKE ?) "
+                "ORDER BY updated_at DESC LIMIT 25", (q, q)).fetchall()
+            for p in pages:
+                if p["workspace_id"] is None or p["workspace_id"] in accessible:
+                    resultats.append({
+                        "id": p["id"], "kind": "page", "title": p["title"],
+                        "workspace_id": p["workspace_id"],
+                        "url": _url_page(p["id"]),
+                        "updated_at": p["updated_at"]})
+            return resultats[:50]
         if nom == "list_repos":
             rows = conn.execute(
                 "SELECT id, name, proj_type, owner, language, description "
                 "FROM projects ORDER BY proj_type, name").fetchall()
             return [dict(r) for r in rows]
+        if nom == "list_projects":
+            # §7.2 : un « projet » du contrat de liens profonds est une
+            # collection. Même périmètre et même forme que
+            # list_collections, vocabulaire du catalogue cible.
+            rows = conn.execute(
+                f"SELECT c.*, (SELECT COUNT(*) FROM collection_pages p "
+                f"WHERE p.collection_id = c.id) AS page_count, "
+                f"{eff_sql} AS eff_ws "
+                "FROM collections c ORDER BY c.name").fetchall()
+            return [{"id": r["id"], "title": r["name"],
+                     "description": r["description"], "icon": r["icon"],
+                     "is_task": bool(r["is_task"]) if "is_task" in r.keys() else False,
+                     "page_count": r["page_count"],
+                     "url": _url_projet(r["id"]),
+                     "updated_at": r["updated_at"]}
+                    for r in rows
+                    if r["eff_ws"] is None or r["eff_ws"] in accessible]
+        if nom == "get_project":
+            coll = conn.execute("SELECT * FROM collections WHERE id = ?",
+                                (int(args.get("id", 0)),)).fetchone()
+            if coll is None or not collection_visible(coll["id"]):
+                return {"erreur": "projet introuvable"}
+            schema = _props(coll["schema_json"]) \
+                if coll["schema_json"] else []
+            titres = conn.execute(
+                "SELECT id, title FROM collection_pages "
+                "WHERE collection_id = ? ORDER BY position, id LIMIT 300",
+                (coll["id"],)).fetchall()
+            return {"project": {"id": coll["id"], "title": coll["name"],
+                                "description": coll["description"],
+                                "icon": coll["icon"],
+                                "is_task": bool(coll["is_task"]) if "is_task" in coll.keys() else False,
+                                "url": _url_projet(coll["id"]),
+                                "updated_at": coll["updated_at"]},
+                    "schema": schema,
+                    "pages": [{"id": t["id"], "title": t["title"]}
+                              for t in titres]}
+        if nom == "read_page":
+            try:
+                pid = int(args.get("id"))
+            except (TypeError, ValueError):
+                return {"erreur": "id requis (entier)"}
+            kind = str(args.get("kind") or "")
+            if kind in ("page", ""):
+                page = conn.execute(
+                    "SELECT * FROM pages WHERE id = ? AND deleted_at IS NULL",
+                    (pid,)).fetchone()
+                if page is not None and (
+                        page["workspace_id"] is None
+                        or page["workspace_id"] in accessible):
+                    contenu, tronque = _tronquer(page["content"] or "")
+                    return {"id": page["id"], "kind": "page",
+                            "title": page["title"],
+                            "workspace_id": page["workspace_id"],
+                            "content": contenu, "truncated": tronque,
+                            "url": _url_page(page["id"]),
+                            "updated_at": page["updated_at"]}
+                if kind == "page":
+                    return {"erreur": "page introuvable"}
+            if kind in ("collection_page", ""):
+                p = conn.execute(
+                    "SELECT * FROM collection_pages WHERE id = ?",
+                    (pid,)).fetchone()
+                if p is None or not collection_visible(p["collection_id"]):
+                    return {"erreur": "page introuvable"}
+                res = _page_resume(p)
+                liee = conn.execute(
+                    "SELECT content FROM pages "
+                    "WHERE collection_row_id = ? AND deleted_at IS NULL",
+                    (pid,)).fetchone()
+                if liee is not None:
+                    contenu, tronque = _tronquer(liee["content"] or "")
+                    res["content"] = contenu
+                    res["truncated"] = tronque
+                return res
+            return {"erreur": f"kind inconnu : {kind}"}
+        if nom == "list_project_tasks":
+            from app.services.task_databases import collect_tasks, is_done
+
+            projet_id = args.get("project_id")
+            if projet_id is not None:
+                try:
+                    projet_id = int(projet_id)
+                except (TypeError, ValueError):
+                    return {"erreur": "project_id doit être un entier"}
+                cible = conn.execute(
+                    "SELECT id, is_task FROM collections WHERE id = ?",
+                    (projet_id,)).fetchone()
+                if cible is None or not collection_visible(projet_id):
+                    return {"erreur": "projet introuvable"}
+                if "is_task" in cible.keys() and not cible["is_task"]:
+                    return {"erreur": "ce projet n'est pas une base de tâches"}
+            taches = collect_tasks(conn, user_id, mine_only=False)
+            sortie = []
+            for t in taches:
+                if projet_id is not None and t["collection_id"] != projet_id:
+                    continue
+                sortie.append({
+                    "id": t["id"], "title": t["title"],
+                    "status": t["status"], "done": t["done"],
+                    "due": t["due"], "assignees": t["assignees"],
+                    "project_id": t["collection_id"],
+                    "project_name": t["collection_name"],
+                    "url": _url_projet(t["collection_id"])})
+            if projet_id is not None and not sortie:
+                # Base de tâches non « configurée » (mapping My Tasks
+                # incomplet) : collect_tasks la saute par conception,
+                # mais le serveur MCP répond quand même — lignes brutes,
+                # statut lu sur la propriété mappée ou nommée Statut.
+                coll = conn.execute(
+                    "SELECT * FROM collections WHERE id = ?",
+                    (projet_id,)).fetchone()
+                prop_statut = None
+                if "task_status_prop" in coll.keys() and coll["task_status_prop"]:
+                    prop_statut = coll["task_status_prop"]
+                else:
+                    pr = conn.execute(
+                        "SELECT id FROM collection_properties "
+                        "WHERE collection_id = ? AND LOWER(name) IN "
+                        "('statut', 'status', 'état', 'etat') "
+                        "ORDER BY position LIMIT 1", (projet_id,)).fetchone()
+                    if pr is not None:
+                        prop_statut = pr["id"]
+                lignes = conn.execute(
+                    "SELECT * FROM collection_pages WHERE collection_id = ? "
+                    "ORDER BY position, id", (projet_id,)).fetchall()
+                for ligne in lignes:
+                    props = _props(ligne["property_values_json"])
+                    statut = ""
+                    if isinstance(props, dict) and prop_statut is not None:
+                        v = props.get(prop_statut, props.get(str(prop_statut), ""))
+                        if isinstance(v, list):
+                            v = v[0] if v else ""
+                        statut = str(v or "")
+                    sortie.append({
+                        "id": ligne["id"], "title": ligne["title"],
+                        "status": statut, "done": is_done(statut),
+                        "due": "", "assignees": [],
+                        "project_id": projet_id,
+                        "project_name": coll["name"],
+                        "url": _url_projet(projet_id)})
+            return sortie
         if nom == "create_page":
             try:
                 collection_id = int(args.get("collection_id"))
@@ -297,7 +546,8 @@ def _appel_outil(nom: str, args: dict, user_id: int):
             emettre("flowdeck.collection.created", f"collection:{new_id}", name,
                     url=url_collection(new_id), actor="agent:mcp")
             return {"id": coll["id"], "name": coll["name"],
-                    "description": coll["description"], "icon": coll["icon"]}
+                    "description": coll["description"], "icon": coll["icon"],
+                    "url": _url_projet(coll["id"])}
     return {"erreur": f"outil inconnu : {nom}"}
 
 
@@ -310,11 +560,45 @@ def _version() -> str:
         return "dev"
 
 
+def _scopes_jeton(token_id: int) -> str:
+    """Scopes du jeton MCP (colonne ``api_tokens.scopes``, défaut hérité
+    ``read,write`` posé par la migration v6.3.0)."""
+    try:
+        with get_conn() as conn:
+            row = conn.execute("SELECT scopes FROM api_tokens WHERE id = ?",
+                               (token_id,)).fetchone()
+        if row is not None and "scopes" in row.keys() and row["scopes"]:
+            return str(row["scopes"])
+    except Exception:  # noqa: BLE001 — base ancienne sans la colonne
+        pass
+    return "read,write"
+
+
+def _journaliser(proprietaire: dict, outil: str, arguments: dict,
+                 request: Request, statut: str) -> None:
+    """Journal d'audit §7.4 : qui (jeton), quel outil, quand — les
+    valeurs d'arguments ne sont pas journalisées, seules leurs clés."""
+    try:
+        from app.services.api_v2_helpers import audit_log
+
+        cles = ",".join(sorted(str(k) for k in (arguments or {}).keys()))
+        audit_log({"id": proprietaire["user_id"],
+                   "_token_id": proprietaire["id"]},
+                  f"mcp:{outil}", "mcp_tool", outil,
+                  f"statut={statut}; jeton={proprietaire.get('name', '')}; args=[{cles}]",
+                  request)
+    except Exception:  # noqa: BLE001 — l'audit ne casse jamais l'appel
+        pass
+
+
 @router.get("/mcp")
 def mcp_info():
+    noms = [t["name"] for t in OUTILS]
     return {"name": "flowdeck", "version": _version(),
             "transport": "streamable-http (JSON-RPC 2.0 over POST)",
-            "read_only": False, "tools": [t["name"] for t in OUTILS]}
+            "read_only": False, "tools": noms,
+            "read_tools": [n for n in noms if n not in OUTILS_ECRITURE],
+            "write_tools": [n for n in noms if n in OUTILS_ECRITURE]}
 
 
 @router.post("/mcp")
@@ -349,8 +633,25 @@ async def mcp(request: Request):
         return {"jsonrpc": "2.0", "id": rid, "result": {"tools": OUTILS}}
     if methode == "tools/call":
         params = data.get("params") or {}
-        res = _appel_outil(params.get("name", ""), params.get("arguments") or {},
-                           proprietaire["user_id"])
+        outil = params.get("name", "")
+        arguments = params.get("arguments") or {}
+        # §7.4 : la portée du jeton fait la loi — un jeton « read »
+        # (ex. clé dédiée à un pont de lecture) ne peut pas écrire.
+        if outil in OUTILS_ECRITURE:
+            from app.services.api_v2_helpers import has_scope
+
+            if not has_scope(_scopes_jeton(proprietaire["id"]), "write"):
+                _journaliser(proprietaire, outil, arguments, request,
+                             "refuse_portee")
+                return {"jsonrpc": "2.0", "id": rid,
+                        "error": {"code": -32002,
+                                  "message": "Jeton en lecture seule : "
+                                             "l'outil d'écriture "
+                                             f"« {outil} » exige le scope write"}}
+        res = _appel_outil(outil, arguments, proprietaire["user_id"])
+        _journaliser(proprietaire, outil, arguments, request,
+                     "erreur" if isinstance(res, dict) and "erreur" in res
+                     else "ok")
         return {"jsonrpc": "2.0", "id": rid,
                 "result": {"content": [{"type": "text",
                                         "text": json.dumps(res, ensure_ascii=False)}]}}

app/services/agent_engine.py

@@ -22,7 +22,7 @@ from app.services import agent_memory
 from app.services.agent_policies import check_tool, get_policy
 from app.services.context_builder import ContextBuilder
 from app.services.llm_client import LLMClient
-from app.services.permission_manager import WRITE_TOOLS, PermissionManager
+from app.services.permission_manager import PermissionManager, is_write_tool
 from app.services.tool_registry import ToolRegistry
 
 logger = logging.getLogger(__name__)
@@ -248,7 +248,7 @@ class AgentEngine:
                     # v7.2.0 — workspace tool scope + human approval gate, checked
                     # *before* permissions so a scoped-out tool never reaches ACLs.
                     gov = check_tool(self.user_id, self.workspace_id, tool,
-                                     is_write=tool in WRITE_TOOLS,
+                                     is_write=is_write_tool(tool),
                                      conversation_id=conversation_id)
                     if not gov.get("allowed"):
                         detail = gov.get("reason") or "Refusé par la politique agent"

app/services/connectors.py

@@ -24,7 +24,11 @@ from app.services.sso_provisioning import decrypt_secret, encrypt_secret
 logger = logging.getLogger(__name__)
 
 NATIVE_KINDS = ("gitea", "github", "web", "google", "ms365")
-CUSTOM_KINDS = ("custom", "discord", "telegram", "mcp", "tiktak")
+CUSTOM_KINDS = ("custom", "discord", "telegram", "mcp", "tiktak", "obsigate")
+#: Kinds personnels qui parlent MCP (probe = handshake + tools/list,
+#: outils dynamiques ``mcp_<serveur>_*``) — générique + les deux presets
+#: de fédération du programme (TikTak ACC-2.1, ObsiGate ACC-2.2).
+MCP_KINDS = ("mcp", "tiktak", "obsigate")
 AUTH_SCHEMES = ("bearer", "bot", "none")
 # Presets proposés par le formulaire du menu ＋ (le jeton reste toujours saisi
 # par l'utilisateur — jamais codé en dur ici).
@@ -45,6 +49,16 @@ def _presets() -> dict:
         # v7.82.0 : URL dérivée de la config (TIKTAK_URL ; défaut officiel).
         "tiktak": {"url": f"{draco_urls.tiktak_url()}/mcp", "auth": "bearer",
                    "hint": "TikTak → Réglages → Intégrations → clé API tck_…"},
+        # Fusion ACC-2.2 — pont MCP vers ObsiGate (plan « étape 1 :
+        # lectures seules », tests T0–T5) : serveur MCP d'ObsiGate sur
+        # /mcp (Streamable HTTP à session — le client gère le
+        # Mcp-Session-Id). La clé se crée dans ObsiGate →
+        # Configurations → Clés API & MCP, dédiée au pont et limitée
+        # à UN SEUL vault au départ : c'est le jeton qui fait la
+        # portée (garde-fou n° 1 du plan), pas ce preset.
+        "obsigate": {"url": f"{draco_urls.obsigate_url()}/mcp", "auth": "bearer",
+                     "hint": "ObsiGate → Configurations → Clés API & MCP → "
+                             "clé dédiée au pont, limitée à un seul vault"},
     }
 
 
@@ -293,7 +307,7 @@ async def probe(connector: str, user_id: int | None = None) -> dict:
     """Teste un connecteur et **persiste** le résultat (personnel uniquement)."""
     kind, ref = _resolve(connector)
     try:
-        if kind in ("mcp", "tiktak"):
+        if kind in MCP_KINDS:
             from app.services import mcp_client
             tools = await mcp_client.initialize_and_list(int(ref))
             status, detail = "ok", f"MCP · {len(tools)} outil(s)"
@@ -343,7 +357,7 @@ async def connector_fetch(connector: str, path: str = "", query: str = "",
     path = (path or "").strip()
     if kind in OAUTH_KINDS:
         return await oauth_connectors.api_get(kind, user_id, path)
-    if kind in ("mcp", "tiktak"):
+    if kind in MCP_KINDS:
         from app.services import mcp_client
         return {"status": "ok", "text": mcp_client.tools_text(int(ref))}
     if kind == "gitea":

app/services/draco_urls.py

@@ -18,15 +18,77 @@ Priorité de résolution, à chaque appel (les tests et les modules qui
 lisaient déjà ``os.environ`` au moment de l'appel gardent ce
 comportement) : variable d'environnement > ``app.config.settings`` >
 défaut officiel. Aucun défaut ne contient ``-dev``.
+
+**Dérivation d'environnement (v7.82.3)** — pour les quatre URL
+sœurs (ObsiGate, Flowdeck, TikTak, Kanbante), entre la configuration
+explicite et le défaut officiel : si l'origine de l'agent telle que
+l'app la connaît déjà (``AGENT_EVENEMENTS_URL``, via
+``agent_evenements_url()``) a un hôte de la forme
+``agent<SUFFIXE>.dracodev.net`` (SUFFIXE éventuellement vide,
+typiquement ``-dev``), l'URL sœur est dérivée :
+``https://<slug><SUFFIXE>.dracodev.net`` (slugs : ``og``,
+``flowdeck``, ``tiktak``, ``kanbante``). Ainsi une installation dev
+qui ne déclare que l'agent pointe ses liens inter-apps en ``-dev``
+sans quatre variables de plus. Si l'hôte de l'agent ne correspond
+pas au motif (autre domaine, LAN, IP), **aucune dérivation** : on
+ne fabrique jamais un lien deviné. L'URL Gitea ne change pas.
+
+**Résolution par l'hôte de la requête (v7.82.4)** — la v7.82.3
+laissait deux trous, prouvés en direct sur flowdeck-dev (CSP
+``frame-src 'self' https://agent.dracodev.net`` alors que la prod
+n'existe pas encore → console du panneau = page morte) :
+
+1. ``agent_url()`` n'avait **aucune** dérivation : même avec
+   ``AGENT_EVENEMENTS_URL`` en ``-dev``, l'agent lui-même retombait
+   sur le nom officiel ;
+2. **aucune** résolution ne lisait l'hôte de la requête : un
+   ``.env`` dev sans ``DRACO_AGENT_URL`` ni ``AGENT_EVENEMENTS_URL``
+   retombait entièrement sur la prod.
+
+Pour tout ce qui est **rendu à une requête** (liens de la barre /
+``DRACO_URLS`` servi, source de l'iframe du panneau, origine agent
+des CSP ``frame-src``/``connect-src``, boutons et redirections
+SSO), l'ordre de résolution devient :
+
+1. variable explicite de l'installation — prioritaire, inchangée ;
+2. **hôte public de la requête** (posé par le middleware CSP dans
+   ``HOTE_REQUETE``, même patron que le nonce CSP) : motif fermé
+   ``^(og|flowdeck|tiktak|kanbante|agent)(<SUFFIXE>)\\.dracodev\\.net$``
+   (suffixe optionnel, p. ex. ``-dev``, vide en prod) →
+   l'environnement = ce suffixe, appliqué à toutes les sœurs **et
+   à l'agent**. Jamais de suffixe inventé depuis un hôte arbitraire ;
+   hors motif (LAN, IP) → étape suivante ;
+3. dérivation par les adresses de l'agent connues en configuration
+   (URL d'événements **ou** base du service) — aussi la résolution
+   des contextes **sans requête**, inchangée sinon ;
+4. défaut officiel.
+
+L'iframe du panneau et la CSP de la même réponse consomment la
+même origine résolue : les deux lisent ``agent_url()`` pendant la
+même requête, avec le même ``HOTE_REQUETE``.
 """
 
 from __future__ import annotations
 
 import os
+import re
+from contextvars import ContextVar
 from urllib.parse import urlparse
 
 from app.config import settings
 
+# Hôte public de la requête en cours (``X-Forwarded-Host`` sinon
+# ``Host``), posé par le middleware CSP avant le traitement — même
+# patron que ``CSP_NONCE`` : le contexte est copié dans la tâche qui
+# rend les templates. Vide hors requête (tâches de fond, CLI, tests
+# unitaires) → la résolution retombe sur la configuration.
+HOTE_REQUETE: ContextVar[str] = ContextVar("draco_hote_requete", default="")
+
+
+def definir_hote_requete(hote: str) -> None:
+    """Mémorise l'hôte public de la requête (premier de la liste)."""
+    HOTE_REQUETE.set((hote or "").split(",")[0].strip().lower())
+
 OFFICIELLES = {
     "obsigate": "https://og.dracodev.net",
     "agent": "https://agent.dracodev.net",
@@ -51,22 +113,165 @@ def _resoudre(env_name: str, settings_attr: str, app: str) -> str:
     return OFFICIELLES[app]
 
 
+# Slugs d'hôte des applications sœurs (dérivation d'environnement).
+_SLUGS_SOEURS = {
+    "obsigate": "og",
+    "flowdeck": "flowdeck",
+    "tiktak": "tiktak",
+    "kanbante": "kanbante",
+}
+
+# Hôte de l'agent : agent<SUFFIXE>.dracodev.net (SUFFIXE = ``-dev``…).
+_MOTIF_HOTE_AGENT = re.compile(r"^agent(-[a-z0-9-]+)?\.dracodev\.net$")
+
+# Hôte public d'une requête : motif FERMÉ des cinq applications.
+# Le suffixe capturé (groupe 2, p. ex. ``-dev``) est l'environnement
+# de TOUTE l'installation — jamais un suffixe inventé depuis un
+# hôte arbitraire : hors motif (autre domaine, LAN, IP), None.
+_MOTIF_HOTE_PUBLIC = re.compile(
+    r"^(og|flowdeck|tiktak|kanbante|agent)(-[a-z0-9-]+)?\.dracodev\.net$")
+
+# Slugs pour la résolution par l'hôte de la requête (l'agent inclus).
+_SLUGS_TOUTES = {**_SLUGS_SOEURS, "agent": "agent"}
+
+
+def _suffixe_environnement() -> str | None:
+    """Suffixe d'environnement porté par l'hôte de l'agent, ou None.
+
+    Source : l'origine de l'agent telle que l'app la connaît déjà —
+    ``agent_evenements_url()`` (``AGENT_EVENEMENTS_URL`` explicite,
+    sinon dérivée de la base de l'agent). Retourne ``""`` pour
+    l'hôte officiel, ``"-dev"`` pour ``agent-dev.dracodev.net``, et
+    None si l'hôte ne correspond pas au motif (aucune dérivation).
+    """
+    hote_agent = (urlparse(agent_evenements_url()).hostname or "").lower()
+    m = _MOTIF_HOTE_AGENT.match(hote_agent)
+    if not m:
+        return None
+    return m.group(1) or ""
+
+
+def _url_derivee(app: str) -> str:
+    """URL sœur dérivée de l'environnement de l'agent ("" si aucune)."""
+    slug = _SLUGS_SOEURS.get(app)
+    if not slug:
+        return ""
+    suffixe = _suffixe_environnement()
+    if suffixe is None:
+        return ""
+    return f"https://{slug}{suffixe}.dracodev.net"
+
+
+def _suffixe_requete() -> str | None:
+    """Suffixe d'environnement porté par l'hôte de la requête, ou None.
+
+    Motif fermé ``_MOTIF_HOTE_PUBLIC`` sur l'hôte public mémorisé
+    (port retiré) : ``""`` pour un nom officiel, ``"-dev"`` pour un
+    nom en ``-dev``, None hors motif ou hors requête.
+    """
+    hote_req = HOTE_REQUETE.get().split(":")[0]
+    if not hote_req:
+        return None
+    m = _MOTIF_HOTE_PUBLIC.match(hote_req)
+    if not m:
+        return None
+    return m.group(2) or ""
+
+
+def _url_depuis_requete(app: str) -> str:
+    """URL d'``app`` dérivée de l'hôte de la requête ("" si aucune)."""
+    slug = _SLUGS_TOUTES.get(app)
+    if not slug:
+        return ""
+    suffixe = _suffixe_requete()
+    if suffixe is None:
+        return ""
+    return f"https://{slug}{suffixe}.dracodev.net"
+
+
+def _suffixe_evenements_explicite() -> str | None:
+    """Suffixe porté par l'URL d'événements **explicite**, ou None.
+
+    Lit ``AGENT_EVENEMENTS_URL`` (environnement puis settings) sans
+    passer par ``agent_evenements_url()`` : pas de récursion avec
+    ``agent_url()``, dont c'est une source de dérivation.
+    """
+    for candidate in (os.environ.get("AGENT_EVENEMENTS_URL", ""),
+                      getattr(settings, "agent_evenements_url", "") or ""):
+        candidate = _nettoyer(candidate)
+        if candidate:
+            hote_agent = (urlparse(candidate).hostname or "").lower()
+            m = _MOTIF_HOTE_AGENT.match(hote_agent)
+            return (m.group(1) or "") if m else None
+    return None
+
+
+def _resoudre_soeur(env_name: str, settings_attr: str, app: str) -> str:
+    """Résolution d'une URL sœur, dans l'ordre du chantier :
+
+    explicite > **hôte de la requête** > dérivation par l'agent
+    connu en configuration > officiel.
+
+    La valeur ``settings`` égale au nom officiel est le *défaut* de
+    ``app/config.py``, pas un choix explicite : elle ne doit pas
+    bloquer les dérivations. Une valeur settings différente (dev,
+    LAN, domaine propre) reste prioritaire.
+    """
+    env = _nettoyer(os.environ.get(env_name, ""))
+    if env:
+        return env
+    cfg = _nettoyer(getattr(settings, settings_attr, "") or "")
+    if cfg and cfg != OFFICIELLES[app]:
+        return cfg
+    depuis_requete = _url_depuis_requete(app)
+    if depuis_requete:
+        return depuis_requete
+    derivee = _url_derivee(app)
+    if derivee:
+        return derivee
+    if cfg:
+        return cfg
+    return OFFICIELLES[app]
+
+
 def obsigate_url() -> str:
     """ObsiGate — aussi l'émettrice SSO (DRACO_SSO_URL, déjà en service)."""
-    return _resoudre("DRACO_SSO_URL", "draco_sso_url", "obsigate")
+    return _resoudre_soeur("DRACO_SSO_URL", "draco_sso_url", "obsigate")
 
 
 def agent_url() -> str:
-    """Service d'agents (Spark) — base sans chemin."""
-    return _resoudre("DRACO_AGENT_URL", "draco_agent_url", "agent")
+    """Service d'agents (Spark) — base sans chemin.
+
+    Même ordre que les sœurs (la v7.82.3 l'avait laissé sans aucune
+    dérivation — trou n° 1 du mode console mort en dev) : explicite
+    (``DRACO_AGENT_URL`` / settings non officiel) > hôte de la
+    requête > dérivation par l'URL d'événements explicite
+    (``AGENT_EVENEMENTS_URL`` : ``agent-dev…`` → base ``agent-dev…``)
+    > défaut officiel.
+    """
+    env = _nettoyer(os.environ.get("DRACO_AGENT_URL", ""))
+    if env:
+        return env
+    cfg = _nettoyer(getattr(settings, "draco_agent_url", "") or "")
+    if cfg and cfg != OFFICIELLES["agent"]:
+        return cfg
+    depuis_requete = _url_depuis_requete("agent")
+    if depuis_requete:
+        return depuis_requete
+    suffixe = _suffixe_evenements_explicite()
+    if suffixe is not None:
+        return f"https://agent{suffixe}.dracodev.net"
+    if cfg:
+        return cfg
+    return OFFICIELLES["agent"]
 
 
 def tiktak_url() -> str:
-    return _resoudre("TIKTAK_URL", "tiktak_url", "tiktak")
+    return _resoudre_soeur("TIKTAK_URL", "tiktak_url", "tiktak")
 
 
 def kanbante_url() -> str:
-    return _resoudre("KANBANTE_URL", "kanbante_url", "kanbante")
+    return _resoudre_soeur("KANBANTE_URL", "kanbante_url", "kanbante")
 
 
 def gitea_url() -> str:
@@ -78,7 +283,8 @@ def flowdeck_url() -> str:
 
     FLOWDECK_URL > origine de OAUTH_REDIRECT_URI (déjà déclarée en
     dev : ``https://flowdeck-dev.dracodev.net/auth/callback``) >
-    APP_BASE_URL si elle n'est pas locale > nom officiel.
+    hôte de la requête (motif fermé) > APP_BASE_URL si elle n'est
+    pas locale > dérivation par l'agent > nom officiel.
     """
     env = _nettoyer(os.environ.get("FLOWDECK_URL", ""))
     if env:
@@ -95,9 +301,19 @@ def flowdeck_url() -> str:
             p = urlparse(candidate)
             if p.scheme and p.netloc:
                 return f"{p.scheme}://{p.netloc}"
+    # Étape 2 du chantier : l'hôte de la requête — c'est l'origine
+    # que le navigateur utilise déjà pour cette instance, donc
+    # aussi celle du callback SSO (``/auth/draco``) quand rien
+    # d'explicite n'est déclaré.
+    depuis_requete = _url_depuis_requete("flowdeck")
+    if depuis_requete:
+        return depuis_requete
     base = _nettoyer(getattr(settings, "app_base_url", "") or "")
     if base and "localhost" not in base and "127.0.0.1" not in base:
         return base
+    derivee = _url_derivee("flowdeck")
+    if derivee:
+        return derivee
     return OFFICIELLES["flowdeck"]
 
 
@@ -129,7 +345,12 @@ def hote(url: str) -> str:
 
 
 def toutes() -> dict[str, str]:
-    """Toutes les URL résolues — injectées aux templates (hôte = config)."""
+    """Toutes les URL résolues — injectées aux templates.
+
+    Résolution à l'appel : dans une requête, l'hôte public de la
+    requête fait foi après les variables explicites ; hors requête,
+    la configuration seule (voir le docstring du module).
+    """
     return {
         "obsigate": obsigate_url(),
         "agent": agent_url(),

app/services/mcp_client.py

@@ -6,10 +6,30 @@ JSON-RPC 2.0 : ``initialize`` → ``notifications/initialized`` → ``tools/list
 ``ToolRegistry`` sous des noms ``mcp_<serveur>_<outil>`` — outils dynamiques,
 sans état en mémoire.
 
-ponytail : **1 seul seam** ``_rpc()`` (monkeypatché en test) ; on lit soit la
-réponse JSON directe, soit la première ligne ``data:`` d'une
-``text/event-stream`` — pas de client SSE à l'état (les méthodes utilisées
-répondent en JSON chez la grande majorité des serveurs).
+**Sessions (ACC-2.2, pont ObsiGate — test T0 du plan d'étape 1).**
+Certains serveurs Streamable HTTP (ObsiGate) sont **à session** : ils
+renvoient un en-tête ``Mcp-Session-Id`` à ``initialize`` et refusent
+tout appel qui ne le réémet pas. Le client capture donc cet en-tête à
+chaque réponse et le réémet sur les appels suivants de la même
+opération ; ``call_tool`` fait désormais le handshake complet avant
+``tools/call`` (les serveurs sans état comme TikTak l'acceptent sans
+frais). Réponses **JSON ou SSE** : dans un flux ``text/event-stream``
+qui porte plusieurs messages (notifications + réponse), c'est le
+message **de réponse** (id correspondant, sinon porteur de
+``result``/``error``) qui est retenu — pas le premier ``data:`` venu.
+
+**Classification des outils (garde-fou n° 2 du plan du pont).** Chaque
+outil découvert est classé au moment du cache : **lecture par défaut** ;
+écrivant si le serveur le déclare (annotation ``readOnlyHint: false``)
+ou si son nom porte un verbe d'écriture (``create_``, ``propose_``,
+``apply_``…). La classe est consommée par les portes de gouvernance
+(``agent_policies.check_tool`` et ``PermissionManager`` via
+``permission_manager.is_write_tool``) : un outil MCP écrivant suit le
+même régime d'approbation et de rôles que les outils internes.
+
+ponytail : **1 seul seam** ``_rpc()`` (monkeypatché en test) ; pas de
+client SSE à l'état (les méthodes utilisées répondent en JSON ou en
+événements ponctuels chez les serveurs du programme).
 """
 
 from __future__ import annotations
@@ -19,7 +39,7 @@ import logging
 import re
 
 from app.db import get_conn
-from app.services.connectors import _headers, _with_secret, get
+from app.services.connectors import MCP_KINDS, _headers, _with_secret, get
 
 logger = logging.getLogger(__name__)
 
@@ -27,6 +47,36 @@ logger = logging.getLogger(__name__)
 PROTOCOL_VERSION = "2024-11-05"
 CLIENT_INFO = {"name": "FlowDeck", "version": "7"}
 
+#: Verbes d'écriture reconnus en tête d'un nom d'outil MCP quand le
+#: serveur ne déclare pas d'annotation ``readOnlyHint``. ``propose_`` /
+#: ``apply_`` sont le patron d'écriture approuvée d'ObsiGate : les deux
+#: sont classés écrivants (voir aussi le blocage d'``apply_`` dans
+#: ``tool_registry.McpTool``).
+_PREFIXES_ECRITURE = (
+    "create_", "update_", "delete_", "add_", "remove_", "set_", "move_",
+    "write_", "edit_", "rename_", "propose_", "apply_", "archive_",
+    "unarchive_", "send_", "publish_", "close_", "sync_", "trigger_",
+    "import_",
+)
+
+#: Derniers ``Mcp-Session-Id`` capturés, par URL de serveur. Mémoire de
+#: processus volontairement éphémère : une session perdue se rouvre par
+#: un handshake, jamais par une persistance.
+_sessions: dict[str, str] = {}
+
+
+def classer_outil(nom_original: str, annotations: dict | None = None) -> bool:
+    """True si l'outil MCP est **écrivant**.
+
+    L'annotation du serveur fait foi quand elle existe
+    (``readOnlyHint``) ; sinon, heuristique de nom. Lecture par défaut —
+    c'est la règle du plan du pont : un outil ne devient écrivant que
+    s'il est déclaré comme tel.
+    """
+    if isinstance(annotations, dict) and "readOnlyHint" in annotations:
+        return not bool(annotations["readOnlyHint"])
+    return (nom_original or "").lower().startswith(_PREFIXES_ECRITURE)
+
 
 def tool_name(server_name: str, tool_name: str) -> str:
     """Nom LLM stable : ``mcp_<serveur>_<outil>`` (slug minuscules/underscores)."""
@@ -36,14 +86,37 @@ def tool_name(server_name: str, tool_name: str) -> str:
     return re.sub(r"_{2,}", "_", joined)
 
 
-def parse_body(content_type: str, body: str, status: int) -> dict:
-    """Corps MCP → dict JSON (JSON direct ou événement ``data:`` d'un flux SSE)."""
+def parse_body(content_type: str, body: str, status: int,
+               expected_id=None) -> dict:
+    """Corps MCP → dict JSON (JSON direct ou événements ``data:`` SSE).
+
+    Dans un flux SSE multi-messages, le message retenu est celui dont
+    l'``id`` correspond à la requête (``expected_id``), sinon le
+    premier porteur de ``result``/``error``, sinon le premier message.
+    """
     text = body or ""
+    data = None
     if "text/event-stream" in (content_type or ""):
+        candidats: list = []
         for line in text.splitlines():
             if line.startswith("data:"):
-                text = line[5:].strip()
-                break
+                try:
+                    candidats.append(json.loads(line[5:].strip()))
+                except ValueError:
+                    continue
+        if expected_id is not None:
+            data = next((c for c in candidats
+                         if isinstance(c, dict) and c.get("id") == expected_id),
+                        None)
+        if data is None:
+            data = next((c for c in candidats
+                         if isinstance(c, dict)
+                         and ("result" in c or "error" in c)), None)
+        if data is None and candidats:
+            data = candidats[0]
+        if data is None:
+            raise ValueError(f"Réponse MCP illisible (HTTP {status})")
+    else:
         try:
             data = json.loads(text)
         except ValueError as exc:
@@ -54,6 +127,25 @@ def parse_body(content_type: str, body: str, status: int) -> dict:
     return data if isinstance(data, dict) else {}
 
 
+def _capturer_session(url: str, headers) -> None:
+    """Mémorise le ``Mcp-Session-Id`` d'une réponse, s'il est présent."""
+    try:
+        sid = headers.get("mcp-session-id") if headers else None
+    except AttributeError:
+        sid = None
+    if sid:
+        _sessions[url] = sid
+
+
+def _session_headers(url: str, headers: dict | None) -> dict:
+    """En-têtes d'appel + session connue pour cette URL (si capturée)."""
+    out = dict(headers or {})
+    sid = _sessions.get(url)
+    if sid and not any(k.lower() == "mcp-session-id" for k in out):
+        out["Mcp-Session-Id"] = sid
+    return out
+
+
 async def _rpc(url: str, payload: dict, headers: dict | None = None) -> dict:
     """POST JSON-RPC gardé SSRF → réponse décodée. Point d'injection des tests."""
     from app.services.http_client import shared_client
@@ -65,8 +157,9 @@ async def _rpc(url: str, payload: dict, headers: dict | None = None) -> dict:
             **(headers or {})}
     async with shared_client(timeout=15, headers=hdrs) as client:
         resp = await client.post(safe, json=payload)
+        _capturer_session(url, resp.headers)
         return parse_body(resp.headers.get("content-type", ""), resp.text or "",
-                          resp.status_code)
+                          resp.status_code, expected_id=payload.get("id"))
 
 
 async def _notify(url: str, payload: dict, headers: dict | None = None) -> None:
@@ -77,11 +170,30 @@ async def _notify(url: str, payload: dict, headers: dict | None = None) -> None:
         logger.debug("MCP notification ignorée: %s", exc)
 
 
+async def _handshake(url: str, headers: dict | None) -> dict:
+    """``initialize`` + ``notifications/initialized`` → en-têtes de session.
+
+    Obligatoire avant tout ``tools/call`` contre un serveur à session
+    (ObsiGate) ; sans effet de bord contre un serveur sans état. Les
+    en-têtes retournés portent le ``Mcp-Session-Id`` capturé pendant
+    ``initialize`` quand le serveur en a émis un.
+    """
+    await _rpc(url, {
+        "jsonrpc": "2.0", "id": 1, "method": "initialize",
+        "params": {"protocolVersion": PROTOCOL_VERSION, "capabilities": {},
+                   "clientInfo": CLIENT_INFO},
+    }, headers)
+    session = _session_headers(url, headers)
+    await _notify(url, {"jsonrpc": "2.0", "method": "notifications/initialized"},
+                  session)
+    return session
+
+
 def _server(connector_id: int) -> dict:
     row = get(connector_id)
     if row is None:
         raise ValueError("Serveur MCP introuvable")
-    if row["kind"] not in ("mcp", "tiktak"):
+    if row["kind"] not in MCP_KINDS:
         raise ValueError("Ce connecteur n'est pas un serveur MCP")
     if not row["enabled"]:
         raise ValueError("Serveur MCP désactivé")
@@ -102,6 +214,7 @@ def _normalize(server: dict, raw: list) -> list[dict]:
             "original": str(t["name"]),
             "description": str(t.get("description") or "Outil MCP"),
             "parameters": params,
+            "write": classer_outil(str(t["name"]), t.get("annotations")),
         })
     return out
 
@@ -117,15 +230,9 @@ async def initialize_and_list(connector_id: int) -> list[dict]:
     """Handshake + ``tools/list`` → met à jour le cache de la base."""
     srv = _server(connector_id)
     url = _with_secret(srv["url"], connector_id)
-    await _rpc(url, {
-        "jsonrpc": "2.0", "id": 1, "method": "initialize",
-        "params": {"protocolVersion": PROTOCOL_VERSION, "capabilities": {},
-                   "clientInfo": CLIENT_INFO},
-    }, srv["headers"])
-    await _notify(url, {"jsonrpc": "2.0", "method": "notifications/initialized"},
-                  srv["headers"])
+    session = await _handshake(url, srv["headers"])
     data = await _rpc(url, {"jsonrpc": "2.0", "id": 2, "method": "tools/list",
-                            "params": {}}, srv["headers"])
+                            "params": {}}, session)
     result = data.get("result") if isinstance(data.get("result"), dict) else {}
     tools = _normalize(srv, result.get("tools") or [])
     _save_tools(connector_id, tools)
@@ -137,7 +244,7 @@ def cached_tools() -> list[dict]:
     with get_conn() as conn:
         rows = conn.execute(
             "SELECT id, tools_json FROM agent_connectors "
-            "WHERE kind IN ('mcp', 'tiktak') AND enabled=1"
+            "WHERE kind IN ('mcp', 'tiktak', 'obsigate') AND enabled=1"
         ).fetchall()
     out: list[dict] = []
     for r in rows:
@@ -170,13 +277,18 @@ def _content_text(result: dict) -> str:
 
 
 async def call_tool(connector_id: int, tool: str, arguments: dict | None = None) -> dict:
-    """``tools/call`` → {status, text} (isError → error, borné à 20 000 car.)."""
+    """``tools/call`` → {status, text} (isError → error, borné à 20 000 car.).
+
+    Le handshake (avec session éventuelle) précède systématiquement
+    l'appel : c'est la condition T0 des serveurs à session.
+    """
     srv = _server(connector_id)
     url = _with_secret(srv["url"], connector_id)
+    session = await _handshake(url, srv["headers"])
     data = await _rpc(url, {
         "jsonrpc": "2.0", "id": 3, "method": "tools/call",
         "params": {"name": tool, "arguments": arguments or {}},
-    }, srv["headers"])
+    }, session)
     result = data.get("result") if isinstance(data.get("result"), dict) else {}
     text = _content_text(result)[:20000]
     return {"status": "error" if result.get("isError") else "ok", "text": text}

app/services/permission_manager.py

@@ -55,6 +55,28 @@ DESTRUCTIVE_TOOLS = {
 }
 
 
+def is_write_tool(tool: str) -> bool:
+    """L'outil est-il écrivant ? Outils internes (``WRITE_TOOLS``) et
+    outils **MCP dynamiques** classés écrivants à la découverte
+    (``mcp_client.classer_outil`` — garde-fou n° 2 du pont ObsiGate,
+    ACC-2.2 : avant, un ``mcp_*`` passait toujours pour une lecture,
+    donc un ``apply_*`` distant échappait aux deux portes).
+    """
+    if tool in WRITE_TOOLS:
+        return True
+    if tool.startswith("mcp_"):
+        try:
+            from app.services import mcp_client
+
+            for entry in mcp_client.cached_tools():
+                if entry.get("name") == tool:
+                    return bool(entry.get("write"))
+        except Exception:  # noqa: BLE001 — la classification ne casse rien
+            logger.debug("is_write_tool: classification MCP indisponible",
+                         exc_info=True)
+    return False
+
+
 class PermissionManager:
     """Resolves workspace role and gates agent + granular ACL checks."""
 
@@ -186,7 +208,7 @@ class PermissionManager:
         - destructive tools: admin/owner, or requires confirm approval mode.
         """
         role = self.role_in_workspace(workspace_id)
-        if tool in WRITE_TOOLS and role not in WRITE_ROLES:
+        if is_write_tool(tool) and role not in WRITE_ROLES:
             raise HTTPException(
                 status_code=403,
                 detail=f"Agent tool '{tool}' requires editor+ role (user is '{role}')",

app/services/tool_registry.py

@@ -1230,9 +1230,23 @@ class McpTool(Tool):
     async def execute(self, args, *, user_id=None) -> ToolResult:
         from app.services import mcp_client
 
+        # Garde-fou du pont (ACC-2.2, patron §7.5) : un outil distant
+        # « apply_* » EXÉCUTE une écriture croisée — il n'est jamais
+        # utilisable par un simple appel d'outil depuis un agent
+        # Flowdeck. L'écriture passe par « propose_* » (classé
+        # écrivant : rôles + approbation) puis par l'approbation dans
+        # l'application propriétaire de la donnée.
+        original = self.entry.get("original") or ""
+        if original.lower().startswith("apply_"):
+            return ToolResult(
+                status="error", tool=self.name, target_type="mcp",
+                message=(f"Outil « {original} » bloqué côté client : une "
+                         "écriture croisée ne s'exécute pas par un appel "
+                         "d'outil — propose_* puis approbation dans "
+                         "l'application propriétaire (patron §7.5)."))
         try:
             res = await mcp_client.call_tool(self.entry["connector_id"],
-                                             self.entry.get("original") or "", args)
+                                             original, args)
         except ValueError as exc:
             return ToolResult(status="error", tool=self.name, message=str(exc))
         except Exception as exc:  # noqa: BLE001 — réseau : erreur outil, pas run

app/templates/agent_panel.html

@@ -452,6 +452,7 @@ body.fd-ap-resizing *{cursor:col-resize!important}
             <option value="telegram">Telegram (bot)</option>
             <option value="mcp">Serveur MCP (streamable HTTP)</option>
             <option value="tiktak">TikTak — tâches (MCP, Fusion)</option>
+            <option value="obsigate">ObsiGate — notes (MCP, Fusion, pont lecture)</option>
           </select>
           <label for="fd-plus-cn-name">Nom</label>
           <input id="fd-plus-cn-name" type="text" x-model="connectorForm.name" placeholder="ex. Base de connaissances" @keydown.enter.prevent="connectorCreate()">
@@ -533,5 +534,5 @@ body.fd-ap-resizing *{cursor:col-resize!important}
   </div>
 </div>
 
-<script nonce="{{ csp_nonce() }}">window.FLOWDECK_TIKTAK_MCP_URL = {{ (draco_urls().tiktak + "/mcp") | tojson }};</script>
+<script nonce="{{ csp_nonce() }}">window.FLOWDECK_TIKTAK_MCP_URL = {{ (draco_urls().tiktak + "/mcp") | tojson }};window.FLOWDECK_OBSIGATE_MCP_URL = {{ (draco_urls().obsigate + "/mcp") | tojson }};</script>
 <script data-cfasync="false" nonce="{{ csp_nonce() }}" src="/static/js/agent_panel_2.js?v={{ asset_version }}"></script>

static/js/agent_panel_2.js

@@ -1425,7 +1425,9 @@
                     hint:'BotFather → Bot Token ({secret} complète l\u2019URL)'},
           mcp:     {url:'', auth:'bearer', hint:'URL du endpoint MCP (streamable HTTP)'},
           tiktak:  {url:(window.FLOWDECK_TIKTAK_MCP_URL || 'https://tiktak.dracodev.net/mcp'), auth:'bearer', name:'TikTak',
-                    hint:'TikTak → Réglages → Intégrations → clé API tck_…'}
+                    hint:'TikTak → Réglages → Intégrations → clé API tck_…'},
+          obsigate:{url:(window.FLOWDECK_OBSIGATE_MCP_URL || 'https://og.dracodev.net/mcp'), auth:'bearer', name:'ObsiGate',
+                    hint:'ObsiGate → Configurations → Clés API & MCP → clé dédiée au pont, limitée à un seul vault'}
         };
         var p = presets[f.kind] || presets.custom;
         f.url = p.url; f.auth = p.auth; f.hint = p.hint;

static/js/draco-panneau.js

@@ -46,6 +46,7 @@
   var etatConnu = false;
   var root = null, msgsEl = null, etatEl = null, inputEl = null, sendBtn = null;
   var titreEl = null, tiroirEl = null;
+  var versionEl = null, versionConnue = "", versionMaj = 0;
 
   function domaine() {
     try { return localStorage.getItem("draco_domain") || "pro"; } catch (e) { return "pro"; }
@@ -274,6 +275,7 @@
       '  <input type="text" placeholder="Écrire à Spark…" autocomplete="off" aria-label="Message à Spark">' +
       '  <button type="submit">Envoyer</button></form>' +
       '  <div class="draco-panneau-note">Conversation réelle, sous-agents (J3) et missions longues (J5) vers les exécuteurs configurés ; toute écriture dans les apps passe par une approbation.</div>' +
+      '  <div class="draco-panneau-version" style="margin-top:6px;font-size:11.5px;opacity:.6" hidden></div>' +
       '</div>';
     document.body.appendChild(root);
     msgsEl = root.querySelector(".draco-panneau-msgs");
@@ -282,6 +284,7 @@
     sendBtn = root.querySelector("button[type=submit]");
     titreEl = root.querySelector(".d-titre");
     tiroirEl = root.querySelector(".draco-discussions");
+    versionEl = root.querySelector(".draco-panneau-version");
     root.querySelector(".fermer").addEventListener("click", fermer);
     root.querySelector('[data-disc="liste"]').addEventListener("click", basculerTiroir);
     root.querySelector('[data-disc="nouvelle"]').addEventListener("click", nouvelleDiscussion);
@@ -336,6 +339,35 @@
     titreEl.title = titreEl.textContent;
   }
 
+  /* ── Version du service : lue en direct dans la santé de
+   * l'agent (GET /healthz → champ « version », la même source que
+   * la console) à l'origine que le panneau connaît déjà. Cache de
+   * session (5 min), rafraîchie à l'ouverture du panneau, au même
+   * rythme que la vérification d'identité. Jamais de valeur codée
+   * en dur : source injoignable ou réponse illisible = la ligne
+   * reste masquée — pas de fausse version, pas d'erreur visible. */
+  function afficherVersion() {
+    if (!versionEl) return;
+    if (versionConnue) {
+      versionEl.textContent = "Service d'agents v" + versionConnue;
+      versionEl.hidden = false;
+    }
+  }
+  function chargerVersion() {
+    if (!API || !versionEl) return;
+    if (versionConnue && Date.now() - versionMaj < 300000) { afficherVersion(); return; }
+    fetch(API + "/healthz")
+      .then(function (r) { if (!r.ok) throw new Error("HTTP " + r.status); return r.json(); })
+      .then(function (d) {
+        if (d && typeof d.version === "string" && /^[0-9]+\.[0-9]+\.[0-9]+/.test(d.version)) {
+          versionConnue = d.version;
+          versionMaj = Date.now();
+          afficherVersion();
+        }
+      })
+      .catch(function () { /* service injoignable : rien d'affiché */ });
+  }
+
   function verifierIdentite() {
     if (etatConnu) return;
     fetch(API + "/api/v1/me", { credentials: "include" })
@@ -812,6 +844,7 @@
     root.classList.add("open");
     appliquerTheme();
     verifierIdentite();
+    chargerVersion();
     setTimeout(function () { inputEl.focus(); }, 300);
   }
   function fermer() { if (root) root.classList.remove("open"); }

tests/test_agent_web_tools.py

@@ -24,6 +24,7 @@ import contextlib
 import ipaddress
 import json
 import socket
+import sys
 
 import httpx
 import pytest
@@ -108,6 +109,14 @@ def mock_http(handler):
         from app.services.importers import url_fetch
         if getattr(url_fetch, "shared_client", None) is build:
             url_fetch.shared_client = _REAL_SHARED_CLIENT
+        # Même fuite via og_fetcher : url_fetch l'importe paresseusement
+        # (parse_og) PENDANT la fenêtre, et son `from … import shared_client`
+        # fige alors `build` — le handler du test courant répondait ensuite
+        # aux fetch_og_metadata des tests suivants du même worker xdist
+        # (flake test_v55 : titre « Titre de la page » au lieu du sien).
+        og_fetcher = sys.modules.get("app.services.og_fetcher")
+        if og_fetcher is not None and getattr(og_fetcher, "shared_client", None) is build:
+            og_fetcher.shared_client = _REAL_SHARED_CLIENT
 
 
 def run(coro):

tests/test_draco_urls_devprod.py

@@ -110,6 +110,72 @@ def test_preset_tiktak_depuis_config(monkeypatch):
         "https://tiktak-dev.dracodev.net/mcp"
 
 
+def test_derivation_dev_depuis_agent_evenements_seul(monkeypatch):
+    # Cas réel de la dev : seul AGENT_EVENEMENTS_URL est posé —
+    # les quatre URLs sœurs servies suivent en ``-dev``, sans
+    # déclarer quatre variables de plus. Gitea ne change pas.
+    monkeypatch.setenv(
+        "AGENT_EVENEMENTS_URL",
+        "https://agent-dev.dracodev.net/api/v1/evenements")
+    urls = draco_urls.toutes()
+    assert urls["obsigate"] == "https://og-dev.dracodev.net"
+    assert urls["flowdeck"] == "https://flowdeck-dev.dracodev.net"
+    assert urls["tiktak"] == "https://tiktak-dev.dracodev.net"
+    assert urls["kanbante"] == "https://kanbante-dev.dracodev.net"
+    assert urls["gitea"] == "https://git.dracodev.net"
+    # Les accesseurs individuels partagent la même résolution.
+    assert draco_urls.obsigate_url() == "https://og-dev.dracodev.net"
+    assert draco_urls.flowdeck_url() == "https://flowdeck-dev.dracodev.net"
+    assert draco_urls.tiktak_url() == "https://tiktak-dev.dracodev.net"
+    assert draco_urls.kanbante_url() == "https://kanbante-dev.dracodev.net"
+
+
+def test_derivation_explicite_gagne_sur_derivation(monkeypatch):
+    monkeypatch.setenv(
+        "AGENT_EVENEMENTS_URL",
+        "https://agent-dev.dracodev.net/api/v1/evenements")
+    # Explicites (contrat inchangé) : elles priment sur la dérivation,
+    # y compris quand elles pointent le nom officiel.
+    monkeypatch.setenv("DRACO_SSO_URL", "https://og.dracodev.net")
+    monkeypatch.setenv("TIKTAK_URL", "https://tiktak.example.net")
+    urls = draco_urls.toutes()
+    assert urls["obsigate"] == "https://og.dracodev.net"
+    assert urls["tiktak"] == "https://tiktak.example.net"
+    # Non déclarées : dérivées.
+    assert urls["flowdeck"] == "https://flowdeck-dev.dracodev.net"
+    assert urls["kanbante"] == "https://kanbante-dev.dracodev.net"
+
+
+def test_derivation_hote_agent_hors_motif_defauts_officiels(monkeypatch):
+    # Autre domaine, LAN, IP : aucune dérivation, jamais un lien deviné.
+    for url_agent in (
+        "https://agent.example.com/api/v1/evenements",
+        "https://agent-dev.example.net/api/v1/evenements",
+        "https://monagent-dev.dracodev.net/api/v1/evenements",
+        "https://agent-dev.dracodev.net.evil.com/api/v1/evenements",
+        "http://192.168.30.101:8790/api/v1/evenements",
+        "http://127.0.0.1:8790/api/v1/evenements",
+        "http://agent-dev.local:8790/api/v1/evenements",
+    ):
+        monkeypatch.setenv("AGENT_EVENEMENTS_URL", url_agent)
+        urls = draco_urls.toutes()
+        assert urls["obsigate"] == "https://og.dracodev.net", url_agent
+        assert urls["flowdeck"] == "https://flowdeck.dracodev.net", url_agent
+        assert urls["tiktak"] == "https://tiktak.dracodev.net", url_agent
+        assert urls["kanbante"] == "https://kanbante.dracodev.net", url_agent
+
+
+def test_derivation_agent_officiel_suffixe_vide(monkeypatch):
+    # Hôte agent officiel : suffixe vide, les sœurs restent officielles.
+    monkeypatch.setenv(
+        "AGENT_EVENEMENTS_URL", "https://agent.dracodev.net/api/v1/evenements")
+    urls = draco_urls.toutes()
+    assert urls["obsigate"] == "https://og.dracodev.net"
+    assert urls["flowdeck"] == "https://flowdeck.dracodev.net"
+    assert urls["tiktak"] == "https://tiktak.dracodev.net"
+    assert urls["kanbante"] == "https://kanbante.dracodev.net"
+
+
 def test_bouton_sso_login_depuis_config(monkeypatch):
     from app.routers.auth import _sso_login_url
 
@@ -122,3 +188,185 @@ def test_bouton_sso_login_depuis_config(monkeypatch):
     assert _sso_login_url() == (
         "https://og-dev.dracodev.net/sso/authorize?return_to="
         "https%3A%2F%2Fflowdeck-dev.dracodev.net%2Fauth%2Fdraco")
+
+
+# ── v7.82.4 — résolution par l'hôte de la requête ─────────────────────
+# Correctif du mode console mort sur flowdeck-dev : l'agent n'avait
+# aucune dérivation (v7.82.3) et rien ne lisait l'hôte de la requête.
+# Ordre : explicite > hôte public de la requête (motif fermé) >
+# dérivation par l'agent connu en configuration > officiel.
+
+@pytest.fixture
+def avec_hote():
+    """Pose l'hôte de requête comme le middleware CSP le fait."""
+    jetons = []
+
+    def poser(hote):
+        jetons.append(draco_urls.HOTE_REQUETE.set(hote))
+
+    yield poser
+    for jeton in reversed(jetons):
+        draco_urls.HOTE_REQUETE.reset(jeton)
+
+
+def test_agent_url_derivee_depuis_evenements_seul(monkeypatch):
+    # Trou n° 1 de la v7.82.3 : seul AGENT_EVENEMENTS_URL posé (cas
+    # réel de la dev) → l'agent lui-même suit en ``-dev``, pas
+    # seulement les sœurs.
+    monkeypatch.setenv(
+        "AGENT_EVENEMENTS_URL",
+        "https://agent-dev.dracodev.net/api/v1/evenements")
+    assert draco_urls.agent_url() == "https://agent-dev.dracodev.net"
+    assert draco_urls.agent_evenements_url() == \
+        "https://agent-dev.dracodev.net/api/v1/evenements"
+
+
+def test_hote_requete_dev_toutes_urls_sans_env(avec_hote):
+    # Aucune variable d'environnement (fixture _env_propre) : l'hôte
+    # public suffit — sœurs ET agent en ``-dev``, Gitea inchangée.
+    avec_hote("flowdeck-dev.dracodev.net")
+    urls = draco_urls.toutes()
+    assert urls["obsigate"] == "https://og-dev.dracodev.net"
+    assert urls["agent"] == "https://agent-dev.dracodev.net"
+    assert urls["tiktak"] == "https://tiktak-dev.dracodev.net"
+    assert urls["kanbante"] == "https://kanbante-dev.dracodev.net"
+    assert urls["flowdeck"] == "https://flowdeck-dev.dracodev.net"
+    assert urls["gitea"] == "https://git.dracodev.net"
+
+
+def test_hote_requete_quelconque_des_cinq_slugs(avec_hote):
+    # L'environnement est celui de l'installation, pas de l'app
+    # demandée : un hôte ``agent-dev`` ou ``og-dev`` vaut ``-dev``
+    # pour toutes, port éventuel ignoré.
+    for hote in ("agent-dev.dracodev.net", "og-dev.dracodev.net:443",
+                 "tiktak-dev.dracodev.net", "kanbante-dev.dracodev.net"):
+        avec_hote(hote)
+        assert draco_urls.agent_url() == "https://agent-dev.dracodev.net", hote
+        assert draco_urls.obsigate_url() == "https://og-dev.dracodev.net", hote
+
+
+def test_hote_requete_prod_noms_officiels(avec_hote):
+    avec_hote("flowdeck.dracodev.net")
+    urls = draco_urls.toutes()
+    assert urls["obsigate"] == "https://og.dracodev.net"
+    assert urls["agent"] == "https://agent.dracodev.net"
+    assert urls["flowdeck"] == "https://flowdeck.dracodev.net"
+    assert all("-dev" not in u for u in urls.values())
+
+
+def test_hote_requete_motif_ferme_aucune_derivation(avec_hote):
+    # Jamais de suffixe inventé depuis un hôte arbitraire.
+    for hote in ("evil-dev.dracodev.net",
+                 "flowdeck-dev.dracodev.net.evil.com",
+                 "monflowdeck-dev.dracodev.net",
+                 "flowdeckdev.dracodev.net",
+                 "flowdeck-dev.example.net"):
+        avec_hote(hote)
+        urls = draco_urls.toutes()
+        assert urls["agent"] == "https://agent.dracodev.net", hote
+        assert urls["obsigate"] == "https://og.dracodev.net", hote
+        assert urls["flowdeck"] == "https://flowdeck.dracodev.net", hote
+
+
+def test_hote_requete_lan_repli_configuration(avec_hote, monkeypatch):
+    # Hors motif (LAN, IP) : repli configuration/défauts, comme
+    # hors requête — ici la dérivation par AGENT_EVENEMENTS_URL.
+    avec_hote("192.168.30.101:8080")
+    urls = draco_urls.toutes()
+    assert urls["agent"] == "https://agent.dracodev.net"
+    assert urls["obsigate"] == "https://og.dracodev.net"
+    monkeypatch.setenv(
+        "AGENT_EVENEMENTS_URL",
+        "https://agent-dev.dracodev.net/api/v1/evenements")
+    urls = draco_urls.toutes()
+    assert urls["agent"] == "https://agent-dev.dracodev.net"
+    assert urls["obsigate"] == "https://og-dev.dracodev.net"
+    assert urls["tiktak"] == "https://tiktak-dev.dracodev.net"
+
+
+def test_hote_requete_explicite_gagne_sur_hote(avec_hote, monkeypatch):
+    avec_hote("flowdeck-dev.dracodev.net")
+    # Explicites prioritaires, y compris quand elles pointent le
+    # nom officiel ou un domaine propre.
+    monkeypatch.setenv("DRACO_AGENT_URL", "https://agent.dracodev.net")
+    monkeypatch.setenv("DRACO_SSO_URL", "https://og.example.net")
+    urls = draco_urls.toutes()
+    assert urls["agent"] == "https://agent.dracodev.net"
+    assert urls["obsigate"] == "https://og.example.net"
+    # Non déclarées : l'hôte de la requête fait foi.
+    assert urls["tiktak"] == "https://tiktak-dev.dracodev.net"
+    assert urls["kanbante"] == "https://kanbante-dev.dracodev.net"
+    assert urls["flowdeck"] == "https://flowdeck-dev.dracodev.net"
+
+
+def test_hote_requete_bouton_sso_tout_dev(avec_hote):
+    from app.routers.auth import _sso_login_url
+
+    avec_hote("flowdeck-dev.dracodev.net")
+    # Émettrice via DRACO_SSO_URL résolu par l'hôte + callback =
+    # propre origine résolue de l'instance.
+    assert _sso_login_url() == (
+        "https://og-dev.dracodev.net/sso/authorize?return_to="
+        "https%3A%2F%2Fflowdeck-dev.dracodev.net%2Fauth%2Fdraco")
+
+
+def test_http_dev_sans_env_tout_en_dev(client):
+    # Requête complète, aucune variable d'environnement : barre,
+    # panneau (config de l'iframe console), CSP et SSO en ``-dev``.
+    r = client.get("/workspaces",
+                   headers={"Host": "flowdeck-dev.dracodev.net"})
+    assert r.status_code == 200
+    html = r.text
+    # Barre inter-apps (les quatre sœurs).
+    for soeur in ("og-dev", "flowdeck-dev", "tiktak-dev", "kanbante-dev"):
+        assert f'href="https://{soeur}.dracodev.net/"' in html
+    # Panneau : la source de l'iframe console vient de cette config.
+    assert 'agent_url: "https://agent-dev.dracodev.net"' in html
+    # CSP de la MÊME réponse : même origine agent résolue.
+    csp = r.headers.get("content-security-policy", "")
+    assert "frame-src 'self' https://agent-dev.dracodev.net" in csp
+    assert "connect-src 'self' https://agent-dev.dracodev.net" in csp
+    # Aucun nom officiel nu ne subsiste dans la réponse.
+    for officiel in ("https://og.dracodev.net", "https://agent.dracodev.net",
+                     "https://tiktak.dracodev.net",
+                     "https://kanbante.dracodev.net",
+                     "https://flowdeck.dracodev.net"):
+        assert officiel not in html.replace("-dev.dracodev.net", ""), officiel
+        assert officiel not in csp.replace("-dev.dracodev.net", ""), officiel
+    # Bouton SSO de la page de connexion, même requête type.
+    r = client.get("/auth/login?provider=local",
+                   headers={"Host": "flowdeck-dev.dracodev.net"})
+    assert ("https://og-dev.dracodev.net/sso/authorize?return_to="
+            "https%3A%2F%2Fflowdeck-dev.dracodev.net%2Fauth%2Fdraco") in r.text
+
+
+def test_http_prod_noms_officiels(client):
+    r = client.get("/workspaces",
+                   headers={"Host": "flowdeck.dracodev.net"})
+    assert r.status_code == 200
+    assert 'agent_url: "https://agent.dracodev.net"' in r.text
+    assert 'href="https://og.dracodev.net/"' in r.text
+    csp = r.headers.get("content-security-policy", "")
+    assert "frame-src 'self' https://agent.dracodev.net" in csp
+    assert "-dev.dracodev.net" not in r.text
+    assert "-dev.dracodev.net" not in csp
+
+
+def test_http_lan_repli_defauts(client):
+    r = client.get("/workspaces",
+                   headers={"Host": "192.168.30.101:8080"})
+    assert r.status_code == 200
+    assert 'agent_url: "https://agent.dracodev.net"' in r.text
+    csp = r.headers.get("content-security-policy", "")
+    assert "frame-src 'self' https://agent.dracodev.net" in csp
+
+
+def test_http_explicite_gagne_sur_hote(client, monkeypatch):
+    monkeypatch.setenv("DRACO_AGENT_URL", "https://agent.dracodev.net")
+    r = client.get("/workspaces",
+                   headers={"Host": "flowdeck-dev.dracodev.net"})
+    assert 'agent_url: "https://agent.dracodev.net"' in r.text
+    csp = r.headers.get("content-security-policy", "")
+    assert "frame-src 'self' https://agent.dracodev.net;" in csp
+    # Le reste suit l'hôte de la requête.
+    assert 'href="https://og-dev.dracodev.net/"' in r.text

tests/test_fusion_mcp_server.py

@@ -70,8 +70,9 @@ def test_mcp_handshake_et_outils(client, jeton):
     noms = [t["name"] for t in r.json()["result"]["tools"]]
     assert noms == ["list_workspaces", "get_workspace",
                     "list_collections", "get_collection", "get_page",
-                    "search_pages", "list_repos", "create_page",
-                    "update_page", "create_collection"]
+                    "search_pages", "list_repos", "list_projects",
+                    "get_project", "read_page", "list_project_tasks",
+                    "create_page", "update_page", "create_collection"]
 
 
 def test_mcp_workspaces(client, jeton):
@@ -107,7 +108,13 @@ def test_mcp_collections_et_pages(client, jeton, donnees):
     assert titres == ["Page de cadrage", "Revue hebdo"]
     assert detail["pages"][0]["properties"] == {"statut": "en cours"}
     res = _appel(client, jeton, 3, "search_pages", {"query": "cadrage"})
-    assert [p["title"] for p in res] == ["Page de cadrage"]
+    # ACC-5.1 : la recherche est plein texte et couvre aussi les pages
+    # classiques — on vérifie la ligne de collection, sa famille et son
+    # URL profonde (§7.2 : l'URL est systématique).
+    lignes = [p for p in res if p["kind"] == "collection_page"]
+    assert [p["title"] for p in lignes] == ["Page de cadrage"]
+    assert all(p["url"].endswith(f"/collections/{donnees}") for p in lignes)
+    assert all("url" in p and "kind" in p for p in res)
     res = _appel(client, jeton, 4, "get_collection", {"id": 99999})
     assert "introuvable" in res["erreur"]
 
@@ -237,3 +244,160 @@ def test_mcp_info_get(client, jeton):
     assert "create_page" in infos["tools"]
     assert "update_page" in infos["tools"]
     assert "create_collection" in infos["tools"]
+    # ACC-5.1 : le catalogue distingue lectures et écritures.
+    assert "list_projects" in infos["read_tools"]
+    assert "read_page" in infos["read_tools"]
+    assert infos["write_tools"] == ["create_page", "update_page",
+                                     "create_collection"]
+
+
+# ── ACC-5.1 : catalogue de lecture §7.2 ────────────────────────────────────
+
+
+@pytest.fixture()
+def projet_taches(client):
+    """Un workspace + une base de tâches configurée (mapping complet)."""
+    with get_conn() as conn:
+        conn.execute("INSERT OR IGNORE INTO users (id, login) VALUES (1, 'tester')")
+        cur = conn.execute(
+            "INSERT INTO workspaces (name, owner_id) VALUES ('Atelier MCP', 1)")
+        wid = cur.lastrowid
+        cur = conn.execute(
+            "INSERT INTO collections (name, description, icon, workspace_id, is_task) "
+            "VALUES ('Suivi projet', 'Base de tâches', '🗂️', ?, 1)", (wid,))
+        cid = cur.lastrowid
+        props = {}
+        for nom, typ, opts in (
+                ("Assigné à", "person", "[]"),
+                ("Statut", "status", json.dumps(
+                    [{"name": "À faire"}, {"name": "Terminé"}])),
+                ("Échéance", "date", "[]")):
+            cur = conn.execute(
+                "INSERT INTO collection_properties "
+                "(collection_id, name, prop_type, options_json) VALUES (?, ?, ?, ?)",
+                (cid, nom, typ, opts))
+            props[nom] = cur.lastrowid
+        conn.execute(
+            "UPDATE collections SET task_assignee_prop=?, task_status_prop=?, "
+            "task_due_prop=? WHERE id=?",
+            (props["Assigné à"], props["Statut"], props["Échéance"], cid))
+        conn.execute(
+            "INSERT INTO collection_pages (collection_id, title, icon, position, "
+            "property_values_json) VALUES (?, 'Relire le pont MCP', 'file', 1, ?)",
+            (cid, json.dumps({str(props["Statut"]): "À faire",
+                              str(props["Échéance"]): "2026-10-30"})))
+        conn.execute(
+            "INSERT INTO collection_pages (collection_id, title, icon, position, "
+            "property_values_json) VALUES (?, 'Brancher ObsiGate', 'file', 2, ?)",
+            (cid, json.dumps({str(props["Statut"]): "Terminé"})))
+        conn.commit()
+    return {"workspace_id": wid, "collection_id": cid, "props": props}
+
+
+def test_mcp_projets_catalogue(client, jeton, donnees, projet_taches):
+    projets = _appel(client, jeton, 90, "list_projects")
+    alpha = [p for p in projets if p["title"] == "Projet Alpha"]
+    assert alpha and alpha[0]["id"] == donnees
+    assert alpha[0]["url"].endswith(f"/collections/{donnees}")
+    assert alpha[0]["page_count"] == 2
+    suivi = [p for p in projets if p["title"] == "Suivi projet"]
+    assert suivi and suivi[0]["is_task"] is True
+
+    detail = _appel(client, jeton, 91, "get_project", {"id": donnees})
+    assert detail["project"]["title"] == "Projet Alpha"
+    assert detail["project"]["url"].endswith(f"/collections/{donnees}")
+    # Détail sans contenu complet : titres seulement.
+    assert [p["title"] for p in detail["pages"]] == ["Page de cadrage",
+                                                     "Revue hebdo"]
+    assert all(set(p) == {"id", "title"} for p in detail["pages"])
+    res = _appel(client, jeton, 92, "get_project", {"id": 99999})
+    assert "introuvable" in res["erreur"]
+
+
+def test_mcp_list_project_tasks(client, jeton, projet_taches):
+    cid = projet_taches["collection_id"]
+    taches = _appel(client, jeton, 93, "list_project_tasks",
+                    {"project_id": cid})
+    assert [t["title"] for t in taches] == ["Relire le pont MCP",
+                                            "Brancher ObsiGate"]
+    assert taches[0]["status"] == "À faire" and taches[0]["done"] is False
+    assert taches[0]["due"] == "2026-10-30"
+    assert taches[1]["done"] is True
+    assert taches[0]["url"].endswith(f"/collections/{cid}")
+    toutes = _appel(client, jeton, 94, "list_project_tasks")
+    assert {t["title"] for t in toutes} >= {"Relire le pont MCP"}
+    res = _appel(client, jeton, 95, "list_project_tasks",
+                 {"project_id": 99999})
+    assert "introuvable" in res["erreur"]
+
+
+def test_mcp_read_page_classique_et_troncature(client, jeton, projet_taches):
+    long_contenu = "mot " * 6000  # 24 000 caractères > borne 20 000
+    with get_conn() as conn:
+        cur = conn.execute(
+            "INSERT INTO pages (workspace, workspace_id, title, content, "
+            "parent_section) VALUES ('Atelier MCP', ?, 'Grande note', ?, 'Private')",
+            (projet_taches["workspace_id"], long_contenu))
+        pid = cur.lastrowid
+        conn.commit()
+    res = _appel(client, jeton, 96, "read_page", {"id": pid})
+    assert res["kind"] == "page" and res["title"] == "Grande note"
+    assert res["truncated"] is True and len(res["content"]) <= 20000
+    assert res["url"].endswith(f"/pages/{pid}")
+    res = _appel(client, jeton, 97, "read_page", {"id": 99999, "kind": "page"})
+    assert "introuvable" in res["erreur"]
+
+
+def test_mcp_read_page_ligne_de_collection(client, jeton, donnees):
+    with get_conn() as conn:
+        ligne = conn.execute(
+            "SELECT id FROM collection_pages WHERE collection_id=? AND "
+            "title='Page de cadrage'", (donnees,)).fetchone()["id"]
+        conn.execute(
+            "INSERT INTO pages (workspace, title, content, collection_row_id) "
+            "VALUES ('', 'contenu', 'Le contenu porté de la ligne', ?)",
+            (ligne,))
+        conn.commit()
+    res = _appel(client, jeton, 98, "read_page",
+                 {"id": ligne, "kind": "collection_page"})
+    assert res["kind"] == "collection_page"
+    assert res["content"] == "Le contenu porté de la ligne"
+    assert res["truncated"] is False
+    assert res["properties"] == {"statut": "en cours"}
+
+
+def test_mcp_jeton_lecture_seule_refuse_les_ecritures(client, jeton, donnees):
+    """§7.4 : un jeton au scope « read » (pont de lecture) ne peut pas
+    écrire via MCP, mais il peut toujours lire."""
+    lecture = "fd_token_lecture_seule"
+    with get_conn() as conn:
+        conn.execute(
+            "INSERT INTO api_tokens (user_id, name, token_hash, token_prefix, "
+            "revoked, scopes) VALUES (1, 'pont-lecture', ?, ?, 0, 'read')",
+            (hashlib.sha256(lecture.encode()).hexdigest(), lecture[:12]))
+        conn.commit()
+    cols = _appel(client, lecture, 100, "list_projects")
+    assert any(c["title"] == "Projet Alpha" for c in cols)
+    r = _rpc(client, lecture, 101, "tools/call",
+             {"name": "create_page",
+              "arguments": {"collection_id": donnees, "title": "Interdite"}})
+    assert r.status_code == 200
+    erreur = r.json()["error"]
+    assert erreur["code"] == -32002 and "lecture seule" in erreur["message"]
+    with get_conn() as conn:
+        n = conn.execute(
+            "SELECT COUNT(*) FROM collection_pages WHERE title='Interdite'"
+        ).fetchone()[0]
+    assert n == 0
+
+
+def test_mcp_journalisation_des_appels(client, jeton, donnees):
+    """§7.4 : chaque tools/call laisse une trace dans api_audit_log."""
+    _appel(client, jeton, 110, "list_projects")
+    with get_conn() as conn:
+        rows = conn.execute(
+            "SELECT action, user_id, token_id, detail FROM api_audit_log "
+            "WHERE action='mcp:list_projects'").fetchall()
+    assert rows, "aucune ligne d'audit pour l'appel MCP"
+    assert rows[-1]["user_id"] == 1 and rows[-1]["token_id"] is not None
+    assert "statut=ok" in rows[-1]["detail"]

tests/test_fusion_pont_obsigate.py

@@ -0,0 +1,320 @@
+"""Fusion ACC-2.2 — pont MCP Flowdeck → ObsiGate (plan « étape 1 :
+lectures seules », tests T0–T5 du plan ``pont-mcp-etape1``).
+
+Ce que le code livre ici (la preuve réelle T0–T5 contre l'ObsiGate de
+Bruno exige sa clé dédiée limitée à un vault + ses arbitrages Q-A/Q-B) :
+
+- **T0 (connectivité)** : le client MCP parle à un serveur **à session**
+  façon ObsiGate — ``initialize`` renvoie ``Mcp-Session-Id``, tout
+  appel sans cet en-tête est refusé ; le handshake précède aussi
+  ``tools/call``. Réponses SSE multi-messages décodées correctement.
+- **Preset « obsigate »** : URL ``<base ObsiGate>/mcp`` dérivée de la
+  configuration (dev/prod), clé chiffrée jamais renvoyée.
+- **Garde-fou n° 2** : les outils MCP sont **classés** à la découverte
+  (annotations ``readOnlyHint``, sinon heuristique de nom) — les
+  écrivants (``propose_*``…) passent par les portes existantes
+  (politique d'approbation + rôles) ; ``apply_*`` est **bloqué** côté
+  client (patron §7.5 : l'écriture croisée s'approuve dans l'app
+  propriétaire).
+"""
+
+import asyncio
+import json
+
+import httpx
+import pytest
+
+from app.db import get_conn
+from app.services import connectors, http_client, mcp_client
+from app.services.importers import url_fetch
+from app.services.tool_registry import ToolRegistry
+
+SESSION_ID = "sess-og-test-123"
+URL_PONT = "https://og.dracodev.net/mcp"
+
+
+@pytest.fixture(autouse=True)
+def _dns_monde_reel(monkeypatch):
+    """Même neutralisation DNS que test_fusion_tiktak (hôtes publics
+    résolus en 198.18.0.0/15 localement, refusés par la garde SSRF)."""
+    monkeypatch.setattr(url_fetch, "_validate_url", lambda url: url.strip())
+    monkeypatch.setattr(mcp_client, "_sessions", {})
+
+
+# Outils tels qu'ObsiGate les déclare sur son serveur MCP : lectures,
+# puis le patron d'écriture propose/apply (spec d'accès centralisé §7.5).
+OUTILS_OBSIGATE = [
+    {"name": "list_vaults",
+     "description": "Liste les vaults accessibles au jeton.",
+     "inputSchema": {"type": "object", "properties": {}}},
+    {"name": "search_fulltext",
+     "description": "Recherche plein texte dans le vault du jeton.",
+     "inputSchema": {"type": "object",
+                     "properties": {"query": {"type": "string"}},
+                     "required": ["query"]},
+     "annotations": {"readOnlyHint": True}},
+    {"name": "read_file",
+     "description": "Lit un fichier du vault.",
+     "inputSchema": {"type": "object",
+                     "properties": {"path": {"type": "string"}},
+                     "required": ["path"]}},
+    {"name": "propose_write_file",
+     "description": "Propose une écriture (brouillon à approuver).",
+     "inputSchema": {"type": "object",
+                     "properties": {"path": {"type": "string"},
+                                    "content": {"type": "string"}},
+                     "required": ["path", "content"]},
+     "annotations": {"readOnlyHint": False}},
+    {"name": "apply_write_file",
+     "description": "Applique une écriture approuvée.",
+     "inputSchema": {"type": "object",
+                     "properties": {"draft_id": {"type": "string"}},
+                     "required": ["draft_id"]}},
+]
+
+
+def _gestionnaire_obsigate(appels):
+    """Rejoue le serveur MCP d'ObsiGate : Streamable HTTP **à session
+    stricte** (tout appel post-initialize sans ``Mcp-Session-Id``
+    valide → erreur -32000), ``tools/list`` répondu en SSE
+    multi-messages (une notification précède la réponse)."""
+    def handler(request: httpx.Request) -> httpx.Response:
+        body = json.loads(request.content or b"{}")
+        method = body.get("method")
+        appels.append((method, dict(request.headers)))
+        if method == "initialize":
+            return httpx.Response(
+                200,
+                json={"jsonrpc": "2.0", "id": body.get("id"),
+                      "result": {"protocolVersion": "2025-03-26",
+                                 "capabilities": {"tools": {}},
+                                 "serverInfo": {"name": "obsigate",
+                                                "version": "2.78"}}},
+                headers={"Mcp-Session-Id": SESSION_ID})
+        if request.headers.get("mcp-session-id") != SESSION_ID:
+            return httpx.Response(
+                400,
+                json={"jsonrpc": "2.0", "id": body.get("id"),
+                      "error": {"code": -32000,
+                                "message": "Mcp-Session-Id manquant ou invalide"}})
+        if method == "notifications/initialized":
+            return httpx.Response(202, content=b"")
+        if method == "tools/list":
+            sse = (
+                'event: message\n'
+                'data: {"jsonrpc": "2.0", "method": "notifications/progress",'
+                ' "params": {"message": "scan des vaults"}}\n\n'
+                'event: message\n'
+                f'data: {json.dumps({"jsonrpc": "2.0", "id": body.get("id"), "result": {"tools": OUTILS_OBSIGATE}})}\n\n'
+            )
+            return httpx.Response(200, content=sse.encode(),
+                                  headers={"Content-Type": "text/event-stream"})
+        if method == "tools/call":
+            nom = (body.get("params") or {}).get("name")
+            if nom == "search_fulltext":
+                res = [{"path": "Notes/Pont MCP.md", "vault": "home-bruno",
+                        "extraits": ["le pont MCP relie Flowdeck à ObsiGate"]}]
+            elif nom == "read_file":
+                res = {"path": "Notes/Pont MCP.md",
+                       "content": "Contenu du fichier Pont MCP."}
+            elif nom == "list_vaults":
+                res = [{"name": "home-bruno"}]
+            else:
+                res = {"ok": True}
+            return httpx.Response(
+                200, json={"jsonrpc": "2.0", "id": body.get("id"),
+                           "result": {"content": [
+                               {"type": "text",
+                                "text": json.dumps(res, ensure_ascii=False)}]}})
+        return httpx.Response(200, json={"jsonrpc": "2.0",
+                                          "id": body.get("id"), "result": {}})
+    return handler
+
+
+@pytest.fixture
+def faux_obsigate(monkeypatch):
+    """Branche le vrai ``_rpc`` du client sur le faux serveur (transport
+    httpx simulé) : sessions, en-têtes et SSE sont exercés pour de vrai."""
+    appels: list = []
+    transport = httpx.MockTransport(_gestionnaire_obsigate(appels))
+    vraie_classe = http_client.shared_client
+
+    class FauxShared:
+        def __init__(self, **kwargs):
+            kwargs["transport"] = transport
+            self._inner = vraie_classe(**kwargs)
+
+        async def __aenter__(self):
+            return await self._inner.__aenter__()
+
+        async def __aexit__(self, *exc):
+            return await self._inner.__aexit__(*exc)
+
+    monkeypatch.setattr(http_client, "shared_client", FauxShared)
+    return appels
+
+
+def _creer_obsigate(client, **kw):
+    body = {"name": "ObsiGate", "kind": "obsigate", "url": URL_PONT,
+            "secret": "og_cle_pont_test", "auth": "bearer", **kw}
+    r = client.post("/api/agent/connectors", json=body)
+    assert r.status_code == 200, r.text
+    return r.json()
+
+
+def _sonder(client):
+    row = _creer_obsigate(client)
+    res = asyncio.run(connectors.probe(str(row["id"])))
+    assert res["status"] == "ok", res
+    return row, res
+
+
+# ── Preset ───────────────────────────────────────────────────────────────
+
+def test_preset_obsigate_et_secret_chiffre(client):
+    row = _creer_obsigate(client)
+    assert row["kind"] == "obsigate" and row["auth"] == "bearer"
+    assert row["has_secret"] is True
+    assert "og_cle_pont_test" not in json.dumps(row)      # jamais renvoyée
+    # URL par défaut = défaut officiel ; en dev, draco_urls la dérive
+    # en og-dev (résolution par configuration, chantier dev/prod).
+    assert connectors.PRESETS["obsigate"]["url"] == "https://og.dracodev.net/mcp"
+    hdrs = connectors._headers(row["id"])
+    assert hdrs["Authorization"] == "Bearer og_cle_pont_test"
+
+
+# ── T0 : session + handshake + SSE ───────────────────────────────────────
+
+def test_t0_session_handshake_et_cache(client, faux_obsigate):
+    row, res = _sonder(client)
+    assert "5 outil(s)" in res["detail"]
+    methodes = [m for m, _ in faux_obsigate]
+    assert methodes[:3] == ["initialize", "notifications/initialized", "tools/list"]
+    # L'appel tools/list porte la session émise à initialize — sans
+    # elle, le faux serveur refuse (session stricte) et le probe échoue.
+    entetes_list = [h for m, h in faux_obsigate if m == "tools/list"][0]
+    assert entetes_list.get("mcp-session-id") == SESSION_ID
+    assert entetes_list.get("authorization") == "Bearer og_cle_pont_test"
+
+    with get_conn() as conn:
+        brut = conn.execute(
+            "SELECT tools_json FROM agent_connectors WHERE id=?",
+            (row["id"],)).fetchone()["tools_json"]
+    outils = {t["original"]: t for t in json.loads(brut)}
+    assert set(outils) == {"list_vaults", "search_fulltext", "read_file",
+                           "propose_write_file", "apply_write_file"}
+    # Classification à la découverte (garde-fou n° 2).
+    assert outils["search_fulltext"]["write"] is False   # annotation readOnlyHint
+    assert outils["read_file"]["write"] is False          # défaut = lecture
+    assert outils["propose_write_file"]["write"] is True  # annotation + préfixe
+    assert outils["apply_write_file"]["write"] is True    # préfixe apply_
+
+
+def test_t1_appel_outil_lecture_avec_session(client, faux_obsigate):
+    _sonder(client)
+    reg = ToolRegistry()
+    noms = {t["name"] for t in reg.schema()}
+    assert "mcp_obsigate_search_fulltext" in noms
+    assert "mcp_obsigate_read_file" in noms
+
+    avant = len(faux_obsigate)
+    res = asyncio.run(reg.execute("mcp_obsigate_search_fulltext",
+                                  {"query": "pont MCP"}))
+    assert res.status == "success", res
+    assert "Pont MCP.md" in res.data["text"]
+    # call_tool refait le handshake complet, session réémise partout.
+    sequence = [m for m, _ in faux_obsigate[avant:]]
+    assert sequence == ["initialize", "notifications/initialized", "tools/call"]
+    entetes_call = [h for m, h in faux_obsigate if m == "tools/call"][0]
+    assert entetes_call.get("mcp-session-id") == SESSION_ID
+
+    res2 = asyncio.run(reg.execute("mcp_obsigate_read_file",
+                                   {"path": "Notes/Pont MCP.md"}))
+    assert res2.status == "success"
+    assert "Contenu du fichier Pont MCP." in res2.data["text"]
+
+
+def test_parse_body_sse_multi_messages():
+    sse = (
+        ': keep-alive\n'
+        'event: message\n'
+        'data: {"jsonrpc": "2.0", "method": "notifications/progress", "params": {}}\n\n'
+        'event: message\n'
+        'data: {"jsonrpc": "2.0", "id": 2, "result": {"tools": []}}\n\n'
+    )
+    # Sans id attendu : le message porteur de result gagne sur la notification.
+    assert mcp_client.parse_body("text/event-stream", sse, 200) == {
+        "jsonrpc": "2.0", "id": 2, "result": {"tools": []}}
+    # Avec id attendu : c'est lui qui tranche.
+    assert mcp_client.parse_body("text/event-stream", sse, 200,
+                                 expected_id=2)["result"] == {"tools": []}
+
+
+def test_classer_outil():
+    assert mcp_client.classer_outil("search_fulltext") is False
+    assert mcp_client.classer_outil("read_file", {"readOnlyHint": True}) is False
+    assert mcp_client.classer_outil("propose_write_file") is True
+    assert mcp_client.classer_outil("apply_write_file") is True
+    assert mcp_client.classer_outil("create_note") is True
+    # L'annotation du serveur fait foi, même contre l'heuristique.
+    assert mcp_client.classer_outil("create_note", {"readOnlyHint": True}) is False
+    assert mcp_client.classer_outil("get_stats", {"readOnlyHint": False}) is True
+
+
+# ── Garde-fous : portes de gouvernance + blocage apply_* ────────────────
+
+def test_apply_bloque_cote_client(client, faux_obsigate):
+    _sonder(client)
+    reg = ToolRegistry()
+    res = asyncio.run(reg.execute("mcp_obsigate_apply_write_file",
+                                  {"draft_id": "b-1"}))
+    assert res.status == "error"
+    assert "bloqué" in res.message
+    # Aucun tools/call n'a quitté le client pour cet outil.
+    assert [m for m, _ in faux_obsigate].count("tools/call") == 0
+
+
+def test_portes_gouvernance_outils_mcp(client, faux_obsigate):
+    from app.services.agent_policies import check_tool
+    from app.services.permission_manager import PermissionManager, is_write_tool
+
+    row, _res = _sonder(client)
+    assert row["id"]
+    assert is_write_tool("mcp_obsigate_propose_write_file") is True
+    assert is_write_tool("mcp_obsigate_apply_write_file") is True
+    assert is_write_tool("mcp_obsigate_search_fulltext") is False
+    assert is_write_tool("create_page") is True       # interne, inchangé
+    assert is_write_tool("search") is False
+
+    # Politique globale « approbation requise » : l'écrivant MCP est
+    # intercepté comme un écrivant interne, la lecture passe.
+    with get_conn() as conn:
+        conn.execute(
+            "INSERT INTO agent_policies (workspace_id, require_approval)"
+            " VALUES (NULL, 1)")
+        conn.commit()
+    gate = check_tool(1, None, "mcp_obsigate_propose_write_file",
+                      is_write=is_write_tool("mcp_obsigate_propose_write_file"))
+    assert gate["allowed"] is False and gate["approval_id"]
+    ok = check_tool(1, None, "mcp_obsigate_search_fulltext",
+                    is_write=is_write_tool("mcp_obsigate_search_fulltext"))
+    assert ok["allowed"] is True
+
+    # Rôles : un viewer ne peut pas lancer un outil MCP écrivant.
+    from fastapi import HTTPException
+    with get_conn() as conn:
+        admin = conn.execute("SELECT id FROM users ORDER BY id LIMIT 1").fetchone()
+        conn.execute("INSERT INTO users (login, is_admin) VALUES ('lecteur_pont', 0)")
+        lecteur = conn.execute(
+            "SELECT id FROM users WHERE login='lecteur_pont'").fetchone()
+        conn.execute("INSERT INTO workspaces (name, owner_id) VALUES ('ws-pont', ?)",
+                     (admin["id"],))
+        ws = conn.execute("SELECT id FROM workspaces WHERE name='ws-pont'").fetchone()
+        conn.execute(
+            "INSERT INTO workspace_members (workspace_id, user_id, role)"
+            " VALUES (?,?, 'viewer')", (ws["id"], lecteur["id"]))
+        conn.commit()
+    pm = PermissionManager(lecteur["id"])
+    with pytest.raises(HTTPException):
+        pm.assert_can("mcp_obsigate_propose_write_file", {}, ws["id"])
+    pm.assert_can("mcp_obsigate_search_fulltext", {}, ws["id"])

tests/test_v511_v512_wiki_templates.py

@@ -396,7 +396,11 @@ def test_editor_front_end_wired(client):
     assert "gtTok" in src and "fd-wiki-chip" in src          # chip render + reader
     assert "wiki/pages" in src and "wiki/titles" in src      # picker + label resolution
     assert "WM={" in src and "fdpage:" in src                # [[ picker module
-    assert "_openTemplatePicker" in base and "page-templates" in base
+    # Depuis la refonte templates (fc85481), le picker de base.html passe par
+    # le registre unifié (/api/templates) ; le câblage « page-templates » vit
+    # dans l'éditeur (fetch /board/api/page-templates) — chaque marqueur est
+    # donc vérifié dans le fichier qui le porte réellement.
+    assert "_openTemplatePicker" in base and "page-templates" in src
     assert "toggleLock" in src and "saveAsTemplate" in src
     css = pathlib.Path("static/css/app.css").read_text(encoding="utf-8")
     assert ".fd-wiki-chip" in css and ".fd-lock-banner" in css and ".tpl-picker-modal" in css