diff --git a/.claude/rules/file-map.md b/.claude/rules/file-map.md index c4ce62ca..941aac8d 100644 --- a/.claude/rules/file-map.md +++ b/.claude/rules/file-map.md @@ -6,7 +6,7 @@ Full module index for LocalChat. **Keep this current** — update in the same co |------|------| | `app.py` | Entry point; `main()` + `create_uvicorn_app()` for dev/prod (FastAPI + Uvicorn) | | `src/app_fastapi.py` | `create_app()` — FastAPI factory, pure wiring only (no I/O); safe to call in tests | -| `src/app_bootstrap.py` | `bootstrap_app(app)` — all startup I/O (Ollama, DB, caching, plugins, connectors, reranker) plus `_clear_upload_staging()`, which removes staged uploads an interrupted ingest left behind; called from `app.py` only | +| `src/app_bootstrap.py` | `bootstrap_app(app)` — all startup I/O (Ollama, DB, caching, plugins, connectors, the core scheduler with the weekly reranker fine-tune as its first subscriber) plus `_clear_upload_staging()`, which removes staged uploads an interrupted ingest left behind; called from `app.py` only | | `src/config.py` | All configuration constants, loads `.env` | | `src/models.py` | Pydantic request/response models | | `src/security_fastapi.py` | JWT (`PyJWT`), rate limiting (`slowapi`), CORS (Starlette middleware), `setup_security_headers()` (CSP + nosniff + referrer + framing + HSTS); `resolve_principal()` — the one place a caller is identified: decode, fail-closed revocation, and the role read from the database | @@ -174,11 +174,13 @@ Full module index for LocalChat. **Keep this current** — update in the same co | `src/tools/executor.py` | Ollama tool-call loop (multi-turn until final response) | | `src/tools/registry.py` | Tool registration with JSON schemas | | `src/tools/builtin.py` | Built-in tools: document search, calculator, datetime | -| `src/tools/plugin_loader.py` | Loads `.py` plugins from `plugins/` at startup; one with `setup(services)` gets its manifest validated and its services built, one without loads as legacy. A failing plugin is disabled alone, with any tool it registered removed | +| `src/tools/plugin_loader.py` | Loads `.py` plugins from `plugins/` at startup; one with `setup(services)` gets its manifest validated, its services built and its hooks subscribed, one without loads as legacy. A failing plugin is disabled alone, with any tool it registered removed; unloading also ends its subscriptions | | `src/plugins/manifest.py` | PC-1 — `read_manifest()`: `PLUGIN_SCOPE`, `PLUGIN_MIN_ROLE`, `PLUGIN_HOOKS` (checked against the catalogue), `PLUGIN_CONFIG`; `PLUGIN_CONTRIBUTES` must be False. Raises `ManifestError` | | `src/plugins/services.py` | PC-1 — the handles a plugin's `setup()` receives: `retrieval` (the request's workspace only, tier capped by the manifest), `llm.embed`, `config`, `identity` (read only), `tools.register` (refuses a taken name) | +| `src/plugins/hooks.py` | PC-2 — `HookBus` (`subscribe`/`unsubscribe` per owner, `emit` skipping a subscriber that raises), `Scheduler` (one daemon thread emitting `on_scheduler_tick`), the `HOOKS` catalogue and the `hook_bus` singleton | | `src/plugins/__init__.py` | `build_services()` — types the plugin's settings from `PLUGIN__` and assembles its handles | | `plugins/example_plugin.py` | Example plugin on the PC-1 contract — `setup(services)`, a manifest and a setting (`word_count`, `reverse_text` tools) | +| `tests/unit/test_hook_bus.py` | PC-2 — a raising subscriber skipped while the next runs, a copy of the payload each, unknown hooks refused, numbered ticks that stop; the loader subscribing a plugin's declared hooks, once across a reload, never for a failed setup, and unsubscribing on unload; the exact payloads of an ingest (new and replaced, none on failure) and of a permitted tool, none for a refused one | | `tests/unit/test_plugin_contract.py` | PC-1 — the manifest (every malformed field refused by name), each handle (retrieval reads the request's workspace and refuses a tier beyond its reach or a key's), the executor hiding and refusing a tool below its role, the loader disabling one plugin alone and removing its half-registered tools, and the guard recording who it admitted | | `plugins/README.md` | Plugin authoring guide | | **Graph / Memory / Performance** | | diff --git a/.claude/rules/plugins.md b/.claude/rules/plugins.md index 50c9c50c..eb3aab02 100644 --- a/.claude/rules/plugins.md +++ b/.claude/rules/plugins.md @@ -3,14 +3,14 @@ How plugins extend LocalChat without destabilising the core. Read alongside the "Plugin Contract" section in `CLAUDE.md` — that states the principle; this is the rule. -> ## ⚠ Status: partly built (PC-1 built 2026-11-11; PC-4 2026-10-23) +> ## ⚠ Status: partly built (PC-1 built 2026-11-11; PC-2 2026-11-13; PC-4 2026-10-23) > > | Described here | Reality | > |---|---| > | The manifest (`PLUGIN_SCOPE`, `PLUGIN_MIN_ROLE`, `PLUGIN_HOOKS`, `PLUGIN_CONFIG`) | **Built** (PC-1) in `src/plugins/manifest.py`; a malformed one disables that plugin only | > | The services catalogue | **Built** (PC-1) in `src/plugins/services.py`: `retrieval`, `llm` (`embed` only), `config`, `identity`, and `tools` for registering. `storage` and `llm.complete` are deferred until a consumer needs them | > | `PLUGIN_MIN_ROLE` | **Enforced by the core** (PC-1): the tool executor hides and refuses a tool below its minimum role | -> | The hooks catalogue (`on_document_ingested`, `on_scheduler_tick`, …) | No hook bus (PC-2). `PLUGIN_HOOKS` is validated against the names below and fires nothing | +> | The hooks catalogue | **Built** (PC-2) in `src/plugins/hooks.py`: `on_document_ingested`, `on_scheduler_tick`, `on_tool_invocation`. `on_route_mount` is deferred until a consumer needs it | > | `plugins/_echo/` reference plugin | Absent (PC-3). `plugins/` holds `example_plugin.py`, on the PC-1 contract | > | The `core-without-plugins` CI job | **Built** (2026-10-23) in `.github/workflows/tests.yml`; a required check since 2026-10-30 | > @@ -61,18 +61,27 @@ describes the capability generically. A `pricing_lookup` service is a leak; a ## Hooks catalogue -A plugin subscribes via its manifest. The core calls all subscribers and is unaffected -if there are none. +A plugin subscribes via its manifest: each hook named in `PLUGIN_HOOKS` is a module +function of the same name taking the payload, `def on_document_ingested(payload): ...`. A +hook named without its function is a malformed manifest. The core calls all subscribers +and is unaffected if there are none. -| Hook | Fires when | Typical use | -|------|-----------|-------------| -| `on_document_ingested` | A document finishes ingest in `SyncWorker._handle_event()` or `processor.py` | Post-process, extract structured data | -| `on_scheduler_tick` | The core scheduler emits a periodic tick | Background evaluation, batch jobs | -| `on_tool_invocation` | An LLM tool call is dispatched (`src/agent/tool_router.py`) | Register LLM-callable tools | -| `on_route_mount` | App startup, router assembly (`src/app_fastapi.py`) | Contribute a FastAPI `APIRouter` | +| Hook | Fires when | Payload | +|------|-----------|---------| +| `on_document_ingested` | A document finishes ingest in `DocumentProcessor.ingest_document()`, the one path uploads and connectors share | `document_id`, `filename`, `workspace_id`, `chunks`, `replaced`, `source_id` | +| `on_scheduler_tick` | The core scheduler ticks, every `SCHEDULER_TICK_SECONDS` | `tick` (1, 2, …), `at` (ISO 8601, UTC) | +| `on_tool_invocation` | An LLM tool call the caller was permitted has run (`src/tools/executor.py`) | `tool`, `source` (`builtin` or the plugin) | +| `on_route_mount` | *Deferred (2026-11-13).* Plugins load after the app is assembled, so a router they contribute needs a later mount | — | A subscriber that raises is logged and skipped; the core operation completes normally. -Hooks are best-effort from the core's side. +Hooks are best-effort from the core's side. `emit()` is synchronous and runs subscribers in +the emitting thread, so a slow one slows that operation (an ingest, the scheduler); the tool +loop emits through `asyncio.to_thread`, so a subscriber never stalls the event loop. The +payloads carry identifiers, not content: a tool's arguments hold the user's question, and no +consumer needs them yet. + +The weekly reranker fine-tune is the scheduler's first subscriber, a core one: the +scheduler was proven against an existing consumer, not a plugin. ## The manifest diff --git a/CHANGELOG.md b/CHANGELOG.md index 214181a0..fff20f54 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,19 @@ reasoning attached, in [docs/history/LESSONS_LEARNED.md](docs/history/LESSONS_LE it, and is refused if the model names it anyway. A malformed manifest disables that plugin alone. **Upgrading:** nothing to do. A plugin without `setup()` still loads as before, now listed as legacy in `/api/plugins` with a warning in the log. +- **Plugins can react to the core through hooks (PC-2).** A plugin names a hook in + `PLUGIN_HOOKS` and defines a function of that name: `on_document_ingested` after every + ingest, by upload or connector; `on_scheduler_tick` from a new core scheduler, every + `SCHEDULER_TICK_SECONDS` (default 3600); `on_tool_invocation` after a permitted tool ran. A + subscriber that raises is logged and skipped. **Upgrading:** a manifest naming + `on_route_mount`, or a hook without its function, now disables that plugin; + `on_route_mount` is not offered yet. + +### Changed + +- **The weekly reranker fine-tune runs on the core scheduler.** It is the scheduler's first + subscriber and runs on the first tick a week after its last attempt, so up to + `SCHEDULER_TICK_SECONDS` later than the timer it replaces. ### Fixed diff --git a/CLAUDE.md b/CLAUDE.md index fa4f714b..7b23e7a3 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -156,7 +156,7 @@ LocalChat supports plugins (LLM tools, connectors, background work) that extend The enforcing procedure is a CI job, `core-without-plugins`, that verifies architectural integrity: the core builds, lints, and passes its full test suite **with the `plugins/` directory and all private plugin code absent.** If that job goes red, a plugin dependency has leaked into the core. -> **It exists since 2026-10-23** (PC-4, built with QP-3): it deletes everything in `plugins/` but its README, then runs `ruff` and the fast suite. It has been a required check since 2026-10-30 (see Pull Requests). The manifest and the services catalogue were built on 2026-11-11 (PC-1, `src/plugins/`), with `PLUGIN_MIN_ROLE` enforced by the core; `storage` and `llm.complete` wait for a consumer. The hook bus (PC-2) and the `_echo` reference plugin (PC-3) are designed and unbuilt. What ships is described in [`plugins/README.md`](plugins/README.md). See the status banner in [.claude/rules/plugins.md](.claude/rules/plugins.md). +> **It exists since 2026-10-23** (PC-4, built with QP-3): it deletes everything in `plugins/` but its README, then runs `ruff` and the fast suite. It has been a required check since 2026-10-30 (see Pull Requests). The manifest and the services catalogue were built on 2026-11-11 (PC-1, `src/plugins/`), with `PLUGIN_MIN_ROLE` enforced by the core; `storage` and `llm.complete` wait for a consumer. The hook bus and the core scheduler were built on 2026-11-13 (PC-2, `src/plugins/hooks.py`), with `on_route_mount` deferred until a consumer needs it. The `_echo` reference plugin (PC-3) is designed and unbuilt. What ships is described in [`plugins/README.md`](plugins/README.md). See the status banner in [.claude/rules/plugins.md](.claude/rules/plugins.md). Full rules -> [.claude/rules/plugins.md](.claude/rules/plugins.md) diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 1d093a8d..8f271cf9 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -463,6 +463,7 @@ to empty, in which case the router falls back to the active model. > fixes this product at one node and one process; scale with a second node instead. | `LOG_FILE` | `logs/app.log` | Path for the file sink | | `PLUGINS_DIR` | `plugins` | Directory scanned for plugins | +| `SCHEDULER_TICK_SECONDS` | `3600` | Interval of the core scheduler's `on_scheduler_tick`; the weekly reranker fine-tune is one of its subscribers, so it runs at most this late | | `PLUGIN__` | the plugin's declared default | A setting a plugin declares in `PLUGIN_CONFIG`. `` is its file name upper-cased, with anything but letters and digits made `_`. Read by `plugin_setting()` in `config.py`; see [plugins/README.md](../plugins/README.md) | | `PRESENCE_TTL_SECONDS` | `30` | How long a workspace presence entry stays live | | `CONNECTOR_LOCAL_ROOTS` | *(empty)* | Comma-separated absolute directories the `local_folder` connector may watch. **Empty disables that connector type** — see below | diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index b6cd9cfc..661a7b5e 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -639,7 +639,26 @@ PC-1 **extends** this foundation: it adds the manifest contract, service injecti --- -### PC-2 — Hook bus and core scheduler +### PC-2 — Hook bus and core scheduler ✅ (built 2026-11-13) + +> **As built.** Three points were decided by the maintainer on 2026-11-13, and +> `.claude/rules/plugins.md` was amended to match: +> - `emit()` is synchronous. The tool loop calls it through `asyncio.to_thread`, so a slow +> subscriber cannot stall the event loop. +> - `on_route_mount` is deferred until a consumer needs it: plugins load in `bootstrap_app()`, +> after `create_app()` has assembled the routers, so a contributed router needs a later mount. +> - One daemon thread ticks every `SCHEDULER_TICK_SECONDS` (default 3600). The reranker's +> fine-tune subscribes and runs on the first tick a week after its last attempt, on a +> monotonic clock; the `threading.Timer` it replaces is gone. +> +> Two departures from the spec below. `on_document_ingested` fires from +> `DocumentProcessor.ingest_document()` only: `SyncWorker` ingests through it, so emitting in +> both would announce one document twice. And `on_tool_invocation` fires from the LLM tool +> loop (`src/tools/executor.py`), not `src/agent/tool_router.py`, which dispatches the +> aggregator's retrieval rather than tool calls; only a tool the caller was permitted is +> announced, with its name and source and not its arguments. A plugin subscribes with a +> module function named after the hook. + **What it builds:** the core-to-plugin event surface, and the generic scheduler that replaces ad-hoc timers. @@ -2056,7 +2075,7 @@ the generation path is a deterministic pipeline unless an answer-level eval says | PG-0..PG-8 | **Production-grade hardening** ✅ — see [PRODUCTION_PLAN.md](history/PRODUCTION_PLAN.md). Gated everything below; **gate lifted 2026-08-31**. | done | | 8 | GKB-1 (schema + two-tier retrieval) ✅ 2026-10-09 — read-only global tier under RLS (0019) | 1 week | | 9 | GKB-2 (contribution workflow) ✓ backend 2026-10-19 (admin-only contribute, migration 0020), UI 2026-10-20 | 1 week | -| 10 | PC-1 ✅ 2026-11-11 (manifest, services, the core's role gate) + PC-2 (hooks, scheduler) | 1 week | +| 10 | PC-1 ✅ 2026-11-11 (manifest, services, the core's role gate) + PC-2 ✅ 2026-11-13 (hooks, scheduler) | 1 week | | 11 | PC-3 + PC-4 (echo plugin, CI gate) | 1 week | | 12 | PR-1 (pricing plugin — private repo) | 1–2 weeks | | 13 | CONN-1 (connector authorisation model — decision, no code) | 2–3 days | diff --git a/plugins/README.md b/plugins/README.md index ec4c7690..c57fa58b 100644 --- a/plugins/README.md +++ b/plugins/README.md @@ -1,8 +1,8 @@ # LocalChat Plugin System > Verified against `src/tools/`, `src/plugins/` and `src/app_bootstrap.py` on 2026-11-11. -> PC-1 of the contract in `.claude/rules/plugins.md` ships: the manifest and the services -> below. The hook bus (PC-2) and the `_echo` reference plugin (PC-3) do not exist yet. +> PC-1 and PC-2 of the contract in `.claude/rules/plugins.md` ship: the manifest, the +> services and the hooks below. The `_echo` reference plugin (PC-3) does not exist yet. Drop a `.py` file here — LocalChat loads it automatically at startup. @@ -28,7 +28,7 @@ PLUGIN_META = {"name": "My Plugin", "version": "1.0.0", "description": "Shown in # The manifest. Every name is optional; these are the defaults. PLUGIN_SCOPE = "local" # retrieval reach: local | global | hybrid PLUGIN_MIN_ROLE = "viewer" # who may use its tools: viewer | editor | owner | admin -PLUGIN_HOOKS = [] # validated against the catalogue; nothing fires them until PC-2 +PLUGIN_HOOKS = [] # each hook named here is a function of that name; see Hooks PLUGIN_CONFIG = { # KEY: (type, default); type is str | int | float | bool "TOP_K": ("int", 5), } @@ -73,6 +73,32 @@ API key's) and then admin (a global administrator). for a plugin that needs it. Tools run synchronously inside the chat stream, so a completion from inside one needs that changed first. `PLUGIN_CONTRIBUTES` must be `False`. +### Hooks + +A plugin reacts to the core through hooks. Name each one in `PLUGIN_HOOKS` and define a +function of the same name taking one argument, the payload (a fresh `dict` per subscriber): + +```python +PLUGIN_HOOKS = ["on_document_ingested", "on_scheduler_tick"] + +def on_document_ingested(payload): # document_id, filename, workspace_id, chunks, replaced, source_id + ... + +def on_scheduler_tick(payload): # tick, at — every SCHEDULER_TICK_SECONDS (default 3600) + ... +``` + +| Hook | Fires when | +|---|---| +| `on_document_ingested` | A document finished ingesting, by upload or connector | +| `on_scheduler_tick` | The core scheduler ticks | +| `on_tool_invocation` | A tool the caller was permitted has run; payload `tool`, `source` | + +A hook named without its function disables the plugin. A subscriber that raises is logged +and skipped, and the core carries on. Subscribers run in the thread that emitted, so keep +them short: a slow `on_document_ingested` slows every ingest. A plugin's subscriptions end +when it is unloaded and are made afresh on reload. `on_route_mount` is not offered yet. + ### Legacy plugins A file without `setup(services)` loads the way every plugin did before PC-1: it imports @@ -97,6 +123,7 @@ role. Move it to `setup(services)` — `example_plugin.py` shows the result. |---|---|---| | `PLUGINS_ENABLED` | `true` | Set to `false` to disable plugin loading | | `PLUGINS_DIR` | `plugins` | Directory to scan (relative to repo root) | +| `SCHEDULER_TICK_SECONDS` | `3600` | Interval of `on_scheduler_tick` | | `PLUGIN__` | the manifest's default | A plugin's declared setting | --- diff --git a/pyproject.toml b/pyproject.toml index a464bcbc..d2fd756e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -135,6 +135,7 @@ module = [ "src.services.workspaces", "src.routes_fastapi._services", "src.plugins", + "src.plugins.hooks", "src.plugins.manifest", "src.plugins.services", "src.utils.caller", diff --git a/src/app_bootstrap.py b/src/app_bootstrap.py index 285a0588..a4d6e693 100644 --- a/src/app_bootstrap.py +++ b/src/app_bootstrap.py @@ -9,7 +9,7 @@ import os import secrets import shutil -import threading +import time from collections.abc import Iterator from typing import Any @@ -163,7 +163,7 @@ def bootstrap_app(app: Any) -> None: _check_graphrag() _load_plugins(app) _init_connectors(app, db, doc_processor) - _init_reranker_scheduler(app, db) + _start_scheduler(app, db) # Cleanup handlers registered last so sync_worker is already set _setup_cleanup_handlers(app) @@ -319,19 +319,35 @@ def _load_plugins(app: Any) -> None: app.state.plugin_loader = plugin_loader -def _init_reranker_scheduler(app: Any, db: Any) -> None: - """Start a weekly background thread for reranker fine-tuning (opt-in).""" +def _start_scheduler(app: Any, db: Any) -> None: + """Start the core scheduler, with the weekly reranker fine-tune as its first subscriber.""" + from .plugins.hooks import Scheduler, hook_bus + + _subscribe_reranker_finetune(db, hook_bus) + scheduler = Scheduler(hook_bus, config.SCHEDULER_TICK_SECONDS) + scheduler.start() + app.state.scheduler = scheduler + + +_WEEK_SECONDS = 7 * 24 * 3600 + + +def _subscribe_reranker_finetune(db: Any, bus: Any) -> None: + """Fine-tune the reranker on the first tick a week or more after the last attempt.""" if not config.RERANKER_ENABLED: return try: import sentence_transformers # noqa: F401 — check availability except ImportError: - logger.info("[Reranker] sentence-transformers not installed — scheduler skipped") + logger.info("[Reranker] sentence-transformers not installed — weekly fine-tune skipped") return - _WEEK_SECONDS = 7 * 24 * 3600 + # Monotonic, so a wall-clock change cannot skip a week or run two in a row. + due = [time.monotonic() + _WEEK_SECONDS] - def _weekly_train() -> None: + def _weekly_train(_payload: dict[str, object]) -> None: + if time.monotonic() < due[0]: + return try: from .rag.feedback_pipeline import ( export_training_pairs, @@ -352,16 +368,11 @@ def _weekly_train() -> None: except Exception: logger.exception("[Reranker] Weekly fine-tune failed") finally: - t = threading.Timer(_WEEK_SECONDS, _weekly_train) - t.daemon = True - t.start() - app.state._reranker_timer = t # type: ignore[attr-defined] + # A failed attempt waits a week too, as the timer it replaced did. + due[0] = time.monotonic() + _WEEK_SECONDS - t = threading.Timer(_WEEK_SECONDS, _weekly_train) - t.daemon = True - t.start() - app.state._reranker_timer = t # type: ignore[attr-defined] - logger.info("[Reranker] Weekly fine-tune scheduler started") + bus.subscribe("on_scheduler_tick", "core.reranker", _weekly_train) + logger.info("[Reranker] Weekly fine-tune subscribed to the scheduler") def _seed_admin_user(db: Any) -> None: @@ -492,6 +503,9 @@ def cleanup() -> None: # shutdown means writing to a closing stderr. from .services.chat import stop_ollama_liveness stop_ollama_liveness() + scheduler = getattr(app.state, "scheduler", None) + if scheduler is not None: + scheduler.stop() sync_worker = getattr(app.state, "sync_worker", None) if sync_worker is not None: logger.info("Stopping connector sync worker...") diff --git a/src/config.py b/src/config.py index 1485b159..85c48d03 100644 --- a/src/config.py +++ b/src/config.py @@ -519,6 +519,9 @@ def _production_errors(errors: list[str]) -> None: PLUGINS_ENABLED: bool = os.environ.get('PLUGINS_ENABLED', 'True').lower() == 'true' # Directory (relative to repo root) that is scanned for .py plugin files. PLUGINS_DIR: str = os.environ.get('PLUGINS_DIR', 'plugins') +# Seconds between on_scheduler_tick events. A subscriber's own period is a multiple of it: +# the weekly reranker fine-tune runs at most this late. +SCHEDULER_TICK_SECONDS: int = int(os.environ.get('SCHEDULER_TICK_SECONDS', '3600')) def plugin_setting(plugin: str, key: str) -> str | None: diff --git a/src/plugins/__init__.py b/src/plugins/__init__.py index ffbde4a8..569fcf6d 100644 --- a/src/plugins/__init__.py +++ b/src/plugins/__init__.py @@ -6,8 +6,8 @@ from .. import config from ..tools.registry import ToolRegistry +from .hooks import HOOKS from .manifest import ( - HOOKS, SCOPES, ConfigValue, Manifest, diff --git a/src/plugins/hooks.py b/src/plugins/hooks.py new file mode 100644 index 00000000..b527d520 --- /dev/null +++ b/src/plugins/hooks.py @@ -0,0 +1,92 @@ +"""The core-to-plugin event surface (PC-2): a hook bus and the scheduler that ticks it. + +Core code announces an event with ``hook_bus.emit(hook, payload)``; whoever subscribed +reacts. The core never depends on there being a subscriber, and a subscriber that raises +is logged and skipped, so the operation that emitted completes as if it had none. + +``emit`` is synchronous and runs every subscriber in the emitting thread. Async code +calls it through ``asyncio.to_thread`` so a slow subscriber cannot stall the event loop. +""" + +from __future__ import annotations + +import threading +from collections.abc import Callable +from datetime import UTC, datetime + +from ..utils.logging_config import get_logger + +logger = get_logger(__name__) + +#: The hooks the core fires. ``on_route_mount`` is deferred until a plugin needs one: +#: plugins load after the app is assembled, so a router they contribute needs a later mount. +HOOKS = ("on_document_ingested", "on_scheduler_tick", "on_tool_invocation") + +Payload = dict[str, object] +Subscriber = Callable[[Payload], None] + + +class HookBus: + """Subscribers per hook, each tagged with its owner so a plugin's can be removed together.""" + + def __init__(self) -> None: + self._lock = threading.Lock() + self._subscribers: dict[str, list[tuple[str, Subscriber]]] = {h: [] for h in HOOKS} + + def subscribe(self, hook: str, owner: str, subscriber: Subscriber) -> None: + if hook not in HOOKS: + raise ValueError(f"unknown hook {hook!r}") + with self._lock: + self._subscribers[hook].append((owner, subscriber)) + + def unsubscribe(self, owner: str) -> None: + with self._lock: + for hook in HOOKS: + self._subscribers[hook] = [s for s in self._subscribers[hook] if s[0] != owner] + + def subscribers(self, hook: str) -> list[str]: + """The owners subscribed to *hook*, in the order they are called.""" + with self._lock: + return [owner for owner, _ in self._subscribers[hook]] + + def emit(self, hook: str, payload: Payload) -> None: + if hook not in HOOKS: + raise ValueError(f"unknown hook {hook!r}") + with self._lock: + subscribers = list(self._subscribers[hook]) + for owner, subscriber in subscribers: + try: + # Each gets its own copy: one subscriber mutating the payload must not + # change what the next one sees. + subscriber(dict(payload)) + except Exception: # noqa: BLE001 — a subscriber is best-effort; the core operation completes without it + logger.exception("[HOOKS] %s subscriber '%s' raised; skipped", hook, owner) + + +class Scheduler: + """One daemon thread emitting ``on_scheduler_tick`` every *interval* seconds.""" + + def __init__(self, bus: HookBus, interval: float) -> None: + self._bus = bus + self._interval = interval + self._stop = threading.Event() + self._thread: threading.Thread | None = None + + def start(self) -> None: + self._thread = threading.Thread(target=self._run, name="core-scheduler", daemon=True) + self._thread.start() + logger.info("[HOOKS] Scheduler ticking every %ss", self._interval) + + def stop(self) -> None: + self._stop.set() + if self._thread is not None: + self._thread.join(timeout=15) + + def _run(self) -> None: + tick = 0 + while not self._stop.wait(self._interval): + tick += 1 + self._bus.emit("on_scheduler_tick", {"tick": tick, "at": datetime.now(UTC).isoformat()}) + + +hook_bus = HookBus() diff --git a/src/plugins/manifest.py b/src/plugins/manifest.py index 1a4b7de4..8c7995ee 100644 --- a/src/plugins/manifest.py +++ b/src/plugins/manifest.py @@ -12,11 +12,10 @@ from typing import Any from ..utils.caller import ROLE_LEVELS +from .hooks import HOOKS #: How far the plugin's retrieval reaches: its workspace, the global tier, or both. SCOPES = ("local", "global", "hybrid") -#: The hooks PC-2 will fire. Named now so a manifest is validated against the catalogue. -HOOKS = ("on_document_ingested", "on_scheduler_tick", "on_tool_invocation", "on_route_mount") _CONFIG_TYPES: dict[str, type] = {"str": str, "int": int, "float": float, "bool": bool} _KEY = re.compile(r"^[A-Z][A-Z0-9_]*$") @@ -68,6 +67,11 @@ def read_manifest(module: ModuleType) -> Manifest: unknown = sorted(set(hooks) - set(HOOKS)) if unknown: raise ManifestError(f"PLUGIN_HOOKS names unknown hook(s): {', '.join(unknown)}") + # The subscriber is the module function named after the hook: declared without one, + # the plugin would look subscribed while nothing could ever be called. + missing = [h for h in hooks if not callable(getattr(module, h, None))] + if missing: + raise ManifestError(f"PLUGIN_HOOKS names hook(s) with no function: {', '.join(missing)}") return Manifest( scope=scope, diff --git a/src/rag/processor.py b/src/rag/processor.py index 285810b8..c3aa4a74 100644 --- a/src/rag/processor.py +++ b/src/rag/processor.py @@ -20,6 +20,7 @@ # Read via globals() in __init__ so tests can patch src.rag.processor.ollama_client; # ruff cannot see that access, hence the noqa. from ..ollama_client import ollama_client # noqa: F401 +from ..plugins.hooks import hook_bus from ..utils.logging_config import get_logger from ..utils.scope import ALL_WORKSPACES, Scope from .chunking import Chunker @@ -537,6 +538,16 @@ def ingest_document( logger.info("Chunks inserted successfully") self._extract_entities(doc_id, chunk_ids, chunks_data) + # Here and not in SyncWorker: a connector ingests through this method too, so + # emitting in both would announce one document twice. + hook_bus.emit("on_document_ingested", { + "document_id": doc_id, + "filename": filename, + "workspace_id": workspace_id, + "chunks": len(chunks_data), + "replaced": replace_doc_id is not None, + "source_id": source_id, + }) success_msg = f"Successfully ingested {filename} ({len(chunks_data)} chunks)" logger.info(success_msg) diff --git a/src/tools/executor.py b/src/tools/executor.py index 0b1cf1d4..87020cf9 100644 --- a/src/tools/executor.py +++ b/src/tools/executor.py @@ -19,12 +19,14 @@ from __future__ import annotations +import asyncio import json from collections.abc import AsyncGenerator from typing import Any from .. import config from ..ollama_client import OllamaClient +from ..plugins.hooks import Payload, hook_bus from ..utils.caller import caller_has_role from ..utils.logging_config import get_logger from .registry import ToolRegistry @@ -197,7 +199,10 @@ async def _execute_round( if not inline_mode: working_messages.append(assistant_msg) - self._append_tool_results(tool_calls, inline_mode, working_messages) + invoked = self._append_tool_results(tool_calls, inline_mode, working_messages) + for event in invoked: + # Off the event loop: a slow subscriber must not stall every other stream. + await asyncio.to_thread(hook_bus.emit, "on_tool_invocation", event) return False, None, inline_mode def _handle_no_tool_calls( @@ -226,13 +231,19 @@ def _append_tool_results( tool_calls: list, inline_mode: bool, working_messages: list[dict[str, Any]], - ) -> None: - """Execute each tool call and append results to working_messages.""" + ) -> list[Payload]: + """Execute each tool call, append results to working_messages, return what ran.""" + invoked: list[Payload] = [] for tc in tool_calls: func = tc.get("function", {}) name = func.get("name", "") arguments = self._parse_tool_arguments(func.get("arguments", {})) result = self._run_tool(name, arguments) + spec = self._registry.get(name) + # A refused tool is announced as nothing, the same answer the caller got. + # The arguments stay out: they carry the user's question, which no hook needs yet. + if spec is not None and self._permitted(name): + invoked.append({"tool": name, "source": spec.source}) if inline_mode: working_messages.append({ "role": "user", @@ -244,6 +255,7 @@ def _append_tool_results( }) else: working_messages.append({"role": "tool", "content": result}) + return invoked def _permitted(self, name: str) -> bool: spec = self._registry.get(name) diff --git a/src/tools/plugin_loader.py b/src/tools/plugin_loader.py index 5362355a..301f8b54 100644 --- a/src/tools/plugin_loader.py +++ b/src/tools/plugin_loader.py @@ -50,6 +50,7 @@ from pathlib import Path from typing import Any +from ..plugins.hooks import HookBus, hook_bus from ..utils.logging_config import get_logger from .registry import ToolRegistry, tool_registry @@ -68,10 +69,11 @@ class PluginLoader: Defaults to the module-level ``tool_registry`` singleton. """ - def __init__(self, registry: ToolRegistry | None = None) -> None: + def __init__(self, registry: ToolRegistry | None = None, bus: HookBus | None = None) -> None: # `is None`, not `or`: an empty registry has length 0, so `or` swapped a fresh # one for the global singleton. self._registry = registry if registry is not None else tool_registry + self._bus = bus if bus is not None else hook_bus # plugin_stem -> {"path": str, "tools": [name, ...], "meta": {...}, "error": str|None} self._plugins: dict[str, dict[str, Any]] = {} @@ -319,10 +321,14 @@ def _set_up(self, plugin_key: str, module: Any) -> dict[str, Any] | None: # A tool registered later, from a kept handle, would be one the loader never # recorded: listed nowhere, and left behind by unload and reload. services.tools.close() + # Last, so a plugin whose setup failed is never subscribed. + for hook in manifest.hooks: + self._bus.subscribe(hook, plugin_key, getattr(module, hook)) return manifest.to_dict() def _unload_tools(self, plugin_name: str) -> None: - """Remove all tools registered by *plugin_name* from the registry.""" + """Remove all tools registered by *plugin_name*, and its hook subscriptions.""" + self._bus.unsubscribe(plugin_name) tools = self._plugins.get(plugin_name, {}).get("tools", []) for tool_name in tools: self._registry.unregister(tool_name) diff --git a/tests/unit/test_app_bootstrap_warmup.py b/tests/unit/test_app_bootstrap_warmup.py index e1aafa20..dac71619 100644 --- a/tests/unit/test_app_bootstrap_warmup.py +++ b/tests/unit/test_app_bootstrap_warmup.py @@ -319,138 +319,83 @@ def test_skips_tool_name_lookup_when_zero_plugins_loaded(self): mock_list.assert_not_called() -class TestInitRerankerScheduler: - """Test _init_reranker_scheduler (src/app_bootstrap.py).""" - - def test_skips_when_reranker_disabled(self): - from src.app_bootstrap import _init_reranker_scheduler - - app = MagicMock() +class TestRerankerFinetuneSubscriber: + """The weekly reranker fine-tune, as the scheduler's first subscriber (PC-2).""" + + WEEK = 7 * 24 * 3600 + + def _subscribe(self, db, *, enabled=True, available=True, now=1000.0): + from src.app_bootstrap import _subscribe_reranker_finetune + from src.plugins.hooks import HookBus + + bus = HookBus() + modules = {"sentence_transformers": MagicMock() if available else None} + with patch("src.app_bootstrap.config") as mock_config, patch.dict("sys.modules", modules), patch("src.app_bootstrap.time.monotonic", return_value=now): + mock_config.RERANKER_ENABLED = enabled + _subscribe_reranker_finetune(db, bus) + return bus + + def _tick(self, bus, at, *, pairs=(), min_pairs=1, finetune=None, export_error=None): + finetune = finetune if finetune is not None else {"skipped": True} + export = {"side_effect": export_error} if export_error else {"return_value": list(pairs)} + with patch("src.app_bootstrap.config") as mock_config, patch("src.app_bootstrap.time.monotonic", return_value=at), patch("src.rag.feedback_pipeline.export_training_pairs", **export) as mock_export, patch("src.rag.feedback_pipeline.finetune_reranker", return_value=finetune) as mock_ft, patch("src.rag.feedback_pipeline.persist_reranker_version", return_value="v2") as mock_persist, patch("src.rag.feedback_pipeline.promote_model") as mock_promote: + mock_config.FEEDBACK_FINETUNE_MIN_PAIRS = min_pairs + bus.emit("on_scheduler_tick", {"tick": 1}) + return mock_export, mock_ft, mock_persist, mock_promote + + def test_not_subscribed_when_reranker_disabled(self): + bus = self._subscribe(MagicMock(), enabled=False) + assert bus.subscribers("on_scheduler_tick") == [] + + def test_not_subscribed_when_sentence_transformers_missing(self): + bus = self._subscribe(MagicMock(), available=False) + assert bus.subscribers("on_scheduler_tick") == [] + + def test_subscribed_as_the_core_reranker_when_available(self): + bus = self._subscribe(MagicMock()) + assert bus.subscribers("on_scheduler_tick") == ["core.reranker"] + + def test_a_tick_before_a_week_has_passed_does_nothing(self): db = MagicMock() - with patch("src.app_bootstrap.config") as mock_config: - mock_config.RERANKER_ENABLED = False - with patch("threading.Timer") as mock_timer: - _init_reranker_scheduler(app, db) + bus = self._subscribe(db, now=1000.0) + mock_export, *_ = self._tick(bus, at=1000.0 + self.WEEK - 1, pairs=[1, 2, 3]) + mock_export.assert_not_called() - mock_timer.assert_not_called() - - def test_skips_when_sentence_transformers_not_installed(self): - from src.app_bootstrap import _init_reranker_scheduler - - app = MagicMock() + def test_the_first_tick_a_week_later_trains_and_the_next_one_waits_a_week(self): db = MagicMock() - with patch("src.app_bootstrap.config") as mock_config: - mock_config.RERANKER_ENABLED = True - with patch.dict("sys.modules", {"sentence_transformers": None}): - with patch("threading.Timer") as mock_timer: - _init_reranker_scheduler(app, db) - - mock_timer.assert_not_called() - - def test_starts_weekly_timer_when_enabled_and_available(self): - from src.app_bootstrap import _init_reranker_scheduler - - app = MagicMock() - db = MagicMock() - with patch("src.app_bootstrap.config") as mock_config: - mock_config.RERANKER_ENABLED = True - with patch.dict("sys.modules", {"sentence_transformers": MagicMock()}): - with patch("threading.Timer") as mock_timer_cls: - mock_timer_instance = MagicMock() - mock_timer_cls.return_value = mock_timer_instance - _init_reranker_scheduler(app, db) - - mock_timer_cls.assert_called_once() - mock_timer_instance.start.assert_called_once() - assert app.state._reranker_timer is mock_timer_instance - - -class TestWeeklyRerankerTrain: - """Test the weekly fine-tune closure scheduled by _init_reranker_scheduler.""" - - def _get_weekly_train(self, app, db): - from src.app_bootstrap import _init_reranker_scheduler - - with patch("src.app_bootstrap.config") as mock_config: - mock_config.RERANKER_ENABLED = True - with patch.dict("sys.modules", {"sentence_transformers": MagicMock()}): - with patch("threading.Timer") as mock_timer_cls: - mock_timer_cls.return_value = MagicMock() - _init_reranker_scheduler(app, db) - - return mock_timer_cls.call_args[0][1] + bus = self._subscribe(db, now=1000.0) + due = 1000.0 + self.WEEK + mock_export, *_ = self._tick(bus, at=due, pairs=[1]) + mock_export.assert_called_once_with(db, days=7) + again, *_ = self._tick(bus, at=due + self.WEEK - 1, pairs=[1]) + later, *_ = self._tick(bus, at=due + self.WEEK, pairs=[1]) + assert (again.call_count, later.call_count) == (0, 1) def test_skips_finetune_when_below_minimum_pairs(self): - app = MagicMock() - db = MagicMock() - weekly_train = self._get_weekly_train(app, db) - - with patch("src.app_bootstrap.config") as mock_config: - mock_config.FEEDBACK_FINETUNE_MIN_PAIRS = 50 - with patch("src.rag.feedback_pipeline.export_training_pairs", return_value=[1, 2]) as mock_export, \ - patch("src.rag.feedback_pipeline.finetune_reranker") as mock_finetune, \ - patch("threading.Timer") as mock_timer_cls: - mock_timer_cls.return_value = MagicMock() - weekly_train() - - mock_export.assert_called_once_with(db, days=7) - mock_finetune.assert_not_called() - mock_timer_cls.assert_called_once() # reschedules in finally + bus = self._subscribe(MagicMock(), now=0.0) + _, mock_ft, _, _ = self._tick(bus, at=self.WEEK, pairs=[1, 2], min_pairs=50) + mock_ft.assert_not_called() def test_promotes_model_when_finetune_improves_ndcg(self): - app = MagicMock() db = MagicMock() - weekly_train = self._get_weekly_train(app, db) - - with patch("src.app_bootstrap.config") as mock_config: - mock_config.FEEDBACK_FINETUNE_MIN_PAIRS = 1 - with patch("src.rag.feedback_pipeline.export_training_pairs", return_value=[1, 2, 3]), \ - patch( - "src.rag.feedback_pipeline.finetune_reranker", - return_value={"skipped": False, "ndcg_before": 0.5, "ndcg_after": 0.8}, - ), \ - patch("src.rag.feedback_pipeline.persist_reranker_version", return_value="v2") as mock_persist, \ - patch("src.rag.feedback_pipeline.promote_model") as mock_promote, \ - patch("threading.Timer") as mock_timer_cls: - mock_timer_cls.return_value = MagicMock() - weekly_train() - - mock_persist.assert_called_once() + bus = self._subscribe(db, now=0.0) + result = {"skipped": False, "ndcg_before": 0.5, "ndcg_after": 0.8} + _, _, mock_persist, mock_promote = self._tick(bus, at=self.WEEK, pairs=[1], finetune=result) + mock_persist.assert_called_once_with(db, result) mock_promote.assert_called_once_with(db, "v2") def test_does_not_promote_when_ndcg_does_not_improve(self): - app = MagicMock() - db = MagicMock() - weekly_train = self._get_weekly_train(app, db) - - with patch("src.app_bootstrap.config") as mock_config: - mock_config.FEEDBACK_FINETUNE_MIN_PAIRS = 1 - with patch("src.rag.feedback_pipeline.export_training_pairs", return_value=[1, 2, 3]), \ - patch( - "src.rag.feedback_pipeline.finetune_reranker", - return_value={"skipped": False, "ndcg_before": 0.8, "ndcg_after": 0.5}, - ), \ - patch("src.rag.feedback_pipeline.persist_reranker_version", return_value="v2"), \ - patch("src.rag.feedback_pipeline.promote_model") as mock_promote, \ - patch("threading.Timer") as mock_timer_cls: - mock_timer_cls.return_value = MagicMock() - weekly_train() - + bus = self._subscribe(MagicMock(), now=0.0) + result = {"skipped": False, "ndcg_before": 0.8, "ndcg_after": 0.5} + *_, mock_promote = self._tick(bus, at=self.WEEK, pairs=[1], finetune=result) mock_promote.assert_not_called() - def test_reschedules_after_exception(self): - app = MagicMock() - db = MagicMock() - weekly_train = self._get_weekly_train(app, db) - - with patch("src.app_bootstrap.config") as mock_config: - mock_config.FEEDBACK_FINETUNE_MIN_PAIRS = 1 - with patch("src.rag.feedback_pipeline.export_training_pairs", side_effect=RuntimeError("boom")), \ - patch("threading.Timer") as mock_timer_cls: - mock_timer_cls.return_value = MagicMock() - weekly_train() # must not raise - - mock_timer_cls.assert_called_once() # reschedules in finally despite the failure + def test_a_failed_attempt_also_waits_a_week(self): + bus = self._subscribe(MagicMock(), now=0.0) + self._tick(bus, at=self.WEEK, export_error=RuntimeError("boom")) # must not raise + soon, *_ = self._tick(bus, at=self.WEEK + 1, pairs=[1]) + week_later, *_ = self._tick(bus, at=2 * self.WEEK, pairs=[1]) + assert (soon.call_count, week_later.call_count) == (0, 1) class TestSeedAdminUser: @@ -691,7 +636,7 @@ def _mark_db_ready(app_arg, *_a, **_kw): patch("src.app_bootstrap._seed_admin_user") as mock_seed_admin, \ patch("src.app_bootstrap._load_plugins") as mock_load_plugins, \ patch("src.app_bootstrap._init_connectors") as mock_init_connectors, \ - patch("src.app_bootstrap._init_reranker_scheduler") as mock_init_reranker, \ + patch("src.app_bootstrap._start_scheduler") as mock_start_scheduler, \ patch("src.app_bootstrap._setup_cleanup_handlers") as mock_setup_cleanup: bootstrap_app(app) @@ -704,7 +649,7 @@ def _mark_db_ready(app_arg, *_a, **_kw): mock_seed_admin.assert_called_once_with(app.state.db) mock_load_plugins.assert_called_once_with(app) mock_init_connectors.assert_called_once() - mock_init_reranker.assert_called_once() + mock_start_scheduler.assert_called_once_with(app, app.state.db) mock_setup_cleanup.assert_called_once_with(app) def test_skips_seeding_admin_user_when_database_unavailable(self): @@ -724,7 +669,7 @@ def test_skips_seeding_admin_user_when_database_unavailable(self): patch("src.app_bootstrap._seed_admin_user") as mock_seed_admin, \ patch("src.app_bootstrap._load_plugins"), \ patch("src.app_bootstrap._init_connectors"), \ - patch("src.app_bootstrap._init_reranker_scheduler"), \ + patch("src.app_bootstrap._start_scheduler"), \ patch("src.app_bootstrap._setup_cleanup_handlers"): bootstrap_app(app) diff --git a/tests/unit/test_hook_bus.py b/tests/unit/test_hook_bus.py new file mode 100644 index 00000000..64ecbf58 --- /dev/null +++ b/tests/unit/test_hook_bus.py @@ -0,0 +1,247 @@ +"""PC-2 — the hook bus, the core scheduler, and the three places the core emits. + +The core announces; a plugin reacts. A subscriber that raises is skipped and the core +operation completes as if there were none, and a plugin's subscriptions go with it when +it is unloaded or fails to load. +""" + +from __future__ import annotations + +import hashlib +import sys +import textwrap +import threading +from pathlib import Path +from types import ModuleType +from unittest.mock import MagicMock, patch + +import pytest + +from src.plugins import ManifestError, read_manifest +from src.plugins.hooks import HOOKS, HookBus, Scheduler +from src.tools.executor import ToolExecutor +from src.tools.plugin_loader import PluginLoader +from src.tools.registry import ToolRegistry +from src.utils.caller import RequestCaller, request_caller + +pytestmark = pytest.mark.unit + + +# --------------------------------------------------------------------------- +# The bus +# --------------------------------------------------------------------------- + +class TestHookBus: + def test_emit_with_no_subscribers_calls_nothing_and_returns(self): + bus = HookBus() + bus.emit("on_document_ingested", {"document_id": 1}) + assert [bus.subscribers(h) for h in HOOKS] == [[], [], []] + + def test_a_raising_subscriber_is_skipped_and_the_next_still_runs(self): + bus, seen = HookBus(), [] + + def boom(payload): + raise RuntimeError("plugin bug") + + bus.subscribe("on_document_ingested", "bad", boom) + bus.subscribe("on_document_ingested", "good", lambda p: seen.append(p["document_id"])) + bus.emit("on_document_ingested", {"document_id": 7}) + assert seen == [7] + + def test_each_subscriber_gets_its_own_copy_of_the_payload(self): + bus, seen = HookBus(), [] + bus.subscribe("on_tool_invocation", "a", lambda p: p.update(tool="changed")) + bus.subscribe("on_tool_invocation", "b", lambda p: seen.append(p["tool"])) + payload = {"tool": "calculate"} + bus.emit("on_tool_invocation", payload) + assert (seen, payload) == (["calculate"], {"tool": "calculate"}) + + def test_unsubscribe_removes_only_that_owners_subscribers(self): + bus = HookBus() + bus.subscribe("on_scheduler_tick", "a", lambda p: None) + bus.subscribe("on_document_ingested", "a", lambda p: None) + bus.subscribe("on_scheduler_tick", "b", lambda p: None) + bus.unsubscribe("a") + assert (bus.subscribers("on_scheduler_tick"), bus.subscribers("on_document_ingested")) == ( + ["b"], []) + + def test_subscribing_to_an_unknown_hook_is_refused(self): + with pytest.raises(ValueError, match="unknown hook"): + HookBus().subscribe("on_route_mount", "a", lambda p: None) + + def test_emitting_an_unknown_hook_is_refused(self): + with pytest.raises(ValueError, match="unknown hook"): + HookBus().emit("on_route_mount", {}) + + +class TestScheduler: + def test_ticks_are_numbered_and_stop_ends_them(self): + bus, ticks, third = HookBus(), [], threading.Event() + + def record(payload): + ticks.append(payload["tick"]) + if len(ticks) == 3: + third.set() + + bus.subscribe("on_scheduler_tick", "t", record) + scheduler = Scheduler(bus, 0.01) + scheduler.start() + assert third.wait(5) + scheduler.stop() + stopped_at = len(ticks) + threading.Event().wait(0.05) + assert (ticks[:3], len(ticks)) == ([1, 2, 3], stopped_at) + + +# --------------------------------------------------------------------------- +# The manifest and the loader +# --------------------------------------------------------------------------- + +def test_a_declared_hook_without_its_function_is_refused(): + module = ModuleType("p") + module.PLUGIN_HOOKS = ["on_document_ingested"] + with pytest.raises(ManifestError, match="no function: on_document_ingested"): + read_manifest(module) + + +def test_on_route_mount_is_not_in_the_catalogue_yet(): + module = ModuleType("p") + module.PLUGIN_HOOKS = ["on_route_mount"] + module.on_route_mount = lambda payload: None + with pytest.raises(ManifestError, match="unknown hook"): + read_manifest(module) + + +_SUBSCRIBER = ''' +PLUGIN_HOOKS = ["on_document_ingested"] +SEEN = [] + +def on_document_ingested(payload): + SEEN.append(payload["document_id"]) + +def setup(services): + pass +''' + + +def _loader(tmp_path: Path, body: str = _SUBSCRIBER) -> tuple[PluginLoader, HookBus]: + (tmp_path / "listener.py").write_text(textwrap.dedent(body), encoding="utf-8") + bus = HookBus() + return PluginLoader(registry=ToolRegistry(), bus=bus), bus + + +class TestLoaderSubscribes: + def test_a_loaded_plugin_receives_the_hook_it_declared(self, tmp_path): + loader, bus = _loader(tmp_path) + loader.load_all(tmp_path) + bus.emit("on_document_ingested", {"document_id": 5}) + seen = sys.modules["localchat_plugin_listener"].SEEN + loader.unload("listener") + assert seen == [5] + + def test_unload_removes_the_subscription(self, tmp_path): + loader, bus = _loader(tmp_path) + loader.load_all(tmp_path) + loader.unload("listener") + assert bus.subscribers("on_document_ingested") == [] + + def test_reload_leaves_one_subscription_not_two(self, tmp_path): + loader, bus = _loader(tmp_path) + loader.load_all(tmp_path) + loader.reload("listener") + owners = bus.subscribers("on_document_ingested") + loader.unload("listener") + assert owners == ["listener"] + + def test_a_plugin_whose_setup_fails_is_not_subscribed(self, tmp_path): + loader, bus = _loader(tmp_path, _SUBSCRIBER.replace(" pass", " raise RuntimeError('no')")) + assert (loader.load_all(tmp_path), bus.subscribers("on_document_ingested")) == (0, []) + + +# --------------------------------------------------------------------------- +# Where the core emits +# --------------------------------------------------------------------------- + +def _processor(embedding_model: str | None = "embed", existing: dict | None = None): + from src.rag.processor import DocumentProcessor + + processor = DocumentProcessor(db=MagicMock(), ollama_client=MagicMock()) + processor._db.documents.exists.return_value = (True, existing) if existing else (False, {}) + processor._db.documents.insert.return_value = 42 + processor._ollama_client.get_embedding_model.return_value = embedding_model + return processor + + +def _ingest(processor, path: Path) -> tuple[tuple, list[dict]]: + path.write_text("content", encoding="utf-8") + bus, seen = HookBus(), [] + bus.subscribe("on_document_ingested", "t", seen.append) + with ( + patch("src.rag.processor.hook_bus", bus), + patch("src.rag.processor._compute_file_hash", return_value=hashlib.sha256(b"new").hexdigest()), + patch.object(processor, "_load_document_chunks", + return_value=(True, None, [{"text": "a", "metadata": {}}] * 2, "content", "TXT", "v1")), + patch.object(processor, "_run_embedding_pipeline", + return_value=([{"chunk_text": "a"}, {"chunk_text": "b"}], 0)), + ): + result = processor.ingest_document(str(path), workspace_id="ws-1", source_id="src-1") + return result, seen + + +class TestDocumentIngestedIsEmitted: + def test_a_new_document_is_announced_once_with_its_id_and_workspace(self, tmp_path): + (ok, _, doc_id), seen = _ingest(_processor(), tmp_path / "report.txt") + assert (ok, doc_id, seen) == (True, 42, [{ + "document_id": 42, "filename": "report.txt", "workspace_id": "ws-1", + "chunks": 2, "replaced": False, "source_id": "src-1", + }]) + + def test_a_replaced_document_is_announced_as_replaced_under_its_own_id(self, tmp_path): + processor = _processor(existing={"id": 9, "content_hash": "old"}) + (ok, _, doc_id), seen = _ingest(processor, tmp_path / "report.txt") + assert (ok, doc_id, [(e["document_id"], e["replaced"]) for e in seen]) == (True, 9, [(9, True)]) + + def test_a_failed_ingest_announces_nothing(self, tmp_path): + (ok, _, _), seen = _ingest(_processor(embedding_model=None), tmp_path / "report.txt") + assert (ok, seen) == (False, []) + + +def _registry() -> ToolRegistry: + reg = ToolRegistry() + reg.register("open_tool", "anyone", {"type": "object", "properties": {}})(lambda: "open") + reg.register("owner_tool", "owners", {"type": "object", "properties": {}}, + source="gated", min_role="owner")(lambda: "secret") + return reg + + +async def _announced(calls: list[str], role: str) -> list[dict]: + replies = iter([ + {"message": {"content": "", "tool_calls": [ + {"function": {"name": n, "arguments": {}}} for n in calls]}}, + {"message": {"content": "done"}}, + ]) + + async def completion(model, messages, tools=None): + return next(replies) + + client = MagicMock() + client.generate_chat_completion = completion + bus, seen = HookBus(), [] + bus.subscribe("on_tool_invocation", "t", seen.append) + with patch("src.tools.executor.hook_bus", bus), request_caller(RequestCaller("u", role)): + out = [c async for c in ToolExecutor(client, _registry()).execute( + "m", [{"role": "user", "content": "hi"}])] + if "".join(out) != "done": + raise AssertionError(f"the tool loop did not finish: {out!r}") + return seen + + +class TestToolInvocationIsEmitted: + async def test_each_tool_that_ran_is_announced_with_its_source(self): + seen = await _announced(["open_tool", "owner_tool"], "owner") + assert seen == [{"tool": "open_tool", "source": "builtin"}, + {"tool": "owner_tool", "source": "gated"}] + + async def test_a_refused_or_unknown_tool_is_not_announced(self): + seen = await _announced(["owner_tool", "no_such_tool", "open_tool"], "editor") + assert seen == [{"tool": "open_tool", "source": "builtin"}] diff --git a/tests/unit/test_plugin_contract.py b/tests/unit/test_plugin_contract.py index 6f8d8125..5c9f6234 100644 --- a/tests/unit/test_plugin_contract.py +++ b/tests/unit/test_plugin_contract.py @@ -83,6 +83,7 @@ def test_declared_values_are_kept(self): m = read_manifest(_module( PLUGIN_SCOPE="hybrid", PLUGIN_MIN_ROLE="owner", PLUGIN_HOOKS=["on_scheduler_tick"], PLUGIN_CONFIG={"LIMIT": ("int", 3)}, + on_scheduler_tick=lambda payload: None, )) assert (m.scope, m.min_role, m.hooks, m.config) == ( "hybrid", "owner", ("on_scheduler_tick",), {"LIMIT": ("int", 3)},