Skip to content

feat: expose durable feedback history #8

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kachofugetsu09 merged 2 commits into from
Aug 25, 2026
Merged

feat: expose durable feedback history #8

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7784740
feat: expose durable feedback history
kachofugetsu09 Aug 23, 2026
531eae4
fix: validate feedback history boundary
kachofugetsu09 Aug 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions .github/workflows/plugin-api-v3.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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
40 changes: 20 additions & 20 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,24 +17,24 @@ 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:<row_id>`，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:<cursor>`，`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 或消费者数据库。已知正式旧表
若尚无三个 preview 列，history 会把这三个稳定字段投影为 `null` 后参与 canonical hash；
旧表经既有 `open_db` 逐列 `ALTER` 后形成的 SQLite schema 也属于同一已知 lineage。

非引用评分使用 Core 正式运行时的共享 HTTP resources。嵌入配置从
`AKASHIC_CONFIG` 指向的 Core 配置加载，不从插件 checkout 的当前目录猜测配置。
Expand Down Expand Up @@ -64,8 +64,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 暴露
已持久化反馈。

## 移动端看板

Expand Down
211 changes: 64 additions & 147 deletions db.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@
from dataclasses import dataclass
from pathlib import Path

from .history import accepted_payload_hash


@dataclass(frozen=True)
class FeedbackEvent:
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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(
Expand Down
Loading
Loading