From 7784740d35856bdfa3d0cccd7ab93f8826130a86 Mon Sep 17 00:00:00 2001 From: huashen <2494946808@qq.com> Date: Sun, 23 Aug 2026 20:10:37 +0800 Subject: [PATCH 1/2] feat: expose durable feedback history --- .github/workflows/plugin-api-v3.yml | 6 +- README.md | 38 ++-- db.py | 211 ++++++------------ history.py | 248 +++++++++++++++++++++ plugin.py | 64 +----- tests/test_plugin.py | 333 +++++++++++++--------------- 6 files changed, 500 insertions(+), 400 deletions(-) create mode 100644 history.py diff --git a/.github/workflows/plugin-api-v3.yml b/.github/workflows/plugin-api-v3.yml index dfab451..d4dc204 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: bad54a58135712793957cafede0b3e3ca89501ae + ref: 9da3a988a2bf62b0f550bd4f6bb98c4eeb1f56f5 path: .akashic-core - uses: actions/setup-python@v5 with: @@ -62,9 +62,9 @@ jobs: AKASHIC_AGENT_ROOT: .akashic-core PYTHONPATH: .akashic-core run: >- - pyright --level error plugin.py dashboard.py db.py scorer.py + pyright --level error plugin.py dashboard.py db.py history.py scorer.py scripts/migrate_v2_data.py scripts/migrate_feedback_previews.py tests - name: Compile Python sources - run: python -m compileall -q plugin.py dashboard.py db.py scorer.py tests scripts + run: python -m compileall -q plugin.py dashboard.py db.py history.py scorer.py tests scripts - name: Check diff formatting run: git diff --check diff --git a/README.md b/README.md index 30dcae1..896cd8a 100644 --- a/README.md +++ b/README.md @@ -17,24 +17,22 @@ Akashic proactive feedback plugin. assistant ID 写入 inbox，正文只在评分内存中重建。候选 generation 不执行 discovery， 因而不会写正式 DB 或事件。 -### Durable typed event - -Core exact `20062a715d2c5822228b327863b51c8d036119b3` 提供唯一的 -`agent.turn_events.proactive_feedback.PROACTIVE_FEEDBACK_COMMITTED` Observe seam。 -每次评分结果都在一次 SQLite commit 中同时写入 `proactive_feedback_events` 和 -`proactive_feedback_outbox`；commit 返回后才调用 -`ctx.observe(PROACTIVE_FEEDBACK_COMMITTED, ProactiveFeedbackCommitted(...))`。 - -事件 `event_id` 固定为 `proactive_feedback:`，DTO 使用 Core 的 -`session_key`、ordered user message identity（DTO 使用该 Turn 最后一条 user ID）、 -assistant/proactive message identity、评分、`reason` 和最多 -2400 字符的 user/assistant/proactive preview。全文不进入事件。发布成功后同库的 -`proactive_feedback_published_cursor` 与 outbox receipt 一起推进；发布失败、进程内 -取消或 Core 重启都会保留 pending 行，正式 generation 启动时按 row 顺序重放。消费方 -必须按 `event_id` 幂等。已发布的 projection/outbox receipt 是不可变事实；重复的 -同一 identity 即使评分不同，也不会改写已发布 DTO。待发布行才允许在同一 identity -内更新；若关联 proactive identity 改变，则保留旧 published row 并创建新的 row/event。 -本插件不再向 `TurnCommitted.extra` 写入反馈，也不提供 marker fallback。 +### Durable history pull + +插件提供普通服务 `proactive-feedback.history.v1`。消费者按单调 `cursor` 调用 +`page(after_cursor, max_items)`；每条 accepted feedback 的 `event_id` 固定为 +`proactive_feedback:`，`payload_hash` 是稳定字段 canonical JSON 的 SHA-256。 +分页最多 100 条，严格按 SQLite row id 递增。 + +`proactive_feedback_events` 是 accepted feedback 的唯一 owner。第一次 accepted payload +写入后不可 UPDATE 或 DELETE；相同 Turn 的完全相同 payload 返回原 identity，任何字段 +漂移都 fail-loud。评分重试的中间计算不是新的领域事实，不另造 history。input inbox +保留 Turn identity 与处理状态。旧 outbox/cursor schema 和既有行冻结保留，新链不再写 +outbox，也不调用 Core Observe event。 + +history reader 始终使用 SQLite `mode=ro`。数据库不存在表示合法空历史且不会创建目录； +数据库存在但损坏、schema 异构或字段类型无效会 fail-loud，不能伪装成空页。插件不向 +`TurnCommitted.extra` 回写结果，也不依赖 Wake、Content 或消费者数据库。 非引用评分使用 Core 正式运行时的共享 HTTP resources。嵌入配置从 `AKASHIC_CONFIG` 指向的 Core 配置加载，不从插件 checkout 的当前目录猜测配置。 @@ -64,8 +62,8 @@ python scripts/migrate_feedback_previews.py \ ``` 插件不再声明 v2 `Plugin` class、EventBus listener、`ProactiveFeedbackRecorded` 或 tool -ABI；v3 运行路径只观察 Core 的 committed Turn，并通过上述 typed event 发布已持久化 -反馈。 +ABI；v3 运行路径只观察 Core 的 committed Turn，并通过上述只读 history service 暴露 +已持久化反馈。 ## 移动端看板 diff --git a/db.py b/db.py index 6a7c48f..578a2e3 100644 --- a/db.py +++ b/db.py @@ -5,6 +5,8 @@ from dataclasses import dataclass from pathlib import Path +from .history import accepted_payload_hash + @dataclass(frozen=True) class FeedbackEvent: @@ -137,30 +139,27 @@ def open_db(path: Path) -> sqlite3.Connection: def insert_feedback(conn: sqlite3.Connection, event: FeedbackEvent) -> int | None: - """Atomically replace one feedback row and enqueue its typed event.""" + """Append one immutable accepted feedback fact or verify an exact duplicate.""" # 1. Reject a proactive message already owned by another user reply. if _feedback_owned_by_other(conn, event): return None - # 2. Keep one row identity for duplicate committed Turns. - existing_id = _existing_feedback_id(conn, event) - if existing_id is not None: - try: - if not _feedback_is_published(conn, existing_id): - _update_feedback_row(conn, event, existing_id) - _upsert_feedback_outbox(conn, event, existing_id) - conn.commit() - except (sqlite3.Error, RuntimeError, TypeError, ValueError): - conn.rollback() - raise - return existing_id + # 2. The first accepted payload owns the Turn identity forever. + existing = _existing_feedback(conn, event) + if existing is not None: + expected_hash = accepted_payload_hash(_accepted_payload(event)) + actual_hash = accepted_payload_hash(_accepted_payload_from_row(existing)) + if actual_hash != expected_hash: + raise RuntimeError( + "accepted feedback payload 漂移: " + f"proactive_feedback:{int(existing['id'])}" + ) + return int(existing["id"]) - # 3. Replace this user's previous projection and pending outbox row together. + # 3. Append the accepted fact without touching the frozen legacy outbox. try: - _remove_previous_feedback(conn, event.user_message_id) row_id = _insert_feedback_row(conn, event) - _upsert_feedback_outbox(conn, event, row_id) conn.commit() except (sqlite3.Error, RuntimeError, TypeError, ValueError): conn.rollback() @@ -390,91 +389,68 @@ def _feedback_owned_by_other( return row is not None -def _existing_feedback_id( +def _existing_feedback( conn: sqlite3.Connection, event: FeedbackEvent, -) -> int | None: +) -> sqlite3.Row | None: row = conn.execute( """ - SELECT id + SELECT id, session_key, user_message_id, assistant_message_id, + proactive_message_id, feedback_type, confidence, + pa_score, pua_score, lag_seconds, candidate_count, + matched_by, reason, user_content_preview, + assistant_content_preview, proactive_content_preview FROM proactive_feedback_events - WHERE user_message_id = ? AND proactive_message_id IS ? + WHERE session_key = ? AND user_message_id = ? + ORDER BY id ASC LIMIT 1 """, - (event.user_message_id, event.proactive_message_id), + (event.session_key, event.user_message_id), ).fetchone() - return None if row is None else int(row["id"]) + return row -def _feedback_is_published(conn: sqlite3.Connection, row_id: int) -> bool: - row = conn.execute( - """ - SELECT published_at - FROM proactive_feedback_outbox - WHERE row_id = ? - """, - (row_id,), - ).fetchone() - return row is not None and row["published_at"] is not None - - -def _update_feedback_row( - conn: sqlite3.Connection, - event: FeedbackEvent, - row_id: int, -) -> None: - _ = conn.execute( - """ - UPDATE proactive_feedback_events - SET session_key = ?, assistant_message_id = ?, feedback_type = ?, - confidence = ?, pa_score = ?, pua_score = ?, lag_seconds = ?, - candidate_count = ?, matched_by = ?, reason = ?, - user_content_preview = ?, assistant_content_preview = ?, - proactive_content_preview = ? - WHERE id = ? - """, - ( - event.session_key, - event.assistant_message_id, - event.feedback_type, - event.confidence, - event.pa_score, - event.pua_score, - event.lag_seconds, - event.candidate_count, - event.matched_by, - event.reason, - event.user_content_preview, - event.assistant_content_preview, - event.proactive_content_preview, - row_id, - ), - ) +def _accepted_payload(event: FeedbackEvent) -> dict[str, object]: + return { + "session_key": event.session_key, + "user_message_id": event.user_message_id, + "assistant_message_id": event.assistant_message_id, + "proactive_message_id": event.proactive_message_id, + "feedback_type": event.feedback_type, + "confidence": event.confidence, + "pa_score": event.pa_score, + "pua_score": event.pua_score, + "lag_seconds": event.lag_seconds, + "candidate_count": event.candidate_count, + "matched_by": event.matched_by, + "reason": event.reason, + "user_content_preview": event.user_content_preview, + "assistant_content_preview": event.assistant_content_preview, + "proactive_content_preview": event.proactive_content_preview, + } -def _remove_previous_feedback(conn: sqlite3.Connection, user_message_id: str) -> None: - pending = conn.execute( - """ - SELECT events.id, outbox.row_id - FROM proactive_feedback_events AS events - LEFT JOIN proactive_feedback_outbox AS outbox - ON outbox.row_id = events.id - WHERE events.user_message_id = ? - AND (outbox.row_id IS NULL OR outbox.published_at IS NULL) - """, - (user_message_id,), - ).fetchall() - for row in pending: - _ = conn.execute( - "DELETE FROM proactive_feedback_events WHERE id = ?", - (int(row["id"]),), - ) - if row["row_id"] is None: - continue - _ = conn.execute( - "DELETE FROM proactive_feedback_outbox WHERE row_id = ?", - (int(row["row_id"]),), +def _accepted_payload_from_row(row: sqlite3.Row) -> dict[str, object]: + return { + field: row[field] + for field in ( + "session_key", + "user_message_id", + "assistant_message_id", + "proactive_message_id", + "feedback_type", + "confidence", + "pa_score", + "pua_score", + "lag_seconds", + "candidate_count", + "matched_by", + "reason", + "user_content_preview", + "assistant_content_preview", + "proactive_content_preview", ) + } def _insert_feedback_row(conn: sqlite3.Connection, event: FeedbackEvent) -> int: @@ -511,65 +487,6 @@ def _insert_feedback_row(conn: sqlite3.Connection, event: FeedbackEvent) -> int: return int(cursor.lastrowid) -def _upsert_feedback_outbox( - conn: sqlite3.Connection, - event: FeedbackEvent, - row_id: int, -) -> None: - event_id = f"proactive_feedback:{row_id}" - payload_json = json.dumps( - _feedback_payload(event_id, event), - ensure_ascii=False, - separators=(",", ":"), - ) - outbox = conn.execute( - """ - SELECT published_at - FROM proactive_feedback_outbox - WHERE row_id = ? AND event_id = ? - """, - (row_id, event_id), - ).fetchone() - if outbox is None: - _ = conn.execute( - """ - INSERT INTO proactive_feedback_outbox(row_id, event_id, payload_json) - VALUES (?, ?, ?) - """, - (row_id, event_id, payload_json), - ) - elif outbox["published_at"] is None: - _ = conn.execute( - """ - UPDATE proactive_feedback_outbox - SET payload_json = ? - WHERE row_id = ? AND event_id = ? - """, - (payload_json, row_id, event_id), - ) - - -def _feedback_payload(event_id: str, event: FeedbackEvent) -> dict[str, object]: - return { - "event_id": event_id, - "session_key": event.session_key, - "user_message_id": event.user_message_id, - "assistant_message_id": event.assistant_message_id, - "proactive_message_id": event.proactive_message_id, - "feedback_type": event.feedback_type, - "confidence": event.confidence, - "pa_score": event.pa_score, - "pua_score": event.pua_score, - "lag_seconds": event.lag_seconds, - "candidate_count": event.candidate_count, - "matched_by": event.matched_by, - "reason": event.reason, - "user_content_preview": event.user_content_preview, - "assistant_content_preview": event.assistant_content_preview, - "proactive_content_preview": event.proactive_content_preview, - } - - def _feedback_input_record(row: sqlite3.Row) -> FeedbackInputRecord: user_message_id = str(row["user_message_id"]) return FeedbackInputRecord( diff --git a/history.py b/history.py new file mode 100644 index 0000000..05a8d2a --- /dev/null +++ b/history.py @@ -0,0 +1,248 @@ +from __future__ import annotations + +import hashlib +import json +import sqlite3 +from dataclasses import dataclass +from pathlib import Path +from typing import Protocol + +from agent.plugin_composition import ServiceKey + + +_MAX_PAGE_SIZE = 100 +_FEEDBACK_TYPES = frozenset( + {"explicit_quote", "topic_follow", "no_topic_follow", "unscored"} +) +_CONFIDENCE = frozenset({"gold", "high", "medium", "low"}) +_REQUIRED_COLUMNS = frozenset( + { + "id", + "session_key", + "user_message_id", + "assistant_message_id", + "proactive_message_id", + "feedback_type", + "confidence", + "pa_score", + "pua_score", + "lag_seconds", + "candidate_count", + "matched_by", + "reason", + "user_content_preview", + "assistant_content_preview", + "proactive_content_preview", + } +) + + +@dataclass(frozen=True, slots=True) +class FeedbackHistoryRecord: + cursor: int + event_id: str + payload_hash: str + session_key: str + user_message_id: str + assistant_message_id: str + proactive_message_id: str | None + feedback_type: str + confidence: str + pa_score: float | None + pua_score: float | None + lag_seconds: int | None + candidate_count: int + matched_by: str + reason: str + user_content_preview: str | None + assistant_content_preview: str | None + proactive_content_preview: str | None + + +@dataclass(frozen=True, slots=True) +class FeedbackHistoryPage: + after_cursor: int + records: tuple[FeedbackHistoryRecord, ...] + + +class FeedbackHistory(Protocol): + def page(self, *, after_cursor: int, max_items: int) -> FeedbackHistoryPage: ... + + +PROACTIVE_FEEDBACK_HISTORY = ServiceKey[FeedbackHistory]( + "proactive-feedback.history.v1" +) + + +class SqliteFeedbackHistory: + """Read immutable accepted feedback from the plugin-owned SQLite history.""" + + def __init__(self, db_path: Path) -> None: + self._db_path = db_path + + def page(self, *, after_cursor: int, max_items: int) -> FeedbackHistoryPage: + """Return one stable cursor-ordered page without creating or migrating data.""" + + # 1. Validate the cross-plugin request before opening plugin data. + if isinstance(after_cursor, bool) or not isinstance(after_cursor, int): + raise TypeError("after_cursor 必须是整数") + if after_cursor < 0: + raise ValueError("after_cursor 不得小于零") + if isinstance(max_items, bool) or not isinstance(max_items, int): + raise TypeError("max_items 必须是整数") + if max_items < 1 or max_items > _MAX_PAGE_SIZE: + raise ValueError(f"max_items 必须在 1..{_MAX_PAGE_SIZE} 之间") + if not self._db_path.exists(): + return FeedbackHistoryPage(after_cursor=after_cursor, records=()) + + # 2. Existing bytes must satisfy the exact read schema; no DDL is allowed. + connection = sqlite3.connect( + self._db_path.resolve().as_uri() + "?mode=ro", + uri=True, + ) + connection.row_factory = sqlite3.Row + try: + _validate_schema(connection) + rows = connection.execute( + """ + SELECT id, session_key, user_message_id, assistant_message_id, + proactive_message_id, feedback_type, confidence, + pa_score, pua_score, lag_seconds, candidate_count, + matched_by, reason, user_content_preview, + assistant_content_preview, proactive_content_preview + FROM proactive_feedback_events + WHERE id > ? + ORDER BY id ASC + LIMIT ? + """, + (after_cursor, max_items), + ).fetchall() + finally: + connection.close() + + # 3. Decode every row at the trust boundary and freeze its content hash. + records = tuple(_record_from_row(row) for row in rows) + return FeedbackHistoryPage(after_cursor=after_cursor, records=records) + + +def accepted_payload_hash(payload: dict[str, object]) -> str: + encoded = json.dumps( + payload, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _validate_schema(connection: sqlite3.Connection) -> None: + tables = { + str(row[0]) + for row in connection.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ) + } + if "proactive_feedback_events" not in tables: + raise RuntimeError("proactive_feedback history 缺少 events 表") + columns = { + str(row[1]) + for row in connection.execute("PRAGMA table_info(proactive_feedback_events)") + } + missing = sorted(_REQUIRED_COLUMNS - columns) + if missing: + raise RuntimeError( + "proactive_feedback history schema 缺少列: " + ", ".join(missing) + ) + + +def _record_from_row(row: sqlite3.Row) -> FeedbackHistoryRecord: + cursor = _positive_int(row["id"], "id") + payload = { + "session_key": _required_text(row["session_key"], "session_key"), + "user_message_id": _required_text( + row["user_message_id"], "user_message_id" + ), + "assistant_message_id": _required_text( + row["assistant_message_id"], "assistant_message_id" + ), + "proactive_message_id": _optional_text( + row["proactive_message_id"], "proactive_message_id" + ), + "feedback_type": _enum_text( + row["feedback_type"], "feedback_type", _FEEDBACK_TYPES + ), + "confidence": _enum_text(row["confidence"], "confidence", _CONFIDENCE), + "pa_score": _optional_score(row["pa_score"], "pa_score"), + "pua_score": _optional_score(row["pua_score"], "pua_score"), + "lag_seconds": _optional_nonnegative_int( + row["lag_seconds"], "lag_seconds" + ), + "candidate_count": _nonnegative_int( + row["candidate_count"], "candidate_count" + ), + "matched_by": _required_text(row["matched_by"], "matched_by"), + "reason": _required_text(row["reason"], "reason"), + "user_content_preview": _optional_text( + row["user_content_preview"], "user_content_preview" + ), + "assistant_content_preview": _optional_text( + row["assistant_content_preview"], "assistant_content_preview" + ), + "proactive_content_preview": _optional_text( + row["proactive_content_preview"], "proactive_content_preview" + ), + } + return FeedbackHistoryRecord( + cursor=cursor, + event_id=f"proactive_feedback:{cursor}", + payload_hash=accepted_payload_hash(payload), + **payload, + ) + + +def _required_text(value: object, field: str) -> str: + if not isinstance(value, str) or not value: + raise ValueError(f"history {field} 必须是非空字符串") + return value + + +def _optional_text(value: object, field: str) -> str | None: + if value is None: + return None + if not isinstance(value, str): + raise TypeError(f"history {field} 必须是字符串或 null") + return value + + +def _enum_text(value: object, field: str, choices: frozenset[str]) -> str: + text = _required_text(value, field) + if text not in choices: + raise ValueError(f"history {field} 不支持: {text}") + return text + + +def _positive_int(value: object, field: str) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value < 1: + raise ValueError(f"history {field} 必须是正整数") + return value + + +def _nonnegative_int(value: object, field: str) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise ValueError(f"history {field} 必须是非负整数") + return value + + +def _optional_nonnegative_int(value: object, field: str) -> int | None: + return None if value is None else _nonnegative_int(value, field) + + +def _optional_score(value: object, field: str) -> float | None: + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise TypeError(f"history {field} 必须是数字或 null") + score = float(value) + if score < -1.0 or score > 1.0: + raise ValueError(f"history {field} 必须在 -1..1 之间") + return score diff --git a/plugin.py b/plugin.py index 0dc704f..0188c95 100644 --- a/plugin.py +++ b/plugin.py @@ -4,7 +4,7 @@ import json import logging import os -from collections.abc import Awaitable, Callable, Iterable, Iterator +from collections.abc import Callable, Iterable, Iterator from pathlib import Path from typing import Any, cast @@ -19,10 +19,6 @@ UI_SLOTS, ) from agent.turn_events.after_turn import AFTER_TURN_COMMITTED -from agent.turn_events.proactive_feedback import ( - PROACTIVE_FEEDBACK_COMMITTED, - ProactiveFeedbackCommitted, -) from bus.events_lifecycle import TurnCommitted from core.net.http import get_default_http_requester from memory2.embedder import Embedder @@ -36,11 +32,9 @@ insert_feedback, insert_feedback_input, mark_feedback_input_processed, - mark_feedback_published, open_db, pending_feedback_input, pending_feedback_inputs, - pending_feedback_outbox, ) from .scorer import ( MessageRow, @@ -50,6 +44,7 @@ recent_proactive_messages_from_rows, score_followup, ) +from .history import PROACTIVE_FEEDBACK_HISTORY, SqliteFeedbackHistory logger = logging.getLogger("plugin.proactive_feedback") @@ -61,8 +56,6 @@ _DISCOVERY_SESSION_LIMIT = 64 _DISCOVERY_TURN_LIMIT = 256 -FeedbackPublisher = Callable[[ProactiveFeedbackCommitted], Awaitable[None]] - api_version = 3 name = "proactive_feedback" version = "3.0.0" @@ -87,10 +80,10 @@ async def apply(ctx: Context, config: object) -> None: session_read=session_read, workspace=ctx.runtime.workspace, db_path=db_path, - publish_feedback=lambda event: ctx.observe( - PROACTIVE_FEEDBACK_COMMITTED, - event, - ), + ) + _ = await ctx.provide( + PROACTIVE_FEEDBACK_HISTORY, + SqliteFeedbackHistory(db_path), ) # 2. Candidate Root 保留同一 listener 拓扑，但不启动持久 worker。 @@ -120,13 +113,11 @@ def __init__( session_read: SessionReadService, workspace: Path, db_path: Path, - publish_feedback: FeedbackPublisher | None = None, session_keys: Callable[[], Iterable[str]] | None = None, ) -> None: self._session_read = session_read self._workspace = workspace self._db_path = db_path - self._publish_feedback = publish_feedback self._session_keys = session_keys or ( lambda: _formal_session_keys_from_read_service(session_read) ) @@ -169,14 +160,6 @@ async def run_worker(self) -> None: self._discovery_done = True while True: # 2. Replay every durable payload before waiting for new Turns. - try: - await self._publish_pending() - except asyncio.CancelledError: - raise - except Exception: - logger.exception("proactive_feedback outbox publish failed") - await asyncio.sleep(_OUTBOX_RETRY_SECONDS) - continue try: has_pending_inputs = await self._process_pending_inputs() except asyncio.CancelledError: @@ -189,7 +172,7 @@ async def run_worker(self) -> None: await asyncio.sleep(_OUTBOX_RETRY_SECONDS) continue - # 3. Process one Core-committed Turn and drain its transaction's outbox. + # 3. Process one Core-committed Turn into immutable accepted history. input_row_id = await self._queue.get() try: await self._process_input_row(input_row_id) @@ -310,7 +293,6 @@ async def _process( reason="scoring_failed", ) self._complete_input(input_row_id) - await self._publish_pending() return if scored is None: self._complete_input(input_row_id) @@ -330,7 +312,6 @@ async def _process( reason=scored.reason, ) self._complete_input(input_row_id) - await self._publish_pending() async def _persist_feedback( self, @@ -468,37 +449,6 @@ async def _process_input_record(self, record: FeedbackInputRecord) -> None: input_row_id=record.row_id, ) - async def _publish_pending(self) -> None: - """Publish durable rows and advance their SQLite cursor after receipt.""" - - # 1. A candidate with no database and tests without a publisher stay inert. - if self._publish_feedback is None or not self._db_path.exists(): - return - # 2. Publish outside SQLite; only a returned receipt advances state. - while True: - sink = open_db(self._db_path) - try: - pending = pending_feedback_outbox(sink, limit=_OUTBOX_BATCH_SIZE) - finally: - sink.close() - if not pending: - return - for record in pending: - payload = _decode_outbox_payload(record.payload_json) - if payload.get("event_id") != record.event_id: - raise ValueError("proactive_feedback outbox event_id 不一致") - feedback = ProactiveFeedbackCommitted(**cast(Any, payload)) - await self._publish_feedback(feedback) - sink = open_db(self._db_path) - try: - mark_feedback_published( - sink, - row_id=record.row_id, - event_id=record.event_id, - ) - finally: - sink.close() - def _get_embedder(self) -> Embedder: if self._embedder is None: self._embedder = _build_embedder(self._workspace) diff --git a/tests/test_plugin.py b/tests/test_plugin.py index 1f78ce2..4044543 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -21,7 +21,6 @@ from agent.plugins.mobile_ui import PluginMobileUiProvider from agent.plugins.manifest import write_plugin_manifest from agent.turn_events.after_turn import AFTER_TURN_COMMITTED -from agent.turn_events.proactive_feedback import ProactiveFeedbackCommitted from bus.events_lifecycle import TurnCommitted from bus.event_bus import EventBus @@ -217,6 +216,9 @@ def test_v2_runtime_symbols_are_not_used_by_module() -> None: assert "event_bus" not in source assert "sessions.db" not in source assert "ProactiveFeedbackRecorded" not in source + assert "PROACTIVE_FEEDBACK_COMMITTED" not in source + assert "CONTENT_SOURCE" not in source + assert "Content" not in source assert "event.extra" not in source assert "【你当前新消息】" not in source @@ -287,40 +289,28 @@ async def test_committed_turn_writes_plugin_owned_projection(tmp_path: Path) -> @pytest.mark.asyncio -async def test_feedback_commit_publishes_typed_event_and_cursor(tmp_path: Path) -> None: - published: list[ProactiveFeedbackCommitted] = [] - - async def publish(event: ProactiveFeedbackCommitted) -> None: - published.append(event) - +async def test_feedback_commit_is_readable_from_stable_history(tmp_path: Path) -> None: runtime = module.ProactiveFeedbackRuntime( session_read=SessionReadService( lambda _key: (cast(Any, _session_state()), None) ), workspace=tmp_path, db_path=tmp_path / "data" / "proactive_feedback.db", - publish_feedback=publish, ) await runtime._process(_event()) await runtime._process(_event()) - assert [event.event_id for event in published] == ["proactive_feedback:1"] - assert published[0].event_id == "proactive_feedback:1" - assert published[0].session_key == "mobile:test" - assert published[0].user_content_preview is not None - conn = sqlite3.connect(runtime._db_path) - try: - outbox = conn.execute( - "SELECT published_at FROM proactive_feedback_outbox" - ).fetchone() - cursor = conn.execute( - "SELECT row_id FROM proactive_feedback_published_cursor " - "WHERE name = 'proactive_feedback'" - ).fetchone() - finally: - conn.close() - assert outbox is not None and outbox[0] is not None - assert cursor == (1,) + page = module.SqliteFeedbackHistory(runtime._db_path).page( + after_cursor=0, + max_items=10, + ) + assert [record.event_id for record in page.records] == ["proactive_feedback:1"] + assert page.records[0].session_key == "mobile:test" + assert len(page.records[0].payload_hash) == 64 + with sqlite3.connect(runtime._db_path) as conn: + assert conn.execute( + "SELECT count(*) FROM proactive_feedback_outbox" + ).fetchone() == (0,) def test_committed_turn_identity_is_durable_without_message_text(tmp_path: Path) -> None: @@ -369,11 +359,6 @@ def test_candidate_enqueue_fails_before_any_write(tmp_path: Path) -> None: @pytest.mark.asyncio async def test_durable_input_replays_after_runtime_restart(tmp_path: Path) -> None: - published: list[ProactiveFeedbackCommitted] = [] - - async def publish(event: ProactiveFeedbackCommitted) -> None: - published.append(event) - original = module.ProactiveFeedbackRuntime( session_read=SessionReadService( lambda _key: (cast(Any, _session_state()), None) @@ -387,10 +372,11 @@ async def publish(event: ProactiveFeedbackCommitted) -> None: session_read=original._session_read, workspace=tmp_path, db_path=original._db_path, - publish_feedback=publish, ) assert await restarted._process_pending_inputs() is False - assert [event.event_id for event in published] == ["proactive_feedback:1"] + assert len(module.SqliteFeedbackHistory(original._db_path).page( + after_cursor=0, max_items=10 + ).records) == 1 conn = sqlite3.connect(original._db_path) try: @@ -405,11 +391,6 @@ async def publish(event: ProactiveFeedbackCommitted) -> None: async def test_durable_input_keeps_ordered_two_user_ids_and_scores_one_turn( tmp_path: Path, ) -> None: - published: list[ProactiveFeedbackCommitted] = [] - - async def publish(event: ProactiveFeedbackCommitted) -> None: - published.append(event) - session_read = SessionReadService( lambda _key: ( cast(Any, SimpleNamespace( @@ -430,7 +411,6 @@ async def publish(event: ProactiveFeedbackCommitted) -> None: session_read=session_read, workspace=tmp_path, db_path=original._db_path, - publish_feedback=publish, ) assert await restarted._process_pending_inputs() is False @@ -450,20 +430,12 @@ async def publish(event: ProactiveFeedbackCommitted) -> None: assert inbox[2] is not None assert projection[0] == "u2" assert "被回复消息" in projection[1] and "当前新消息" in projection[1] - assert [event.event_id for event in published] == ["proactive_feedback:1"] @pytest.mark.asyncio async def test_formal_boot_discovers_committed_turn_without_callback_once( tmp_path: Path, ) -> None: - published: list[ProactiveFeedbackCommitted] = [] - published_event = asyncio.Event() - - async def publish(event: ProactiveFeedbackCommitted) -> None: - published.append(event) - published_event.set() - session_read = SessionReadService( lambda _key: ( cast(Any, _session_state()), @@ -474,18 +446,21 @@ async def publish(event: ProactiveFeedbackCommitted) -> None: session_read=session_read, workspace=tmp_path, db_path=tmp_path / "data" / "proactive_feedback.db", - publish_feedback=publish, session_keys=lambda: ("mobile:test",), ) worker = asyncio.create_task(restarted.run_worker()) - await asyncio.wait_for(published_event.wait(), timeout=1) + for _ in range(100): + if restarted._db_path.exists() and module.SqliteFeedbackHistory( + restarted._db_path + ).page(after_cursor=0, max_items=10).records: + break + await asyncio.sleep(0.01) worker.cancel() with pytest.raises(asyncio.CancelledError): await worker restarted._discover_committed_inputs() - assert [event.event_id for event in published] == ["proactive_feedback:1"] conn = sqlite3.connect(restarted._db_path) try: assert conn.execute( @@ -643,7 +618,7 @@ async def blocked_scoring(**_kwargs: object) -> None: conn.close() -def test_published_feedback_identity_is_immutable(tmp_path: Path) -> None: +def test_accepted_feedback_identity_is_immutable_and_drift_fails(tmp_path: Path) -> None: sink = module.open_db(tmp_path / "proactive_feedback.db") try: first = FeedbackEvent( @@ -665,11 +640,6 @@ def test_published_feedback_identity_is_immutable(tmp_path: Path) -> None: ) row_id = module.insert_feedback(sink, first) assert row_id == 1 - module.mark_feedback_published( - sink, - row_id=row_id, - event_id="proactive_feedback:1", - ) second = FeedbackEvent( **{ **first.__dict__, @@ -683,25 +653,25 @@ def test_published_feedback_identity_is_immutable(tmp_path: Path) -> None: "proactive_content_preview": "second proactive", } ) - assert module.insert_feedback(sink, second) == row_id + with pytest.raises(RuntimeError, match="payload 漂移"): + module.insert_feedback(sink, second) third = FeedbackEvent( **{**first.__dict__, "proactive_message_id": "p2"} ) - assert module.insert_feedback(sink, third) == 2 + with pytest.raises(RuntimeError, match="payload 漂移"): + module.insert_feedback(sink, third) + assert module.insert_feedback(sink, first) == row_id assert tuple(sink.execute( "SELECT count(*) FROM proactive_feedback_events" - ).fetchone()) == (2,) + ).fetchone()) == (1,) projection = tuple(sink.execute( "SELECT feedback_type, confidence, pa_score, pua_score, reason, " "user_content_preview, assistant_content_preview, proactive_content_preview " "FROM proactive_feedback_events WHERE id = 1" ).fetchone()) - payload = json.loads( - sink.execute( - "SELECT payload_json FROM proactive_feedback_outbox " - "WHERE row_id = 1" - ).fetchone()[0] - ) + assert tuple(sink.execute( + "SELECT count(*) FROM proactive_feedback_outbox" + ).fetchone()) == (0,) finally: sink.close() assert projection == ( @@ -714,85 +684,6 @@ def test_published_feedback_identity_is_immutable(tmp_path: Path) -> None: "first answer", "first proactive", ) - assert payload["reason"] == "first_reason" - assert payload["pa_score"] == 1.0 - - -@pytest.mark.asyncio -async def test_failed_publication_is_replayed_after_restart(tmp_path: Path) -> None: - async def fail(_event: ProactiveFeedbackCommitted) -> None: - raise RuntimeError("observer unavailable") - - runtime = module.ProactiveFeedbackRuntime( - session_read=SessionReadService( - lambda _key: (cast(Any, _session_state()), None) - ), - workspace=tmp_path, - db_path=tmp_path / "data" / "proactive_feedback.db", - publish_feedback=fail, - ) - with pytest.raises(RuntimeError, match="observer unavailable"): - await runtime._process(_event()) - - conn = sqlite3.connect(runtime._db_path) - try: - assert conn.execute( - "SELECT published_at FROM proactive_feedback_outbox" - ).fetchone() == (None,) - finally: - conn.close() - - replayed: list[ProactiveFeedbackCommitted] = [] - - async def publish(event: ProactiveFeedbackCommitted) -> None: - replayed.append(event) - - restarted = module.ProactiveFeedbackRuntime( - session_read=SessionReadService.candidate_validation(), - workspace=tmp_path, - db_path=runtime._db_path, - publish_feedback=publish, - ) - await restarted._publish_pending() - assert [event.event_id for event in replayed] == ["proactive_feedback:1"] - conn = sqlite3.connect(runtime._db_path) - try: - assert conn.execute( - "SELECT published_at FROM proactive_feedback_outbox" - ).fetchone()[0] is not None - finally: - conn.close() - - -@pytest.mark.asyncio -async def test_publication_cancellation_keeps_pending_receipt(tmp_path: Path) -> None: - started = asyncio.Event() - - async def blocked(_event: ProactiveFeedbackCommitted) -> None: - started.set() - await asyncio.Future() - - runtime = module.ProactiveFeedbackRuntime( - session_read=SessionReadService( - lambda _key: (cast(Any, _session_state()), None) - ), - workspace=tmp_path, - db_path=tmp_path / "data" / "proactive_feedback.db", - publish_feedback=blocked, - ) - task = asyncio.create_task(runtime._process(_event())) - await started.wait() - task.cancel() - with pytest.raises(asyncio.CancelledError): - await task - - conn = sqlite3.connect(runtime._db_path) - try: - assert conn.execute( - "SELECT published_at FROM proactive_feedback_outbox" - ).fetchone() == (None,) - finally: - conn.close() @pytest.mark.asyncio @@ -940,47 +831,134 @@ def test_dashboard_reads_preview_projection_without_sessions_database(tmp_path: assert items[0]["assistant_preview"] == "回答" -def test_feedback_projection_and_outbox_commit_atomically(tmp_path: Path) -> None: +def test_new_feedback_does_not_touch_frozen_legacy_outbox(tmp_path: Path) -> None: sink = module.open_db(tmp_path / "proactive_feedback.db") try: _ = sink.execute( """ - CREATE TRIGGER reject_feedback_outbox - BEFORE INSERT ON proactive_feedback_outbox - BEGIN - SELECT RAISE(ABORT, 'outbox unavailable'); - END; + INSERT INTO proactive_feedback_outbox( + row_id, event_id, payload_json, published_at + ) VALUES(91, 'legacy:91', '{"legacy":true}', '2026-08-01T00:00:00Z') """ ) - with pytest.raises(sqlite3.IntegrityError, match="outbox unavailable"): - module.insert_feedback( - sink, - FeedbackEvent( - session_key="mobile:test", - user_message_id="u1", - assistant_message_id="a1", - proactive_message_id="p1", - feedback_type="explicit_quote", - confidence="gold", - pa_score=1.0, - pua_score=1.0, - lag_seconds=8, - candidate_count=1, - matched_by="explicit_quote", - reason="explicit_quote", - user_content_preview="继续", - assistant_content_preview="回答", - proactive_content_preview="主题", - ), - ) + sink.commit() + before = tuple(sink.execute( + "SELECT * FROM proactive_feedback_outbox" + ).fetchone()) + module.insert_feedback( + sink, + FeedbackEvent( + session_key="mobile:test", + user_message_id="u1", + assistant_message_id="a1", + proactive_message_id="p1", + feedback_type="explicit_quote", + confidence="gold", + pa_score=1.0, + pua_score=1.0, + lag_seconds=8, + candidate_count=1, + matched_by="explicit_quote", + reason="explicit_quote", + user_content_preview="继续", + assistant_content_preview="回答", + proactive_content_preview="主题", + ), + ) assert tuple(sink.execute( "SELECT count(*) FROM proactive_feedback_events" - ).fetchone()) == (0,) + ).fetchone()) == (1,) assert tuple(sink.execute( - "SELECT count(*) FROM proactive_feedback_outbox" - ).fetchone()) == (0,) + "SELECT * FROM proactive_feedback_outbox" + ).fetchone()) == before + finally: + sink.close() + + +def test_history_missing_database_is_empty_without_creation(tmp_path: Path) -> None: + path = tmp_path / "missing" / "proactive_feedback.db" + page = module.SqliteFeedbackHistory(path).page(after_cursor=0, max_items=10) + assert page.records == () + assert not path.parent.exists() + + +def test_history_corrupt_or_incompatible_database_fails_loud(tmp_path: Path) -> None: + corrupt = tmp_path / "corrupt.db" + corrupt.write_bytes(b"not sqlite") + before = corrupt.read_bytes() + with pytest.raises(sqlite3.DatabaseError): + module.SqliteFeedbackHistory(corrupt).page(after_cursor=0, max_items=10) + assert corrupt.read_bytes() == before + + incompatible = tmp_path / "incompatible.db" + with sqlite3.connect(incompatible) as conn: + conn.execute("CREATE TABLE unrelated(value TEXT)") + with pytest.raises(RuntimeError, match="缺少 events 表"): + module.SqliteFeedbackHistory(incompatible).page( + after_cursor=0, max_items=10 + ) + + invalid_payload = tmp_path / "invalid-payload.db" + connection = module.open_db(invalid_payload) + try: + module.insert_feedback( + connection, + FeedbackEvent( + session_key="mobile:test", + user_message_id="u1", + assistant_message_id="a1", + proactive_message_id="p1", + feedback_type="topic_follow", + confidence="high", + pa_score=0.8, + pua_score=0.7, + lag_seconds=1, + candidate_count=1, + matched_by="pua", + reason="fixture", + ), + ) + connection.execute( + "UPDATE proactive_feedback_events SET confidence='unknown' WHERE id=1" + ) + connection.commit() + finally: + connection.close() + with pytest.raises(ValueError, match="confidence 不支持"): + module.SqliteFeedbackHistory(invalid_payload).page( + after_cursor=0, max_items=10 + ) + + +def test_history_pages_are_stable_and_new_rows_wait_for_next_page(tmp_path: Path) -> None: + path = tmp_path / "proactive_feedback.db" + sink = module.open_db(path) + try: + for index in range(3): + event = FeedbackEvent( + session_key="mobile:test", + user_message_id=f"u{index}", + assistant_message_id=f"a{index}", + proactive_message_id=f"p{index}", + feedback_type="topic_follow", + confidence="high", + pa_score=0.8, + pua_score=0.7, + lag_seconds=index, + candidate_count=1, + matched_by="pua", + reason="fixture", + ) + module.insert_feedback(sink, event) finally: sink.close() + history = module.SqliteFeedbackHistory(path) + first = history.page(after_cursor=0, max_items=2) + assert [row.cursor for row in first.records] == [1, 2] + repeated = history.page(after_cursor=0, max_items=2) + assert repeated == first + second = history.page(after_cursor=2, max_items=2) + assert [row.cursor for row in second.records] == [3] def test_plugin_runtime_does_not_move_legacy_database(tmp_path: Path) -> None: @@ -1008,6 +986,7 @@ async def test_manager_stable_candidate_ui_dashboard_and_cleanup(tmp_path: Path) "plugin.py", "dashboard.py", "db.py", + "history.py", "scorer.py", "mobile_panel.js", "mobile_panel.css", @@ -1044,6 +1023,10 @@ async def test_manager_stable_candidate_ui_dashboard_and_cleanup(tmp_path: Path) "proactive_feedback" ).data_dir assert not (formal_data / "proactive_feedback.db").exists() + formal_history = stable.composition_root.context.require( + module.PROACTIVE_FEEDBACK_HISTORY + ) + assert formal_history.page(after_cursor=0, max_items=10).records == () formal_database = formal_data / "proactive_feedback.db" formal_database.parent.mkdir(parents=True, exist_ok=True) formal_connection = module.open_db(formal_database) @@ -1081,6 +1064,10 @@ async def test_manager_stable_candidate_ui_dashboard_and_cleanup(tmp_path: Path) assert not (binding.runtime_data_root / "proactive_feedback.db").exists() candidate_root = candidate_snapshot.composition_root assert candidate_root is not None + candidate_history = candidate_root.context.require( + module.PROACTIVE_FEEDBACK_HISTORY + ) + assert candidate_history.page(after_cursor=0, max_items=10).records == () assert candidate_snapshot.composition_topology is not None assert stable.composition_topology is not None assert ( From 531eae4e4ac4714aad5417b8257a724007728345 Mon Sep 17 00:00:00 2001 From: huashen <2494946808@qq.com> Date: Sun, 23 Aug 2026 20:27:51 +0800 Subject: [PATCH 2/2] fix: validate feedback history boundary --- README.md | 4 +- history.py | 93 +++++++++++++++++++++++++++++++++---- tests/test_plugin.py | 106 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 193 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index 896cd8a..f3c885f 100644 --- a/README.md +++ b/README.md @@ -32,7 +32,9 @@ outbox，也不调用 Core Observe event。 history reader 始终使用 SQLite `mode=ro`。数据库不存在表示合法空历史且不会创建目录； 数据库存在但损坏、schema 异构或字段类型无效会 fail-loud，不能伪装成空页。插件不向 -`TurnCommitted.extra` 回写结果，也不依赖 Wake、Content 或消费者数据库。 +`TurnCommitted.extra` 回写结果，也不依赖 Wake、Content 或消费者数据库。已知正式旧表 +若尚无三个 preview 列，history 会把这三个稳定字段投影为 `null` 后参与 canonical hash； +旧表经既有 `open_db` 逐列 `ALTER` 后形成的 SQLite schema 也属于同一已知 lineage。 非引用评分使用 Core 正式运行时的共享 HTTP resources。嵌入配置从 `AKASHIC_CONFIG` 指向的 Core 配置加载，不从插件 checkout 的当前目录猜测配置。 diff --git a/history.py b/history.py index 05a8d2a..a137bf4 100644 --- a/history.py +++ b/history.py @@ -2,6 +2,7 @@ import hashlib import json +import math import sqlite3 from dataclasses import dataclass from pathlib import Path @@ -15,6 +16,59 @@ {"explicit_quote", "topic_follow", "no_topic_follow", "unscored"} ) _CONFIDENCE = frozenset({"gold", "high", "medium", "low"}) + + +def _normalize_sql(sql: str) -> str: + return "".join(sql.lower().split()) + + +_CURRENT_EVENTS_SQL = """ +CREATE TABLE proactive_feedback_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + session_key TEXT NOT NULL, + user_message_id TEXT NOT NULL, + assistant_message_id TEXT NOT NULL, + proactive_message_id TEXT, + feedback_type TEXT NOT NULL, + confidence TEXT NOT NULL, + pa_score REAL, + pua_score REAL, + lag_seconds INTEGER, + candidate_count INTEGER NOT NULL, + matched_by TEXT NOT NULL, + reason TEXT NOT NULL, + user_content_preview TEXT, + assistant_content_preview TEXT, + proactive_content_preview TEXT, + UNIQUE(user_message_id, proactive_message_id) +) +""" +_LEGACY_EVENTS_SQL = """ +CREATE TABLE proactive_feedback_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + session_key TEXT NOT NULL, + user_message_id TEXT NOT NULL, + assistant_message_id TEXT NOT NULL, + proactive_message_id TEXT, + feedback_type TEXT NOT NULL, + confidence TEXT NOT NULL, + pa_score REAL, + pua_score REAL, + lag_seconds INTEGER, + candidate_count INTEGER NOT NULL, + matched_by TEXT NOT NULL, + reason TEXT NOT NULL, + UNIQUE(user_message_id, proactive_message_id) +) +""" +_ALLOWED_EVENTS_SQL = frozenset( + { + _normalize_sql(_CURRENT_EVENTS_SQL), + _normalize_sql(_LEGACY_EVENTS_SQL), + } +) _REQUIRED_COLUMNS = frozenset( { "id", @@ -102,14 +156,21 @@ def page(self, *, after_cursor: int, max_items: int) -> FeedbackHistoryPage: ) connection.row_factory = sqlite3.Row try: - _validate_schema(connection) + columns = _validate_schema(connection) + previews = ", ".join( + column if column in columns else f"NULL AS {column}" + for column in ( + "user_content_preview", + "assistant_content_preview", + "proactive_content_preview", + ) + ) rows = connection.execute( - """ + f""" SELECT id, session_key, user_message_id, assistant_message_id, proactive_message_id, feedback_type, confidence, pa_score, pua_score, lag_seconds, candidate_count, - matched_by, reason, user_content_preview, - assistant_content_preview, proactive_content_preview + matched_by, reason, {previews} FROM proactive_feedback_events WHERE id > ? ORDER BY id ASC @@ -131,28 +192,42 @@ def accepted_payload_hash(payload: dict[str, object]) -> str: ensure_ascii=False, sort_keys=True, separators=(",", ":"), + allow_nan=False, ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() -def _validate_schema(connection: sqlite3.Connection) -> None: +def _validate_schema(connection: sqlite3.Connection) -> frozenset[str]: tables = { - str(row[0]) + str(row[0]): str(row[1]) for row in connection.execute( - "SELECT name FROM sqlite_master WHERE type='table'" + "SELECT name, sql FROM sqlite_master WHERE type='table'" ) } if "proactive_feedback_events" not in tables: raise RuntimeError("proactive_feedback history 缺少 events 表") + if _normalize_sql(tables["proactive_feedback_events"]) not in _ALLOWED_EVENTS_SQL: + raise RuntimeError("proactive_feedback history events table schema 不匹配") columns = { str(row[1]) for row in connection.execute("PRAGMA table_info(proactive_feedback_events)") } - missing = sorted(_REQUIRED_COLUMNS - columns) + required = _REQUIRED_COLUMNS.difference( + { + "user_content_preview", + "assistant_content_preview", + "proactive_content_preview", + } + ) + missing = sorted(required - columns) if missing: raise RuntimeError( "proactive_feedback history schema 缺少列: " + ", ".join(missing) ) + checks = tuple(tuple(row) for row in connection.execute("PRAGMA quick_check")) + if checks != (("ok",),): + raise RuntimeError("proactive_feedback history SQLite quick_check failed") + return frozenset(columns) def _record_from_row(row: sqlite3.Row) -> FeedbackHistoryRecord: @@ -243,6 +318,6 @@ def _optional_score(value: object, field: str) -> float | None: if isinstance(value, bool) or not isinstance(value, (int, float)): raise TypeError(f"history {field} 必须是数字或 null") score = float(value) - if score < -1.0 or score > 1.0: + if not math.isfinite(score) or score < -1.0 or score > 1.0: raise ValueError(f"history {field} 必须在 -1..1 之间") return score diff --git a/tests/test_plugin.py b/tests/test_plugin.py index 4044543..5e836c4 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -898,6 +898,28 @@ def test_history_corrupt_or_incompatible_database_fails_loud(tmp_path: Path) -> after_cursor=0, max_items=10 ) + malformed = tmp_path / "same-columns-without-constraints.db" + valid = tmp_path / "valid-schema.db" + connection = module.open_db(valid) + try: + table_sql = connection.execute( + "SELECT sql FROM sqlite_master " + "WHERE type='table' AND name='proactive_feedback_events'" + ).fetchone()[0] + finally: + connection.close() + malformed_sql = table_sql.replace( + "id INTEGER PRIMARY KEY AUTOINCREMENT", "id INTEGER" + ).replace( + ",\n UNIQUE(user_message_id, proactive_message_id)", "" + ) + with sqlite3.connect(malformed) as connection: + connection.execute(malformed_sql) + with pytest.raises(RuntimeError, match="events table schema 不匹配"): + module.SqliteFeedbackHistory(malformed).page( + after_cursor=0, max_items=10 + ) + invalid_payload = tmp_path / "invalid-payload.db" connection = module.open_db(invalid_payload) try: @@ -929,6 +951,90 @@ def test_history_corrupt_or_incompatible_database_fails_loud(tmp_path: Path) -> after_cursor=0, max_items=10 ) + invalid_score = tmp_path / "invalid-score.db" + connection = module.open_db(invalid_score) + try: + module.insert_feedback( + connection, + FeedbackEvent( + session_key="mobile:test", + user_message_id="u1", + assistant_message_id="a1", + proactive_message_id="p1", + feedback_type="topic_follow", + confidence="high", + pa_score=0.8, + pua_score=0.7, + lag_seconds=1, + candidate_count=1, + matched_by="pua", + reason="fixture", + ), + ) + connection.execute( + "UPDATE proactive_feedback_events SET pa_score=? WHERE id=1", + (float("inf"),), + ) + connection.commit() + finally: + connection.close() + with pytest.raises(ValueError, match="pa_score 必须在"): + module.SqliteFeedbackHistory(invalid_score).page( + after_cursor=0, max_items=10 + ) + + history_module = sys.modules["proactive_feedback_v3_test.plugin.history"] + with pytest.raises(ValueError): + history_module.accepted_payload_hash({"score": float("nan")}) + + +def test_history_reads_exact_legacy_and_altered_table_lineage(tmp_path: Path) -> None: + path = tmp_path / "legacy.db" + with sqlite3.connect(path) as connection: + connection.executescript( + """ + CREATE TABLE proactive_feedback_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + session_key TEXT NOT NULL, + user_message_id TEXT NOT NULL, + assistant_message_id TEXT NOT NULL, + proactive_message_id TEXT, + feedback_type TEXT NOT NULL, + confidence TEXT NOT NULL, + pa_score REAL, + pua_score REAL, + lag_seconds INTEGER, + candidate_count INTEGER NOT NULL, + matched_by TEXT NOT NULL, + reason TEXT NOT NULL, + UNIQUE(user_message_id, proactive_message_id) + ); + INSERT INTO proactive_feedback_events( + session_key, user_message_id, assistant_message_id, + proactive_message_id, feedback_type, confidence, pa_score, + pua_score, lag_seconds, candidate_count, matched_by, reason + ) VALUES ( + 'mobile:test', 'u1', 'a1', 'p1', 'topic_follow', 'high', + 0.8, 0.7, 1, 1, 'pua', 'legacy fixture' + ); + """ + ) + + legacy = module.SqliteFeedbackHistory(path).page(after_cursor=0, max_items=10) + assert legacy.records[0].user_content_preview is None + + migrated = module.open_db(path) + migrated.close() + with sqlite3.connect(path) as connection: + table_sql = connection.execute( + "SELECT sql FROM sqlite_master " + "WHERE type='table' AND name='proactive_feedback_events'" + ).fetchone()[0] + assert table_sql.index("user_content_preview") < table_sql.index("UNIQUE(") + page = module.SqliteFeedbackHistory(path).page(after_cursor=0, max_items=10) + assert page.records == legacy.records + def test_history_pages_are_stable_and_new_rows_wait_for_next_page(tmp_path: Path) -> None: path = tmp_path / "proactive_feedback.db"