diff --git a/.github/workflows/plugin-api-v2.yml b/.github/workflows/plugin-api-v2.yml deleted file mode 100644 index 7c1876b..0000000 --- a/.github/workflows/plugin-api-v2.yml +++ /dev/null @@ -1,28 +0,0 @@ -name: plugin-api-v2 - -on: - pull_request: - push: - branches: - - main - -permissions: - contents: read - -jobs: - contract: - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - - uses: actions/checkout@v4 - with: - repository: akashic-plugins/plugin-contracts - ref: 24543445c7b99ca63fcd90b5828f754a148b184c - path: .plugin-contracts - - uses: actions/setup-python@v5 - with: - python-version: "3.13" - - name: Check Plugin API v2 - env: - PYTHONPATH: .plugin-contracts - run: python -m akashic_plugin_contracts check plugin.py diff --git a/.github/workflows/plugin-api-v3.yml b/.github/workflows/plugin-api-v3.yml new file mode 100644 index 0000000..7139e00 --- /dev/null +++ b/.github/workflows/plugin-api-v3.yml @@ -0,0 +1,71 @@ +name: plugin-api-v3 + +on: + pull_request: + push: + branches: + - main + workflow_dispatch: + +permissions: + contents: read + +jobs: + contract: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/checkout@v4 + with: + repository: akashic-plugins/plugin-contracts + ref: 4dd69dd621e029e51e99aa428443fa3a4ec1f6cf + path: .plugin-contracts + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + - name: Check Plugin API v3 + env: + PYTHONPATH: .plugin-contracts + run: python -m akashic_plugin_contracts check plugin.py + + plugin-tests: + runs-on: ubuntu-latest + timeout-minutes: 20 + steps: + - uses: actions/checkout@v4 + - uses: actions/checkout@v4 + with: + repository: kachofugetsu09/akashic-agent + # 5624 lacks the v3 domain-effect lookup export. Keep this exact + # Core commit until it is published; checkout failure is intentional + # release blocking, not permission to weaken the plugin oracle. + ref: 3005f838bcd96e2cbc58616aede46e4f39df4523 + path: .akashic-core + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + cache: pip + cache-dependency-path: | + .akashic-core/requirements.txt + .akashic-core/requirements-dev.txt + - name: Install exact Core runtime + run: python -m pip install -r .akashic-core/requirements.txt -r .akashic-core/requirements-dev.txt + - name: Run focused plugin tests + env: + AKASHIC_AGENT_ROOT: .akashic-core + PYTHONPATH: .akashic-core + run: python -m pytest -q tests + - uses: actions/setup-node@v4 + with: + node-version: "22" + - name: Verify mobile panel + run: node --test tests/test_mobile_panel.mjs + - name: Check v3 source types + env: + AKASHIC_AGENT_ROOT: .akashic-core + PYTHONPATH: .akashic-core + run: pyright --level error plugin.py dashboard.py db.py tests + - name: Compile Python sources + run: python -m compileall -q plugin.py dashboard.py db.py drift tests + - name: Check diff formatting + run: git diff --check diff --git a/README.md b/README.md index 09555e6..03227fb 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,31 @@ Akashic emotion and proactive tuning plugin. +## v3 接入 + +入口是 module-level `api_version = 3` 与 `apply(ctx, config)`。Emotion 通过 Core +声明以下能力： + +- `PROACTIVE_COMPONENTS`：在 exact generation 中形成 VAD prompt projection；formal + 运行由 `emotion.state` domain effect 提交 SQLite，candidate 不打开数据库。 +- `BACKGROUND_JOBS`：`feedback-preference-context` Drift 完成后，使用 Core 的 LLM + lease 和窄 documents port 合并 `PROACTIVE_CONTEXT.md` / `proactive_pending.md`。 +- `AFTER_TURN_COMMITTED`：消费 Core 已提交的 typed Turn。上游若提供 + `extra.proactive_feedback`，按其稳定 identity 幂等写入；显式引用消息则按 Turn + 自带标记写入 gold feedback。 +- `UI_SLOTS` 与 C09 Dashboard：移动端和桌面端只读 Emotion 自有投影，不读取 + `sessions.db`，不取得任意 workspace 句柄。 + +插件不再声明 v2 `Plugin`、EventBus listener、固定 `proactive_modules()` / `jobs()` +或旧 mobile/dashboard ABI。旧数据库不会在 import/apply 时自动迁移；切换前应先 +停用旧 runtime 并使用独立迁移脚本（尚未将旧源删除）。 + +CI 的 Core pin 是 20062a715d2c5822228b327863b51c8d036119b3，因为旧 pin +5624a059348406c1f97993612adfec886b158158 没有 domain_effect_lookup_export。 +该 commit 尚未发布到 Core 的公共默认分支前，CI checkout 失败属于明确的发布阻塞； +本插件必须继续在 integration Core exact worktree 上验证，不得删除 lookup seam 或放宽 +candidate/formal oracle。 + ## 移动端看板 插件通过通用移动 UI 生命周期注册“主动状态”入口，说明用户反馈如何改变 Agent 的语气 diff --git a/akashic.plugin.toml b/akashic.plugin.toml new file mode 100644 index 0000000..af23561 --- /dev/null +++ b/akashic.plugin.toml @@ -0,0 +1,5 @@ +schema_version = 1 +name = "emotion" +version = "3.0.0" +api_version = 3 +entrypoint = "plugin.py" diff --git a/dashboard.py b/dashboard.py index c029bc5..2c18e92 100644 --- a/dashboard.py +++ b/dashboard.py @@ -1,5 +1,4 @@ from __future__ import annotations - import json import sqlite3 import threading @@ -9,13 +8,14 @@ from fastapi import FastAPI +from agent.plugin_composition import DashboardContext + from .db import EmotionState, describe_behavior class EmotionDashboardReader: - def __init__(self, workspace: Path) -> None: - self.db_path = workspace / "emotion" / "emotion.db" - self.sessions_db_path = workspace / "sessions.db" + def __init__(self, emotion_root: Path) -> None: + self.db_path = emotion_root / "emotion.db" self._lock = threading.RLock() def get_overview(self) -> dict[str, Any]: @@ -36,11 +36,10 @@ def list_influences(self, *, limit: int = 30) -> list[dict[str, Any]]: with _connect(self.db_path) as db: rows = _influence_rows(db, safe_limit) - # 2. 用事件已持有的消息 ID 补齐可读预览 + # 2. 事件 payload 已经是插件自己的完整投影；不越过 Core workspace 边界读 sessions.db。 decoded = [_decode_influence(row) for row in rows] - previews = self._load_user_previews(decoded) for item in decoded: - item["user_preview"] = _preview(previews.get(str(item["user_message_id"]))) + item["user_preview"] = "" return decoded def get_mobile_bootstrap(self, *, limit: int = 30) -> dict[str, Any]: @@ -56,10 +55,9 @@ def get_mobile_bootstrap(self, *, limit: int = 30) -> dict[str, Any]: overview = _overview_from_db(db) decoded = [_decode_influence(row) for row in _influence_rows(db, safe_limit)] - # 2. 会话预览是独立数据源，只补充文案，不参与 emotion 状态一致性 - previews = self._load_user_previews(decoded) + # 2. Mobile 只返回 Emotion 自有投影，不取得 Session 持久化 owner。 for item in decoded: - item["user_preview"] = _preview(previews.get(str(item["user_message_id"]))) + item["user_preview"] = "" return {"overview": overview, "items": decoded} def list_effects( @@ -98,22 +96,10 @@ def get_effect(self, effect_id: int) -> dict[str, Any] | None: ).fetchone() return _decode_effect(row) if row is not None else None - def _load_user_previews(self, items: list[dict[str, Any]]) -> dict[str, str]: - if not items or not self.sessions_db_path.exists(): - return {} - ids = list(dict.fromkeys(str(item["user_message_id"]) for item in items)) - placeholders = ",".join("?" for _ in ids) - with _connect(self.sessions_db_path) as db: - rows = db.execute( - f"SELECT id, content FROM messages WHERE id IN ({placeholders})", - ids, - ).fetchall() - return {str(row["id"]): str(row["content"] or "") for row in rows} - +def register(app: FastAPI, context: DashboardContext) -> None: + """Register Emotion read-only routes against the exact v3 workspace root.""" -def register(app: FastAPI, plugin_dir: Path, workspace: Path) -> None: - _ = plugin_dir - reader = EmotionDashboardReader(workspace) + reader = EmotionDashboardReader(context.workspace_root("emotion")) @app.get("/api/dashboard/emotion/overview") def get_emotion_overview() -> dict[str, Any]: @@ -246,10 +232,3 @@ def _decode_influence(row: sqlite3.Row) -> dict[str, Any]: metadata = json.loads(str(payload.pop("payload_json"))) payload["user_message_id"] = str(metadata["user_message_id"]) return payload - - -def _preview(value: str | None, limit: int = 180) -> str: - text = str(value or "").replace("\n", " ").strip() - if len(text) <= limit: - return text - return text[:limit].rstrip() + "..." diff --git a/db.py b/db.py index 73b34eb..7e9288d 100644 --- a/db.py +++ b/db.py @@ -8,7 +8,29 @@ from pathlib import Path from typing import Any -from proactive_v2.energy import compute_energy + +def compute_energy( + last_user_at: datetime | None, + now: datetime | None = None, + *, + alpha: float = 0.50, + beta: float = 0.35, + gamma: float = 0.15, + tau1_min: float = 30.0, + tau2_min: float = 240.0, + tau3_min: float = 2880.0, +) -> float: + """Return the current interaction energy without depending on Core internals.""" + + if last_user_at is None: + return 0.0 + now = now or datetime.now(timezone.utc) + minutes = max(0.0, (now - last_user_at).total_seconds() / 60.0) + return ( + alpha * math.exp(-minutes / tau1_min) + + beta * math.exp(-minutes / tau2_min) + + gamma * math.exp(-minutes / tau3_min) + ) @dataclass(frozen=True) @@ -34,6 +56,19 @@ class EmotionBehavior: expected_effect: str +@dataclass(frozen=True) +class EmotionDomainEffect: + """表示一次已由 Emotion SQLite 提交的幂等领域效果。""" + + semantic_job_id: str + event_id: str + invocation_id: str + effect_id: str + idempotency_key: str + attempt: int + result_digest: str + + def open_db(path: Path) -> sqlite3.Connection: path.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(path) @@ -70,6 +105,27 @@ def open_db(path: Path) -> sqlite3.Connection: payload_json TEXT NOT NULL ); + CREATE TABLE IF NOT EXISTS emotion_feedback_samples ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + source_event_id TEXT NOT NULL UNIQUE, + session_key TEXT NOT NULL, + user_message_id TEXT, + assistant_message_id TEXT, + 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, + matched_by TEXT, + reason TEXT, + user_content_preview TEXT, + assistant_content_preview TEXT, + proactive_content_preview TEXT + ); + CREATE TABLE IF NOT EXISTS emotion_effects ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at TEXT NOT NULL DEFAULT (datetime('now')), @@ -86,6 +142,19 @@ def open_db(path: Path) -> sqlite3.Connection: prompt_section TEXT NOT NULL, metadata_json TEXT NOT NULL ); + + CREATE TABLE IF NOT EXISTS emotion_domain_effects ( + semantic_job_id TEXT NOT NULL, + event_id TEXT NOT NULL, + invocation_id TEXT NOT NULL, + effect_id TEXT NOT NULL, + idempotency_key TEXT NOT NULL, + attempt INTEGER NOT NULL CHECK (attempt >= 1), + result_digest TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + PRIMARY KEY (invocation_id, effect_id, idempotency_key), + UNIQUE (semantic_job_id, event_id, effect_id) + ); """ ) now = datetime.now(timezone.utc).isoformat() @@ -100,6 +169,124 @@ def open_db(path: Path) -> sqlite3.Connection: return conn +def commit_domain_effect( + conn: sqlite3.Connection, + *, + semantic_job_id: str, + event_id: str, + invocation_id: str, + effect_id: str, + idempotency_key: str, + attempt: int, + result_digest: str, +) -> EmotionDomainEffect: + """在 Emotion 事务内幂等提交一次 job 领域效果及其 durable receipt。""" + + effect = EmotionDomainEffect( + semantic_job_id=_required_text(semantic_job_id, "semantic_job_id"), + event_id=_required_text(event_id, "event_id"), + invocation_id=_required_text(invocation_id, "invocation_id"), + effect_id=_required_text(effect_id, "effect_id"), + idempotency_key=_required_text(idempotency_key, "idempotency_key"), + attempt=_required_attempt(attempt), + result_digest=_required_text(result_digest, "result_digest"), + ) + + # 1. 锁定同一语义事件，避免新 invocation 重复提交领域效果。 + # 若调用方已经开启事务，receipt 必须加入该事务，不能提前提交。 + owns_transaction = not conn.in_transaction + if owns_transaction: + conn.execute("BEGIN IMMEDIATE") + try: + row = conn.execute( + """ + SELECT * FROM emotion_domain_effects + WHERE semantic_job_id = ? AND event_id = ? AND effect_id = ? + """, + (effect.semantic_job_id, effect.event_id, effect.effect_id), + ).fetchone() + if row is not None: + existing = _domain_effect_from_row(row) + if existing != effect: + raise RuntimeError("Emotion domain effect 幂等 identity 漂移") + if owns_transaction: + conn.commit() + return existing + + # 2. receipt 与该 effect 的领域写集在同一 SQLite transaction 提交。 + conn.execute( + """ + INSERT INTO emotion_domain_effects ( + semantic_job_id, event_id, invocation_id, effect_id, + idempotency_key, attempt, result_digest + ) VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + effect.semantic_job_id, + effect.event_id, + effect.invocation_id, + effect.effect_id, + effect.idempotency_key, + effect.attempt, + effect.result_digest, + ), + ) + if owns_transaction: + conn.commit() + except BaseException: + if owns_transaction: + conn.rollback() + raise + return effect + + +def lookup_domain_effect( + conn: sqlite3.Connection, + *, + invocation_id: str, + effect_id: str, + idempotency_key: str, +) -> EmotionDomainEffect | None: + """按 Core 固定的 invocation identity 读取 Emotion durable receipt。""" + + row = conn.execute( + """ + SELECT * FROM emotion_domain_effects + WHERE invocation_id = ? AND effect_id = ? AND idempotency_key = ? + """, + ( + _required_text(invocation_id, "invocation_id"), + _required_text(effect_id, "effect_id"), + _required_text(idempotency_key, "idempotency_key"), + ), + ).fetchone() + return None if row is None else _domain_effect_from_row(row) + + +def lookup_domain_effect_path( + path: Path, + *, + invocation_id: str, + effect_id: str, + idempotency_key: str, +) -> EmotionDomainEffect | None: + """只读查询既有 Emotion DB，不因恢复扫描创建任何文件。""" + + if not path.is_file() or path.is_symlink(): + return None + conn = sqlite3.connect(f"{path.resolve().as_uri()}?mode=ro", uri=True) + conn.row_factory = sqlite3.Row + try: + return lookup_domain_effect( + conn, + invocation_id=invocation_id, + effect_id=effect_id, + idempotency_key=idempotency_key, + ) + finally: + conn.close() + + def classify_feedback_delta(feedback_type: str, confidence: str) -> FeedbackDelta: if feedback_type == "explicit_quote": return FeedbackDelta(0.03, 0.08, "explicit_quote") @@ -163,11 +350,67 @@ def apply_feedback( ) except sqlite3.IntegrityError: return before + if feedback_type in {"topic_follow", "explicit_quote"}: + _insert_feedback_sample( + conn, + source_event_id=source_event_id, + session_key=session_key, + feedback_type=feedback_type, + confidence=confidence, + payload=payload, + ) _save_state(conn, after) conn.commit() return after +def _insert_feedback_sample( + conn: sqlite3.Connection, + *, + source_event_id: str, + session_key: str, + feedback_type: str, + confidence: str, + payload: dict[str, Any], +) -> None: + """Persist the bounded typed-Turn sample consumed by the Emotion Drift skill.""" + + _ = conn.execute( + """ + INSERT INTO emotion_feedback_samples ( + source_event_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 + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + source_event_id, + session_key, + _payload_text(payload, "user_message_id"), + _payload_text(payload, "assistant_message_id"), + _payload_text(payload, "proactive_message_id"), + feedback_type, + confidence, + payload.get("pa_score"), + payload.get("pua_score"), + payload.get("lag_seconds"), + payload.get("candidate_count"), + _payload_text(payload, "matched_by"), + _payload_text(payload, "reason"), + _payload_text(payload, "user_content_preview"), + _payload_text(payload, "assistant_content_preview"), + _payload_text(payload, "proactive_content_preview"), + ), + ) + + +def _payload_text(payload: dict[str, Any], field: str) -> str | None: + value = payload.get(field) + return value if isinstance(value, str) else None + + def build_effect( conn: sqlite3.Connection, *, @@ -176,6 +419,7 @@ def build_effect( now_utc: datetime, last_user_at: datetime | None, base_threshold: float, + commit: bool = True, ) -> dict[str, Any]: stored = _decay(get_state(conn), now_utc.isoformat()) energy = compute_energy(last_user_at, now_utc) @@ -244,7 +488,8 @@ def build_effect( json.dumps(metadata, ensure_ascii=False), ), ) - conn.commit() + if commit: + conn.commit() return { "provider_name": "emotion", "prompt_section": prompt_section, @@ -253,6 +498,34 @@ def build_effect( } +def lookup_effect( + conn: sqlite3.Connection, + *, + tick_id: str, +) -> dict[str, Any] | None: + """Read one previously committed proactive projection without recomputing it.""" + + row = conn.execute( + """ + SELECT prompt_section, threshold_delta, metadata_json + FROM emotion_effects + WHERE tick_id = ? + """, + (tick_id,), + ).fetchone() + if row is None: + return None + metadata = json.loads(str(row["metadata_json"])) + if not isinstance(metadata, dict): + raise RuntimeError("Emotion effect metadata 必须是 object") + return { + "provider_name": "emotion", + "prompt_section": str(row["prompt_section"]), + "threshold_delta": float(row["threshold_delta"]), + "metadata": metadata, + } + + def get_state(conn: sqlite3.Connection) -> EmotionState: row = conn.execute( """ @@ -272,6 +545,34 @@ def get_state(conn: sqlite3.Connection) -> EmotionState: ) +def _domain_effect_from_row(row: sqlite3.Row) -> EmotionDomainEffect: + return EmotionDomainEffect( + semantic_job_id=str(row["semantic_job_id"]), + event_id=str(row["event_id"]), + invocation_id=str(row["invocation_id"]), + effect_id=str(row["effect_id"]), + idempotency_key=str(row["idempotency_key"]), + attempt=int(row["attempt"]), + result_digest=str(row["result_digest"]), + ) + + +def _required_text(value: object, field: str) -> str: + if not isinstance(value, str): + raise TypeError(f"{field} 必须是字符串") + if not value or value.strip() != value: + raise ValueError(f"{field} 必须是无首尾空白的非空字符串") + return value + + +def _required_attempt(value: object) -> int: + if isinstance(value, bool) or not isinstance(value, int): + raise TypeError("attempt 必须是整数") + if value < 1: + raise ValueError("attempt 必须是正整数") + return value + + def _save_state(conn: sqlite3.Connection, state: EmotionState) -> None: _ = conn.execute( """ diff --git a/drift/skills/feedback-preference-context/SKILL.md b/drift/skills/feedback-preference-context/SKILL.md index 250ad29..d5aa80f 100644 --- a/drift/skills/feedback-preference-context/SKILL.md +++ b/drift/skills/feedback-preference-context/SKILL.md @@ -14,7 +14,7 @@ description: 从 proactive 正反馈中归纳待审核推送偏好候选，追 ## 流程 ```text -feedback events +Emotion feedback samples ├─ evidence bundle │ ├─ proactive text │ ├─ user text @@ -35,15 +35,15 @@ python3 skills/feedback-preference-context/scripts/sample_feedback_context.py sa ```text sample -├─ 读取 drift.db 中 cursor.latest_processed_feedback_id -├─ 查询 ../proactive_feedback/proactive_feedback.db +├─ 读取 drift.db 中 cursor.latest_processed_emotion_sample_id +├─ 查询 ../emotion/emotion.db 的 emotion_feedback_samples │ └─ feedback_type IN ('topic_follow', 'explicit_quote') -├─ WHERE id > latest_processed_feedback_id +├─ WHERE id > latest_processed_emotion_sample_id ├─ ORDER BY id DESC LIMIT 50 -├─ 用 ../sessions.db 回填 proactive/user 原文 -├─ proactive 文本截断到 100 字，user 文本不截断 +├─ 使用 typed Turn 已写入的 bounded proactive/user 文本 +├─ 不读取 sessions.db 或其他插件数据库 ├─ 每次只返回 chunk-size 条 events -└─ 返回 cursor_tail_feedback_id +└─ 返回 cursor_tail_emotion_sample_id ``` 脚本不判断 topic、不判断 effect、不生成 pending 内容。 @@ -57,7 +57,7 @@ sample { "command": "python3 skills/feedback-preference-context/scripts/sample_feedback_context.py sample --drift-dir . --chunk-index 0 --chunk-size 10", "cwd": ".", - "description": "读取 proactive feedback 第 0 个 chunk", + "description": "读取 Emotion feedback sample 第 0 个 chunk", "timeout": 30 } ``` @@ -78,7 +78,7 @@ sample 14. 只有处理并写入所有 chunk，直到某个 chunk 返回 `has_more=false` 后，才能调用 `finish_drift`。 15. 不要等 50 条全部看完再写，也不要为每条 event 单独写文件。 16. 只追加新候选，不修改、不删除已有队列项；写文件时必须保留原文完整前缀，只在末尾增加新 batch/chunk 内容。 -17. 只看 proactive 消息和 user 回复这一对文本。assistant 后续回答不是证据来源。 +17. 只看 typed Turn sample 中的 proactive 消息和 user 回复文本。assistant 后续回答不是证据来源。 18. MEMORY 只用于理解长期兴趣边界和查重；新增候选必须由当前 chunk 的 feedback 证据支撑。 19. `signal_hints` 只是弱提示；最终 topic 粒度、用户态度、effect 都必须由你结合当前 chunk 和 MEMORY 推断。 20. 不要求用户显式说喜欢或讨厌。追问、纠错、补充背景、切换关注对象、持续互动都可以作为态度证据，但必须解释它对推送决策有什么影响。 @@ -110,9 +110,9 @@ sample - `tone`：改变同一候选内容的表达方式。 27. 不写普通生活事实、人设事实、一次性寒暄、测试消息。 28. evidence 必须包含完整 `feedback#id` 和完整 user message id。 -29. 全部 chunk 都写入成功后，`cursor_update.latest_processed_feedback_id` 必须等于第 0 个 chunk 返回的 `cursor_tail_feedback_id`。 +29. 全部 chunk 都写入成功后，`cursor_update.latest_processed_emotion_sample_id` 必须等于第 0 个 chunk 返回的 `cursor_tail_emotion_sample_id`。 30. `finish_drift.briefing` 必须使用实际处理结果，写清总样本数、chunk 数、pending 候选条数，不要估算。 -31. 如果只处理了部分 chunk，必须 `status="paused"`，不要推进 `latest_processed_feedback_id`。 +31. 如果只处理了部分 chunk，必须 `status="paused"`，不要推进 `latest_processed_emotion_sample_id`。 ## proactive_pending.md 格式 @@ -163,9 +163,9 @@ sample "briefing": "根据 proactive 正反馈样本追加 proactive_pending.md 队列", "message_result": "silent", "cursor_update": { - "latest_processed_feedback_id": 123, - "active_cursor_tail_feedback_id": null, - "active_feedback_ids": null + "latest_processed_emotion_sample_id": 123, + "active_cursor_tail_emotion_sample_id": null, + "active_emotion_sample_ids": null }, "journal_append": [ { @@ -173,7 +173,7 @@ sample "key": "1-123", "payload": { "feedback_ids": [1, 2, 3], - "cursor_tail_feedback_id": 123 + "cursor_tail_emotion_sample_id": 123 } } ] @@ -184,8 +184,8 @@ sample - 一次最多处理 50 条反馈。 - `explicit_quote` 必须包含，且视为更强证据，但不是自动规则。 -- 只有 `proactive_pending.md` 成功尾部追加后才能推进 `last_feedback_id`。 +- 只有 `proactive_pending.md` 成功尾部追加后才能推进 `latest_processed_emotion_sample_id`。 - 写入前必须检查旧内容；禁止用新生成内容覆盖整个 pending 文件。 - 不打扰用户，不调用 `message_push`。 - 不读取或写入 `state.json`、`history.json`。 -- 不修改 proactive_feedback 数据库。 +- 只读 Emotion 数据库；不修改 `emotion.db` 或任何其他插件数据库。 diff --git a/drift/skills/feedback-preference-context/scripts/sample_feedback_context.py b/drift/skills/feedback-preference-context/scripts/sample_feedback_context.py index c9147eb..8707abc 100644 --- a/drift/skills/feedback-preference-context/scripts/sample_feedback_context.py +++ b/drift/skills/feedback-preference-context/scripts/sample_feedback_context.py @@ -12,7 +12,7 @@ def _connect(path: Path) -> sqlite3.Connection: - conn = sqlite3.connect(path) + conn = sqlite3.connect(f"{path.resolve().as_uri()}?mode=ro", uri=True) conn.row_factory = sqlite3.Row return conn @@ -39,24 +39,6 @@ def _load_cursor(drift_dir: Path) -> dict[str, Any]: return cast(dict[str, Any], data) if isinstance(data, dict) else {} -def _message_previews(workspace: Path, ids: list[str]) -> dict[str, str]: - db_path = workspace / "sessions.db" - if not ids or not db_path.exists(): - return {} - unique_ids = list(dict.fromkeys(text for text in ids if text)) - placeholders = ",".join("?" for _ in unique_ids) - with _connect(db_path) as conn: - rows = conn.execute( - f""" - SELECT id, content - FROM messages - WHERE id IN ({placeholders}) - """, - unique_ids, - ).fetchall() - return {str(row["id"]): str(row["content"] or "") for row in rows} - - def _clip_text(text: str, limit: int) -> str: clean = " ".join(str(text or "").split()) if len(clean) <= limit: @@ -86,56 +68,62 @@ def sample( chunk_size: int, chunk_index: int, ) -> dict[str, Any]: + """Read bounded feedback samples from the Emotion-owned derived database.""" + workspace = drift_dir.parent - feedback_db = workspace / "proactive_feedback" / "proactive_feedback.db" - if not feedback_db.exists(): - return {"found": False, "reason": "feedback_db_missing"} + emotion_db = workspace / "emotion" / "emotion.db" + if not emotion_db.is_file() or emotion_db.is_symlink(): + return _empty_result(0, "emotion_db_missing", chunk_index, chunk_size) cursor = _load_cursor(drift_dir) - last_feedback_id = int( - cursor.get("latest_processed_feedback_id") - or cursor.get("last_feedback_id") - or 0 - ) + last_sample_id = int(cursor.get("latest_processed_emotion_sample_id") or 0) safe_limit = max(1, min(int(limit), 50)) - with _connect(feedback_db) as conn: - rows = conn.execute( - """ - SELECT - id, - created_at, - session_key, - user_message_id, - proactive_message_id, - feedback_type, - confidence, - pa_score, - pua_score, - lag_seconds, - candidate_count, - matched_by, - reason - FROM proactive_feedback_events - WHERE id > ? - AND feedback_type IN ('topic_follow', 'explicit_quote') - ORDER BY id DESC - LIMIT ? - """, - (last_feedback_id, safe_limit), - ).fetchall() + try: + with _connect(emotion_db) as conn: + rows = conn.execute( + """ + SELECT + id, + created_at, + session_key, + user_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 emotion_feedback_samples + WHERE id > ? + AND feedback_type IN ('topic_follow', 'explicit_quote') + ORDER BY id DESC + LIMIT ? + """, + (last_sample_id, safe_limit), + ).fetchall() + except sqlite3.OperationalError as exc: + if "no such table: emotion_feedback_samples" not in str(exc): + raise + return _empty_result( + last_sample_id, + "emotion_feedback_samples_missing", + chunk_index, + chunk_size, + ) if not rows: - return { - "found": False, - "last_feedback_id": last_feedback_id, - "latest_processed_feedback_id": last_feedback_id, - "cursor_tail_feedback_id": last_feedback_id, - "count": 0, - "chunk_index": max(0, int(chunk_index)), - "chunk_size": max(1, min(int(chunk_size), 10)), - "has_more": False, - "next_chunk_index": None, - } + return _empty_result( + last_sample_id, + "emotion_feedback_samples_empty", + chunk_index, + chunk_size, + ) safe_chunk_size = max(1, min(int(chunk_size), 10)) safe_chunk_index = max(0, int(chunk_index)) @@ -144,22 +132,11 @@ def sample( chunk_rows = rows[chunk_start:chunk_end] has_more = chunk_end < len(rows) - message_ids: list[str] = [] - for row in chunk_rows: - message_ids.extend( - str(row[key] or "") - for key in ( - "proactive_message_id", - "user_message_id", - ) - if row[key] - ) - previews = _message_previews(workspace, message_ids) events: list[dict[str, Any]] = [] for row in chunk_rows: proactive_id = str(row["proactive_message_id"] or "") user_id = str(row["user_message_id"] or "") - user_text = _clean_text(previews.get(user_id, "")) + user_text = _clean_text(str(row["user_content_preview"] or "")) events.append( { "id": int(row["id"]), @@ -183,7 +160,7 @@ def sample( ), "texts": { "proactive": _clip_text( - previews.get(proactive_id, ""), + str(row["proactive_content_preview"] or ""), PROACTIVE_TEXT_LIMIT, ), "user": user_text, @@ -194,10 +171,10 @@ def sample( cursor_tail = max(int(row["id"]) for row in rows) return { "found": True, - "last_feedback_id": last_feedback_id, - "latest_processed_feedback_id": last_feedback_id, + "last_sample_id": last_sample_id, + "latest_processed_emotion_sample_id": last_sample_id, "count": len(rows), - "cursor_tail_feedback_id": cursor_tail, + "cursor_tail_emotion_sample_id": cursor_tail, "feedback_ids": [int(row["id"]) for row in rows], "chunk_index": safe_chunk_index, "chunk_size": safe_chunk_size, @@ -213,6 +190,26 @@ def sample( } +def _empty_result( + last_sample_id: int, + reason: str, + chunk_index: int, + chunk_size: int, +) -> dict[str, Any]: + return { + "found": False, + "reason": reason, + "last_sample_id": last_sample_id, + "latest_processed_emotion_sample_id": last_sample_id, + "cursor_tail_emotion_sample_id": last_sample_id, + "count": 0, + "chunk_index": max(0, int(chunk_index)), + "chunk_size": max(1, min(int(chunk_size), 10)), + "has_more": False, + "next_chunk_index": None, + } + + def evidence_bundle( drift_dir: Path, limit: int, @@ -250,9 +247,9 @@ def evidence_bundle( return { "found": True, - "last_feedback_id": first["last_feedback_id"], + "last_sample_id": first["last_sample_id"], "count": first["count"], - "cursor_tail_feedback_id": first["cursor_tail_feedback_id"], + "cursor_tail_emotion_sample_id": first["cursor_tail_emotion_sample_id"], "feedback_ids": first["feedback_ids"], "events": compact_events, } diff --git a/plugin.py b/plugin.py index 8293349..dc45022 100644 --- a/plugin.py +++ b/plugin.py @@ -1,29 +1,55 @@ from __future__ import annotations -import logging +import hashlib +import json from pathlib import Path -from typing import Any - -from agent.plugins import ( - EventTrigger, - MobileUiContribution, +from collections.abc import Mapping +from typing import Any, cast + +from agent.plugin_composition import ( + BACKGROUND_JOBS, + BackgroundJobDefinition, + Context, + CoreEvent, + CoreEventTrigger, + MobileUiDefinition, MobileUiNavigation, - Plugin, - PluginJobContext, - PluginJobSpec, - tool, + MobileUiRpcInvalidRequest, + PROACTIVE_COMPONENTS, + ProactiveModuleDefinition, + UI_SLOTS, +) +from agent.turn_events.after_turn import AFTER_TURN_COMMITTED +from bus.events_lifecycle import DriftFinished, TurnCommitted +from agent.plugins.generation_proactive_host import ( + ProactiveModuleContext, + ProactiveModuleOutcome, ) -from agent.plugins.mobile_ui import MobileUiRpcInvalidRequest -from bus.events_proactive import ProactiveFeedbackRecorded -from bus.events_lifecycle import DriftFinished -from proactive_v2.frame import ProactiveFrame -from .db import apply_feedback, build_effect, get_state, open_db +from .db import ( + apply_feedback, + build_effect, + commit_domain_effect, + lookup_effect, + lookup_domain_effect, + lookup_domain_effect_path, + open_db, +) from .dashboard import EmotionDashboardReader -logger = logging.getLogger("plugin.emotion") +api_version = 3 +name = "emotion" +version = "3.0.0" +desc = "Proactive VAD state and feedback preference projection." +inject = (BACKGROUND_JOBS, PROACTIVE_COMPONENTS, UI_SLOTS) +workspace_roots = ("emotion",) +drift_skill_roots = ("drift/skills",) +dashboard_module = "dashboard.py" +_v3_emotion_root: Path | None = None +_v3_emotion_module: "EmotionProjectionModule | None" = None _FEEDBACK_CONTEXT_SKILL = "feedback-preference-context" +_FEEDBACK_PREVIEW_MAX_CHARS = 2400 _PROACTIVE_CONTEXT_TEMPLATE = """# Proactive Context 在这里写会影响未来主动推送取舍的稳定偏好。 @@ -76,213 +102,403 @@ """ -class EmotionProactivePromptModule: - slot = "proactive.prompt.emotion" - produces = ( - "proactive:prompt:system_bottom:emotion", - "proactive:effect:emotion", +async def apply(ctx: Context, config: object) -> None: + """登记 Emotion 的 proactive module、typed Turn observer、job 与 UI。""" + + # 1. 只冻结 Core 投影的 generation-local 数据根，不打开数据库或调用模型。 + del config + global _v3_emotion_module, _v3_emotion_root + _v3_emotion_root = ctx.workspace_root("emotion") + emotion_root = _v3_emotion_root + _v3_emotion_module = EmotionProjectionModule(emotion_root) + + # 2. Proactive module 只在 formal domain-effect facade 中提交 SQLite 状态。 + await ctx.require(PROACTIVE_COMPONENTS).register( + ctx, + ProactiveModuleDefinition( + slot="proactive.prompt.emotion", + lifecycle_id="default.proactive.frame.v1", + produces=( + "proactive:prompt:system_bottom:emotion", + "proactive:effect:emotion", + ), + handler_export="run_emotion_prompt_v3", + domain_effect="emotion.state", + domain_effect_lookup_export="lookup_emotion_domain_effect_v3", + ), ) - def __init__(self, plugin: "EmotionPlugin") -> None: - self._plugin = plugin + # 3. JobHost 独占 event admission、LLM lease、effect receipt 与文档提交。 + await ctx.require(BACKGROUND_JOBS).register( + ctx, + BackgroundJobDefinition( + name="merge_proactive_pending", + triggers=(CoreEventTrigger(CoreEvent.DRIFT_FINISHED),), + handler_export="merge_proactive_pending_v3", + documents_scope=("emotion",), + domain_effect="emotion.state", + domain_effect_lookup_export="lookup_emotion_domain_effect_v3", + model_role="agent", + ), + ) - async def run(self, frame: ProactiveFrame) -> ProactiveFrame: - effect = self._plugin.build_proactive_prompt_effect(frame) - if effect is None: - return frame - frame.slots["proactive:prompt:system_bottom:emotion"] = str( - effect.get("prompt_section") or "" - ) - frame.slots["proactive:effect:emotion"] = effect - return frame + # 4. TurnCommitted 是反馈唯一的 typed owner；listener 固定当前 Root。 + def on_turn_committed(event: TurnCommitted) -> None: + _on_turn_committed(event, root=emotion_root) + await ctx.on(AFTER_TURN_COMMITTED, on_turn_committed) -class EmotionPlugin(Plugin): - api_version = 2 - @classmethod - def dashboard_module(cls) -> str | None: - return "dashboard.py" + # 5. Mobile 只读查询与静态资源绑定到同一个 generation Root。 + def mobile_query( + method: str, + payload: dict[str, object], + *, + session_id: str | None, + turn_id: str | None, + ) -> dict[str, object]: + return _mobile_ui_query( + method, + payload, + session_id=session_id, + turn_id=turn_id, + root=emotion_root, + ) - @classmethod - def mobile_ui(cls) -> MobileUiContribution: - return MobileUiContribution( + await ctx.require(UI_SLOTS).register_mobile( + ctx, + MobileUiDefinition( module="mobile_panel.js", stylesheet="mobile_panel.css", navigation=MobileUiNavigation( label="主动状态", description="反馈如何改变 Agent 的语气和主动发送把握", ), - ) - - name = "emotion" - version = "1.1.0" + ), + query=mobile_query, + ) - @classmethod - def drift_skill_roots(cls) -> tuple[str, ...]: - return ("drift/skills",) - def activate(self) -> None: - workspace = self.context.workspace - if workspace is None: - logger.warning("emotion 插件缺少 workspace，跳过加载") - return - self._db_path = workspace / "emotion" / "emotion.db" - conn = open_db(self._db_path) - conn.close() - self.context.event_bus.on(ProactiveFeedbackRecorded, self._on_feedback_recorded) +class EmotionProjectionModule: + """Build one proactive emotion projection and submit its domain receipt.""" - async def terminate(self) -> None: - return None + def __init__(self, root: Path) -> None: + self._root = root - def mobile_ui_query( + async def run( self, - method: str, - payload: dict[str, object], - *, - session_id: str | None, - turn_id: str | None, - ) -> dict[str, object]: - """返回反馈如何调节主动状态的移动投影。""" - - # 1. 在插件 RPC 边界校验方法与列表上限 - _ = session_id, turn_id - if method != "emotion.bootstrap": - raise MobileUiRpcInvalidRequest(f"未知 emotion 移动方法: {method}") - workspace = self.context.workspace - if workspace is None: - raise RuntimeError("emotion 移动看板缺少 workspace") - limit = payload.get("limit", 30) - if not isinstance(limit, int) or isinstance(limit, bool) or not 1 <= limit <= 50: - raise MobileUiRpcInvalidRequest("limit 必须是 1 到 50 的整数") - - # 2. 单个 SQLite 快照返回首屏全部区域 - return EmotionDashboardReader(workspace).get_mobile_bootstrap(limit=limit) - - def proactive_modules(self) -> list[object]: - return [EmotionProactivePromptModule(self)] - - def jobs(self) -> list[PluginJobSpec]: - return [ - PluginJobSpec( - id="merge_proactive_pending", - triggers=[EventTrigger(DriftFinished)], - handler=self.merge_proactive_pending, + context: ProactiveModuleContext, + frame: ProactiveModuleOutcome, + ) -> ProactiveModuleOutcome: + """Persist one formal frame effect through Core's exact domain facade.""" + + # 1. Resolve the only allowed domain effect and derive stable frame inputs. + effects = getattr(context, "domain_effects", None) + if effects is None or not callable(getattr(effects, "run", None)): + raise RuntimeError( + "emotion proactive module 缺少 Core-owned domain effects facade" ) - ] - - async def merge_proactive_pending(self, ctx: PluginJobContext) -> None: - event = ctx.event - if not isinstance(event, DriftFinished): - return - if event.skill_name != _FEEDBACK_CONTEXT_SKILL or event.status != "completed": - return - workspace = self.context.workspace - if workspace is None: - return - - pending_path = workspace / "proactive_pending.md" - context_path = workspace / "PROACTIVE_CONTEXT.md" - pending = self._read_text(pending_path).strip() - if not pending or "- [ ]" not in pending: - return - - current_context = self._read_text(context_path).strip() - if not current_context: - current_context = _PROACTIVE_CONTEXT_TEMPLATE.strip() - prompt = _MERGE_PROACTIVE_CONTEXT_PROMPT.format( - current_context=current_context, - pending=pending, + session_key = str( + frame.slots.get("proactive:session_key") or frame.input.session_key ) - merged = await ctx.llm.generate_text( - system=_MERGE_PROACTIVE_CONTEXT_SYSTEM, - prompt=prompt, - max_tokens=4096, + base_threshold = float( + frame.slots.get("proactive:base_judge_send_threshold") or 0.60 ) - if not merged: - return - _ = context_path.write_text(merged.strip() + "\n", encoding="utf-8") - _ = pending_path.write_text("", encoding="utf-8") - logger.info("emotion proactive pending 已合并到 PROACTIVE_CONTEXT.md") - - @staticmethod - def _read_text(path: Path) -> str: - if not path.exists(): - return "" - return path.read_text(encoding="utf-8") - - def build_proactive_prompt_effect( - self, - frame: ProactiveFrame, - ) -> dict[str, Any] | None: - db_path = getattr(self, "_db_path", None) - if db_path is None: - return None - conn = open_db(Path(db_path)) - try: - return build_effect( - conn, - tick_id=f"frame:{frame.input.started_at.isoformat()}", - session_key=str( - frame.slots.get("proactive:session_key") - or frame.input.session_key - ), - now_utc=frame.input.started_at, - last_user_at=frame.slots.get("proactive:last_user_at"), - base_threshold=float( - frame.slots.get("proactive:base_judge_send_threshold") or 0.60 - ), - ) - finally: - conn.close() - - def _on_feedback_recorded(self, event: ProactiveFeedbackRecorded) -> None: - db_path = getattr(self, "_db_path", None) - if db_path is None: - return - payload: dict[str, Any] = { - "feedback_event_id": event.event_id, - "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, - "pua_score": event.pua_score, - "lag_seconds": event.lag_seconds, - "matched_by": event.matched_by, - } - conn = open_db(Path(db_path)) + last_user_at = frame.slots.get("proactive:last_user_at") + if last_user_at is not None and not hasattr(last_user_at, "tzinfo"): + raise TypeError("emotion last_user_at 必须是 datetime 或 None") + tick_id = f"{frame.input.session_key}:{frame.input.started_at.isoformat()}" + result: dict[str, Any] = {} + + # 2. Core signs the effect only after this SQLite transaction has a durable receipt. + def transaction(effect_context: object) -> None: + exact_context = cast(Any, effect_context) + conn = open_db(self._root / "emotion.db") + try: + conn.execute("BEGIN IMMEDIATE") + if ( + exact_context.event_id != tick_id + or exact_context.tick_id != tick_id + ): + raise RuntimeError("Emotion proactive tick identity 与 Core 不一致") + effect: dict[str, Any] + result["effect"] = build_effect( + conn, + tick_id=tick_id, + session_key=session_key, + now_utc=frame.input.started_at, + last_user_at=cast(Any, last_user_at), + base_threshold=base_threshold, + commit=False, + ) + effect = cast(dict[str, Any], result["effect"]) + _ = commit_domain_effect( + conn, + semantic_job_id=exact_context.semantic_job_id, + event_id=exact_context.event_id, + invocation_id=exact_context.invocation_id, + effect_id=exact_context.effect_id, + idempotency_key=exact_context.idempotency_key, + attempt=exact_context.attempt, + result_digest=_effect_digest(effect), + ) + conn.commit() + except BaseException: + conn.rollback() + raise + finally: + conn.close() + + await effects.run("emotion.state", transaction) + effect = result.get("effect") + if not isinstance(effect, dict): + conn = open_db(self._root / "emotion.db") + try: + effect = lookup_effect(conn, tick_id=tick_id) + finally: + conn.close() + if not isinstance(effect, dict): + raise RuntimeError("emotion domain effect 未返回 frame projection") + frame.slots["proactive:prompt:system_bottom:emotion"] = str( + effect.get("prompt_section") or "" + ) + frame.slots["proactive:effect:emotion"] = effect + return frame + + +def _effect_digest(effect: Mapping[str, object]) -> str: + """Return the stable digest Core records for one committed projection.""" + + payload = json.dumps( + effect, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + return hashlib.sha256(payload).hexdigest() + + +async def run_emotion_prompt_v3( + context: ProactiveModuleContext, + frame: ProactiveModuleOutcome, +) -> ProactiveModuleOutcome: + """Run the exact-generation emotion proactive module.""" + + module = _v3_emotion_module + if module is None: + raise RuntimeError("emotion v3 generation 尚未完成 apply") + return await module.run(context, frame) + + +def _on_turn_committed(event: TurnCommitted, *, root: Path | None = None) -> None: + """Project one typed committed Turn into Emotion's idempotent SQLite state.""" + + feedback = _feedback_from_turn(event) + if feedback is None: + return + root = _require_v3_emotion_root() if root is None else root + conn = open_db(root / "emotion.db") + try: + apply_feedback( + conn, + source_event_id=feedback["source_event_id"], + session_key=event.session_key, + feedback_type=feedback["feedback_type"], + confidence=feedback["confidence"], + payload=feedback["payload"], + ) + finally: + conn.close() + + +def _feedback_from_turn(event: TurnCommitted) -> dict[str, Any] | None: + """Read an optional typed feedback result, with explicit quote as the local fallback.""" + + raw = event.extra.get("proactive_feedback") + if isinstance(raw, Mapping): + feedback_type = raw.get("feedback_type") + confidence = raw.get("confidence") + if isinstance(feedback_type, str) and isinstance(confidence, str): + source = raw.get("event_id") or event.turn_id or event.persisted_user_message_id + if isinstance(source, str) and source: + payload = { + "feedback_event_id": str(source), + "user_message_id": event.persisted_user_message_id, + "assistant_message_id": event.assistant_message_id, + "proactive_message_id": raw.get("proactive_message_id"), + "feedback_type": feedback_type, + "confidence": confidence, + "pua_score": raw.get("pua_score"), + "lag_seconds": raw.get("lag_seconds"), + "matched_by": raw.get("matched_by", "typed_turn"), + "candidate_count": raw.get("candidate_count"), + "pa_score": raw.get("pa_score"), + "reason": raw.get("reason", "typed_turn"), + "user_content_preview": _feedback_preview( + raw.get("user_content_preview") + or event.persisted_user_message + or event.input_message + ), + "assistant_content_preview": _feedback_preview( + raw.get("assistant_content_preview") or event.assistant_response + ), + "proactive_content_preview": _feedback_preview( + raw.get("proactive_content_preview") + or _quoted_proactive_text(event.input_message) + ), + } + return { + "source_event_id": f"proactive_feedback:{source}", + "feedback_type": feedback_type, + "confidence": confidence, + "payload": payload, + } + + # A quote is already an explicit feedback signal carried by the committed user message. + marker = "【你当前新消息】" + source = event.turn_id or event.persisted_user_message_id + if marker not in event.input_message or not isinstance(source, str) or not source: + return None + payload = { + "feedback_event_id": source, + "user_message_id": event.persisted_user_message_id, + "assistant_message_id": event.assistant_message_id, + "proactive_message_id": None, + "feedback_type": "explicit_quote", + "confidence": "gold", + "pua_score": 1.0, + "lag_seconds": None, + "matched_by": "explicit_quote", + "candidate_count": None, + "pa_score": None, + "reason": "explicit_quote", + "user_content_preview": _feedback_preview( + event.persisted_user_message or event.input_message + ), + "assistant_content_preview": _feedback_preview(event.assistant_response), + "proactive_content_preview": _feedback_preview( + _quoted_proactive_text(event.input_message) + ), + } + return { + "source_event_id": f"proactive_feedback:{source}", + "feedback_type": "explicit_quote", + "confidence": "gold", + "payload": payload, + } + + +def _feedback_preview(value: object) -> str | None: + if not isinstance(value, str): + return None + clean = " ".join(value.split()) + if len(clean) <= _FEEDBACK_PREVIEW_MAX_CHARS: + return clean + return clean[:_FEEDBACK_PREVIEW_MAX_CHARS].rstrip() + "..." + + +def _quoted_proactive_text(value: str) -> str | None: + marker = "【你当前新消息】" + quoted = value.split(marker, 1)[0].strip() + prefix = "被回复消息：" + if quoted.startswith(prefix): + quoted = quoted[len(prefix) :].strip() + return quoted or None + + +def _mobile_ui_query( + method: str, + payload: dict[str, object], + *, + session_id: str | None, + turn_id: str | None, + root: Path | None = None, +) -> dict[str, object]: + """Return a read-only bounded Emotion mobile projection.""" + + _ = session_id, turn_id + if method != "emotion.bootstrap": + raise MobileUiRpcInvalidRequest(f"未知 emotion 移动方法: {method}") + limit = payload.get("limit", 30) + if not isinstance(limit, int) or isinstance(limit, bool) or not 1 <= limit <= 50: + raise MobileUiRpcInvalidRequest("limit 必须是 1 到 50 的整数") + emotion_root = _require_v3_emotion_root() if root is None else root + return EmotionDashboardReader(emotion_root).get_mobile_bootstrap( + limit=limit + ) + + +async def merge_proactive_pending_v3(ctx: Any) -> None: + """形成 merge 内容，并经 Core receipt fence 发布两份 proactive 文档。""" + + # 1. 非目标 Drift completion 不读取文档、不调用模型。 + event = ctx.event + if not isinstance(event, DriftFinished): + return + if event.skill_name != _FEEDBACK_CONTEXT_SKILL or event.status != "completed": + return + if ctx.documents is None or ctx.domain_effects is None: + raise RuntimeError("emotion merge job 缺少 Core documents/domain effects") + + # 2. 从窄 port 读取脱离 bytes，生成新文档并先持久化完整 intent。 + expected, current = ctx.documents.read_pair() + pending = current.pending.decode("utf-8").strip() + if not pending or "- [ ]" not in pending: + return + current_context = current.context.decode("utf-8").strip() + if not current_context: + current_context = _PROACTIVE_CONTEXT_TEMPLATE.strip() + prompt = _MERGE_PROACTIVE_CONTEXT_PROMPT.format( + current_context=current_context, + pending=pending, + ) + merged = await ctx.llm.generate_text( + system=_MERGE_PROACTIVE_CONTEXT_SYSTEM, + prompt=prompt, + max_tokens=4096, + ) + if not merged: + return + pair = { + "context": merged.strip().encode("utf-8") + b"\n", + "pending": b"", + } + intent = await ctx.documents.prepare_pair(expected, pair) + result_digest = hashlib.sha256(pair["context"] + b"\0" + pair["pending"]).hexdigest() + + # 3. Emotion SQLite 提交领域 receipt 后，Core 才能向前提交文档。 + async def transaction(effect_ctx: Any) -> None: + db = open_db(_require_v3_emotion_root() / "emotion.db") try: - _ = apply_feedback( - conn, - source_event_id=f"proactive_feedback:{event.event_id}", - session_key=event.session_key, - feedback_type=event.feedback_type, - confidence=event.confidence, - payload=payload, + _ = commit_domain_effect( + db, + semantic_job_id=effect_ctx.semantic_job_id, + event_id=effect_ctx.event_id or event.event_id, + invocation_id=effect_ctx.invocation_id, + effect_id=effect_ctx.effect_id, + idempotency_key=effect_ctx.idempotency_key, + attempt=effect_ctx.attempt, + result_digest=result_digest, ) finally: - conn.close() + db.close() - @tool( - "get_emotion_state", - risk="read-only", - search_hint="查询 proactive VAD 情绪状态", + receipt = await ctx.domain_effects.run("emotion.state", transaction) + _ = await ctx.documents.commit_after(intent, receipt) + + +def lookup_emotion_domain_effect_v3(effect_ctx: Any) -> object | None: + """只读返回 Emotion durable receipt，供 Core 重签 exact capability。""" + + return lookup_domain_effect_path( + _require_v3_emotion_root() / "emotion.db", + invocation_id=effect_ctx.invocation_id, + effect_id=effect_ctx.effect_id, + idempotency_key=effect_ctx.idempotency_key, ) - async def get_emotion_state(self, event: Any) -> dict[str, Any]: - """查询 proactive VAD 情绪状态。""" - _ = event - db_path = getattr(self, "_db_path", None) - if db_path is None: - return {"available": False} - conn = open_db(Path(db_path)) - try: - state = get_state(conn) - finally: - conn.close() - return { - "available": True, - "valence": state.valence, - "arousal": state.arousal, - "dominance": state.dominance, - "updated_at": state.updated_at, - } + + +def _require_v3_emotion_root() -> Path: + if _v3_emotion_root is None: + raise RuntimeError("emotion v3 generation 尚未绑定 workspace root") + return _v3_emotion_root diff --git a/tests/test_feedback_sample.py b/tests/test_feedback_sample.py new file mode 100644 index 0000000..d203a57 --- /dev/null +++ b/tests/test_feedback_sample.py @@ -0,0 +1,108 @@ +from __future__ import annotations + +import importlib.util +import sys +from datetime import datetime, timezone +from pathlib import Path + +from bus.events_lifecycle import TurnCommitted + + +def _load_module(path: Path, name: str): + spec = importlib.util.spec_from_file_location( + name, + path, + submodule_search_locations=[str(path.parent)], + ) + if spec is None or spec.loader is None: + raise ImportError(str(path)) + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + return module + + +PLUGIN = _load_module(Path(__file__).parents[1] / "plugin.py", "test_emotion_sample_plugin") +SAMPLE = _load_module( + Path(__file__).parents[1] + / "drift/skills/feedback-preference-context/scripts/sample_feedback_context.py", + "test_emotion_feedback_sample", +) + + +def _turn() -> TurnCommitted: + return TurnCommitted( + session_key="sample:test", + channel="test", + chat_id="chat", + input_message="被回复消息：主动提醒关于迁移的主题\n\n【你当前新消息】继续这个主题", + persisted_user_message="被回复消息：主动提醒关于迁移的主题\n\n【你当前新消息】继续这个主题", + assistant_response="好的，我继续整理。", + tools_used=[], + turn_id="turn-sample-1", + persisted_user_message_id="user-sample-1", + assistant_message_id="assistant-sample-1", + timestamp=datetime.now(timezone.utc), + extra={ + "proactive_feedback": { + "event_id": "feedback-sample-1", + "feedback_type": "topic_follow", + "confidence": "high", + "proactive_message_id": "proactive-sample-1", + "proactive_content_preview": "主动提醒关于迁移的主题", + "pua_score": 0.91, + "candidate_count": 2, + "matched_by": "typed_turn", + "reason": "topic_follow_high", + } + }, + ) + + +def test_typed_turn_feedback_is_readable_without_legacy_databases(tmp_path: Path) -> None: + emotion_root = tmp_path / "emotion" + PLUGIN._on_turn_committed(_turn(), root=emotion_root) + + old_feedback = tmp_path / "proactive_feedback" + old_feedback.mkdir() + (old_feedback / "proactive_feedback.db").write_text("not sqlite", encoding="utf-8") + (tmp_path / "sessions.db").write_text("not sqlite", encoding="utf-8") + + result = SAMPLE.sample(tmp_path / "drift", 50, 10, 0) + + assert result["found"] is True + assert result["count"] == 1 + event = result["events"][0] + assert event["feedback_type"] == "topic_follow" + assert event["message_ids"] == { + "proactive": "proactive-sample-1", + "user": "user-sample-1", + } + assert event["texts"] == { + "proactive": "主动提醒关于迁移的主题", + "user": "被回复消息：主动提醒关于迁移的主题 【你当前新消息】继续这个主题", + } + + +def test_feedback_sample_missing_does_not_fallback_to_legacy_owner(tmp_path: Path) -> None: + old_feedback = tmp_path / "proactive_feedback" + old_feedback.mkdir() + (old_feedback / "proactive_feedback.db").write_text("not sqlite", encoding="utf-8") + (tmp_path / "sessions.db").write_text("not sqlite", encoding="utf-8") + + result = SAMPLE.sample(tmp_path / "drift", 50, 10, 0) + + assert result["found"] is False + assert result["reason"] == "emotion_db_missing" + assert not (tmp_path / "emotion" / "emotion.db").exists() + + +def test_feedback_sample_empty_is_closed_without_creating_legacy_reads(tmp_path: Path) -> None: + db = PLUGIN.open_db(tmp_path / "emotion" / "emotion.db") + db.close() + + result = SAMPLE.sample(tmp_path / "drift", 50, 10, 0) + + assert result["found"] is False + assert result["reason"] == "emotion_feedback_samples_empty" + assert not (tmp_path / "sessions.db").exists() diff --git a/tests/test_plugin.py b/tests/test_plugin.py index c548050..14a29fe 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -1,16 +1,52 @@ from __future__ import annotations +import asyncio +import hashlib import importlib.util +import inspect +import os +import shutil +import subprocess import sys from datetime import datetime, timedelta, timezone from pathlib import Path +from types import SimpleNamespace +from typing import Any, cast import pytest -from agent.plugins.context import PluginContext, PluginKVStore -from agent.plugins.scope import PluginScope, ScopedEventBus +from agent.plugin_composition import ( + BACKGROUND_JOBS, + PROACTIVE_COMPONENTS, + UI_SLOTS, + CompositionRoot, + PluginRuntime, +) +from agent.plugin_composition.background_jobs import ( + PluginBackgroundJobs, + _freeze_plugin_background_jobs, +) +from agent.plugin_composition.proactive import ( + PluginProactiveComponents, + _freeze_plugin_proactive_components, +) +from agent.plugin_composition.ui_slots import PluginUiSlots +from agent.plugins.generation_activity_host import ActivityHost +from agent.plugins.generation_proactive_bridge import CommittedProactiveBridge +from agent.plugins.generation_proactive_host import ( + ProactiveActivityAdapter, + ProactiveModuleOutcome, +) +from agent.plugins.manager import PluginManager +from agent.plugins.proactive_documents import ( + ProactiveDocumentDigests, + ProactiveDocumentPair, +) +from bus.events_lifecycle import DriftFinished, TurnCommitted from bus.event_bus import EventBus -from bus.events_proactive import ProactiveFeedbackRecorded +from proactive_v2.frame import ProactiveTickInput + +ProactiveFrame = ProactiveModuleOutcome def _load_plugin_module(): @@ -29,127 +65,723 @@ def _load_plugin_module(): module = _load_plugin_module() -EmotionPlugin = module.EmotionPlugin +async def _mount_runtime(tmp_path: Path) -> tuple[CompositionRoot, Path]: + workspace = tmp_path / "workspace" + emotion_root = workspace / "emotion" + emotion_root.mkdir(parents=True) + root = CompositionRoot("emotion-v3") + jobs = PluginBackgroundJobs(root.instance_token) + proactive = PluginProactiveComponents(root.instance_token) + ui = PluginUiSlots() + _ = await root.context.provide(BACKGROUND_JOBS, jobs) + _ = await root.context.provide(PROACTIVE_COMPONENTS, proactive) + _ = await root.context.provide(UI_SLOTS, ui) + return root, emotion_root + + +def _copy_emotion_plugin(tmp_path: Path) -> Path: + """Copy the plugin into an isolated discovery root for Manager tests.""" + + source = Path(__file__).parents[1] + target = tmp_path / "plugins" / "emotion" + shutil.copytree( + source, + target, + ignore=shutil.ignore_patterns( + ".git", ".akashic-core", "__pycache__", ".pytest_cache" + ), + ) + return target -def _plugin_context(tmp_path: Path) -> PluginContext: - scope = PluginScope("emotion") - return PluginContext( - event_bus=ScopedEventBus(EventBus(), scope), +async def _start_emotion_manager( + tmp_path: Path, + plugin_dir: Path | None = None, +) -> PluginManager: + """Boot one real Manager generation with Core's ActivityHost owner.""" + + plugin_dir = plugin_dir or _copy_emotion_plugin(tmp_path) + manager = PluginManager( + plugin_dirs=[plugin_dir.parent], + event_bus=EventBus(), tool_registry=None, - plugin_id="emotion", - plugin_dir=tmp_path, - data_dir=tmp_path, - kv_store=PluginKVStore(tmp_path / ".kv.json"), - workspace=tmp_path, - scope=scope, + workspace=tmp_path / "workspace", + installed_cache_root=tmp_path / "cache", + ) + manager.bind_activity_host(ActivityHost((ProactiveActivityAdapter(),))) + await manager.load_all() + return manager + + +async def _run_manager_tick( + manager: PluginManager, + frame: ProactiveFrame, +) -> ProactiveFrame: + """Run one proactive module through the exact stable Root and lease.""" + + snapshot = manager.current_snapshot + activity = manager.activity_host + if snapshot is None or activity is None: + raise AssertionError("Manager 未发布 stable snapshot/activity") + lease = await manager.snapshot_store.acquire(snapshot.snapshot_id) + admission = activity.acquire(lease) + bridge = CommittedProactiveBridge(activity) + token = bridge.bind_execution(lease) + try: + runtime = bridge.runtime_for(snapshot) + modules = bridge.lifecycle_modules( + runtime, + lifecycle_id="default.proactive.frame.v1", + ) + if len(modules) != 1: + raise AssertionError(f"Emotion proactive module 数量异常: {len(modules)}") + return await cast(Any, modules[0]).run(frame) + finally: + bridge.reset_execution(token) + await admission.release() + await lease.release() + + +def test_module_exports_pure_v3_contract() -> None: + assert module.api_version == 3 + assert module.name == "emotion" + assert module.version == "3.0.0" + assert inspect.signature(module.apply).parameters.keys() == {"ctx", "config"} + module_file = module.__file__ + assert isinstance(module_file, str) + source = Path(module_file).read_text(encoding="utf-8") + db_source = Path(module_file).with_name("db.py").read_text(encoding="utf-8") + assert "class EmotionPlugin" not in source + assert "ProactiveFeedbackRecorded" not in source + production_source = f"{source}\n{db_source}" + assert "proactive_v2.frame" not in production_source + assert "proactive_v2.energy" not in production_source + assert "EventBus" not in source + assert "from agent.plugins import" not in source + + +@pytest.mark.asyncio +async def test_v3_apply_freezes_all_catalogs_without_opening_db(tmp_path: Path) -> None: + root, emotion_root = await _mount_runtime(tmp_path) + _ = await root.mount( + lambda ctx: module.apply(ctx, object()), + name="emotion", + inject=(BACKGROUND_JOBS, PROACTIVE_COMPONENTS, UI_SLOTS), + runtime=PluginRuntime( + plugin_id="emotion", + plugin_dir=Path(__file__).parents[1], + data_dir=tmp_path / "plugin-data", + workspace=emotion_root.parent, + config=None, + workspace_roots=("emotion",), + ), + ) + jobs = root.context.get(BACKGROUND_JOBS) + proactive = root.context.get(PROACTIVE_COMPONENTS) + assert jobs is not None and proactive is not None + job_catalog = _freeze_plugin_background_jobs(jobs, root.instance_token) + proactive_catalog = _freeze_plugin_proactive_components( + proactive, + root.instance_token, ) + assert job_catalog.job("emotion:merge_proactive_pending") is not None + module_binding = proactive_catalog.module("emotion:proactive.prompt.emotion") + assert module_binding is not None + assert module_binding.definition.domain_effect == "emotion.state" + assert not (emotion_root / "emotion.db").exists() + await root.dispose() + + +@pytest.mark.asyncio +async def test_typed_turn_feedback_is_idempotent_and_mobile_read_only( + tmp_path: Path, +) -> None: + root, emotion_root = await _mount_runtime(tmp_path) + _ = await root.mount( + lambda ctx: module.apply(ctx, object()), + name="emotion", + inject=(BACKGROUND_JOBS, PROACTIVE_COMPONENTS, UI_SLOTS), + runtime=PluginRuntime( + plugin_id="emotion", + plugin_dir=Path(__file__).parents[1], + data_dir=tmp_path / "plugin-data", + workspace=emotion_root.parent, + config=None, + workspace_roots=("emotion",), + ), + ) + event = TurnCommitted( + session_key="mobile:test", + channel="test", + chat_id="chat", + input_message="被回复消息：主动提醒某个主题\n\n【你当前新消息】继续这个主题", + persisted_user_message="被回复消息：主动提醒某个主题\n\n【你当前新消息】继续这个主题", + assistant_response="继续回答", + tools_used=[], + turn_id="turn-1", + persisted_user_message_id="u1", + assistant_message_id="a1", + ) + module._on_turn_committed(event) + module._on_turn_committed(event) + before = sorted(path.relative_to(tmp_path).as_posix() for path in tmp_path.rglob("*") if path.is_file()) + bootstrap = module._mobile_ui_query( + "emotion.bootstrap", + {"limit": 10}, + session_id=None, + turn_id=None, + ) + after = sorted(path.relative_to(tmp_path).as_posix() for path in tmp_path.rglob("*") if path.is_file()) + assert before == after + assert bootstrap["overview"]["event_count"] == 1 + assert bootstrap["overview"]["influence_count"] == 1 + assert bootstrap["items"][0]["source_type"] == "explicit_quote" + await root.dispose() + + +@pytest.mark.asyncio +async def test_proactive_projection_requires_and_uses_domain_effect_facade( + tmp_path: Path, +) -> None: + emotion_root = tmp_path / "emotion" + emotion_root.mkdir() + projection = module.EmotionProjectionModule(emotion_root) + frame = ProactiveFrame( + input=ProactiveTickInput( + session_key="proactive:test", + started_at=datetime(2026, 8, 17, tzinfo=timezone.utc), + ) + ) + calls: list[str] = [] + + class Effects: + async def run(self, effect_id: str, transaction): + calls.append(effect_id) + effect_context = SimpleNamespace( + semantic_job_id="emotion:proactive.prompt.emotion", + event_id="proactive:test:2026-08-17T00:00:00+00:00", + invocation_id=( + "proactive:emotion:proactive.prompt.emotion:" + "proactive:test:2026-08-17T00:00:00+00:00" + ), + effect_id=effect_id, + idempotency_key=( + "proactive:test:2026-08-17T00:00:00+00:00:" + "emotion:proactive.prompt.emotion" + ), + attempt=1, + tick_id="proactive:test:2026-08-17T00:00:00+00:00", + ) + result = transaction(effect_context) + if inspect.isawaitable(result): + await result + return object() + + result = await projection.run(SimpleNamespace(domain_effects=Effects()), frame) + assert result.slots["proactive:prompt:system_bottom:emotion"] + assert result.slots["proactive:effect:emotion"]["metadata"]["expected_effect"] == "tone_only" + assert calls == ["emotion.state"] + db = module.open_db(emotion_root / "emotion.db") + try: + assert db.execute("SELECT COUNT(*) FROM emotion_effects").fetchone()[0] == 1 + assert db.execute("SELECT COUNT(*) FROM emotion_domain_effects").fetchone()[0] == 1 + finally: + db.close() + with pytest.raises(RuntimeError, match="domain effects facade"): + await projection.run(SimpleNamespace(domain_effects=None), frame) @pytest.mark.asyncio -async def test_emotion_plugin_activates_and_reads_state(tmp_path: Path) -> None: - plugin = EmotionPlugin() - plugin.context = _plugin_context(tmp_path) - plugin.activate() +async def test_manager_proactive_failure_rolls_back_then_retries_idempotently( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + manager = await _start_emotion_manager(tmp_path) try: - plugin._on_feedback_recorded( - ProactiveFeedbackRecorded( - event_id=1, - session_key="telegram:1", - user_message_id="u1", - assistant_message_id="a1", - proactive_message_id="p1", - feedback_type="topic_follow", - confidence="high", - pua_score=0.7, - lag_seconds=1, - matched_by="recent_pua", + generation = manager.generation("emotion") + assert generation is not None + plugin_module = cast(Any, generation.instance).module + frame = ProactiveFrame( + input=ProactiveTickInput( + session_key="proactive:emotion", + started_at=datetime(2026, 8, 17, tzinfo=timezone.utc), ) ) - state = await plugin.get_emotion_state(None) + original_build_effect = plugin_module.build_effect + + def fail_after_writes(conn, **kwargs): + result = original_build_effect(conn, **kwargs) + raise RuntimeError("synthetic precommit failure") + + monkeypatch.setattr(plugin_module, "build_effect", fail_after_writes) + with pytest.raises(RuntimeError, match="synthetic precommit failure"): + await _run_manager_tick(manager, frame) + + db = module.open_db(tmp_path / "workspace" / "emotion" / "emotion.db") + try: + assert db.execute("SELECT COUNT(*) FROM emotion_effects").fetchone()[0] == 0 + assert ( + db.execute("SELECT COUNT(*) FROM emotion_domain_effects").fetchone()[0] + == 0 + ) + state = db.execute( + "SELECT valence, arousal, dominance FROM emotion_state WHERE id = 1" + ).fetchone() + assert state is not None and tuple(state) == (0.0, 0.0, 0.0) + finally: + db.close() + + monkeypatch.setattr(plugin_module, "build_effect", original_build_effect) + first = await _run_manager_tick(manager, frame) + second = await _run_manager_tick(manager, frame) + assert first.slots["proactive:effect:emotion"] == second.slots[ + "proactive:effect:emotion" + ] + + db = module.open_db(tmp_path / "workspace" / "emotion" / "emotion.db") + try: + assert db.execute("SELECT COUNT(*) FROM emotion_effects").fetchone()[0] == 1 + assert ( + db.execute("SELECT COUNT(*) FROM emotion_domain_effects").fetchone()[0] + == 1 + ) + tick_id = "proactive:emotion:2026-08-17T00:00:00+00:00" + receipt = db.execute( + """ + SELECT semantic_job_id, event_id, invocation_id, effect_id, + idempotency_key, attempt + FROM emotion_domain_effects + """ + ).fetchone() + assert receipt is not None + assert tuple(receipt) == ( + "emotion:proactive.prompt.emotion", + tick_id, + f"proactive:emotion:proactive.prompt.emotion:{tick_id}", + "emotion.state", + f"{tick_id}:emotion:proactive.prompt.emotion", + 1, + ) + finally: + db.close() finally: - await plugin.terminate() - assert state["available"] is True + await manager.terminate_all() @pytest.mark.asyncio -async def test_mobile_projection_returns_state_and_real_influences(tmp_path: Path) -> None: - plugin = EmotionPlugin() - plugin.context = _plugin_context(tmp_path) - plugin.activate() +async def test_manager_proactive_commit_survives_cancellation_and_reentry( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + manager = await _start_emotion_manager(tmp_path) try: - base = datetime(2026, 7, 17, tzinfo=timezone.utc) - db = module.open_db(tmp_path / "emotion" / "emotion.db") + from agent.plugins.generation_job_host import ProactiveDomainEffects + + frame = ProactiveFrame( + input=ProactiveTickInput( + session_key="proactive:emotion-cancel", + started_at=datetime(2026, 8, 17, 0, 1, tzinfo=timezone.utc), + ) + ) + original_lookup = ProactiveDomainEffects._lookup_committed + lookup_calls = 0 + + async def cancel_after_commit(self): + nonlocal lookup_calls + lookup_calls += 1 + record = await original_lookup(self) + if lookup_calls == 2: + raise asyncio.CancelledError + return record + + monkeypatch.setattr( + ProactiveDomainEffects, + "_lookup_committed", + cancel_after_commit, + ) + with pytest.raises(asyncio.CancelledError): + await _run_manager_tick(manager, frame) + assert lookup_calls == 2 + + monkeypatch.setattr( + ProactiveDomainEffects, + "_lookup_committed", + original_lookup, + ) + resumed = await _run_manager_tick(manager, frame) + assert resumed.slots["proactive:prompt:system_bottom:emotion"] + + db = module.open_db(tmp_path / "workspace" / "emotion" / "emotion.db") try: - for index in range(10): - module.build_effect( - db, - tick_id=f"noise:{index}", - session_key="proactive:default", - now_utc=base + timedelta(minutes=index), - last_user_at=base, - base_threshold=0.6, - ) + assert db.execute("SELECT COUNT(*) FROM emotion_effects").fetchone()[0] == 1 + assert ( + db.execute("SELECT COUNT(*) FROM emotion_domain_effects").fetchone()[0] + == 1 + ) finally: db.close() - for event_id in range(1, 5): - plugin._on_feedback_recorded( - ProactiveFeedbackRecorded( - event_id=event_id, - session_key="mobile:test", - user_message_id=f"u-mobile-{event_id}", - assistant_message_id=f"a-mobile-{event_id}", - proactive_message_id=f"p-mobile-{event_id}", - feedback_type="explicit_quote", - confidence="gold", - pua_score=None, - lag_seconds=4, - matched_by="quote", + finally: + await manager.terminate_all() + + +def test_manager_proactive_receipt_survives_core_process_crash_and_reentry( + tmp_path: Path, +) -> None: + plugin_dir = _copy_emotion_plugin(tmp_path) + workspace = tmp_path / "workspace" + script = """ +import asyncio +import os +import sys +from datetime import datetime, timezone +from pathlib import Path + +from agent.plugins.generation_activity_host import ActivityHost +from agent.plugins.generation_proactive_bridge import CommittedProactiveBridge +from agent.plugins.generation_proactive_host import ProactiveActivityAdapter +from agent.plugins.generation_job_host import ProactiveDomainEffects +from agent.plugins.manager import PluginManager +from bus.event_bus import EventBus +from proactive_v2.frame import ProactiveFrame, ProactiveTickInput + + +async def run_tick(manager, frame): + snapshot = manager.current_snapshot + activity = manager.activity_host + assert snapshot is not None and activity is not None + lease = await manager.snapshot_store.acquire(snapshot.snapshot_id) + admission = activity.acquire(lease) + bridge = CommittedProactiveBridge(activity) + token = bridge.bind_execution(lease) + try: + runtime = bridge.runtime_for(snapshot) + modules = bridge.lifecycle_modules( + runtime, + lifecycle_id="default.proactive.frame.v1", + ) + assert len(modules) == 1 + await modules[0].run(frame) + finally: + bridge.reset_execution(token) + await admission.release() + await lease.release() + + +async def main(): + plugin_parent = Path(sys.argv[1]) + workspace = Path(sys.argv[2]) + manager = PluginManager( + plugin_dirs=[plugin_parent], + event_bus=EventBus(), + tool_registry=None, + workspace=workspace, + installed_cache_root=workspace.parent / "cache", + ) + manager.bind_activity_host(ActivityHost((ProactiveActivityAdapter(),))) + original_lookup = ProactiveDomainEffects._lookup_committed + lookup_calls = 0 + + async def crash_after_commit(self): + nonlocal lookup_calls + lookup_calls += 1 + if lookup_calls == 2: + os._exit(137) + return await original_lookup(self) + + ProactiveDomainEffects._lookup_committed = crash_after_commit + await manager.load_all() + frame = ProactiveFrame( + input=ProactiveTickInput( + session_key="proactive:emotion-crash", + started_at=datetime(2026, 8, 17, 0, 2, tzinfo=timezone.utc), + ) + ) + await run_tick(manager, frame) + + +asyncio.run(main()) +""" + env = dict(os.environ) + core_root = os.environ.get("AKASHIC_AGENT_ROOT") or str( + Path(__file__).parents[3] / "akasic-agent" + ) + env["PYTHONPATH"] = core_root + os.pathsep + env.get("PYTHONPATH", "") + crashed = subprocess.run( + [ + sys.executable, + "-c", + script, + str(plugin_dir.parent), + str(workspace), + ], + env=env, + check=False, + ) + assert crashed.returncode == 137 + + db = module.open_db(workspace / "emotion" / "emotion.db") + try: + assert db.execute("SELECT COUNT(*) FROM emotion_effects").fetchone()[0] == 1 + assert ( + db.execute("SELECT COUNT(*) FROM emotion_domain_effects").fetchone()[0] + == 1 + ) + finally: + db.close() + + async def reenter() -> None: + manager = await _start_emotion_manager(tmp_path, plugin_dir) + try: + frame = ProactiveFrame( + input=ProactiveTickInput( + session_key="proactive:emotion-crash", + started_at=datetime(2026, 8, 17, 0, 2, tzinfo=timezone.utc), ) ) - plugin._on_feedback_recorded( - ProactiveFeedbackRecorded( - event_id=5, - session_key="mobile:test", - user_message_id="u-neutral", - assistant_message_id="a-neutral", - proactive_message_id="p-neutral", - feedback_type="unscored", - confidence="low", - pua_score=None, - lag_seconds=4, - matched_by="recent_pua", - ) + resumed = await _run_manager_tick(manager, frame) + assert resumed.slots["proactive:prompt:system_bottom:emotion"] + finally: + await manager.terminate_all() + + asyncio.run(reenter()) + db = module.open_db(workspace / "emotion" / "emotion.db") + try: + assert db.execute("SELECT COUNT(*) FROM emotion_effects").fetchone()[0] == 1 + assert ( + db.execute("SELECT COUNT(*) FROM emotion_domain_effects").fetchone()[0] + == 1 ) - bootstrap = plugin.mobile_ui_query( - "emotion.bootstrap", - {"limit": 10}, - session_id=None, - turn_id=None, + finally: + db.close() + + +def test_domain_effect_receipt_is_atomic_idempotent_and_durable(tmp_path: Path) -> None: + db_path = tmp_path / "emotion" / "emotion.db" + conn = module.open_db(db_path) + try: + digest = hashlib.sha256(b"merged-documents").hexdigest() + committed = module.commit_domain_effect( + conn, + semantic_job_id="emotion:merge_proactive_pending", + event_id="drift-1", + invocation_id="invocation-1", + effect_id="emotion.state", + idempotency_key="emotion:merge_proactive_pending:event:drift-1", + attempt=1, + result_digest=digest, + ) + repeated = module.commit_domain_effect( + conn, + semantic_job_id="emotion:merge_proactive_pending", + event_id="drift-1", + invocation_id="invocation-1", + effect_id="emotion.state", + idempotency_key="emotion:merge_proactive_pending:event:drift-1", + attempt=1, + result_digest=digest, ) - overview = bootstrap["overview"] - history = {"items": bootstrap["items"]} finally: - await plugin.terminate() + conn.close() - assert overview["effect_count"] == 10 - assert overview["event_count"] == 5 - assert overview["influence_count"] == 4 - assert overview["last_effect"]["expected_effect"] == "tone_only" - assert overview["current_behavior"]["expected_effect"] == "lower_send_bar" - assert len(history["items"]) == 4 - assert {item["source_type"] for item in history["items"]} == {"explicit_quote"} + restarted = module.open_db(db_path) + try: + found = module.lookup_domain_effect( + restarted, + invocation_id="invocation-1", + effect_id="emotion.state", + idempotency_key="emotion:merge_proactive_pending:event:drift-1", + ) + rows = restarted.execute( + "SELECT COUNT(*) FROM emotion_domain_effects" + ).fetchone() + with pytest.raises(RuntimeError, match="identity 漂移"): + module.commit_domain_effect( + restarted, + semantic_job_id="emotion:merge_proactive_pending", + event_id="drift-1", + invocation_id="invocation-2", + effect_id="emotion.state", + idempotency_key="emotion:merge_proactive_pending:event:drift-1", + attempt=1, + result_digest=digest, + ) + finally: + restarted.close() + assert committed == repeated == found + assert rows is not None and int(rows[0]) == 1 -@pytest.mark.asyncio -async def test_mobile_projection_rejects_invalid_limit(tmp_path: Path) -> None: - plugin = EmotionPlugin() - plugin.context = _plugin_context(tmp_path) - - with pytest.raises(ValueError, match="limit 必须"): - plugin.mobile_ui_query( - "emotion.bootstrap", - {"limit": True}, - session_id=None, - turn_id=None, + +def test_domain_effect_receipt_survives_core_process_crash_and_restart( + tmp_path: Path, +) -> None: + db_path = tmp_path / "emotion" / "emotion.db" + plugin_path = Path(__file__).parents[1] / "plugin.py" + script = """ +import importlib.util +import os +import sys +import types +from pathlib import Path + +path = Path(sys.argv[1]) +package_name = "emotion_crash_test" +package = types.ModuleType(package_name) +package.__path__ = [str(path.parent)] +sys.modules[package_name] = package +spec = importlib.util.spec_from_file_location( + package_name + ".plugin", + path, + submodule_search_locations=[str(path.parent)], +) +assert spec is not None and spec.loader is not None +module = importlib.util.module_from_spec(spec) +sys.modules[spec.name] = module +spec.loader.exec_module(module) +conn = module.open_db(Path(sys.argv[2])) +module.commit_domain_effect( + conn, + semantic_job_id="emotion:merge_proactive_pending", + event_id="crash-event", + invocation_id="crash-invocation", + effect_id="emotion.state", + idempotency_key="emotion:merge_proactive_pending:event:crash-event", + attempt=1, + result_digest="crash-digest", +) +conn.close() +os._exit(137) + """ + env = dict(os.environ) + core_root = os.environ.get("AKASHIC_AGENT_ROOT") or str( + Path(__file__).parents[3] / "akasic-agent" + ) + env["PYTHONPATH"] = core_root + os.pathsep + str(plugin_path.parent) + result = subprocess.run( + [sys.executable, "-c", script, str(plugin_path), str(db_path)], + env=env, + check=False, + ) + assert result.returncode == 137 + restarted = module.open_db(db_path) + try: + found = module.lookup_domain_effect( + restarted, + invocation_id="crash-invocation", + effect_id="emotion.state", + idempotency_key="emotion:merge_proactive_pending:event:crash-event", ) + finally: + restarted.close() + assert found is not None + assert found.result_digest == "crash-digest" + + +@pytest.mark.asyncio +async def test_v3_apply_registers_job_without_opening_emotion_db(tmp_path: Path) -> None: + root, emotion_root = await _mount_runtime(tmp_path) + + _ = await root.mount( + lambda ctx: module.apply(ctx, object()), + name="emotion", + inject=(BACKGROUND_JOBS, PROACTIVE_COMPONENTS, UI_SLOTS), + runtime=PluginRuntime( + plugin_id="emotion", + plugin_dir=Path(__file__).parents[1], + data_dir=tmp_path / "plugin-data", + workspace=emotion_root.parent, + config=None, + workspace_roots=("emotion",), + ), + ) + jobs = root.context.get(BACKGROUND_JOBS) + assert jobs is not None + catalog = _freeze_plugin_background_jobs(jobs, root.instance_token) + binding = catalog.job("emotion:merge_proactive_pending") + assert binding is not None + assert binding.definition.documents_scope == ("emotion",) + assert binding.definition.domain_effect == "emotion.state" + assert not (emotion_root / "emotion.db").exists() + await root.dispose() + + +@pytest.mark.asyncio +async def test_v3_merge_uses_core_ports_and_durable_emotion_receipt( + tmp_path: Path, +) -> None: + emotion_root = tmp_path / "workspace" / "emotion" + emotion_root.mkdir(parents=True) + setattr(module, "_v3_emotion_root", emotion_root) + calls: list[str] = [] + prepared_intent = object() + issued_receipt = object() + + class Documents: + def read_pair(self): + calls.append("read") + return ( + ProactiveDocumentDigests(context=None, pending=None), + ProactiveDocumentPair( + context=b"# Proactive Context\n", + pending=b"- [ ] prefer calm summaries\n", + ), + ) + + async def prepare_pair(self, expected, pair): + calls.append("prepare") + assert expected.pending is None + assert pair["pending"] == b"" + return prepared_intent + + async def commit_after(self, intent, receipt): + calls.append("documents") + assert intent is prepared_intent + assert receipt is issued_receipt + return object() + + class Effects: + async def run(self, effect_id, transaction): + calls.append("effect") + effect_ctx = SimpleNamespace( + semantic_job_id="emotion:merge_proactive_pending", + event_id="drift-v3-1", + invocation_id="invocation-v3-1", + effect_id=effect_id, + idempotency_key="emotion:merge_proactive_pending:event:drift-v3-1", + attempt=1, + ) + await transaction(effect_ctx) + durable = module.lookup_emotion_domain_effect_v3(effect_ctx) + assert durable is not None + return issued_receipt + + class Llm: + async def generate_text(self, **kwargs): + calls.append("llm") + assert "prefer calm summaries" in kwargs["prompt"] + return "# Proactive Context\n\n- Prefer calm summaries." + + event = DriftFinished( + event_id="drift-v3-1", + session_key="session", + skill_name="feedback-preference-context", + status="completed", + briefing="briefing", + message_result="ok", + timestamp=datetime.now(timezone.utc), + ) + ctx = SimpleNamespace( + event=event, + documents=Documents(), + domain_effects=Effects(), + llm=Llm(), + ) + + await module.merge_proactive_pending_v3(ctx) + + assert calls == ["read", "llm", "prepare", "effect", "documents"] + db = module.open_db(emotion_root / "emotion.db") + try: + assert db.execute("SELECT COUNT(*) FROM emotion_domain_effects").fetchone()[0] == 1 + finally: + db.close()