From 38ce15c42090eaaaf245343c52b13eef24622c77 Mon Sep 17 00:00:00 2001 From: huashen <2494946808@qq.com> Date: Sat, 29 Aug 2026 22:50:06 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=B9=E7=94=A8=E6=A8=A1=E5=9E=8B=E5=B5=8C?= =?UTF-8?q?=E5=85=A5=E8=B5=84=E6=BA=90=E8=AF=84=E5=88=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .github/workflows/plugin-api-v3.yml | 2 +- README.md | 7 +- plugin.py | 44 +++---- tests/test_plugin.py | 192 +++++++++++++++++++--------- 4 files changed, 152 insertions(+), 93 deletions(-) diff --git a/.github/workflows/plugin-api-v3.yml b/.github/workflows/plugin-api-v3.yml index d4dc204..a5d03fa 100644 --- a/.github/workflows/plugin-api-v3.yml +++ b/.github/workflows/plugin-api-v3.yml @@ -36,7 +36,7 @@ jobs: - uses: actions/checkout@v4 with: repository: kachofugetsu09/akashic-agent - ref: 9da3a988a2bf62b0f550bd4f6bb98c4eeb1f56f5 + ref: 0607e546de923e9377b04fb841d01384b1a666c1 path: .akashic-core - uses: actions/setup-python@v5 with: diff --git a/README.md b/README.md index f3c885f..2f6ed15 100644 --- a/README.md +++ b/README.md @@ -36,10 +36,9 @@ history reader 始终使用 SQLite `mode=ro`。数据库不存在表示合法空 若尚无三个 preview 列，history 会把这三个稳定字段投影为 `null` 后参与 canonical hash； 旧表经既有 `open_db` 逐列 `ALTER` 后形成的 SQLite schema 也属于同一已知 lineage。 -非引用评分使用 Core 正式运行时的共享 HTTP resources。嵌入配置从 -`AKASHIC_CONFIG` 指向的 Core 配置加载，不从插件 checkout 的当前目录猜测配置。 -embedding 继续使用既有 Core provider 数据流；API key 只作为运行时认证，不进入 -inbox、projection 或 typed event，完整正文也不进入这些持久/发布边界。 +非引用评分只注入公共 `EMBEDDINGS` service，不读取 Core 配置、provider 或凭据。 +后台评分每次都在当前 generation 的 `runtime_scope` 内执行 `bind → embed`，因此热更新 +前后的任务不会混用模型快照。向量和完整正文不进入 inbox、projection 或 history。 旧 v2 `scripts/backfill_proactive_feedback.py` 已移除：它直接操作 `workspace/proactive_feedback/proactive_feedback.db`，而 `--clear` 会删除旧 DB、WAL diff --git a/plugin.py b/plugin.py index 0188c95..d27808f 100644 --- a/plugin.py +++ b/plugin.py @@ -3,14 +3,14 @@ import asyncio import json import logging -import os from collections.abc import Callable, Iterable, Iterator from pathlib import Path from typing import Any, cast -from agent.config_models import Config as CoreConfig from agent.plugin_composition import ( Context, + EMBEDDINGS, + Embeddings, MobileUiDefinition, MobileUiNavigation, MobileUiRpcInvalidRequest, @@ -20,8 +20,6 @@ ) from agent.turn_events.after_turn import AFTER_TURN_COMMITTED from bus.events_lifecycle import TurnCommitted -from core.net.http import get_default_http_requester -from memory2.embedder import Embedder from .dashboard import ProactiveFeedbackDashboardReader from .db import ( @@ -37,6 +35,7 @@ pending_feedback_inputs, ) from .scorer import ( + EmbedBatch, MessageRow, message_rows_from_snapshot, parse_quote_parts, @@ -61,7 +60,7 @@ version = "3.0.0" desc = "记录主动消息被继续的反馈，并提供桌面与移动只读投影。" author = "Akashic" -inject = (SESSION_READ, UI_SLOTS) +inject = (SESSION_READ, UI_SLOTS, EMBEDDINGS) skill_roots: tuple[str, ...] = () drift_skill_roots: tuple[str, ...] = () workspace_roots: tuple[str, ...] = () @@ -75,10 +74,11 @@ async def apply(ctx: Context, config: object) -> None: _ = config session_read = ctx.require(SESSION_READ) ui_slots = ctx.require(UI_SLOTS) + embeddings = ctx.require(EMBEDDINGS) db_path = ctx.data_root / _FEEDBACK_DB_NAME runtime = ProactiveFeedbackRuntime( session_read=session_read, - workspace=ctx.runtime.workspace, + embed_batch=_bind_embeddings(embeddings, ctx), db_path=db_path, ) _ = await ctx.provide( @@ -111,18 +111,17 @@ def __init__( self, *, session_read: SessionReadService, - workspace: Path, + embed_batch: EmbedBatch, db_path: Path, session_keys: Callable[[], Iterable[str]] | None = None, ) -> None: self._session_read = session_read - self._workspace = workspace + self._embed_batch = embed_batch self._db_path = db_path self._session_keys = session_keys or ( lambda: _formal_session_keys_from_read_service(session_read) ) self._queue: asyncio.Queue[int] = asyncio.Queue(maxsize=_QUEUE_MAX) - self._embedder: Embedder | None = None self._discovery_done = False def observe_committed(self, event: TurnCommitted) -> None: @@ -270,7 +269,7 @@ async def _process( # 3. Persist one deduplicated projection, including bounded display text. try: scored = await score_followup( - embed_batch=self._get_embedder().embed_batch if allow_pua else _no_embed, + embed_batch=self._embed_batch if allow_pua else _no_embed, user=user, assistant=assistant, candidates=candidates, @@ -449,11 +448,6 @@ async def _process_input_record(self, record: FeedbackInputRecord) -> None: input_row_id=record.row_id, ) - def _get_embedder(self) -> Embedder: - if self._embedder is None: - self._embedder = _build_embedder(self._workspace) - return self._embedder - def _discover_committed_inputs(self) -> None: """Discover bounded eligible Turns committed before the callback fanout.""" @@ -687,18 +681,14 @@ def _bounded_session_keys( return (*unique[: limit - 1], unique[-1]) -def _build_embedder(workspace: Path) -> Embedder: - config_path = os.environ.get("AKASHIC_CONFIG", "").strip() - if not config_path: - raise RuntimeError("proactive_feedback 需要 Core 的 AKASHIC_CONFIG") - embedding = CoreConfig.load(path=config_path, workspace=workspace).memory.embedding - return Embedder( - base_url=embedding.base_url, - api_key=embedding.api_key, - model=embedding.model, - output_dimensionality=embedding.output_dimensionality, - requester=get_default_http_requester("external_default"), - ) +def _bind_embeddings(embeddings: Embeddings, ctx: Context) -> EmbedBatch: + async def embed_batch(texts: list[str]) -> list[list[float]]: + async with ctx.runtime_scope(): + async with embeddings.bind() as bound: + result = await bound.embed(texts) + return [list(vector) for vector in result.vectors] + + return embed_batch def _decode_outbox_payload(payload_json: str) -> dict[str, object]: diff --git a/tests/test_plugin.py b/tests/test_plugin.py index 5e836c4..e1046aa 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -5,8 +5,10 @@ import importlib.util import inspect import json +import os import shutil import sqlite3 +import subprocess import sys from pathlib import Path from types import SimpleNamespace @@ -17,6 +19,7 @@ from agent.plugin_composition import SessionReadService, SessionReadSnapshot from agent.plugins.composable import ComposablePlugin from agent.plugins.dashboard_host import DashboardBinding, PluginDashboardHost +from agent.plugins.install import install_git_plugin from agent.plugins.manager import PluginManager from agent.plugins.mobile_ui import PluginMobileUiProvider from agent.plugins.manifest import write_plugin_manifest @@ -47,6 +50,28 @@ def _load_plugin_module(): FeedbackEvent = module.FeedbackEvent +async def _embed_batch(texts: list[str]) -> list[list[float]]: + return [[1.0, 0.0] for _ in texts] + + +def _commit_plugin(repo: Path) -> None: + for args in ( + ("init",), + ("config", "user.name", "test"), + ("config", "user.email", "test@example.com"), + ("add", "."), + ("commit", "-m", "fixture"), + ): + result = subprocess.run( + ("git", *args), + cwd=repo, + capture_output=True, + text=True, + env=os.environ.copy(), + ) + assert result.returncode == 0, result.stderr + + def _event(*, quoted: bool = True) -> TurnCommitted: user = ( "被回复消息：主动提醒某个很长很长的主题\n\n【你当前新消息】我继续这个主题" @@ -223,43 +248,44 @@ def test_v2_runtime_symbols_are_not_used_by_module() -> None: assert "【你当前新消息】" not in source -def test_embedder_uses_core_config_and_shared_http_requester( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - config_path = tmp_path / "runtime.toml" - requester = object() - embedding = SimpleNamespace( - base_url="https://embedding.example/v1", - api_key="test-key", - model="text-embedding-v3", - output_dimensionality=1024, +@pytest.mark.asyncio +async def test_embedding_service_is_bound_inside_the_generation_scope() -> None: + events: list[str] = [] + + class Scope: + def __init__(self, value: object = None) -> None: + self.value = value + + async def __aenter__(self) -> object: + events.append("enter") + return self.value + + async def __aexit__(self, *_args: object) -> None: + events.append("exit") + + class Bound: + async def embed(self, texts: list[str]) -> object: + events.append("embed") + assert texts == ["one", "two"] + return SimpleNamespace(vectors=((1.0, 0.0), (0.0, 1.0))) + + class Embeddings: + def bind(self) -> Scope: + events.append("bind") + return Scope(Bound()) + + class Context: + def runtime_scope(self) -> Scope: + events.append("scope") + return Scope() + + embed_batch = module._bind_embeddings( + cast(Any, Embeddings()), + cast(Any, Context()), ) - seen: list[tuple[Path, Path]] = [] - - def fake_load(path: str | Path, *, workspace: str | Path) -> object: - seen.append((Path(path), Path(workspace))) - return SimpleNamespace(memory=SimpleNamespace(embedding=embedding)) - - class FakeEmbedder: - def __init__(self, **kwargs: object) -> None: - self.kwargs = kwargs - monkeypatch.setenv("AKASHIC_CONFIG", str(config_path)) - monkeypatch.setattr(module.CoreConfig, "load", fake_load) - monkeypatch.setattr(module, "get_default_http_requester", lambda _profile: requester) - monkeypatch.setattr(module, "Embedder", FakeEmbedder) - - embedder = module._build_embedder(tmp_path) - - assert seen == [(config_path, tmp_path)] - assert embedder.kwargs == { - "base_url": embedding.base_url, - "api_key": embedding.api_key, - "model": embedding.model, - "output_dimensionality": embedding.output_dimensionality, - "requester": requester, - } + assert await embed_batch(["one", "two"]) == [[1.0, 0.0], [0.0, 1.0]] + assert events == ["scope", "enter", "bind", "enter", "embed", "exit", "exit"] @pytest.mark.asyncio @@ -268,7 +294,7 @@ async def test_committed_turn_writes_plugin_owned_projection(tmp_path: Path) -> session_read=SessionReadService( lambda _key: (cast(Any, _session_state()), None) ), - workspace=tmp_path, + embed_batch=_embed_batch, db_path=tmp_path / "data" / "proactive_feedback.db", ) await runtime._process(_event()) @@ -294,7 +320,7 @@ async def test_feedback_commit_is_readable_from_stable_history(tmp_path: Path) - session_read=SessionReadService( lambda _key: (cast(Any, _session_state()), None) ), - workspace=tmp_path, + embed_batch=_embed_batch, db_path=tmp_path / "data" / "proactive_feedback.db", ) await runtime._process(_event()) @@ -318,7 +344,7 @@ def test_committed_turn_identity_is_durable_without_message_text(tmp_path: Path) session_read=SessionReadService( lambda _key: (cast(Any, _session_state()), None) ), - workspace=tmp_path, + embed_batch=_embed_batch, db_path=tmp_path / "data" / "proactive_feedback.db", ) runtime.enqueue(_event()) @@ -347,7 +373,7 @@ def test_candidate_enqueue_fails_before_any_write(tmp_path: Path) -> None: db_path = tmp_path / "data" / "proactive_feedback.db" runtime = module.ProactiveFeedbackRuntime( session_read=SessionReadService.candidate_validation(), - workspace=tmp_path, + embed_batch=_embed_batch, db_path=db_path, ) @@ -363,14 +389,14 @@ async def test_durable_input_replays_after_runtime_restart(tmp_path: Path) -> No session_read=SessionReadService( lambda _key: (cast(Any, _session_state()), None) ), - workspace=tmp_path, + embed_batch=_embed_batch, db_path=tmp_path / "data" / "proactive_feedback.db", ) original.enqueue(_event()) restarted = module.ProactiveFeedbackRuntime( session_read=original._session_read, - workspace=tmp_path, + embed_batch=_embed_batch, db_path=original._db_path, ) assert await restarted._process_pending_inputs() is False @@ -402,14 +428,14 @@ async def test_durable_input_keeps_ordered_two_user_ids_and_scores_one_turn( ) original = module.ProactiveFeedbackRuntime( session_read=session_read, - workspace=tmp_path, + embed_batch=_embed_batch, db_path=tmp_path / "data" / "proactive_feedback.db", ) original.enqueue(_two_user_event()) restarted = module.ProactiveFeedbackRuntime( session_read=session_read, - workspace=tmp_path, + embed_batch=_embed_batch, db_path=original._db_path, ) assert await restarted._process_pending_inputs() is False @@ -444,7 +470,7 @@ async def test_formal_boot_discovers_committed_turn_without_callback_once( ) restarted = module.ProactiveFeedbackRuntime( session_read=session_read, - workspace=tmp_path, + embed_batch=_embed_batch, db_path=tmp_path / "data" / "proactive_feedback.db", session_keys=lambda: ("mobile:test",), ) @@ -501,7 +527,7 @@ def lookup(session_key: str) -> tuple[Any, None]: runtime = module.ProactiveFeedbackRuntime( session_read=SessionReadService(lookup), - workspace=tmp_path, + embed_batch=_embed_batch, db_path=db_path, session_keys=lambda: (*old_keys, new_key), ) @@ -560,7 +586,7 @@ def track_insert(*args: Any, **kwargs: Any) -> int: monkeypatch.setattr(module, "insert_feedback_input", track_insert) runtime = module.ProactiveFeedbackRuntime( session_read=SessionReadService(lookup), - workspace=tmp_path, + embed_batch=_embed_batch, db_path=db_path, session_keys=lambda: (*old_keys, new_key), ) @@ -599,7 +625,7 @@ async def blocked_scoring(**_kwargs: object) -> None: session_read=SessionReadService( lambda _key: (cast(Any, _session_state()), None) ), - workspace=tmp_path, + embed_batch=_embed_batch, db_path=tmp_path / "data" / "proactive_feedback.db", ) runtime.enqueue(_event()) @@ -691,7 +717,7 @@ async def test_candidate_session_read_fails_before_any_write(tmp_path: Path) -> db_path = tmp_path / "data" / "proactive_feedback.db" runtime = module.ProactiveFeedbackRuntime( session_read=SessionReadService.candidate_validation(), - workspace=tmp_path, + embed_batch=_embed_batch, db_path=db_path, ) with pytest.raises(RuntimeError, match="禁止读取正式 Session"): @@ -701,23 +727,16 @@ async def test_candidate_session_read_fails_before_any_write(tmp_path: Path) -> @pytest.mark.asyncio async def test_nonquoted_turn_keeps_pua_scoring_path( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: runtime = module.ProactiveFeedbackRuntime( session_read=SessionReadService( lambda _key: (cast(Any, _session_state(quoted=False)), None) ), - workspace=tmp_path, + embed_batch=_embed_batch, db_path=tmp_path / "data" / "proactive_feedback.db", ) - class EmbedderStub: - async def embed_batch(self, texts: list[str]) -> list[list[float]]: - assert len(texts) == 3 - return [[1.0, 0.0], [1.0, 0.0], [1.0, 0.0]] - - monkeypatch.setattr(runtime, "_get_embedder", lambda: EmbedderStub()) await runtime._process(_event(quoted=False)) conn = sqlite3.connect(runtime._db_path) try: @@ -738,7 +757,7 @@ async def test_scoring_failure_records_unscored_feedback( session_read=SessionReadService( lambda _key: (cast(Any, _session_state()), None) ), - workspace=tmp_path, + embed_batch=_embed_batch, db_path=tmp_path / "data" / "proactive_feedback.db", ) @@ -766,7 +785,7 @@ async def test_in_process_cancellation_does_not_persist_partial_feedback( session_read=SessionReadService( lambda _key: (cast(Any, _session_state()), None) ), - workspace=tmp_path, + embed_batch=_embed_batch, db_path=tmp_path / "data" / "proactive_feedback.db", ) started = asyncio.Event() @@ -788,7 +807,7 @@ async def blocked_scoring(**_kwargs: object) -> None: async def test_worker_cancellation_has_no_live_task(tmp_path: Path) -> None: runtime = module.ProactiveFeedbackRuntime( session_read=SessionReadService.candidate_validation(), - workspace=tmp_path, + embed_batch=_embed_batch, db_path=tmp_path / "data" / "proactive_feedback.db", ) task = asyncio.create_task(runtime.run_worker()) @@ -1075,7 +1094,7 @@ def test_plugin_runtime_does_not_move_legacy_database(tmp_path: Path) -> None: target_root = tmp_path / "plugin-data" runtime = module.ProactiveFeedbackRuntime( session_read=cast(Any, object()), - workspace=tmp_path / "workspace", + embed_batch=_embed_batch, db_path=target_root / "proactive_feedback.db", ) @@ -1084,9 +1103,56 @@ def test_plugin_runtime_does_not_move_legacy_database(tmp_path: Path) -> None: assert not target_root.exists() +@pytest.mark.asyncio +async def test_plugin_installs_and_loads_from_ordinary_cache(tmp_path: Path) -> None: + source = tmp_path / "source" + source.mkdir() + for path in Path(__file__).parents[1].iterdir(): + if path.is_file() and path.name != ".git": + shutil.copy2(path, source / path.name) + _commit_plugin(source) + workspace = tmp_path / "workspace" + plugin_home = tmp_path / "home" + installed = install_git_plugin( + workspace=workspace, + source=str(source), + marketplace="ordinary-test", + plugins_home=plugin_home, + ) + core_plugins = Path(inspect.getfile(PluginManager)).parents[2] / "plugins" + manager = PluginManager( + plugin_dirs=[ + core_plugins / "models", + core_plugins / "openai_compatible", + ], + event_bus=EventBus(), + tool_registry=None, + session_manager=_empty_session_manager(), + workspace=workspace, + installed_cache_root=plugin_home / "cache", + ) + try: + await manager.load_all() + generation = manager.generation("proactive_feedback@ordinary-test") + assert generation is not None + assert generation.source_type == "installed" + assert generation.plugin_dir == installed.installed_path + instance = cast(ComposablePlugin, generation.instance) + assert instance.module.__file__ is not None + assert Path(instance.module.__file__).resolve().is_relative_to( + installed.installed_path + ) + assert manager.current_snapshot is not None + assert manager.current_snapshot.composition_root is not None + assert manager.current_snapshot.composition_root.receipt().ready + finally: + await manager.terminate_all() + + @pytest.mark.asyncio async def test_manager_stable_candidate_ui_dashboard_and_cleanup(tmp_path: Path) -> None: plugin_dir = tmp_path / "plugins" / "proactive_feedback" + core_plugins = Path(inspect.getfile(PluginManager)).parents[2] / "plugins" plugin_dir.mkdir(parents=True) for filename in ( "plugin.py", @@ -1104,7 +1170,11 @@ async def test_manager_stable_candidate_ui_dashboard_and_cleanup(tmp_path: Path) plugins_home=tmp_path / "home", ) manager = PluginManager( - plugin_dirs=[tmp_path / "plugins"], + plugin_dirs=[ + tmp_path / "plugins", + core_plugins / "models", + core_plugins / "openai_compatible", + ], event_bus=EventBus(), tool_registry=None, session_manager=_empty_session_manager(),