wuyang
416 lines
13 KiB
Python
416 lines
13 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from app.config import UserSeed
|
|
from app.curator import (
|
|
CURATOR_INSTRUCTIONS,
|
|
_rank_ideas,
|
|
build_curator_prompt,
|
|
)
|
|
from app.db import Database
|
|
from app.schemas import CuratorDecision
|
|
|
|
|
|
ADMIN = UserSeed(
|
|
id="admin-user",
|
|
label="管理员",
|
|
role="admin",
|
|
access_key_hash="hash-admin",
|
|
)
|
|
MEMBER = UserSeed(
|
|
id="member-user",
|
|
label="用户二",
|
|
role="member",
|
|
access_key_hash="hash-member",
|
|
)
|
|
|
|
|
|
def database(tmp_path: Path) -> Database:
|
|
db = Database(tmp_path / "test.sqlite3")
|
|
db.initialize([ADMIN, MEMBER])
|
|
return db
|
|
|
|
|
|
def assessment(
|
|
idea_id: str | None = None,
|
|
maturity: float = 38,
|
|
supporting_fragment_ids: list[str] | None = None,
|
|
):
|
|
return {
|
|
"idea_id": idea_id,
|
|
"title": "无分类的笔记入口",
|
|
"summary": "先保留念头，再让结构在后台形成。",
|
|
"maturity": maturity,
|
|
"confidence": 0.76,
|
|
"motion": "从抱怨聚成原则",
|
|
"position": "已经有清楚的交互原则，还没有碰到真实使用。",
|
|
"tension": "自由输入与系统内部结构之间需要同时成立。",
|
|
"trajectory": "从输入摩擦的感受，收束成了产品边界。",
|
|
"possible_moves": ["做一个只保留原文的输入原型"],
|
|
"relevance": 1,
|
|
"supporting_fragment_ids": supporting_fragment_ids or [],
|
|
}
|
|
|
|
|
|
def test_fragment_is_saved_before_analysis(tmp_path: Path):
|
|
db = database(tmp_path)
|
|
|
|
fragment = db.create_fragment(ADMIN.id, "一个还没有分类的念头")
|
|
|
|
assert db.list_fragments(ADMIN.id) == [
|
|
{
|
|
"id": fragment["id"],
|
|
"content": "一个还没有分类的念头",
|
|
"created_at": fragment["created_at"],
|
|
}
|
|
]
|
|
assert db.get_fragment(fragment["id"], ADMIN.id)["analysis_status"] == "pending"
|
|
|
|
|
|
def test_user_data_is_isolated_at_query_layer(tmp_path: Path):
|
|
db = database(tmp_path)
|
|
admin_fragment = db.create_fragment(ADMIN.id, "管理员的私密念头")
|
|
member_fragment = db.create_fragment(MEMBER.id, "用户二的私密念头")
|
|
db.apply_assessments(ADMIN.id, admin_fragment["id"], [assessment()])
|
|
admin_idea = db.list_ideas(ADMIN.id)[0]
|
|
|
|
assert [item["content"] for item in db.list_fragments(ADMIN.id)] == [
|
|
"管理员的私密念头"
|
|
]
|
|
assert [item["content"] for item in db.list_fragments(MEMBER.id)] == [
|
|
"用户二的私密念头"
|
|
]
|
|
assert db.get_fragment(member_fragment["id"], ADMIN.id) is None
|
|
assert db.get_idea(MEMBER.id, admin_idea["id"]) is None
|
|
assert db.update_idea_override(MEMBER.id, admin_idea["id"], 80) is None
|
|
assert db.curator_context(MEMBER.id)[0] == []
|
|
|
|
|
|
def test_apply_assessment_builds_trajectory(tmp_path: Path):
|
|
db = database(tmp_path)
|
|
first = db.create_fragment(ADMIN.id, "笔记不应该要求先分类")
|
|
db.apply_assessments(ADMIN.id, first["id"], [assessment()])
|
|
ideas = db.list_ideas(ADMIN.id)
|
|
|
|
assert len(ideas) == 1
|
|
assert ideas[0]["maturity"] == 38
|
|
assert ideas[0]["fragment_count"] == 1
|
|
|
|
second = db.create_fragment(ADMIN.id, "AI 的判断不能出现在记录流里")
|
|
db.apply_assessments(
|
|
ADMIN.id,
|
|
second["id"],
|
|
[assessment(ideas[0]["id"], maturity=46)],
|
|
)
|
|
idea = db.get_idea(ADMIN.id, ideas[0]["id"])
|
|
|
|
assert idea["fragment_count"] == 2
|
|
assert len(idea["snapshots"]) == 2
|
|
assert idea["maturity"] == 46
|
|
|
|
|
|
def test_curator_prompt_contains_recent_continuity_and_idea_catalog(
|
|
tmp_path: Path,
|
|
):
|
|
db = database(tmp_path)
|
|
first = db.create_fragment(
|
|
ADMIN.id,
|
|
"我先让 AI 完成综述，接下来会抽取几个感兴趣的地方。",
|
|
)
|
|
db.apply_assessments(ADMIN.id, first["id"], [assessment()])
|
|
current = db.create_fragment(
|
|
ADMIN.id,
|
|
"我识别出了两个新的认知，第一点让我耳目一新。",
|
|
)
|
|
ideas, fragments = db.curator_context(ADMIN.id)
|
|
|
|
prompt = build_curator_prompt(current, ideas, fragments)
|
|
|
|
assert "我先让 AI 完成综述" in prompt
|
|
assert "我识别出了两个新的认知" in prompt
|
|
assert "无分类的笔记入口" in prompt
|
|
assert "不要要求用户显式建立会话" in prompt
|
|
assert "supporting_fragment_ids" in prompt
|
|
assert "是否形成新的未完成方向" in prompt
|
|
|
|
|
|
def test_explicit_user_direction_cannot_be_silently_discarded():
|
|
with pytest.raises(
|
|
ValidationError,
|
|
match="explicit user-declared direction cannot remain standalone",
|
|
):
|
|
CuratorDecision.model_validate(
|
|
{
|
|
"direction_signal": "explicit",
|
|
"standalone": True,
|
|
"reasoning_note": "目标太大，暂时不值得跟踪",
|
|
"assessments": [],
|
|
}
|
|
)
|
|
|
|
assert "有根据的信任" in CURATOR_INSTRUCTIONS
|
|
assert "用户拥有自己话语的严肃程度" in CURATOR_INSTRUCTIONS
|
|
assert "不能成为拒绝创建的门槛" in CURATOR_INSTRUCTIONS
|
|
assert "保持诚实，不做空洞吹捧" in CURATOR_INSTRUCTIONS
|
|
|
|
|
|
def test_later_fragment_can_recover_earlier_standalone_evidence(
|
|
tmp_path: Path,
|
|
):
|
|
db = database(tmp_path)
|
|
earlier = db.create_fragment(
|
|
ADMIN.id, "也许该重新设计小爱的架构，但现在还说不清。"
|
|
)
|
|
db.apply_assessments(ADMIN.id, earlier["id"], [])
|
|
current = db.create_fragment(
|
|
ADMIN.id, "这个新工作确定要做，我先熟悉 harness 5.0。"
|
|
)
|
|
|
|
db.apply_assessments(
|
|
ADMIN.id,
|
|
current["id"],
|
|
[assessment(supporting_fragment_ids=[earlier["id"]])],
|
|
)
|
|
|
|
idea = db.get_idea(ADMIN.id, db.list_ideas(ADMIN.id)[0]["id"])
|
|
assert [item["content"] for item in idea["fragments"]] == [
|
|
earlier["content"],
|
|
current["content"],
|
|
]
|
|
assert idea["snapshots"][-1]["fragment_ids"] == [
|
|
earlier["id"],
|
|
current["id"],
|
|
]
|
|
|
|
|
|
def test_recovered_evidence_cannot_cross_user_boundary(tmp_path: Path):
|
|
db = database(tmp_path)
|
|
member_fragment = db.create_fragment(MEMBER.id, "另一个人的记录")
|
|
current = db.create_fragment(ADMIN.id, "管理员的新方向")
|
|
|
|
try:
|
|
db.apply_assessments(
|
|
ADMIN.id,
|
|
current["id"],
|
|
[
|
|
assessment(
|
|
supporting_fragment_ids=[member_fragment["id"]]
|
|
)
|
|
],
|
|
)
|
|
except ValueError as exc:
|
|
assert str(exc) == "supporting fragment not found for user"
|
|
else:
|
|
raise AssertionError("cross-user supporting fragment was accepted")
|
|
|
|
assert db.list_ideas(ADMIN.id) == []
|
|
|
|
|
|
def test_delayed_reanalysis_keeps_later_input_out_of_prior_context(
|
|
tmp_path: Path,
|
|
):
|
|
db = database(tmp_path)
|
|
previous = db.create_fragment(ADMIN.id, "当时真正的上文")
|
|
target = db.create_fragment(ADMIN.id, "后来需要重新整理的这一条")
|
|
future = db.create_fragment(ADMIN.id, "这句话发生在目标片段之后")
|
|
timeline = (
|
|
(previous, "2026-07-28T10:00:00.000+00:00"),
|
|
(target, "2026-07-28T10:10:00.000+00:00"),
|
|
(future, "2026-07-28T10:20:00.000+00:00"),
|
|
)
|
|
with db.connect() as connection:
|
|
for fragment, created_at in timeline:
|
|
fragment["created_at"] = created_at
|
|
connection.execute(
|
|
"UPDATE fragments SET created_at = ? WHERE id = ?",
|
|
(created_at, fragment["id"]),
|
|
)
|
|
ideas, fragments = db.curator_context(ADMIN.id)
|
|
|
|
prompt = build_curator_prompt(target, ideas, fragments)
|
|
recent_section = prompt.split("<recent_context>", 1)[1].split(
|
|
"</recent_context>", 1
|
|
)[0]
|
|
later_section = prompt.split("<later_context>", 1)[1].split(
|
|
"</later_context>", 1
|
|
)[0]
|
|
|
|
assert previous["content"] in recent_section
|
|
assert future["content"] not in recent_section
|
|
assert future["content"] in later_section
|
|
assert "延迟重整" in prompt
|
|
|
|
|
|
def test_idea_search_falls_back_to_recent_catalog():
|
|
idea = {
|
|
"id": "idea-one",
|
|
"title": "先综述再深入",
|
|
"summary": "一种理解策略",
|
|
"position": "正在实践",
|
|
"tension": "可靠性",
|
|
"trajectory": "从原则走向实践",
|
|
"updated_at": "2026-07-28",
|
|
"recent_fragments": [],
|
|
}
|
|
|
|
candidates, fallback = _rank_ideas(
|
|
[idea], "agent architecture long task reliability"
|
|
)
|
|
|
|
assert fallback is True
|
|
assert candidates == [idea]
|
|
|
|
|
|
def test_idea_versions_preserve_complete_state(tmp_path: Path):
|
|
db = database(tmp_path)
|
|
first = db.create_fragment(ADMIN.id, "第一条原始证据")
|
|
db.apply_assessments(ADMIN.id, first["id"], [assessment(maturity=30)])
|
|
idea = db.list_ideas(ADMIN.id)[0]
|
|
first_version = db.get_idea(ADMIN.id, idea["id"])["snapshots"][-1]
|
|
|
|
assert first_version["title"] == "无分类的笔记入口"
|
|
assert first_version["summary"] == "先保留念头，再让结构在后台形成。"
|
|
assert first_version["confidence"] == 0.76
|
|
assert first_version["change_kind"] == "analysis"
|
|
assert first_version["fragment_ids"] == [first["id"]]
|
|
|
|
db.update_idea_override(ADMIN.id, idea["id"], 61)
|
|
versions = db.get_idea(ADMIN.id, idea["id"])["snapshots"]
|
|
|
|
assert versions[-1]["change_kind"] == "manual_calibration"
|
|
assert versions[-1]["maturity_override"] == 61
|
|
assert versions[-1]["fragment_ids"] == [first["id"]]
|
|
|
|
|
|
def test_manual_position_is_authoritative(tmp_path: Path):
|
|
db = database(tmp_path)
|
|
fragment = db.create_fragment(ADMIN.id, "一个想法")
|
|
db.apply_assessments(ADMIN.id, fragment["id"], [assessment(maturity=30)])
|
|
idea = db.list_ideas(ADMIN.id)[0]
|
|
|
|
overridden = db.update_idea_override(ADMIN.id, idea["id"], 61)
|
|
assert overridden["maturity"] == 61
|
|
assert overridden["maturity_ai"] == 30
|
|
assert overridden["is_overridden"] is True
|
|
|
|
restored = db.update_idea_override(ADMIN.id, idea["id"], None)
|
|
assert restored["maturity"] == 30
|
|
assert restored["is_overridden"] is False
|
|
|
|
|
|
def test_existing_unowned_data_migrates_to_first_admin(tmp_path: Path):
|
|
db = Database(tmp_path / "legacy.sqlite3")
|
|
db.initialize()
|
|
with db.connect() as connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO fragments (
|
|
id, user_id, content, created_at, analysis_status
|
|
) VALUES ('legacy-fragment', NULL, '旧记录', '2026-01-01', 'done')
|
|
"""
|
|
)
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO ideas (
|
|
id, user_id, title, summary, maturity_ai, confidence,
|
|
motion, position, tension, trajectory, possible_moves,
|
|
created_at, updated_at
|
|
) VALUES (
|
|
'legacy-idea', NULL, '旧想法', '摘要', 10, .5,
|
|
'浮现', '位置', '张力', '轨迹', '[]',
|
|
'2026-01-01', '2026-01-01'
|
|
)
|
|
"""
|
|
)
|
|
|
|
db.initialize([ADMIN, MEMBER])
|
|
|
|
assert db.get_fragment("legacy-fragment", ADMIN.id)["content"] == "旧记录"
|
|
assert db.get_idea(ADMIN.id, "legacy-idea")["title"] == "旧想法"
|
|
assert db.get_fragment("legacy-fragment", MEMBER.id) is None
|
|
|
|
|
|
def test_agent_run_records_metrics_and_events(tmp_path: Path):
|
|
db = database(tmp_path)
|
|
fragment = db.create_fragment(ADMIN.id, "可观测的一次分析")
|
|
run_id = db.start_agent_run(
|
|
ADMIN.id,
|
|
fragment["id"],
|
|
"deepseek-v4-pro",
|
|
"prompt.v1",
|
|
"high",
|
|
)
|
|
db.add_agent_event(
|
|
run_id,
|
|
ADMIN.id,
|
|
"tool_search_ideas",
|
|
{"query": "可观测", "result_count": 0},
|
|
duration_ms=3,
|
|
)
|
|
db.finish_agent_run(
|
|
run_id,
|
|
status="success",
|
|
duration_ms=1200,
|
|
attempt_count=1,
|
|
model_rounds=2,
|
|
tool_calls=1,
|
|
search_calls=1,
|
|
inspection_calls=0,
|
|
input_tokens=100,
|
|
output_tokens=50,
|
|
reasoning_tokens=30,
|
|
cached_tokens=20,
|
|
)
|
|
|
|
run = db.get_agent_run(run_id)
|
|
events = db.list_agent_events(run_id)
|
|
overview = db.admin_overview()
|
|
|
|
assert run["status"] == "success"
|
|
assert run["reasoning_tokens"] == 30
|
|
assert [event["event_type"] for event in events] == [
|
|
"run_started",
|
|
"tool_search_ideas",
|
|
]
|
|
assert overview["last_24h"]["success_rate"] == 100
|
|
|
|
|
|
def test_failed_agent_run_can_be_safely_requeued(tmp_path: Path):
|
|
db = database(tmp_path)
|
|
fragment = db.create_fragment(ADMIN.id, "需要重新整理")
|
|
run_id = db.start_agent_run(
|
|
ADMIN.id,
|
|
fragment["id"],
|
|
"deepseek-v4-pro",
|
|
"prompt.v2",
|
|
"high",
|
|
)
|
|
error = RuntimeError("temporary failure")
|
|
db.finish_agent_run(
|
|
run_id,
|
|
status="error",
|
|
duration_ms=100,
|
|
attempt_count=1,
|
|
model_rounds=1,
|
|
tool_calls=0,
|
|
search_calls=0,
|
|
inspection_calls=0,
|
|
input_tokens=10,
|
|
output_tokens=2,
|
|
reasoning_tokens=1,
|
|
cached_tokens=0,
|
|
error=error,
|
|
)
|
|
db.set_fragment_status(fragment["id"], "error", str(error))
|
|
|
|
retry = db.prepare_agent_retry(run_id)
|
|
|
|
assert retry == {
|
|
"user_id": ADMIN.id,
|
|
"fragment_id": fragment["id"],
|
|
}
|
|
assert db.get_fragment(fragment["id"], ADMIN.id)["analysis_status"] == "pending"
|
|
assert db.prepare_agent_retry(run_id) is None
|