mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-30 11:54:50 +08:00
e0642eafa7
* chore(types): clear AniList endpoint mypy debt * chore(types): clear Douban endpoint mypy debt * chore(types): clear recommendation endpoint mypy debt * chore(types): clear subscription notification mypy debt * chore(types): clear SDK string compatibility mypy debt * chore(types): ratchet v3 host mypy across remaining files * chore(types): reduce v3 host mypy debt and retire ledger * chore(types): reconcile host snapshot with upstream HTTP pool * test(types): cover host paths exposed by mypy cleanup
255 lines
12 KiB
Python
255 lines
12 KiB
Python
"""
|
||
有界稳定记忆：整条修订、原子批次、后台删除提案和磁盘漂移保护。
|
||
"""
|
||
|
||
import json
|
||
from collections.abc import Callable
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from uuid import uuid4
|
||
|
||
from app.agent.learning.files import contained_path, mutation_lock, read_text, write_text
|
||
from app.agent.learning.schema import MemoryInput, MemoryOperation
|
||
from app.agent.learning.threats import first_threat_message
|
||
|
||
ENTRY_DELIMITER = '\n§\n'
|
||
LIMITS = {'memory': 2200, 'user': 1375}
|
||
|
||
|
||
def parse_entries(raw: str) -> list[str]:
|
||
"""
|
||
仅按完整分隔符拆条，正文中单独出现的 § 不会切断原文。
|
||
"""
|
||
return [part.strip() for part in raw.split(ENTRY_DELIMITER) if part.strip()]
|
||
|
||
|
||
def locate(entries: list[str], old_text: str, pinned: str | None = None,
|
||
empty_message: str | None = None) -> int:
|
||
"""
|
||
整条精确匹配优先；不同内容的多处子串命中拒绝猜测，审批匹配原始整条。
|
||
"""
|
||
if pinned is not None:
|
||
if pinned not in entries:
|
||
raise ValueError('待确认的原记忆已发生变化，提案保留，请重新核对')
|
||
return entries.index(pinned)
|
||
if not old_text.strip():
|
||
raise ValueError('replace/remove 需要 old_text，且 replace 会替换整条记忆')
|
||
indexes = [i for i, entry in enumerate(entries) if entry == old_text]
|
||
indexes = indexes or [i for i, entry in enumerate(entries) if old_text in entry]
|
||
if not entries and empty_message:
|
||
raise ValueError(empty_message)
|
||
if not indexes or len({entries[i] for i in indexes}) > 1:
|
||
raise ValueError('old_text 未唯一匹配，请重新读取记忆并使用准确片段')
|
||
return indexes[0]
|
||
|
||
|
||
def apply_operations(entries: list[str], operations: list[MemoryOperation],
|
||
pinned: list[str | None] | None = None,
|
||
empty_message: str | None = None) -> tuple[list[str], list[str | None]]:
|
||
"""
|
||
在副本上顺序验证所有操作，返回新列表与每次删除或替换的完整旧条目。
|
||
"""
|
||
working = list(entries)
|
||
previous: list[str | None] = []
|
||
for i, operation in enumerate(operations):
|
||
content = (operation.content if operation.content is not None else operation.new_text or '').strip()
|
||
if operation.action != 'remove':
|
||
if not content:
|
||
raise ValueError('add/replace 需要非空 content；删除请使用 remove')
|
||
if threat := first_threat_message(content, scope='strict'):
|
||
raise ValueError(threat)
|
||
if operation.action == 'add':
|
||
if content not in working:
|
||
working.append(content)
|
||
previous.append(None)
|
||
else:
|
||
index = locate(working, (operation.old_text or '').strip(), pinned[i] if pinned else None,
|
||
empty_message)
|
||
previous.append(working[index])
|
||
working[index:index + 1] = [content] if operation.action == 'replace' else []
|
||
return working, previous
|
||
|
||
|
||
class MemoryStore:
|
||
"""
|
||
一用户一目录；复盘只有新增直接生效，破坏性操作保留精确原文等待人工决定。
|
||
"""
|
||
|
||
def __init__(self, root: Path, *, background: bool = False,
|
||
cancelled: Callable[[], bool] = lambda: False) -> None:
|
||
"""
|
||
身份、后台标志与取消令牌由宿主绑定，工具参数不能改写。
|
||
"""
|
||
self.root, self.background, self.cancelled = root, background, cancelled
|
||
self.failures = 0
|
||
self._snapshot: dict[str, str] | None = None
|
||
|
||
def reset_turn(self) -> None:
|
||
"""
|
||
只在真实新轮次重置连续失败预算。
|
||
"""
|
||
self.failures = 0
|
||
|
||
def _path(self, target: str) -> Path:
|
||
"""
|
||
只允许两个稳定记忆目标，不将模型字符串当文件路径使用。
|
||
"""
|
||
if target not in LIMITS:
|
||
raise ValueError('记忆目标必须是 memory 或 user')
|
||
return contained_path(self.root, 'USER.md' if target == 'user' else 'MEMORY.md')
|
||
|
||
def _read(self, target: str) -> tuple[str, list[str]]:
|
||
"""
|
||
读取失败必须保留原文件，不能把异常当空列表覆盖。
|
||
"""
|
||
path = self._path(target)
|
||
raw = read_text(path) if path.exists() else ''
|
||
return raw, list(dict.fromkeys(parse_entries(raw)))
|
||
|
||
def snapshot(self) -> dict[str, str]:
|
||
"""
|
||
冻结正文由调用方持有；污染条目只从注入快照屏蔽，磁盘保留供人工清理。
|
||
"""
|
||
if self._snapshot is not None:
|
||
return dict(self._snapshot)
|
||
with mutation_lock(self.root):
|
||
result = {}
|
||
for target in LIMITS:
|
||
_, entries = self._read(target)
|
||
safe = ['[BLOCKED: 记忆条目命中持久化注入模式，请人工核对]' if first_threat_message(entry) else entry
|
||
for entry in entries]
|
||
result[str(self._path(target))] = ENTRY_DELIMITER.join(safe)
|
||
self._snapshot = result
|
||
return dict(result)
|
||
|
||
@staticmethod
|
||
def _operations(request: MemoryInput) -> list[MemoryOperation]:
|
||
"""
|
||
批次字段优先，一次调用中的所有操作必须属于同一目标。
|
||
"""
|
||
if request.operations:
|
||
return request.operations
|
||
return [MemoryOperation.model_validate({
|
||
'action': request.action,
|
||
'content': request.content,
|
||
'old_text': request.old_text,
|
||
'new_text': request.new_text,
|
||
})]
|
||
|
||
def _validate(self, target: str, raw: str, entries: list[str], operations: list[MemoryOperation],
|
||
batch: bool, pinned: list[str | None] | None = None) -> tuple[list[str], list[str | None]]:
|
||
"""
|
||
按最终状态校验预算；整批清空与不可往返的外部编辑均不得默默丢失。
|
||
"""
|
||
if any(op.action != 'add' for op in operations):
|
||
parsed = parse_entries(raw)
|
||
if raw.strip() and (raw.strip() != ENTRY_DELIMITER.join(parsed) or max(map(len, parsed), default=0) > LIMITS[target]):
|
||
backup = self._path(target).with_suffix(f'.md.bak.{uuid4().hex}')
|
||
write_text(backup, raw)
|
||
raise ValueError(f'磁盘内容无法按记忆条目无损往返；原文备份到 {backup.name}，请先人工整理')
|
||
file_name = 'USER.md' if target == 'user' else 'MEMORY.md'
|
||
empty_message = (
|
||
f"目标记忆为空或文件不存在：target='{target}' 对应当前用户目录的 {file_name}；"
|
||
"跨任务偏好请使用 target='user'，环境事实请使用 target='memory'。"
|
||
)
|
||
working, previous = apply_operations(entries, operations, pinned, empty_message)
|
||
if batch and entries and not working:
|
||
raise ValueError('批次不能清空全部记忆；需要明确清除时使用单次 remove')
|
||
if len(ENTRY_DELIMITER.join(working)) > LIMITS[target]:
|
||
raise ValueError(f'记忆超出 {LIMITS[target]} 字符预算；在同一批次合并或缩短条目后再试')
|
||
return working, previous
|
||
|
||
def _pending_path(self, identifier: str) -> Path:
|
||
"""
|
||
提案 ID 必须是宿主生成的 UUID，不能用于路径穿越。
|
||
"""
|
||
if len(identifier) != 32 or any(char not in '0123456789abcdef' for char in identifier):
|
||
raise ValueError('无效的记忆提案 ID')
|
||
return contained_path(self.root, f'.pending/{identifier}.json')
|
||
|
||
def _stage(self, request: MemoryInput, operations: list[MemoryOperation], previous: list[str | None]) -> dict[str, Any]:
|
||
"""
|
||
保存当前整条原文以防审批时子串误命中新内容；失败不会退化成直接删除。
|
||
"""
|
||
directory = contained_path(self.root, '.pending')
|
||
if directory.exists() and sum(1 for _ in directory.glob('*.json')) >= 100:
|
||
raise ValueError('待处理记忆提案已达100项，请先处理旧提案')
|
||
identifier = uuid4().hex
|
||
payload = dict(target=request.target, operations=[op.model_dump() for op in operations],
|
||
previous=previous, batch=bool(request.operations))
|
||
if self.cancelled():
|
||
raise RuntimeError('后台学习已取消')
|
||
serialized = json.dumps(payload, ensure_ascii=False)
|
||
if len(serialized.encode('utf-8')) > 64 * 1024:
|
||
raise ValueError('提案超过64KiB，请拆成更小的可审阅变更')
|
||
write_text(self._pending_path(identifier), serialized)
|
||
return dict(success=True, staged=True, proposal_staged=True, pending_id=identifier,
|
||
message=f'后台不会自动删除已有记忆。用 /memory pending 查看，/memory approve {identifier} 应用，或 /memory discard {identifier} 丢弃。')
|
||
|
||
def pending(self, identifier: str | None = None) -> dict[str, Any]:
|
||
"""
|
||
枚举至多100条简短索引，指定 ID 才读取完整提案，避免反复装入全部正文。
|
||
"""
|
||
if identifier:
|
||
return dict(success=True, proposal=json.loads(read_text(self._pending_path(identifier))))
|
||
directory = contained_path(self.root, '.pending')
|
||
proposals = []
|
||
for path in sorted(directory.glob('*.json'))[:100]:
|
||
payload = json.loads(read_text(self._pending_path(path.stem)))
|
||
proposals.append(dict(id=path.stem, target=payload['target'], operations=len(payload['operations']),
|
||
preview=str(payload['operations'][0].get('content') or payload['operations'][0].get('old_text') or '')[:120]))
|
||
return dict(success=True, proposals=proposals, detail_command='/memory pending ID')
|
||
|
||
def manage(self, request: MemoryInput) -> dict[str, Any]:
|
||
"""
|
||
连续四次失败返回终止回执，成功写入不回显全库以免诱发无限整理。
|
||
"""
|
||
if request.action == 'pending':
|
||
return self.pending()
|
||
with mutation_lock(self.root):
|
||
try:
|
||
return self._manage(request)
|
||
except ValueError as error:
|
||
self.failures += 1
|
||
if self.failures > 3:
|
||
return dict(success=False, done=True, error='本轮记忆整理已连续失败四次，请停止重试并继续回复用户')
|
||
_, entries = self._read(request.target)
|
||
return dict(success=False, error=str(error), **({} if request.operations else {'current_entries': entries}))
|
||
|
||
def _manage(self, request: MemoryInput) -> dict[str, Any]:
|
||
"""
|
||
锁内重新读盘并一次提交，后台破坏性批次整体保存为提案。
|
||
"""
|
||
operations = self._operations(request)
|
||
raw, entries = self._read(request.target)
|
||
working, previous = self._validate(request.target, raw, entries, operations, bool(request.operations))
|
||
if self.background and any(op.action != 'add' for op in operations):
|
||
return self._stage(request, operations, previous)
|
||
if self.cancelled():
|
||
raise RuntimeError('后台学习已取消')
|
||
changed = working != entries
|
||
if changed:
|
||
write_text(self._path(request.target), ENTRY_DELIMITER.join(working))
|
||
self.failures = 0
|
||
return dict(success=True, done=True, changed=changed, target=request.target, entry_count=len(working),
|
||
usage=f'{len(ENTRY_DELIMITER.join(working))}/{LIMITS[request.target]}', previous_entries=previous,
|
||
note='写入已完成，请勿重复本次操作')
|
||
|
||
def resolve(self, identifier: str, *, approve: bool) -> dict[str, Any]:
|
||
"""
|
||
仅供宿主显式用户命令调用；模型工具不暴露此入口。
|
||
"""
|
||
with mutation_lock(self.root):
|
||
path = self._pending_path(identifier)
|
||
payload = json.loads(read_text(path))
|
||
if approve:
|
||
operations = [MemoryOperation(**op) for op in payload['operations']]
|
||
raw, entries = self._read(payload['target'])
|
||
working, _ = self._validate(payload['target'], raw, entries, operations, payload['batch'], payload['previous'])
|
||
write_text(self._path(payload['target']), ENTRY_DELIMITER.join(working))
|
||
try:
|
||
path.unlink()
|
||
except OSError:
|
||
return dict(success=True, approved=approve, pending_id=identifier, cleanup_pending=True)
|
||
return dict(success=True, approved=approve, pending_id=identifier)
|