From 634cfe1de01fb102a15c30292ef32d0c9215dfcf Mon Sep 17 00:00:00 2001
From: AmriaLin <325560502+AmriaLin@users.noreply.github.com>
Date: Mon, 19 Oct 2026 09:58:30 +0000
Subject: [PATCH 1/6] =?UTF-8?q?feat(dmn):=20=E7=A9=BA=E9=97=B2=E7=AA=97?=
=?UTF-8?q?=E5=8F=A3=E8=B0=83=E5=BA=A6=E4=B8=8E=20prompt=20=E9=9D=99?=
=?UTF-8?q?=E6=80=81=E5=8C=96=20(0.8.0)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
调度层
- 新增 schedule.py：7×12 热度图（2h 槽），每条用户消息加权并随时间衰减
- 细/中/粗三层（单片星期几、工作日-周末、全局）按有效样本降级，
细层向全局做 n/(n+k) 收缩，避免稀疏样本失真
- 冷槽双条件判定（占比 + 原始计数），环形扫描支持跨午夜冷段
- 段内回到原始时间戳：噪声合并后二分求可用空闲上界，扣 buffer 得窗口
- penalty 语义改为"要求的冷段长度"，不足时减半放宽三次，
最后退到热度最低段，彻底消除静默停摆
- 用户全活跃时不返回空，冷启动走固定 target_hour
- 状态持久化热度图、last_run_ts、last_iter_summary；重排带一分钟节流
prompt 层
- system 剥离全部动态信息：total_runs、精确时间、上轮摘要、
待审建议计数、记忆库膨胀告警
- 日期降到日粒度，同日内 system 前缀逐字节一致，可命中前缀缓存
- 新增 subconscious_get_context 聚合工具，单项失败标 stale 不中断
- 规则中强制每轮先调用它，未调用视为本轮无效
- 旧 penalty 升级时清零一次（语义不可换算）
其他
- 新增 18 个 idle_* 配置项，use_idle_window_planner 可控回退旧路径
- tests/test_schedule.py 13 条单测；pyproject 启用 pytest
- 版本 0.7.1 -> 0.8.0
---
amrita_plugin_memory/config.py | 53 ++
amrita_plugin_memory/rethinking/_state.py | 21 +
amrita_plugin_memory/rethinking/consts.py | 39 +-
amrita_plugin_memory/rethinking/hooks.py | 4 +-
amrita_plugin_memory/rethinking/runner.py | 171 +++---
amrita_plugin_memory/rethinking/schedule.py | 573 ++++++++++++++++++++
amrita_plugin_memory/rethinking/schemas.py | 12 +
amrita_plugin_memory/rethinking/tools.py | 92 +++-
pyproject.toml | 4 +-
tests/test_schedule.py | 161 ++++++
10 files changed, 1026 insertions(+), 104 deletions(-)
create mode 100644 amrita_plugin_memory/rethinking/schedule.py
create mode 100644 tests/test_schedule.py
diff --git a/amrita_plugin_memory/config.py b/amrita_plugin_memory/config.py
index ff5f8cb..0a93e30 100644
--- a/amrita_plugin_memory/config.py
+++ b/amrita_plugin_memory/config.py
@@ -42,6 +42,59 @@ class SubconsciousConfig(BaseModel):
rethink_max_delay_minutes: int = Field(
default=1440, ge=60, le=10080, description="惩罚延迟上限（分钟），默认1天"
)
+ use_idle_window_planner: bool = Field(
+ default=True,
+ description="是否启用空闲窗口调度（热度图 + 冷段二分）；关闭后回退到旧的固定延迟方案",
+ )
+ idle_timezone: str = Field(
+ default="Asia/Shanghai", description="热度图与窗口计算使用的时区"
+ )
+ idle_fine_tau_days: float = Field(
+ default=28.0, ge=1.0, description="单片星期几层的衰减常数（天）"
+ )
+ idle_coarse_tau_days: float = Field(
+ default=14.0, ge=1.0, description="全局层的衰减常数（天）"
+ )
+ idle_shrink_k: float = Field(
+ default=4.0, ge=0.0, description="细层向全局收缩的强度，越大越保守"
+ )
+ idle_min_sample_days_fine: int = Field(
+ default=6, ge=1, description="启用单片星期几层所需的最少活跃天数"
+ )
+ idle_min_sample_days_mid: int = Field(
+ default=3, ge=1, description="启用工作日/周末层所需的最少活跃天数"
+ )
+ idle_cold_pct_max: float = Field(
+ default=0.05, ge=0.0, le=1.0, description="冷槽的占比上限"
+ )
+ idle_cold_abs_max: float = Field(
+ default=1.0, ge=0.0, description="冷槽的原始计数上限，避免稀疏样本全部判冷"
+ )
+ idle_target_hour: int = Field(
+ default=3, ge=0, le=23, description="偏好触发时刻（本地时区小时）"
+ )
+ idle_base_slots: int = Field(
+ default=3, ge=1, description="首次计划要求的最短冷段（槽，1 槽=2h）"
+ )
+ idle_max_slots: int = Field(default=6, ge=1, description="冷段要求的封顶槽数")
+ idle_penalty_multiplier: float = Field(
+ default=1.5, ge=1.0, le=10.0, description="惩罚对冷段要求的放大倍率"
+ )
+ idle_max_relax_steps: int = Field(
+ default=3, ge=0, le=6, description="冷段不足时允许的减半放宽次数"
+ )
+ idle_buffer_minutes: int = Field(
+ default=15, ge=0, description="窗口起止的安全缓冲（分钟）"
+ )
+ idle_run_estimate_minutes: int = Field(
+ default=10, ge=0, description="单轮 DMN 预计耗时（分钟），用于扣减可用窗口"
+ )
+ idle_fallback_minutes: int = Field(
+ default=120, ge=10, description="冷启动或规划失败时的固定窗口长度（分钟）"
+ )
+ idle_noise_gap_minutes: int = Field(
+ default=15, ge=0, description="连发消息合并为单点的最小间隔（分钟）"
+ )
prompt_file: str = Field(
default="prompt/subconscious_main.md.jinja2",
description="主推理提示词文件名，相对于 config/amrita_plugin_memory/",
diff --git a/amrita_plugin_memory/rethinking/_state.py b/amrita_plugin_memory/rethinking/_state.py
index 75917b1..c44c3ca 100644
--- a/amrita_plugin_memory/rethinking/_state.py
+++ b/amrita_plugin_memory/rethinking/_state.py
@@ -18,6 +18,9 @@
_penalty_count: int = 0
# 知识建议队列 — 对话 LLM 提议创建/更新知识，潜意识 Agent 下一轮审查
_knowledge_suggestions: list[dict[str, str]] = []
+# 上一轮 iter_stop 的摘要与结束时间 — 供 subconscious_get_context 读取
+_last_iter_summary: str = ""
+_last_run_at: float = 0.0
# 隔离的工具管理器 — 所有 @on_tools 装饰器通过 bound_to 注册到这里，不污染全局 ToolsManager
_SUBCONSCIOUS_TOOLS = MultiToolsManager()
@@ -89,3 +92,21 @@ def set_knowledge_suggestions(suggestions: list[dict[str, str]]) -> None:
def add_knowledge_suggestion(suggestion: dict[str, str]) -> None:
global _knowledge_suggestions
_knowledge_suggestions.append(suggestion)
+
+
+def get_last_iter_summary() -> str:
+ return _last_iter_summary
+
+
+def set_last_iter_summary(text: str) -> None:
+ global _last_iter_summary
+ _last_iter_summary = text
+
+
+def get_last_run_at() -> float:
+ return _last_run_at
+
+
+def set_last_run_at(ts: float) -> None:
+ global _last_run_at
+ _last_run_at = ts
diff --git a/amrita_plugin_memory/rethinking/consts.py b/amrita_plugin_memory/rethinking/consts.py
index 8458bb8..c77c62e 100644
--- a/amrita_plugin_memory/rethinking/consts.py
+++ b/amrita_plugin_memory/rethinking/consts.py
@@ -6,8 +6,9 @@
DEFAULT_SUBCONSCIOUS_PROMPT = """\
{# 潜意识推理循环 - 系统提示词 #}
-{# 变量: last_run, current_time, target_user_id, total_runs, character_prompt #}
-{# 编辑此文件即可自定义提示词，无需改动代码。 #}
+{# 变量: character_prompt, target_user_id, current_date #}
+{# 本模板只放不变量：所有会随循环变化的信息（时间、待办计数、热度统计） #}
+{# 一律通过 subconscious_get_context 按需拉取，不要写回这里。 #}
你是用户的记忆管家（Memory Curator），在后台静默运行。你的职责是整理、优化和组织用户的长期记忆库。
@@ -16,9 +17,8 @@
{{ character_prompt }}
{% endif %}
-当前时间: {{ current_time }}
+日期: {{ current_date }}
用户ID: {{ target_user_id }}
-这是第 {{ total_runs }} 次推理循环。
@@ -27,15 +27,20 @@
1. 你**必须**使用提供的工具来检查和操作用户的记忆库。不要凭空想象或推测记忆内容。
-2. 每次推理循环中，你应该:
+2. 每轮**第一件事**必须调用 subconscious_get_context 获取当前环境
+ （精确时间、距上轮间隔、上次摘要、待审查建议、记忆库规模）。
+ 不调用则视为本轮无效，不要凭上一轮的印象行动。
+3. 之后你应该:
a. 先调用 subconscious_list_memory 或 subconscious_read_memory 了解记忆库当前状态
b. 识别需要处理的问题（重复记忆、矛盾记忆、过期信息、缺失标签等）
c. 使用 subconscious_write_memory / subconscious_update_memory / subconscious_delete_memory 执行整理操作
-3. 当本轮整理工作完成（或没有新发现需要处理），你**必须**调用 subconscious_iter_stop 来结束本次推理循环。
-4. 利用 subconscious_read_sessions 和 subconscious_read_chat_context 了解用户近况。
-5. 利用 subconscious_get_profile / subconscious_update_profile 渐进式构建用户画像。
-6. 不要在单次循环中做太多事情, 保持每次循环简洁高效。
-7. 所有记忆操作都作用于目标用户（user_id={{ target_user_id }}）, scope 固定为 user。
+4. 当本轮整理工作完成（或没有新发现需要处理），你**必须**调用 subconscious_iter_stop 来结束本次推理循环。
+5. 利用 subconscious_read_sessions 和 subconscious_read_chat_context 了解用户近况。
+6. 利用 subconscious_get_profile / subconscious_update_profile 渐进式构建用户画像。
+7. 不要在单次循环中做太多事情, 保持每次循环简洁高效。
+8. 所有记忆操作都作用于目标用户（user_id={{ target_user_id }}）, scope 固定为 user。
+9. 记忆库规模、待审查建议数量这类数字只以 subconscious_get_context 的返回为准，
+ 不要假设它们在本轮内保持不变，也不要自己推算轮数或进度。
@@ -46,13 +51,6 @@
- 过时信息: 识别明显过时的信息, 考虑更新或降低重要性
-{% if last_run %}
-
-上一次推理循环的摘要:
-{{ last_run }}
-
-{% endif %}
-
你不是在和用户对话。你只通过工具调用（tool calls）和最终调用 subconscious_iter_stop 来表达你的工作。
在 iter_stop 的 summary 字段中, 请简要描述你本轮做了什么（例如: "检查了5条记忆, 合并了2条重复项, 为3条记忆补充了标签"）。
@@ -214,7 +212,7 @@ async def load_character_prompt() -> str:
| 文件 | 用途 | 可用变量 |
| ----------------------------- | ---------------------------------------------- | ------------------------------------------------------------------------------ |
-| `subconscious_main.md.jinja2` | 主推理循环提示词 | `character_prompt`, `last_run`, `current_time`, `target_user_id`, `total_runs` |
+| `subconscious_main.md.jinja2` | 主推理循环提示词 | `character_prompt`, `current_date`, `target_user_id` |
| `subconscious_send.md.jinja2` | 向用户主动发消息的生成模板 | `character_prompt`, `intent`, `memory_context`, `current_time` |
| `knowledge_guide.md.jinja2` | 全局知识库使用指南（注入到主 prompt 末尾） | 无（纯静态文本） |
| `profile_guide.md.jinja2` | 用户画像渐进式构建指南（注入到主 prompt 末尾） | 无（纯静态文本） |
@@ -233,8 +231,9 @@ async def load_character_prompt() -> str:
## 注意事项
- 不要删除 `{% if character_prompt %}` 块——它用于注入 Bot 的角色设定
-- `last_run` 变量只在非首次循环时有值
-- `current_time` 格式为 `YYYY-MM-DD HH:MM:SS UTC`
+- `current_date` 只到日粒度，用来保持 system 前缀跨轮稳定（可命中缓存）
+- 精确时间、上轮摘要、待办计数、记忆库规模一律不进模板，
+ 由工具 `subconscious_get_context` 按需拉取，避免把过期快照写进 system
- `target_user_id` 是配置中指定的数字 ID 字符串
## 默认行为
diff --git a/amrita_plugin_memory/rethinking/hooks.py b/amrita_plugin_memory/rethinking/hooks.py
index 4986817..265dbd2 100644
--- a/amrita_plugin_memory/rethinking/hooks.py
+++ b/amrita_plugin_memory/rethinking/hooks.py
@@ -1,5 +1,7 @@
"""Hook 注册与生命周期管理。"""
+import time
+
from amrita.plugins.chat.runtime import try_get_amrita_ctx
from amrita_core.hook.event import PreCompletionEvent
from amrita_core.hook.on import on_precompletion
@@ -25,7 +27,7 @@ async def _subconscious_pre_hook(event: PreCompletionEvent) -> None:
return
try:
- await runner.cancel_and_reschedule()
+ await runner.cancel_and_reschedule(chat_ts=time.time())
except Exception as e:
# 调度失败不能影响用户这条消息的处理
logger.opt(exception=e, colors=True, raw=True).error(
diff --git a/amrita_plugin_memory/rethinking/runner.py b/amrita_plugin_memory/rethinking/runner.py
index b9a2af2..ccfa6a0 100644
--- a/amrita_plugin_memory/rethinking/runner.py
+++ b/amrita_plugin_memory/rethinking/runner.py
@@ -5,6 +5,7 @@
import asyncio
import contextlib
import json
+import time
from datetime import datetime, timedelta, timezone
from typing import Any
@@ -31,7 +32,6 @@
from ..config import DATA_PATH, SubconsciousConfig
from ..keys import make_scope_id
from ..models import SubconsciousState
-from ..vector import AsyncUserMemory, get_db_conn
from . import _state
from .backend import SubconsciousBackend
from .consts import (
@@ -43,9 +43,12 @@
load_character_prompt,
)
from .knowledge import KnowledgeBaseManager
+from .schedule import IdlePlanner, describe
from .types import ProfileResult, SessionSummary
_STATE_UID = "amrita_memory"
+#: 重排节流窗口（秒）
+_REPLAN_THROTTLE_SECONDS = 60.0
_LEGACY_REPO_UID = (
"amrita_memory" # 旧版本元状态占用的 Bot Memory 表 uid（仅用于升级迁移）
)
@@ -79,6 +82,11 @@ def __init__(self, config: SubconsciousConfig) -> None:
# session 摘要缓存（LRU）：session DB id -> 摘要文本，最多 128 条
self._session_cache: LRUCache[int, str] = LRUCache(128)
self._profile_path = DATA_PATH / "user_profile.md"
+ # 空闲窗口规划器（热度图 + 冷段二分）
+ self._planner = IdlePlanner(config)
+ # 重排节流：一分钟内重复触发只更新惩罚与热度，不重复规划
+ self._last_replan_at: float = 0.0
+ self._replan_throttle: float = _REPLAN_THROTTLE_SECONDS
@property
def is_running(self) -> bool:
@@ -123,26 +131,47 @@ def stop(self) -> None:
if self._manual_task is not None and not self._manual_task.done():
self._manual_task.cancel()
- async def cancel_and_reschedule(self) -> None:
- """取消当前计划，递增惩罚计数器，重新计算延迟并调度一次运行。
+ async def cancel_and_reschedule(self, chat_ts: float | None = None) -> None:
+ """取消当前计划并重新规划下一个空闲窗口。
- 用户每次发消息时调用此方法。不 dump/恢复上下文——
+ 用户每次发消息时调用：先给热度图记一次观测，再递增惩罚，
+ 然后按新的冷段门槛重新选窗口。不 dump/恢复上下文——
用户聊完天后有新内容，旧推理上下文应直接丢弃。
"""
+ # 热度图观测：每条消息都算，加深该时间窗口的热点
with contextlib.suppress(Exception):
- scheduler.remove_job(self._job_id)
+ self._planner.observe(chat_ts)
penalty = _state.increment_penalty()
- base = self._config.rethink_base_delay_minutes
- multiplier = self._config.rethink_penalty_multiplier
- cap = self._config.rethink_max_delay_minutes
- raw = base * (multiplier ** (penalty - 1))
- delay_minutes = min(raw, cap)
- logger.info(
- f"[Subconscious] Rescheduled: penalty=#{penalty}, "
- f"delay={delay_minutes:.0f}min (raw={raw:.0f}min, cap={cap}min)"
- )
- self._schedule_once(int(delay_minutes * 60))
- # 惩罚计数落盘，避免重启后丢掉已累积的退避
+ now_mono = time.monotonic()
+ if now_mono - self._last_replan_at < self._replan_throttle:
+ # 同窗口内的后续消息不重排，避免每条消息都扫一遍热度图
+ with contextlib.suppress(Exception):
+ await self._save_state()
+ return
+ self._last_replan_at = now_mono
+ with contextlib.suppress(Exception):
+ scheduler.remove_job(self._job_id)
+ if self._config.use_idle_window_planner:
+ plan = self._planner.plan(penalty)
+ logger.info(
+ f"[Subconscious] Rescheduled: penalty=#{penalty}, {describe(plan)}"
+ )
+ delay_seconds = int(
+ max(60.0, (plan.start - datetime.now(timezone.utc)).total_seconds())
+ )
+ else:
+ base = self._config.rethink_base_delay_minutes
+ multiplier = self._config.rethink_penalty_multiplier
+ cap = self._config.rethink_max_delay_minutes
+ raw = base * (multiplier ** (penalty - 1))
+ delay_minutes = min(raw, cap)
+ logger.info(
+ f"[Subconscious] Rescheduled(legacy): penalty=#{penalty}, "
+ f"delay={delay_minutes:.0f}min (raw={raw:.0f}min, cap={cap}min)"
+ )
+ delay_seconds = int(delay_minutes * 60)
+ self._schedule_once(delay_seconds)
+ # 惩罚计数与热度图落盘，避免重启后丢掉已累积的退避
with contextlib.suppress(Exception):
await self._save_state()
@@ -173,6 +202,9 @@ async def _run(self) -> None:
return
self._is_running = True
try:
+ # 窗口已实际占用：记一次观测，避免下一轮重复选同一段
+ with contextlib.suppress(Exception):
+ self._planner.note_window_taken(time.time())
await self._run_cycle()
finally:
self._is_running = False
@@ -187,9 +219,9 @@ async def _run_cycle(self) -> None:
return
prompt_text = await self._load_prompt()
user_msg_text = (
- f"现在是第 {self._total_runs + 1} 次潜意识推理循环。"
- f"请检查用户 (ID: {self._config.target_user_id}) 的记忆库，执行必要的整理操作。"
- f"完成后务必调用 subconscious_iter_stop。"
+ f"请先调用 subconscious_get_context 了解当前环境，"
+ f"再检查用户 (ID: {self._config.target_user_id}) 的记忆库，"
+ f"执行必要的整理操作。完成后务必调用 subconscious_iter_stop。"
)
train = Message(role="system", content=prompt_text)
user_input = user_msg_text
@@ -210,7 +242,7 @@ async def _run_cycle(self) -> None:
self._chat_obj = chat_obj
try:
- logger.info(f"[Subconscious] Cycle #{self._total_runs + 1}")
+ logger.info("[Subconscious] Cycle start")
chat_obj.begin()
await chat_obj
except Exception as e:
@@ -245,6 +277,7 @@ async def _post_process(self, chat_obj: ChatObject) -> None:
f"[Subconscious] Abstract saved ({len(self._last_abstracts)}/{self._config.max_abstracts})"
)
self._total_runs += 1
+ _state.set_last_run_at(datetime.now(timezone.utc).timestamp())
# 先持久化，任一失败则不重置惩罚（保留重试机会）
save_ok = True
@@ -368,6 +401,17 @@ async def _load_state(self) -> None:
suggestions = payload.get("knowledge_suggestions", [])
if isinstance(suggestions, list):
_state.set_knowledge_suggestions(suggestions)
+ _state.set_last_iter_summary(str(payload.get("last_iter_summary", "")))
+ _state.set_last_run_at(float(payload.get("last_run_ts", 0.0) or 0.0))
+ self._planner.load(payload)
+ # 旧 penalty 语义是"被推迟的次数"，与新语义（要求的冷段长度）不可换算，
+ # 升级后清零一次，避免带着旧值把门槛顶到上限
+ if "idle_planner" not in payload and _state.get_penalty_count() > 0:
+ logger.info(
+ "[Subconscious] Migrate: reset legacy penalty "
+ f"#{_state.get_penalty_count()} (old delay-based semantics)"
+ )
+ _state.reset_penalty()
except Exception as e:
logger.warning(f"[Subconscious] Load state failed: {e}")
@@ -404,19 +448,27 @@ async def _migrate_legacy_state(self) -> dict[str, Any]:
async def _save_state(self) -> None:
"""保存元状态到 SubconsciousState 表。"""
- await self._write_state_payload(
- _STATE_UID,
- {
- "total_runs": self._total_runs,
- "last_run_time": datetime.now(timezone.utc).isoformat(),
- "penalty": _state.get_penalty_count(),
- "last_abstracts": self._last_abstracts,
- "pending_messages": _state.get_pending(),
- "knowledge_suggestions": _state.get_knowledge_suggestions(),
- },
- )
+ payload: dict[str, Any] = {
+ "total_runs": self._total_runs,
+ "last_run_time": datetime.now(timezone.utc).isoformat(),
+ "last_run_ts": _state.get_last_run_at(),
+ "last_iter_summary": _state.get_last_iter_summary(),
+ "penalty": _state.get_penalty_count(),
+ "last_abstracts": self._last_abstracts,
+ "pending_messages": _state.get_pending(),
+ "knowledge_suggestions": _state.get_knowledge_suggestions(),
+ }
+ payload.update(self._planner.dump())
+ await self._write_state_payload(_STATE_UID, payload)
async def _load_prompt(self) -> str:
+ """渲染 system prompt。
+
+ 只允许不变量进入 system：角色、规则、质量口径、输出格式，
+ 以及静态的画像/知识库指南。精确时间、上轮摘要、待办计数、
+ 记忆库规模等动态信息一律交给 subconscious_get_context 按需拉取，
+ 这样跨轮的 system 前缀逐字节一致，能吃到 provider 的前缀缓存。
+ """
main_path = (self._prompt_dir / self._config.prompt_file).resolve()
kn_path = (self._prompt_dir / self._config.prompt_knowledge_file).resolve()
pr_path = (self._prompt_dir / self._config.prompt_profile_file).resolve()
@@ -427,62 +479,25 @@ async def _load_prompt(self) -> str:
character_prompt = await load_character_prompt()
template: Template = Template(main_path.read_text(encoding="utf-8"))
- now = datetime.now(timezone.utc)
+ # 日粒度日期：每天才变一次，不影响缓存命中，又能给记忆新旧提供锚点
+ current_date = datetime.now(timezone.utc).strftime("%Y-%m-%d")
prompt = await asyncio.to_thread(
template.render,
- last_abstracts=self._last_abstracts,
- last_run=(self._last_abstracts[-1] if self._last_abstracts else ""),
- current_time=now.strftime("%Y-%m-%d %H:%M:%S UTC"),
+ current_date=current_date,
target_user_id=self._config.target_user_id or "(未配置)",
- total_runs=self._total_runs,
character_prompt=character_prompt,
)
- # 注入知识库 & 画像指南（Jinja2 渲染，支持变量插值）
- common_vars = {
- "current_time": now.strftime("%Y-%m-%d %H:%M:%S UTC"),
+ # 指南文件仍走 Jinja2，但只提供不变量：旧文件里的 {{ current_time }}
+ # 会被渲染成空串，而不是原样泄漏到 system 里
+ static_vars = {
+ "current_date": current_date,
"target_user_id": self._config.target_user_id or "(未配置)",
}
- prompt += "\n\n" + await asyncio.to_thread(
- Template(kn_path.read_text(encoding="utf-8")).render, **common_vars
- )
- prompt += "\n\n" + await asyncio.to_thread(
- Template(pr_path.read_text(encoding="utf-8")).render, **common_vars
- )
-
- # 待审查的知识建议 — 注入提示让 Agent 主动拉取
- suggestions = _state.get_knowledge_suggestions()
- if suggestions:
- prompt += (
- f"\n\n📋 有 {len(suggestions)} 条来自对话 LLM 的知识建议等待审查。"
- f"请调用 subconscious_read_suggestions 获取详情，"
- f"审查后对值得保留的用 subconscious_knowledge_create/update 实际写入。"
+ for guide in (kn_path, pr_path):
+ prompt += "\n\n" + await asyncio.to_thread(
+ Template(guide.read_text(encoding="utf-8")).render, **static_vars
)
-
- # 膨胀感知：查 ChromaDB 总量，超阈值注入警告
- try:
- pid = make_scope_id(self._config.target_user_id, is_group=False)
- ope = AsyncUserMemory(get_db_conn())
- await ope.init()
- # 先取轻量计数，超阈值才拉全量算分布
- total = await ope.count_user_notes(pid)
- if total > self._config.memory_warn_threshold:
- result = await ope.get_all_notes(pid, include=["metadatas"])
- low_count = sum(
- 1
- for m in (result.get("metadatas") or [])
- if isinstance(m, dict) and m.get("importance") == "low"
- )
- prompt += (
- f"\n\n⚠️ 当前记忆库有 {total} 条记忆"
- f"（其中低重要性 {low_count} 条）。"
- f"建议本轮优先调用 subconscious_get_memory_stats 分析分布，"
- f"然后用 subconscious_duplicate_helper 获取待整理列表，"
- f"进行合并去重和低质记忆清理。"
- )
- except Exception as e:
- logger.debug(f"[Subconscious] Memory stats query failed: {e}")
-
return prompt
async def _read_recent_sessions(self, n: int = 5) -> list[SessionSummary]:
diff --git a/amrita_plugin_memory/rethinking/schedule.py b/amrita_plugin_memory/rethinking/schedule.py
new file mode 100644
index 0000000..f084105
--- /dev/null
+++ b/amrita_plugin_memory/rethinking/schedule.py
@@ -0,0 +1,573 @@
+"""DMN 空闲窗口调度：7×12 热度图 + 冷段扫描 + 段内二分。
+
+设计要点
+- 观测：用户每条消息让对应 (星期几, 2h 槽) 权重 +1，并随时间衰减。
+- 分层：fine(单片星期几) / mid(工作日-周末) / coarse(全局)，逐级降级。
+- 收缩：fine/mid 向 coarse 按有效样本天数加权，避免稀薄样本失真。
+- 规划：环形找连续冷槽 -> 段内用原始时间戳做分钟级二分 -> 扣 buffer 得可用窗口。
+- 保底：冷段不足门槛时逐次减半放宽，最后强制取最长冷段。
+
+本模块不依赖 nonebot，纯数据 + 纯函数，便于独立单测。
+"""
+
+from __future__ import annotations
+
+import time
+from collections.abc import Sequence
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+from typing import Any
+from zoneinfo import ZoneInfo
+
+SLOT_MINUTES = 120
+SLOTS_PER_DAY = 12
+DAYS = 7
+CELLS = DAYS * SLOTS_PER_DAY
+MINUTES_PER_DAY = SLOT_MINUTES * SLOTS_PER_DAY
+
+#: 最近消息时间戳缓冲上限（分钟级细化用，不能无限涨）
+_MAX_RECENT = 500
+#: 活跃日期登记上限
+_MAX_SEEN_DAYS = 240
+
+_WD_NAMES = ("周一", "周二", "周三", "周四", "周五", "周六", "周日")
+
+
+def _cell(wd: int, slot: int) -> int:
+ return wd * SLOTS_PER_DAY + (slot % SLOTS_PER_DAY)
+
+
+def slot_of(ts: float, tz: ZoneInfo) -> tuple[int, int]:
+ """把 unix 时间戳换成 (星期几 0=周一, 槽位 0..11)。"""
+ dt = datetime.fromtimestamp(ts, tz)
+ wd = dt.weekday()
+ slot = (dt.hour * 60 + dt.minute) // SLOT_MINUTES
+ return wd, slot
+
+
+# ---------------------------------------------------------------- 纯函数工具
+
+
+def merge_noise(ts: Sequence[float], gap_seconds: float) -> list[float]:
+ """把一波连发的消息压成单点：间距小于 gap 的后来者直接并掉。"""
+ out: list[float] = []
+ prev: float | None = None
+ for t in sorted(ts):
+ if prev is not None and t - prev < gap_seconds:
+ continue
+ out.append(t)
+ prev = t
+ return out
+
+
+def free_gaps(ts: Sequence[float], start: float, end: float) -> list[tuple[float, float]]:
+ """返回 [start, end) 内的全部无消息区间。"""
+ if end <= start:
+ return []
+ inner = sorted(t for t in ts if start < t < end)
+ pts = [start, *inner, end]
+ return [(pts[i], pts[i + 1]) for i in range(len(pts) - 1)]
+
+
+def longest_free(
+ ts: Sequence[float], start: float, end: float
+) -> tuple[float, float]:
+ """[start, end) 内最长的无消息区间。"""
+ gaps = free_gaps(ts, start, end)
+ if not gaps:
+ return start, end
+ return max(gaps, key=lambda g: g[1] - g[0])
+
+
+def feasible_minutes(
+ ts: Sequence[float], start: float, end: float, minutes: int
+) -> bool:
+ """是否存在一段连续 >= minutes 的空闲。（随 minutes 单调递减）"""
+ if minutes <= 0:
+ return end > start
+ need = minutes * 60
+ return any(b - a >= need for a, b in free_gaps(ts, start, end))
+
+
+def bisect_free_minutes(
+ ts: Sequence[float], start: float, end: float, hi: int | None = None
+) -> int:
+ """二分求窗口内最长连续空闲（分钟）。判定函数单调，收敛到分钟精度。"""
+ if end <= start:
+ return 0
+ top = hi if hi is not None else int((end - start) // 60)
+ lo = 0
+ while lo < top:
+ mid = (lo + top + 1) // 2
+ if feasible_minutes(ts, start, end, mid):
+ lo = mid
+ else:
+ top = mid - 1
+ return lo
+
+
+def _normalize(values: Sequence[float]) -> list[float]:
+ total = sum(values)
+ if total <= 1e-9:
+ return [0.0] * len(values)
+ return [v / total for v in values]
+
+
+def _runs(mask: Sequence[bool], limit: int) -> list[tuple[int, int]]:
+ """环形找连续 True 段，返回 (起点, 长度)，长度不超过 limit。"""
+ n = len(mask)
+ if n == 0:
+ return []
+ if all(mask):
+ return [(0, limit)]
+ if not any(mask):
+ return []
+ start = mask.index(False)
+ seq = list(mask[start:]) + list(mask[:start])
+ out: list[tuple[int, int]] = []
+ i = 0
+ while i < len(seq):
+ if not seq[i]:
+ i += 1
+ continue
+ j = i
+ while j < len(seq) and seq[j]:
+ j += 1
+ out.append(((start + i) % n, min(j - i, limit)))
+ i = j
+ return out
+
+
+# ---------------------------------------------------------------- 规划结果
+
+
+@dataclass
+class Plan:
+ """一次调度规划的结果。"""
+
+ start: datetime
+ end: datetime
+ length_minutes: int
+ tier: str = "coarse"
+ relaxed_steps: int = 0
+ forced: bool = False
+ penalty: int = 0
+ note: str = ""
+
+ @property
+ def delay_seconds(self) -> int:
+ return max(1, int((self.start - datetime.now(timezone.utc)).total_seconds()))
+
+ def to_dict(self) -> dict[str, Any]:
+ return {
+ "start": self.start.isoformat(),
+ "end": self.end.isoformat(),
+ "length_minutes": self.length_minutes,
+ "tier": self.tier,
+ "relaxed_steps": self.relaxed_steps,
+ "forced": self.forced,
+ "penalty": self.penalty,
+ "note": self.note,
+ }
+
+
+@dataclass
+class PlannerState:
+ """可持久化的热度图状态。"""
+
+ fine: list[float] = field(default_factory=lambda: [0.0] * CELLS)
+ coarse: list[float] = field(default_factory=lambda: [0.0] * SLOTS_PER_DAY)
+ seen_days: list[str] = field(default_factory=list)
+ recent: list[float] = field(default_factory=list)
+ last_update: float = 0.0
+
+ def to_dict(self) -> dict[str, Any]:
+ return {
+ "fine": [round(v, 6) for v in self.fine],
+ "coarse": [round(v, 6) for v in self.coarse],
+ "seen_days": self.seen_days[-_MAX_SEEN_DAYS:],
+ "recent": self.recent[-_MAX_RECENT:],
+ "last_update": self.last_update,
+ }
+
+
+# ---------------------------------------------------------------- 规划器
+
+
+class IdlePlanner:
+ """基于热度图的空闲窗口规划器。"""
+
+ def __init__(self, cfg: Any) -> None:
+ self._cfg = cfg
+ tz_name = getattr(cfg, "idle_timezone", "Asia/Shanghai")
+ try:
+ self.tz = ZoneInfo(tz_name)
+ except Exception:
+ self.tz = timezone(timedelta(hours=8))
+ self.state = PlannerState()
+
+ self.noise_gap = getattr(cfg, "idle_noise_gap_minutes", 15) * 60
+ self.tau_fine = max(1.0, getattr(cfg, "idle_fine_tau_days", 28))
+ self.tau_coarse = max(1.0, getattr(cfg, "idle_coarse_tau_days", 14))
+ self.shrink_k = max(0.0, getattr(cfg, "idle_shrink_k", 4))
+ self.cold_pct_max = getattr(cfg, "idle_cold_pct_max", 0.05)
+ self.cold_abs_max = getattr(cfg, "idle_cold_abs_max", 1.0)
+ self.min_weeks_fine = getattr(cfg, "idle_min_sample_days_fine", 6)
+ self.min_weeks_mid = getattr(cfg, "idle_min_sample_days_mid", 3)
+ self.target_hour = getattr(cfg, "idle_target_hour", 3)
+ self.base_slots = getattr(cfg, "idle_base_slots", 3)
+ self.max_slots = getattr(cfg, "idle_max_slots", 6)
+ self.multiplier = getattr(cfg, "idle_penalty_multiplier", 1.5)
+ self.max_relax = getattr(cfg, "idle_max_relax_steps", 3)
+ self.buffer_minutes = getattr(cfg, "idle_buffer_minutes", 15)
+ self.run_estimate = getattr(cfg, "idle_run_estimate_minutes", 10)
+ self.fallback_minutes = getattr(cfg, "idle_fallback_minutes", 120)
+ self._last_plan: Plan | None = None
+
+ # ------------------------------------------------------------ 持久化
+
+ def load(self, payload: dict[str, Any]) -> None:
+ raw = payload.get("idle_planner") if isinstance(payload, dict) else None
+ if not isinstance(raw, dict):
+ return
+ try:
+ fine = [float(v) for v in raw.get("fine", [])]
+ coarse = [float(v) for v in raw.get("coarse", [])]
+ if len(fine) == CELLS:
+ self.state.fine = fine
+ if len(coarse) == SLOTS_PER_DAY:
+ self.state.coarse = coarse
+ seen = raw.get("seen_days", [])
+ if isinstance(seen, list):
+ self.state.seen_days = [str(d) for d in seen][-_MAX_SEEN_DAYS:]
+ recent = raw.get("recent", [])
+ if isinstance(recent, list):
+ self.state.recent = [float(v) for v in recent][-_MAX_RECENT:]
+ self.state.last_update = float(raw.get("last_update", 0.0))
+ except (TypeError, ValueError):
+ self.state = PlannerState()
+
+ def dump(self) -> dict[str, Any]:
+ return {"idle_planner": self.state.to_dict()}
+
+ # ------------------------------------------------------------ 观测
+
+ def _decay(self, now_ts: float) -> None:
+ if self.state.last_update <= 0:
+ self.state.last_update = now_ts
+ return
+ dt_days = (now_ts - self.state.last_update) / 86400
+ if dt_days <= 1e-6:
+ return
+ f_fine = 2 ** (-dt_days / self.tau_fine)
+ f_coarse = 2 ** (-dt_days / self.tau_coarse)
+ self.state.fine = [v * f_fine for v in self.state.fine]
+ self.state.coarse = [v * f_coarse for v in self.state.coarse]
+ self.state.last_update = now_ts
+
+ def observe(self, ts: float | None = None) -> None:
+ """记录一条用户消息，加深该时间窗口的热点。"""
+ now_ts = time.time() if ts is None else float(ts)
+ self._decay(now_ts)
+ wd, slot = slot_of(now_ts, self.tz)
+ self.state.fine[_cell(wd, slot)] += 1.0
+ self.state.coarse[slot] += 1.0
+ day_key = datetime.fromtimestamp(now_ts, self.tz).strftime("%Y-%m-%d")
+ if day_key not in self.state.seen_days:
+ self.state.seen_days.append(day_key)
+ if len(self.state.seen_days) > _MAX_SEEN_DAYS:
+ self.state.seen_days = self.state.seen_days[-_MAX_SEEN_DAYS:]
+ self.state.recent.append(now_ts)
+ if len(self.state.recent) > _MAX_RECENT:
+ self.state.recent = self.state.recent[-_MAX_RECENT:]
+
+ def last_plan(self) -> Plan | None:
+ return self._last_plan
+
+ def last_plan_dict(self) -> dict[str, Any] | None:
+ plan = self._last_plan
+ if plan is None:
+ return None
+ local_end = plan.end.astimezone(self.tz)
+ return {
+ "window_start": plan.start.strftime("%Y-%m-%d %H:%M"),
+ "window_end": local_end.strftime("%Y-%m-%d %H:%M"),
+ "planned_minutes": plan.length_minutes,
+ "tier": plan.tier,
+ "relaxed_steps": plan.relaxed_steps,
+ "forced": plan.forced,
+ "note": plan.note,
+ }
+
+ def note_window_taken(self, start_ts: float) -> None:
+ """窗口实际被占用后，把该时段计入一次观测（避免反复选同一段）。"""
+ self.observe(start_ts + 60)
+
+ # ------------------------------------------------------------ 分层分值
+
+ def active_days_of(self, weekdays: Sequence[int]) -> int:
+ cnt = 0
+ keys = set(self.state.seen_days)
+ for key in keys:
+ try:
+ dt = datetime.strptime(key, "%Y-%m-%d")
+ except ValueError:
+ continue
+ if dt.weekday() in weekdays:
+ cnt += 1
+ return cnt
+
+ def _shrink(self, cells: Sequence[float], n_days: int) -> list[float]:
+ alpha = n_days / (n_days + self.shrink_k) if (n_days + self.shrink_k) > 0 else 0.0
+ base = self.state.coarse
+ return [alpha * cells[i] + (1 - alpha) * base[i] for i in range(SLOTS_PER_DAY)]
+
+ def tier_table(self, tier: str, wd: int) -> tuple[list[float], list[float], int]:
+ """返回 (槽位权重, 槽位原始计数, 有效样本天数)。"""
+ if tier == "fine":
+ cells = self.state.fine[wd * SLOTS_PER_DAY : (wd + 1) * SLOTS_PER_DAY]
+ n_days = self.active_days_of([wd])
+ return self._shrink(cells, n_days), list(cells), n_days
+ if tier == "mid":
+ days = [0, 1, 2, 3, 4] if wd <= 4 else [5, 6]
+ cells = [
+ sum(self.state.fine[_cell(d, s)] for d in days) / len(days)
+ for s in range(SLOTS_PER_DAY)
+ ]
+ n_days = self.active_days_of(days)
+ return self._shrink(cells, n_days), cells, n_days
+ cells = list(self.state.coarse)
+ return cells, cells, self.active_days_of(list(range(DAYS)))
+
+ def _tiers_for(self, wd: int) -> list[str]:
+ tiers: list[str] = []
+ if self.active_days_of([wd]) >= self.min_weeks_fine:
+ tiers.append("fine")
+ days = [0, 1, 2, 3, 4] if wd <= 4 else [5, 6]
+ if self.active_days_of(days) >= self.min_weeks_mid:
+ tiers.append("mid")
+ tiers.append("coarse")
+ return tiers
+
+ # ------------------------------------------------------------ 冷段
+
+ def _cold_runs(
+ self, tier: str, wd: int
+ ) -> list[tuple[int, int, int]]:
+ """返回 [(环形起点槽, 长度, 归属星期几)]，长度上限一天。"""
+ weights, counts, _ = self.tier_table(tier, wd)
+ pcts = _normalize(weights)
+ mask = [
+ pcts[s] <= self.cold_pct_max and counts[s] <= self.cold_abs_max
+ for s in range(SLOTS_PER_DAY)
+ ]
+ # 跨午夜：接上下一星期几的表，否则 22:00-08:00 的安静段会被切断
+ w2, c2, _ = self.tier_table(tier, (wd + 1) % DAYS)
+ p2 = _normalize(w2)
+ mask2 = [
+ p2[s] <= self.cold_pct_max and c2[s] <= self.cold_abs_max
+ for s in range(SLOTS_PER_DAY)
+ ]
+ ring = mask + mask2
+ return [(start, length, wd) for start, length in _runs(ring, SLOTS_PER_DAY)]
+
+ def _lightest_run(
+ self, tier: str, wd: int, length_slots: int
+ ) -> tuple[int, int] | None:
+ """环形滑动窗口找总权重最低的连续 length_slots 槽段。"""
+ weights, _, _ = self.tier_table(tier, wd)
+ pcts = _normalize(weights)
+ if sum(pcts) <= 1e-9:
+ return None
+ n = SLOTS_PER_DAY
+ ring = pcts + pcts
+ best: tuple[int, int] | None = None
+ best_sum = float("inf")
+ for start in range(n):
+ total = sum(ring[start : start + length_slots])
+ if total < best_sum - 1e-12:
+ best_sum = total
+ best = (start, length_slots)
+ return best
+
+ def _next_occurrence(self, wd: int, now: datetime) -> datetime:
+ """本周/下周该星期几的 00:00（不早于 now 所在日期）。"""
+ today = now.astimezone(self.tz).replace(
+ hour=0, minute=0, second=0, microsecond=0
+ )
+ delta = (wd - today.weekday()) % DAYS
+ cand = today + timedelta(days=delta)
+ if cand + timedelta(days=1) <= now.astimezone(self.tz):
+ cand += timedelta(days=DAYS)
+ return cand
+
+ def _refine(
+ self,
+ start_slot: int,
+ length_slots: int,
+ wd: int,
+ now: datetime,
+ tier: str,
+ penalty: int,
+ relaxed: int,
+ forced: bool,
+ ) -> Plan:
+ """把冷段换成具体时间窗：段内找最长空闲，二分校核，扣 buffer。"""
+ day0 = self._next_occurrence(wd, now)
+ seg_start = day0 + timedelta(minutes=start_slot * SLOT_MINUTES)
+ seg_end = seg_start + timedelta(minutes=length_slots * SLOT_MINUTES)
+ now_local = now.astimezone(self.tz)
+ if seg_end <= now_local:
+ seg_start += timedelta(days=DAYS)
+ seg_end += timedelta(days=DAYS)
+ if seg_start < now_local:
+ seg_start = now_local + timedelta(minutes=1)
+
+ # 只信原始时间戳：连发消息先合并，再取段内最长无消息区间
+ ts = merge_noise(self.state.recent, self.noise_gap)
+ s_ts, e_ts = seg_start.timestamp(), seg_end.timestamp()
+ f_start, f_end = longest_free(ts, s_ts, e_ts)
+ # 二分只为确认这段空闲的上界（段内可能有更细的分布）
+ capacity = bisect_free_minutes(ts, s_ts, e_ts)
+ gap_minutes = int((f_end - f_start) // 60)
+ # 二分给出理论上界，实际可用不得超出真实无消息区间
+ free_minutes = min(capacity, gap_minutes)
+
+ need = self.buffer_minutes + self.run_estimate
+ target = seg_start.replace(
+ hour=self.target_hour, minute=0, second=0, microsecond=0
+ )
+ if target < seg_start:
+ target += timedelta(days=1)
+ # 起点尽量靠近偏好时刻，但必须落在空闲段内且留够运行空间
+ latest_start = max(f_start, f_end - need * 60)
+ start_ts = min(max(f_start, target.timestamp()), latest_start)
+ start_ts = max(start_ts, f_start, now.timestamp() + 60)
+
+ if f_end - start_ts < need * 60:
+ # 段内剩余空间放不下一次运行，标记但不放弃（保底语义）
+ length_minutes = max(1, int((f_end - start_ts) // 60))
+ note = "insufficient_free_space"
+ start_ts = f_start
+ length_minutes = max(1, int((f_end - f_start) // 60))
+ else:
+ # 结束时间严格受空闲段与冷段双重约束
+ length_minutes = int((f_end - start_ts) // 60)
+ if free_minutes < length_minutes: # 保守取真实空闲长度
+ length_minutes = free_minutes
+ note = ""
+
+ start = datetime.fromtimestamp(start_ts, self.tz)
+ end = min(start + timedelta(minutes=length_minutes),
+ datetime.fromtimestamp(f_end, self.tz))
+ return Plan(
+ start=start,
+ end=end,
+ length_minutes=length_minutes,
+ tier=tier,
+ relaxed_steps=relaxed,
+ forced=forced,
+ penalty=penalty,
+ note=note,
+ )
+
+ # ------------------------------------------------------------ 主流程
+
+ def _fallback(self, now: datetime, penalty: int, note: str) -> Plan:
+ local = now.astimezone(self.tz)
+ target = local.replace(
+ hour=self.target_hour, minute=0, second=0, microsecond=0
+ )
+ if target <= local + timedelta(minutes=1):
+ target += timedelta(days=1)
+ return Plan(
+ start=target,
+ end=target + timedelta(minutes=self.fallback_minutes),
+ length_minutes=self.fallback_minutes,
+ tier="fallback",
+ forced=True,
+ penalty=penalty,
+ note=note,
+ )
+
+ def plan(self, penalty: int = 1, now: datetime | None = None) -> Plan:
+ result = self._plan(penalty, now)
+ self._last_plan = result
+ return result
+
+ def _plan(self, penalty: int = 1, now: datetime | None = None) -> Plan:
+ now = now or datetime.now(timezone.utc)
+ local = now.astimezone(self.tz)
+ wd = local.weekday()
+
+ if sum(self.state.coarse) <= 1e-6:
+ return self._fallback(local, penalty, "no_history")
+
+ want = self.base_slots * (self.multiplier ** max(0, penalty - 1))
+ want = max(1, min(round(want), self.max_slots))
+
+ tiers = self._tiers_for(wd)
+ for step in range(self.max_relax + 1):
+ need_slots = max(1, want >> step)
+ for tier in tiers:
+ runs = [
+ r for r in self._cold_runs(tier, wd) if r[1] >= need_slots
+ ]
+ if not runs:
+ continue
+ start_slot, length_slots, owner = max(
+ runs, key=lambda r: (r[1], -r[0])
+ )
+ return self._refine(
+ start_slot,
+ length_slots,
+ owner,
+ now,
+ tier,
+ penalty,
+ step,
+ forced=False,
+ )
+
+ # 全部放宽仍不足：强制取最长冷段
+ best: tuple[int, int, int, str] | None = None
+ for tier in tiers:
+ for run in self._cold_runs(tier, wd):
+ if best is None or run[1] > best[1]:
+ best = (run[0], run[1], run[2], tier)
+ if best is not None:
+ return self._refine(
+ best[0], best[1], best[2], now, best[3], penalty,
+ self.max_relax, forced=True,
+ )
+ # 完全没有冷槽（用户几乎全天活跃）：退到热度最低的连续段
+ for tier in tiers:
+ floor_run = self._lightest_run(tier, wd, 1)
+ if floor_run is not None:
+ return self._refine(
+ floor_run[0], floor_run[1], wd, now, tier, penalty,
+ self.max_relax, forced=True,
+ )
+ return self._fallback(local, penalty, "no_cold_segment")
+ return self._refine(
+ best[0],
+ best[1],
+ best[2],
+ now,
+ best[3],
+ penalty,
+ self.max_relax,
+ forced=True,
+ )
+
+
+def describe(plan: Plan) -> str:
+ """日志用的一行描述。"""
+ local = plan.start
+ wd = _WD_NAMES[local.weekday()]
+ return (
+ f"{wd} {local.strftime('%H:%M')}-{plan.end.strftime('%H:%M')} "
+ f"({plan.length_minutes}min, tier={plan.tier}, "
+ f"relax={plan.relaxed_steps}, forced={plan.forced}"
+ f"{', ' + plan.note if plan.note else ''})"
+ )
diff --git a/amrita_plugin_memory/rethinking/schemas.py b/amrita_plugin_memory/rethinking/schemas.py
index 9c45534..c30b69f 100644
--- a/amrita_plugin_memory/rethinking/schemas.py
+++ b/amrita_plugin_memory/rethinking/schemas.py
@@ -192,6 +192,17 @@
),
)
+GET_CONTEXT_SCHEMA = FunctionDefinitionSchema(
+ name="subconscious_get_context",
+ description=(
+ "获取本轮运行环境：精确时间、距上轮间隔、上轮摘要、待审查知识建议、"
+ "记忆库规模与热度窗口。每轮应先调用本工具，再决定做什么。"
+ "返回中的 stale 字段列出本次未取到的项，不要推测它们的值。"
+ ),
+ parameters=FunctionParametersSchema(type="object", properties={}, required=[]),
+)
+
+
GET_MEMORY_STATS_SCHEMA = FunctionDefinitionSchema(
name="subconscious_get_memory_stats",
description="获取记忆库统计概览：总数、重要性分布、标签分布、时间范围。用于判断是否需要压缩。",
@@ -418,6 +429,7 @@
LIST_MEMORY_SCHEMA,
ITER_STOP_SCHEMA,
DUPLICATE_HELPER_SCHEMA,
+ GET_CONTEXT_SCHEMA,
GET_MEMORY_STATS_SCHEMA,
KNOWLEDGE_LIST_SCHEMA,
KNOWLEDGE_READ_SCHEMA,
diff --git a/amrita_plugin_memory/rethinking/tools.py b/amrita_plugin_memory/rethinking/tools.py
index 06a4d3b..c86fc29 100644
--- a/amrita_plugin_memory/rethinking/tools.py
+++ b/amrita_plugin_memory/rethinking/tools.py
@@ -27,6 +27,7 @@
from .schemas import (
DELETE_MEMORY_SCHEMA,
DUPLICATE_HELPER_SCHEMA,
+ GET_CONTEXT_SCHEMA,
GET_MEMORY_STATS_SCHEMA,
GET_PROFILE_SCHEMA,
ITER_STOP_SCHEMA,
@@ -241,9 +242,9 @@ async def subconscious_list_memory(data: dict[str, Any]) -> str:
@on_tools(ITER_STOP_SCHEMA, strict=True, bound_to=_SUBCONSCIOUS_TOOLS)
async def subconscious_iter_stop(data: dict[str, Any]) -> str:
- logger.info(
- f"[Subconscious] iter_stop: summary={str(data.get('summary', ''))[:80]}"
- )
+ summary = str(data.get("summary", ""))
+ _state.set_last_iter_summary(summary)
+ logger.info(f"[Subconscious] iter_stop: summary={summary[:80]}")
return json.dumps(
{"status": "acknowledged", "message": "推理循环已确认结束"}, ensure_ascii=False
)
@@ -580,6 +581,91 @@ async def subconscious_duplicate_helper(data: dict[str, Any]) -> str:
return _tools_err(str(e))
+@on_tools(GET_CONTEXT_SCHEMA, strict=True, bound_to=_SUBCONSCIOUS_TOOLS)
+async def subconscious_get_context(data: dict[str, Any]) -> str:
+ """聚合本轮运行环境；单项失败只标 stale，不中断。"""
+ stale: list[str] = []
+ runner = _state.get_runner()
+ cfg = runner._config if runner is not None else None
+
+ tz_name = getattr(cfg, "idle_timezone", "Asia/Shanghai") if cfg else "Asia/Shanghai"
+ try:
+ from zoneinfo import ZoneInfo
+
+ tz = ZoneInfo(tz_name)
+ except Exception:
+ tz = timezone.utc
+ stale.append("timezone")
+
+ now = datetime.now(timezone.utc)
+ local = now.astimezone(tz)
+
+ last_at = _state.get_last_run_at()
+ if last_at > 0:
+ gap_minutes = int((now.timestamp() - last_at) // 60)
+ last_run_iso = datetime.fromtimestamp(last_at, tz).isoformat()
+ else:
+ gap_minutes = None
+ last_run_iso = None
+
+ suggested = _state.get_knowledge_suggestions()
+
+ total = None
+ low_count = None
+ threshold = None
+ memory_hint = ""
+ try:
+ ope = _make_operator()
+ await ope.init()
+ pid = _get_partition_id()
+ total = await ope.count_user_notes(pid)
+ threshold = getattr(cfg, "memory_warn_threshold", 100) if cfg else 100
+ if total > threshold:
+ result = await ope.get_all_notes(pid, include=["metadatas"])
+ low_count = sum(
+ 1
+ for m in (result.get("metadatas") or [])
+ if isinstance(m, dict) and m.get("importance") == "low"
+ )
+ memory_hint = (
+ f"记忆库已有 {total} 条（低重要性 {low_count} 条），"
+ "建议本轮优先用 subconscious_duplicate_helper 整理合并。"
+ )
+ except Exception as e:
+ logger.warning(f"[Subconscious] get_context memory stats failed: {e}")
+ stale.append("memory_stats")
+
+ plan_info = None
+ try:
+ if runner is not None and getattr(runner, "_planner", None) is not None:
+ plan_info = runner._planner.last_plan_dict()
+ except Exception:
+ stale.append("idle_plan")
+
+ return json.dumps(
+ {
+ "status": "success",
+ "now": local.strftime("%Y-%m-%d %H:%M:%S"),
+ "timezone": tz_name,
+ "weekday": local.strftime("%A"),
+ "last_run_at": last_run_iso,
+ "minutes_since_last_run": gap_minutes,
+ "last_run_summary": _state.get_last_iter_summary() or None,
+ "pending_knowledge_suggestions": {
+ "count": len(suggested),
+ "ids": [s.get("kid") or s.get("title") or "?" for s in suggested][:20],
+ },
+ "memory_total": total,
+ "memory_low_importance": low_count,
+ "memory_warn_threshold": threshold,
+ "memory_hint": memory_hint,
+ "idle_plan": plan_info,
+ "stale": stale,
+ },
+ ensure_ascii=False,
+ )
+
+
@on_tools(GET_MEMORY_STATS_SCHEMA, strict=True, bound_to=_SUBCONSCIOUS_TOOLS)
async def subconscious_get_memory_stats(data: dict[str, Any]) -> str:
"""返回记忆库统计概览。"""
diff --git a/pyproject.toml b/pyproject.toml
index d9c7888..0043cdb 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,7 +1,7 @@
[project]
name = "amrita_plugin_memory"
description = "Amrita长期记忆插件"
-version = "0.7.1"
+version = "0.8.0"
dependencies = [
"aiohttp>=3.13.3",
"aiologic>=0.16.0",
@@ -37,7 +37,7 @@ module_name = "nonebot.adapters.onebot.v11"
[tool.uv]
dev-dependencies = [
"ruff>=0.12.8",
- #"pytest>=8.4.1",
+ "pytest>=8.4.1",
# `ambot memory` 依赖 0.2.1 起的 entry point extras（[full_load]）支持
"ambot-inlinectl>=0.2.1",
]
diff --git a/tests/test_schedule.py b/tests/test_schedule.py
new file mode 100644
index 0000000..3e538b4
--- /dev/null
+++ b/tests/test_schedule.py
@@ -0,0 +1,161 @@
+"""DMN 空闲窗口调度的纯函数与规划器单测。
+
+只测 schedule.py，不加载 nonebot，便于快速回归。
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+from types import SimpleNamespace as NS
+from zoneinfo import ZoneInfo
+
+from amrita_plugin_memory.rethinking.schedule import (
+ CELLS,
+ SLOTS_PER_DAY,
+ IdlePlanner,
+ _runs,
+ bisect_free_minutes,
+ feasible_minutes,
+ free_gaps,
+ longest_free,
+ merge_noise,
+ slot_of,
+)
+
+TZ = ZoneInfo("Asia/Shanghai")
+
+
+def make_cfg(**over) -> NS:
+ base = dict(
+ idle_timezone="Asia/Shanghai",
+ idle_noise_gap_minutes=15,
+ idle_fine_tau_days=28.0,
+ idle_coarse_tau_days=14.0,
+ idle_shrink_k=4.0,
+ idle_cold_pct_max=0.05,
+ idle_cold_abs_max=1.0,
+ idle_min_sample_days_fine=6,
+ idle_min_sample_days_mid=3,
+ idle_target_hour=3,
+ idle_base_slots=3,
+ idle_max_slots=6,
+ idle_penalty_multiplier=1.5,
+ idle_max_relax_steps=3,
+ idle_buffer_minutes=15,
+ idle_run_estimate_minutes=10,
+ idle_fallback_minutes=120,
+ )
+ base.update(over)
+ return NS(**base)
+
+
+def seed_daytime(planner: IdlePlanner, days: int = 14) -> None:
+ """灌入白天活跃、夜间安静的历史。"""
+ base = datetime(2026, 10, 5, tzinfo=TZ)
+ for d in range(days):
+ for h in range(9, 22, 2):
+ for mi in (0, 40):
+ planner.observe((base + timedelta(days=d, hours=h, minutes=mi)).timestamp())
+
+
+def test_merge_noise_collapses_bursts():
+ assert merge_noise([0, 60, 600, 1000], 900) == [0, 1000]
+
+
+def test_free_gaps_and_longest():
+ assert free_gaps([100, 200], 0, 300) == [(0, 100), (100, 200), (200, 300)]
+ assert longest_free([100, 200], 0, 300) == (0, 100)
+ assert free_gaps([], 0, 0) == []
+
+
+def test_feasible_is_monotonic():
+ ts = [3600 * 2]
+ for minutes in range(0, 240, 10):
+ if feasible_minutes(ts, 0, 3600 * 4, minutes):
+ assert feasible_minutes(ts, 0, 3600 * 4, max(0, minutes - 10))
+
+
+def test_bisect_matches_bruteforce():
+ ts = [3600 * 2]
+ got = bisect_free_minutes(ts, 0, 3600 * 4)
+ brute = max(m for m in range(0, 241) if feasible_minutes(ts, 0, 3600 * 4, m))
+ assert got == brute == 120
+
+
+def test_ring_runs_wrap_around():
+ assert _runs([True] * 24, 12) == [(0, 12)]
+ mask = [True, True] + [False] * 20 + [True, True]
+ assert dict(_runs(mask, 24))[22] == 4
+
+
+def test_slot_of_two_hour_grid():
+ ts = datetime(2026, 10, 19, 23, 30, tzinfo=TZ).timestamp()
+ wd, slot = slot_of(ts, TZ)
+ assert wd == 0 and slot == 11
+
+
+def test_cold_start_falls_back_to_target_hour():
+ planner = IdlePlanner(make_cfg())
+ plan = planner.plan(1, datetime(2026, 10, 19, 9, 0, tzinfo=TZ))
+ assert plan.tier == "fallback"
+ assert plan.forced
+ assert plan.start.hour == 3
+
+
+def test_plan_lands_in_night_and_respects_capacity():
+ planner = IdlePlanner(make_cfg())
+ seed_daytime(planner)
+ now = datetime(2026, 10, 19, 9, 0, tzinfo=TZ)
+ plan = planner.plan(1, now)
+ assert not plan.forced
+ assert plan.start >= now
+ assert plan.end > plan.start
+ assert plan.end.hour <= 9
+
+
+def test_penalty_raises_requirement_but_stays_in_window():
+ planner = IdlePlanner(make_cfg())
+ seed_daytime(planner)
+ now = datetime(2026, 10, 19, 9, 0, tzinfo=TZ)
+ plans = [planner.plan(p, now) for p in (1, 3, 8)]
+ assert all(not p.forced for p in plans)
+ assert all(p.start.hour == 3 for p in plans)
+
+
+def test_no_deadlock_when_fully_active():
+ """用户几乎全天活跃时也必须给出窗口，不能返回空。"""
+ planner = IdlePlanner(make_cfg())
+ base = datetime(2026, 10, 5, tzinfo=TZ)
+ for d in range(21):
+ for s in range(SLOTS_PER_DAY):
+ planner.observe((base + timedelta(days=d, hours=s * 2)).timestamp())
+ plan = planner.plan(6, datetime(2026, 10, 19, 9, 0, tzinfo=TZ))
+ assert plan.start is not None
+ assert plan.forced
+
+
+def test_state_roundtrip():
+ planner = IdlePlanner(make_cfg())
+ seed_daytime(planner, days=5)
+ dumped = planner.dump()
+ assert len(dumped["idle_planner"]["fine"]) == CELLS
+ other = IdlePlanner(make_cfg())
+ other.load(dumped)
+ assert other.state.seen_days == planner.state.seen_days
+ assert abs(sum(other.state.coarse) - sum(planner.state.coarse)) < 1e-2
+
+
+def test_decay_reduces_history():
+ planner = IdlePlanner(make_cfg())
+ seed_daytime(planner, days=7)
+ before = sum(planner.state.coarse)
+ planner.observe(datetime(2026, 11, 30, tzinfo=TZ).timestamp())
+ assert sum(planner.state.coarse) < before
+
+
+def test_sparse_history_still_plans():
+ planner = IdlePlanner(make_cfg())
+ planner.observe(datetime(2026, 10, 12, 15, 0, tzinfo=TZ).timestamp())
+ plan = planner.plan(1, datetime(2026, 10, 19, 9, 0, tzinfo=TZ))
+ assert plan.start is not None
+ assert plan.length_minutes >= 1
From 794e382571b16a37bda5a0de2ef14c08ff6d2782 Mon Sep 17 00:00:00 2001
From: AmriaLin <325560502+AmriaLin@users.noreply.github.com>
Date: Tue, 20 Oct 2026 05:40:48 +0000
Subject: [PATCH 2/6] =?UTF-8?q?feat(profile):=20=E6=97=A5=E5=B8=B8?=
=?UTF-8?q?=E5=AF=B9=E8=AF=9D=E5=BC=80=E6=94=BE=E7=94=A8=E6=88=B7=E7=94=BB?=
=?UTF-8?q?=E5=83=8F=E5=8F=AA=E8=AF=BB=EF=BC=88=E4=BB=85=20DMN=20=E7=9B=AE?=
=?UTF-8?q?=E6=A0=87=E7=94=A8=E6=88=B7=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
- 抽出 profile_store：画像路径 PROFILE_PATH 与纯函数 read_profile，运行器与会话侧共用
- permissions.can_read_profile：比对 subconscious.target_user_id，仅本人可读
- 新增表层工具 read_profile（custom_run 校验），写入仍只留在循环内部
- README 补充只读口径说明
---
README.md | 6 ++
amrita_plugin_memory/permissions.py | 40 ++++++++++++-
amrita_plugin_memory/profile_store.py | 67 ++++++++++++++++++++++
amrita_plugin_memory/rethinking/runner.py | 45 ++-------------
amrita_plugin_memory/rethinking/schemas.py | 21 +++++++
amrita_plugin_memory/rethinking/tools.py | 34 ++++++++++-
amrita_plugin_memory/rethinking/types.py | 10 ----
7 files changed, 168 insertions(+), 55 deletions(-)
create mode 100644 amrita_plugin_memory/profile_store.py
diff --git a/README.md b/README.md
index cf26699..9bbc054 100644
--- a/README.md
+++ b/README.md
@@ -410,11 +410,17 @@ flowchart TD
| `knowledge_read` | kid, start_line?, end_line? |
| `knowledge_search` | query, top_k? |
| `knowledge_suggest` | action, title, summary, body, reason |
+| `read_profile` | start_line?, end_line? |
知识库的三个只读工具 `knowledge_list` / `knowledge_read` / `knowledge_search`
需要 LP 权限节点 `memory.knowledge`；`lp.admin` 与 `.env` 里的超级用户始终持有。
未授权时工具返回错误 JSON，不返回任何知识内容。
+用户画像的只读工具 `read_profile` 走另一条口径：只对离线巩固循环的目标用户
+（`subconscious.target_user_id`）开放，且只能读自己那份画像，不占用权限节点。
+返回值与循环侧一致：`summary` + `content` 行窗口 + `total_lines` / `range_start` / `range_end`。
+画像写入仍只发生在循环内部，表层不开放。
+
### 巩固循环工具（`rethinking/tools.py`，共 20 个）
记忆与 session/画像工具注册在隔离的 `_SUBCONSCIOUS_TOOLS` 上，不污染全局工具管理器。
diff --git a/amrita_plugin_memory/permissions.py b/amrita_plugin_memory/permissions.py
index 4970f83..be91bc3 100644
--- a/amrita_plugin_memory/permissions.py
+++ b/amrita_plugin_memory/permissions.py
@@ -1,8 +1,12 @@
-"""知识库读取的权限校验。
+"""会话侧读取权限校验（知识库 / 用户画像）。
-会话侧（对话 LLM）可以读全局知识库，但需要 LP 节点 ``memory.knowledge``；
+知识库：会话侧（对话 LLM）可以读全局知识库，但需要 LP 节点 ``memory.knowledge``；
``lp.admin`` 与 ``.env`` 里的超级用户始终持有该权限。
-潜意识循环内部调用不受此限制。
+
+用户画像：只对离线巩固循环（DMN）的目标用户开放只读，
+比对 ``subconscious.target_user_id``，不做权限节点扩散。
+
+潜意识循环内部调用均不受此限制。
"""
from __future__ import annotations
@@ -54,6 +58,36 @@ async def can_read_knowledge(event: Event | None) -> bool:
return await _has_node(event, KNOWLEDGE_NODE)
+async def get_dmn_target_user_id() -> str:
+ """当前 DMN 目标用户 ID：优先取运行器中实际生效的值，回退到配置。"""
+ try:
+ from .rethinking import _state
+
+ uid = _state.get_target_user_id()
+ if uid:
+ return str(uid)
+ except Exception as e:
+ logger.debug(f"[Memory] 取 DMN 运行器失败: {e}")
+ try:
+ from .config import DataManager
+
+ cfg = (await DataManager().safe_get_config()).subconscious
+ return str(cfg.target_user_id or "")
+ except Exception as e:
+ logger.debug(f"[Memory] 取 DMN 配置失败: {e}")
+ return ""
+
+
+async def can_read_profile(event: Event | None) -> bool:
+ """是否允许读取用户画像：仅 DMN 目标用户本人。"""
+ if event is None:
+ return False
+ target = await get_dmn_target_user_id()
+ if not target:
+ return False
+ return str(event.get_user_id()) == target
+
+
def get_event_from_tool_ctx(ctx: Any) -> Event | None:
"""从 ToolContext 里取出触发本次调用的原始事件。"""
try:
diff --git a/amrita_plugin_memory/profile_store.py b/amrita_plugin_memory/profile_store.py
new file mode 100644
index 0000000..6f96f4b
--- /dev/null
+++ b/amrita_plugin_memory/profile_store.py
@@ -0,0 +1,67 @@
+"""用户画像文件（Markdown）的路径与解析。
+
+会话侧只读入口与潜意识循环共用同一份解析逻辑，
+避免为读取画像而依赖运行器实例。
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import TypedDict
+
+from nonebot import logger
+
+from .config import DATA_PATH
+
+
+class ProfileResult(TypedDict):
+ """用户画像读取返回值。"""
+
+ summary: str
+ content: str
+ total_lines: int
+ range_start: int
+ range_end: int
+
+
+#: 画像文件路径
+PROFILE_PATH: Path = DATA_PATH / "user_profile.md"
+
+
+def read_profile(
+ path: Path,
+ start_line: int | None = None,
+ end_line: int | None = None,
+) -> ProfileResult:
+ """解析画像 Markdown（summary + "---" + body），按行返回滑动窗口。
+
+ 与运行器解耦：潜意识循环未启动时，会话侧也能读到既有画像文件。
+ """
+ empty: ProfileResult = {
+ "summary": "",
+ "content": "",
+ "total_lines": 0,
+ "range_start": 0,
+ "range_end": 0,
+ }
+ if not path.exists():
+ return empty
+ try:
+ text = path.read_text(encoding="utf-8")
+ parts = text.split("---", 1)
+ summary = parts[0].strip() if parts else ""
+ body = parts[1].strip() if len(parts) > 1 else ""
+ body_lines = body.split("\n") if body else []
+ total = len(body_lines)
+ s = 0 if start_line is None else max(0, start_line)
+ e = total if end_line is None else max(s, min(total, end_line))
+ return {
+ "summary": summary,
+ "content": "\n".join(body_lines[s:e]),
+ "total_lines": total,
+ "range_start": s,
+ "range_end": e,
+ }
+ except Exception as e:
+ logger.warning(f"[Subconscious] Read profile failed: {e}")
+ return empty
diff --git a/amrita_plugin_memory/rethinking/runner.py b/amrita_plugin_memory/rethinking/runner.py
index ccfa6a0..a8f0308 100644
--- a/amrita_plugin_memory/rethinking/runner.py
+++ b/amrita_plugin_memory/rethinking/runner.py
@@ -32,6 +32,7 @@
from ..config import DATA_PATH, SubconsciousConfig
from ..keys import make_scope_id
from ..models import SubconsciousState
+from ..profile_store import PROFILE_PATH, ProfileResult, read_profile
from . import _state
from .backend import SubconsciousBackend
from .consts import (
@@ -44,7 +45,7 @@
)
from .knowledge import KnowledgeBaseManager
from .schedule import IdlePlanner, describe
-from .types import ProfileResult, SessionSummary
+from .types import SessionSummary
_STATE_UID = "amrita_memory"
#: 重排节流窗口（秒）
@@ -81,7 +82,7 @@ def __init__(self, config: SubconsciousConfig) -> None:
self._kb_manager: KnowledgeBaseManager | None = None
# session 摘要缓存（LRU）：session DB id -> 摘要文本，最多 128 条
self._session_cache: LRUCache[int, str] = LRUCache(128)
- self._profile_path = DATA_PATH / "user_profile.md"
+ self._profile_path = PROFILE_PATH
# 空闲窗口规划器（热度图 + 冷段二分）
self._planner = IdlePlanner(config)
# 重排节流：一分钟内重复触发只更新惩罚与热度，不重复规划
@@ -571,45 +572,7 @@ async def _read_profile(
self, start_line: int | None = None, end_line: int | None = None
) -> ProfileResult:
"""读取用户画像 Markdown 文件，支持行数滑动窗口。"""
- if not self._profile_path.exists():
- return {
- "summary": "",
- "content": "",
- "total_lines": 0,
- "range_start": 0,
- "range_end": 0,
- }
- try:
- text = self._profile_path.read_text(encoding="utf-8")
- parts = text.split("---", 1)
- summary = parts[0].strip() if parts else ""
- body = parts[1].strip() if len(parts) > 1 else ""
- body_lines = body.split("\n") if body else []
- total = len(body_lines)
- if start_line is None:
- s = 0
- else:
- s = max(0, start_line)
- if end_line is None:
- e = total
- else:
- e = max(s, min(total, end_line))
- return {
- "summary": summary,
- "content": "\n".join(body_lines[s:e]),
- "total_lines": total,
- "range_start": s,
- "range_end": e,
- }
- except Exception as e:
- logger.warning(f"[Subconscious] Read profile failed: {e}")
- return {
- "summary": "",
- "content": "",
- "total_lines": 0,
- "range_start": 0,
- "range_end": 0,
- }
+ return read_profile(self._profile_path, start_line, end_line)
async def _update_profile(
self,
diff --git a/amrita_plugin_memory/rethinking/schemas.py b/amrita_plugin_memory/rethinking/schemas.py
index c30b69f..59d536b 100644
--- a/amrita_plugin_memory/rethinking/schemas.py
+++ b/amrita_plugin_memory/rethinking/schemas.py
@@ -391,6 +391,27 @@
),
)
+READ_PROFILE_SCHEMA = FunctionDefinitionSchema(
+ name="read_profile",
+ description=(
+ "读取本机离线巩固循环为你维护的用户画像（摘要 + 正文）。"
+ "返回值含 summary、content、total_lines、range_start、range_end；"
+ "正文较长时用 start_line / end_line 分段读取。仅画像所有者本人可读"
+ ),
+ parameters=FunctionParametersSchema(
+ type="object",
+ properties={
+ "start_line": FunctionPropertySchema(
+ type="integer", description="起始行号（0-based），默认全文"
+ ),
+ "end_line": FunctionPropertySchema(
+ type="integer", description="结束行号（不含），默认全文"
+ ),
+ },
+ required=[],
+ ),
+)
+
UPDATE_PROFILE_SCHEMA = FunctionDefinitionSchema(
name="subconscious_update_profile",
description=(
diff --git a/amrita_plugin_memory/rethinking/tools.py b/amrita_plugin_memory/rethinking/tools.py
index c86fc29..dc7a5e4 100644
--- a/amrita_plugin_memory/rethinking/tools.py
+++ b/amrita_plugin_memory/rethinking/tools.py
@@ -20,7 +20,12 @@
from nonebot import logger
from ..keys import make_scope_id
-from ..permissions import can_read_knowledge, get_event_from_tool_ctx
+from ..permissions import (
+ can_read_knowledge,
+ can_read_profile,
+ get_event_from_tool_ctx,
+)
+from ..profile_store import PROFILE_PATH, read_profile
from ..vector import AsyncUserMemory, MemoryMetadata, get_db_conn
from . import _state
from .consts import DEFAULT_SEND_PROMPT, ensure_prompt_file, load_character_prompt
@@ -41,6 +46,7 @@
LIST_MEMORY_SCHEMA,
READ_CHAT_CONTEXT_SCHEMA,
READ_MEMORY_SCHEMA,
+ READ_PROFILE_SCHEMA,
READ_SESSIONS_SCHEMA,
READ_SUGGESTIONS_SCHEMA,
SEND_TO_USER_SCHEMA,
@@ -893,6 +899,32 @@ async def knowledge_search_authorized(ctx: ToolContext) -> str | None:
return await subconscious_knowledge_search(ctx.data)
+# 会话侧只读入口：用户画像，仅画像所有者（DMN 目标用户）可读
+
+
+def _deny_profile() -> str:
+ return _tools_err("无权限读取该画像（仅离线巩固循环的目标用户可读自己的画像）")
+
+
+@on_tools(READ_PROFILE_SCHEMA, strict=True, custom_run=True)
+async def read_profile_authorized(ctx: ToolContext) -> str | None:
+ """对话 LLM 读取用户画像，仅限 DMN 目标用户本人。"""
+ if not await can_read_profile(get_event_from_tool_ctx(ctx)):
+ return _deny_profile()
+ try:
+ profile = read_profile(
+ PROFILE_PATH,
+ start_line=ctx.data.get("start_line"),
+ end_line=ctx.data.get("end_line"),
+ )
+ return _tools_ok(**profile)
+ except Exception as e:
+ logger.opt(exception=e, colors=True, raw=True).exception(
+ f"read_profile error: {e}"
+ )
+ return _tools_err(str(e))
+
+
# 知识建议：对话 LLM -> 潜意识 Agent 审查管线
diff --git a/amrita_plugin_memory/rethinking/types.py b/amrita_plugin_memory/rethinking/types.py
index 9b924ba..c086635 100644
--- a/amrita_plugin_memory/rethinking/types.py
+++ b/amrita_plugin_memory/rethinking/types.py
@@ -67,13 +67,3 @@ class SessionSummary(TypedDict):
created_at: float
summary: str
message_count: int
-
-
-class ProfileResult(TypedDict):
- """用户画像读取返回值。"""
-
- summary: str
- content: str
- total_lines: int
- range_start: int
- range_end: int
From 8486079021add457c3a98614784bb9d433325e62 Mon Sep 17 00:00:00 2001
From: AmriaLin <325560502+AmriaLin@users.noreply.github.com>
Date: Tue, 20 Oct 2026 06:39:06 +0000
Subject: [PATCH 3/6] =?UTF-8?q?chore:=20=E8=A1=A5=20AGENTS.md=20=E5=B9=B6?=
=?UTF-8?q?=E4=BF=AE=E6=97=B6=E5=8C=BA=E7=B1=BB=E5=9E=8B=E6=A0=87=E6=B3=A8?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
- 新增 AGENTS.md：模块边界、静态检查命令、工具注册三条路径、权限口径、数据布局
- schedule.py：slot_of 的 tz 形参放宽为 tzinfo，IdlePlanner.tz 显式标注，消除 pyright 类型错误
---
AGENTS.md | 94 +++++++++++++++++++++
amrita_plugin_memory/rethinking/schedule.py | 5 +-
2 files changed, 97 insertions(+), 2 deletions(-)
create mode 100644 AGENTS.md
diff --git a/AGENTS.md b/AGENTS.md
new file mode 100644
index 0000000..a74f9c8
--- /dev/null
+++ b/AGENTS.md
@@ -0,0 +1,94 @@
+# AGENTS.md
+
+给在本仓库工作的 AI 代理（以及新加入的人）看的约定。只写与改动直接相关的硬性要求，风格问题按现有代码对齐。
+
+## 项目概况
+
+- AmritaBot 长期记忆插件：L1 备忘录、L2 向量记忆、L3 离线巩固循环（DMN）三层。
+- 表层工具（对话 LLM 可用）注册在全局 ToolsManager；`subconscious_*` 通过 `bound_to=_SUBCONSCIOUS_TOOLS` 注册到隔离管理器，不污染全局。
+- 插件入口 `amrita_plugin_memory/__init__.py`，命令入口 `cli.py`（`ambot memory`），对话命令在 `matchers.py`（`/memory`）。
+
+## 模块边界
+
+| 位置 | 职责 |
+| --- | --- |
+| `config.py` | EnvConfig / ConfigFile（pydantic + uniconf），DATA_PATH 等路径常量 |
+| `keys.py` | uni_id 生成与 scope 解析 |
+| `vector.py` | ChromaDB 封装、AsyncUserMemory 记忆读写 |
+| `embedding.py` / `embed.py` | 嵌入模型与指纹校验、重映射 |
+| `memo.py` | L1 备忘录读写 |
+| `decay.py` | 时间记忆退化与分层清理 |
+| `permissions.py` | 会话侧读取权限校验 |
+| `profile_store.py` | 画像路径与 Markdown 解析（纯函数，不依赖 nonebot 运行时） |
+| `rethinking/` | DMN 循环：runner、tools、schemas、knowledge、schedule |
+
+新增纯逻辑优先放进不依赖 nonebot 的模块（如 `profile_store.py`、`rethinking/schedule.py`），运行器只做编排。
+
+## 环境与依赖
+
+- Python 3.10–3.13，包管理只用 uv。改依赖后跑 `uv sync`，不要用 `pip install` 补装单包。
+- 类型检查用 `pyright`，配置在 `pyproject.toml` 的 `[tool.pyright]`（standard 模式）。
+- 本仓库 `.venv` 里通常没有 pyright / ruff，用 `--pythonpath .venv/bin/python` 指向仓库环境，否则会因解析不到 amrita_core / chromadb 报一片 Import 错误。
+
+## 常用命令
+
+```bash
+uv run ruff check amrita_plugin_memory
+uv run ruff format --check amrita_plugin_memory
+uv run pyright --pythonpath .venv/bin/python amrita_plugin_memory
+python -m py_compile amrita_plugin_memory/**/*.py
+```
+
+## 测试要求
+
+本仓库不补单元测试，验证以静态检查加手动冒烟为准：
+
+1. `ruff check` 与 `ruff format --check` 通过
+2. `pyright` 零 error（用上面的 pythonpath 跑）
+3. 改到的模块 `py_compile` 通过
+4. 手动跑一遍改动的命令（`/memory`、`ambot memory`）或工具调用，并在 PR 里说明验证方式与结果
+
+纯逻辑尽量写成可独立执行的形式，方便用 `python -c` 或临时脚本验证；不要把临时脚本提交进仓库。
+`tests/` 下既有的 `test_schedule.py` 是需要 nonebot 初始化才能跑的历史文件，不作为改动门槛。
+
+## 工具注册的三条路径
+
+- 表层可读、需权限：实现放 `rethinking/tools.py`，`@on_tools(SCHEMA, strict=True, custom_run=True)` 注册到全局，进函数先校验 `permissions.py` 里的判定。
+- 仅循环可用：同上但带 `bound_to=_SUBCONSCIOUS_TOOLS`。
+- 新工具要同时加入 `rethinking/schemas.py` 的 `_SUBCONSCIOUS_TOOL_FUNCTIONS`（若循环也要用），并在 README 的工具表里登记。
+
+新增工具时注意：schema 的 `name` 是工具名，函数名可另取；`custom_run=True` 时函数签名收 `ToolContext`，返回值必须是字符串。
+
+## 权限口径
+
+- 知识库只读：`permissions.can_read_knowledge`，超用户 / `lp.admin` / `memory.knowledge` 节点三者之一。
+- 用户画像只读：`permissions.can_read_profile`，仅 `subconscious.target_user_id` 本人，不扩散权限节点。
+- 取值优先用 `_state.get_target_user_id()`（运行器实际生效值），回退 `DataManager` 配置。空值一律拒。
+
+## 数据与持久化
+
+- 画像：`DATA_PATH/user_profile.md`，格式 `summary` + `---` + 正文；行级增量更新，写入只发生在循环内部。
+- 知识库：`DATA_PATH/knowledge/` + `knowledge_index.json` + ChromaDB，文件/索引/向量三方同步。
+- DMN 元状态：`SubconsciousState` 表；向量记忆在 ChromaDB collection。
+- 写入路径优先走读-改-写整体替换，避免嵌入失败时丢数据（参考 `25f1078`）。
+
+## 代码注释风格
+
+- 注释与 docstring 一律中文，单行短句，讲“为什么”和约束，不复述代码本身在做什么。
+- 模块顶部 docstring 写职责、关键约束、与其他模块的边界；函数 docstring 写清参数语义与副作用。
+- 不写分节注释（`# ---- 小节名`、`# ====`），用空行和函数拆分表达结构。
+- 不写连续多行 `#` 注释；需要成段解释时写进模块或函数 docstring。
+- 待办写成 `# TODO(review #)：具体动作`。
+- 类型注解必须齐全；pydantic 字段用 `Field(description=...)`，描述用中文。
+- 不写注释掉的旧代码，不写“新增/修改”这类只在 diff 里有意义的注释。
+
+## 配置改动
+
+- 配置型在 `config.py`（pydantic + `nonebot_plugin_uniconf`）。新字段必须有中文 `description` 和保守默认值，默认值要保证旧配置文件行为不变。
+- 新增开关涉及 DMN 行为时，同步检查 `matchers.py` 的 `/memory dmn` 前置校验与 README 的配置表。
+
+## Git 与 PR
+
+- 分支名用 `feat/` `fix/` `refactor/` 前缀；提交信息中文，首行说明改了什么，正文写动机与影响面。
+- PR 拿到 approval 不等于授权合并；合并默认 squash，等明确指令再动。
+- GitHub raw 域名可能不可达，取仓库内容走 `api.github.com`。
diff --git a/amrita_plugin_memory/rethinking/schedule.py b/amrita_plugin_memory/rethinking/schedule.py
index f084105..3398ffe 100644
--- a/amrita_plugin_memory/rethinking/schedule.py
+++ b/amrita_plugin_memory/rethinking/schedule.py
@@ -15,7 +15,7 @@
import time
from collections.abc import Sequence
from dataclasses import dataclass, field
-from datetime import datetime, timedelta, timezone
+from datetime import datetime, timedelta, timezone, tzinfo
from typing import Any
from zoneinfo import ZoneInfo
@@ -37,7 +37,7 @@ def _cell(wd: int, slot: int) -> int:
return wd * SLOTS_PER_DAY + (slot % SLOTS_PER_DAY)
-def slot_of(ts: float, tz: ZoneInfo) -> tuple[int, int]:
+def slot_of(ts: float, tz: tzinfo) -> tuple[int, int]:
"""把 unix 时间戳换成 (星期几 0=周一, 槽位 0..11)。"""
dt = datetime.fromtimestamp(ts, tz)
wd = dt.weekday()
@@ -199,6 +199,7 @@ class IdlePlanner:
def __init__(self, cfg: Any) -> None:
self._cfg = cfg
+ self.tz: tzinfo
tz_name = getattr(cfg, "idle_timezone", "Asia/Shanghai")
try:
self.tz = ZoneInfo(tz_name)
From 5ae37248a71b0ece64967f4f8e45c43d4701e818 Mon Sep 17 00:00:00 2001
From: AmriaLin <325560502+AmriaLin@users.noreply.github.com>
Date: Tue, 20 Oct 2026 06:42:27 +0000
Subject: [PATCH 4/6] =?UTF-8?q?chore:=20=E7=A7=BB=E9=99=A4=20tests=20?=
=?UTF-8?q?=E7=9B=AE=E5=BD=95=E4=B8=8E=20pytest=20=E4=BE=9D=E8=B5=96?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
- 删掉 tests/test_schedule.py，与不补单测的约定对齐
- pyproject.toml 去掉 pytest dev 依赖
- AGENTS.md 明确本仓库不建 tests/ 目录
---
AGENTS.md | 2 +-
pyproject.toml | 1 -
tests/test_schedule.py | 161 -----------------------------------------
3 files changed, 1 insertion(+), 163 deletions(-)
delete mode 100644 tests/test_schedule.py
diff --git a/AGENTS.md b/AGENTS.md
index a74f9c8..546bbd3 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -49,7 +49,7 @@ python -m py_compile amrita_plugin_memory/**/*.py
4. 手动跑一遍改动的命令（`/memory`、`ambot memory`）或工具调用，并在 PR 里说明验证方式与结果
纯逻辑尽量写成可独立执行的形式，方便用 `python -c` 或临时脚本验证；不要把临时脚本提交进仓库。
-`tests/` 下既有的 `test_schedule.py` 是需要 nonebot 初始化才能跑的历史文件，不作为改动门槛。
+本仓库无 `tests/` 目录，也不要新建。纯逻辑写成可独立执行的形式，用临时脚本验证后不提交。
## 工具注册的三条路径
diff --git a/pyproject.toml b/pyproject.toml
index 0043cdb..c53420a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -37,7 +37,6 @@ module_name = "nonebot.adapters.onebot.v11"
[tool.uv]
dev-dependencies = [
"ruff>=0.12.8",
- "pytest>=8.4.1",
# `ambot memory` 依赖 0.2.1 起的 entry point extras（[full_load]）支持
"ambot-inlinectl>=0.2.1",
]
diff --git a/tests/test_schedule.py b/tests/test_schedule.py
deleted file mode 100644
index 3e538b4..0000000
--- a/tests/test_schedule.py
+++ /dev/null
@@ -1,161 +0,0 @@
-"""DMN 空闲窗口调度的纯函数与规划器单测。
-
-只测 schedule.py，不加载 nonebot，便于快速回归。
-"""
-
-from __future__ import annotations
-
-from datetime import datetime, timedelta
-from types import SimpleNamespace as NS
-from zoneinfo import ZoneInfo
-
-from amrita_plugin_memory.rethinking.schedule import (
- CELLS,
- SLOTS_PER_DAY,
- IdlePlanner,
- _runs,
- bisect_free_minutes,
- feasible_minutes,
- free_gaps,
- longest_free,
- merge_noise,
- slot_of,
-)
-
-TZ = ZoneInfo("Asia/Shanghai")
-
-
-def make_cfg(**over) -> NS:
- base = dict(
- idle_timezone="Asia/Shanghai",
- idle_noise_gap_minutes=15,
- idle_fine_tau_days=28.0,
- idle_coarse_tau_days=14.0,
- idle_shrink_k=4.0,
- idle_cold_pct_max=0.05,
- idle_cold_abs_max=1.0,
- idle_min_sample_days_fine=6,
- idle_min_sample_days_mid=3,
- idle_target_hour=3,
- idle_base_slots=3,
- idle_max_slots=6,
- idle_penalty_multiplier=1.5,
- idle_max_relax_steps=3,
- idle_buffer_minutes=15,
- idle_run_estimate_minutes=10,
- idle_fallback_minutes=120,
- )
- base.update(over)
- return NS(**base)
-
-
-def seed_daytime(planner: IdlePlanner, days: int = 14) -> None:
- """灌入白天活跃、夜间安静的历史。"""
- base = datetime(2026, 10, 5, tzinfo=TZ)
- for d in range(days):
- for h in range(9, 22, 2):
- for mi in (0, 40):
- planner.observe((base + timedelta(days=d, hours=h, minutes=mi)).timestamp())
-
-
-def test_merge_noise_collapses_bursts():
- assert merge_noise([0, 60, 600, 1000], 900) == [0, 1000]
-
-
-def test_free_gaps_and_longest():
- assert free_gaps([100, 200], 0, 300) == [(0, 100), (100, 200), (200, 300)]
- assert longest_free([100, 200], 0, 300) == (0, 100)
- assert free_gaps([], 0, 0) == []
-
-
-def test_feasible_is_monotonic():
- ts = [3600 * 2]
- for minutes in range(0, 240, 10):
- if feasible_minutes(ts, 0, 3600 * 4, minutes):
- assert feasible_minutes(ts, 0, 3600 * 4, max(0, minutes - 10))
-
-
-def test_bisect_matches_bruteforce():
- ts = [3600 * 2]
- got = bisect_free_minutes(ts, 0, 3600 * 4)
- brute = max(m for m in range(0, 241) if feasible_minutes(ts, 0, 3600 * 4, m))
- assert got == brute == 120
-
-
-def test_ring_runs_wrap_around():
- assert _runs([True] * 24, 12) == [(0, 12)]
- mask = [True, True] + [False] * 20 + [True, True]
- assert dict(_runs(mask, 24))[22] == 4
-
-
-def test_slot_of_two_hour_grid():
- ts = datetime(2026, 10, 19, 23, 30, tzinfo=TZ).timestamp()
- wd, slot = slot_of(ts, TZ)
- assert wd == 0 and slot == 11
-
-
-def test_cold_start_falls_back_to_target_hour():
- planner = IdlePlanner(make_cfg())
- plan = planner.plan(1, datetime(2026, 10, 19, 9, 0, tzinfo=TZ))
- assert plan.tier == "fallback"
- assert plan.forced
- assert plan.start.hour == 3
-
-
-def test_plan_lands_in_night_and_respects_capacity():
- planner = IdlePlanner(make_cfg())
- seed_daytime(planner)
- now = datetime(2026, 10, 19, 9, 0, tzinfo=TZ)
- plan = planner.plan(1, now)
- assert not plan.forced
- assert plan.start >= now
- assert plan.end > plan.start
- assert plan.end.hour <= 9
-
-
-def test_penalty_raises_requirement_but_stays_in_window():
- planner = IdlePlanner(make_cfg())
- seed_daytime(planner)
- now = datetime(2026, 10, 19, 9, 0, tzinfo=TZ)
- plans = [planner.plan(p, now) for p in (1, 3, 8)]
- assert all(not p.forced for p in plans)
- assert all(p.start.hour == 3 for p in plans)
-
-
-def test_no_deadlock_when_fully_active():
- """用户几乎全天活跃时也必须给出窗口，不能返回空。"""
- planner = IdlePlanner(make_cfg())
- base = datetime(2026, 10, 5, tzinfo=TZ)
- for d in range(21):
- for s in range(SLOTS_PER_DAY):
- planner.observe((base + timedelta(days=d, hours=s * 2)).timestamp())
- plan = planner.plan(6, datetime(2026, 10, 19, 9, 0, tzinfo=TZ))
- assert plan.start is not None
- assert plan.forced
-
-
-def test_state_roundtrip():
- planner = IdlePlanner(make_cfg())
- seed_daytime(planner, days=5)
- dumped = planner.dump()
- assert len(dumped["idle_planner"]["fine"]) == CELLS
- other = IdlePlanner(make_cfg())
- other.load(dumped)
- assert other.state.seen_days == planner.state.seen_days
- assert abs(sum(other.state.coarse) - sum(planner.state.coarse)) < 1e-2
-
-
-def test_decay_reduces_history():
- planner = IdlePlanner(make_cfg())
- seed_daytime(planner, days=7)
- before = sum(planner.state.coarse)
- planner.observe(datetime(2026, 11, 30, tzinfo=TZ).timestamp())
- assert sum(planner.state.coarse) < before
-
-
-def test_sparse_history_still_plans():
- planner = IdlePlanner(make_cfg())
- planner.observe(datetime(2026, 10, 12, 15, 0, tzinfo=TZ).timestamp())
- plan = planner.plan(1, datetime(2026, 10, 19, 9, 0, tzinfo=TZ))
- assert plan.start is not None
- assert plan.length_minutes >= 1
From d6042b154e321ef9f1ae9d608254f30c80488ae5 Mon Sep 17 00:00:00 2001
From: AmriaLin <325560502+AmriaLin@users.noreply.github.com>
Date: Tue, 20 Oct 2026 07:28:51 +0000
Subject: [PATCH 5/6] =?UTF-8?q?fix(dmn):=20=E5=A4=84=E7=90=86=20Sourcery?=
=?UTF-8?q?=20=E8=AF=84=E5=AE=A1=E7=9A=84=2022=20=E9=A1=B9=E9=97=AE?=
=?UTF-8?q?=E9=A2=98?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
调度正确性
- 两天拼接的冷槽不再按环形相连，避免把隔天两组冷槽误并成一段
- 惩罚指数先封顶再取整，溢出时退到上限而不是抛异常
- 冷段选择改为时间感知：优先尚未到来的窗口，全等权重时按最近时间取
- _refine 尾部 buffer 不计入回报窗口；段内放不下一次运行就换段而非硬报
- 跨午夜扫描时相邻星期几独立挑分层，样本不足的天不会以 0 计数当冷槽
- active_days_of 只看近期，换作息后旧日期不再撑着 fine/mid 层
- _shrink 先各自归一化再混合，细层不再被全局层绝对计数压过
- 时区解析收敛成 resolve_timezone，调度与会话侧共用，回退时如实回报
状态与并发
- _save_state 加锁，并发写不会让旧快照盖掉新观测
- 写入前保留本实例不认识的字段，老版本实例不会再抹掉热度图
- 旧调度模式下不写 idle_planner，升级迁移不再误判已完成
- 仅启用空闲窗口调度时才清零旧语义 penalty
- 上轮运行时间兼容 0.7.1 的 last_run_time
- 热度图记录时区与目标用户，任一变就重建
- 收尾只在惩罚未被新消息改动时才清零
生命周期
- 用户新消息到来时掐断正在跑的周期，旧上下文不再写回记忆
- 节流期内仍顺延待执行任务，旧任务不会撞进新的活动窗口
- 首条消息不再绕过冷启动保底窗口
- 状态写入失败改为记日志，不再静默丢退避
工具
- iter_stop 校验本轮是否取过上下文，空转周期标记无效且不清惩罚
- get_context 的向量库操作加超时，超时标 stale 不阻塞循环
---
AGENTS.md | 1 +
amrita_plugin_memory/rethinking/_state.py | 32 ++
amrita_plugin_memory/rethinking/runner.py | 183 +++++++++--
amrita_plugin_memory/rethinking/schedule.py | 325 ++++++++++++++------
amrita_plugin_memory/rethinking/tools.py | 37 ++-
5 files changed, 446 insertions(+), 132 deletions(-)
diff --git a/AGENTS.md b/AGENTS.md
index 546bbd3..1ce3dda 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -21,6 +21,7 @@
| `permissions.py` | 会话侧读取权限校验 |
| `profile_store.py` | 画像路径与 Markdown 解析（纯函数，不依赖 nonebot 运行时） |
| `rethinking/` | DMN 循环：runner、tools、schemas、knowledge、schedule |
+| `rethinking/schedule.py` | 空闲窗口规划（纯数据 + 纯函数，`resolve_timezone` 是全局唯一的时区解析入口） |
新增纯逻辑优先放进不依赖 nonebot 的模块（如 `profile_store.py`、`rethinking/schedule.py`），运行器只做编排。
diff --git a/amrita_plugin_memory/rethinking/_state.py b/amrita_plugin_memory/rethinking/_state.py
index c44c3ca..1c6b36e 100644
--- a/amrita_plugin_memory/rethinking/_state.py
+++ b/amrita_plugin_memory/rethinking/_state.py
@@ -21,6 +21,10 @@
# 上一轮 iter_stop 的摘要与结束时间 — 供 subconscious_get_context 读取
_last_iter_summary: str = ""
_last_run_at: float = 0.0
+# 本轮是否已调过 subconscious_get_context — iter_stop 据此判定周期是否有效
+_context_fetched: bool = False
+# 本轮是否算一次有效周期 — 空转的周期不重置惩罚
+_cycle_valid: bool = True
# 隔离的工具管理器 — 所有 @on_tools 装饰器通过 bound_to 注册到这里，不污染全局 ToolsManager
_SUBCONSCIOUS_TOOLS = MultiToolsManager()
@@ -103,6 +107,34 @@ def set_last_iter_summary(text: str) -> None:
_last_iter_summary = text
+def mark_context_fetched() -> None:
+ global _context_fetched
+ _context_fetched = True
+
+
+def is_context_fetched() -> bool:
+ return _context_fetched
+
+
+def reset_context_fetched() -> None:
+ global _context_fetched
+ _context_fetched = False
+
+
+def mark_cycle_invalid() -> None:
+ global _cycle_valid
+ _cycle_valid = False
+
+
+def is_cycle_valid() -> bool:
+ return _cycle_valid
+
+
+def reset_cycle_valid() -> None:
+ global _cycle_valid
+ _cycle_valid = True
+
+
def get_last_run_at() -> float:
return _last_run_at
diff --git a/amrita_plugin_memory/rethinking/runner.py b/amrita_plugin_memory/rethinking/runner.py
index a8f0308..818469a 100644
--- a/amrita_plugin_memory/rethinking/runner.py
+++ b/amrita_plugin_memory/rethinking/runner.py
@@ -88,6 +88,13 @@ def __init__(self, config: SubconsciousConfig) -> None:
# 重排节流：一分钟内重复触发只更新惩罚与热度，不重复规划
self._last_replan_at: float = 0.0
self._replan_throttle: float = _REPLAN_THROTTLE_SECONDS
+ # 状态写入串行化：并发 save 先后到达时，旧快照不能盖掉新快照
+ self._state_lock = asyncio.Lock()
+ # 正在跑的周期任务，用户新消息到来时可以掐断
+ self._cycle_task: asyncio.Task | None = None
+ # 本轮是否已取过上下文，供 iter_stop 校验
+ _state.reset_context_fetched()
+ _state.reset_cycle_valid()
@property
def is_running(self) -> bool:
@@ -139,42 +146,92 @@ async def cancel_and_reschedule(self, chat_ts: float | None = None) -> None:
然后按新的冷段门槛重新选窗口。不 dump/恢复上下文——
用户聊完天后有新内容，旧推理上下文应直接丢弃。
"""
- # 热度图观测：每条消息都算，加深该时间窗口的热点
+ # 先记下历史是否存在：首条消息会把热度图从空变成非空，
+ # 若观测后再判断，冷启动就会被自己这条消息绕过
+ had_history = True
with contextlib.suppress(Exception):
+ had_history = self._planner.has_history()
self._planner.observe(chat_ts)
penalty = _state.increment_penalty()
+
+ # 用户在聊天，正在跑的那一轮已经过时了，掐掉避免它把旧结论写回记忆
+ self._abort_running_cycle()
+
now_mono = time.monotonic()
if now_mono - self._last_replan_at < self._replan_throttle:
- # 同窗口内的后续消息不重排，避免每条消息都扫一遍热度图
- with contextlib.suppress(Exception):
- await self._save_state()
+ # 同窗口内的后续消息不重排，但仍把待执行任务往后挪，
+ # 否则旧任务会在用户刚开始的这轮活动里跑起来
+ self._defer_pending_job(penalty, had_history)
+ await self._save_state_logged()
return
self._last_replan_at = now_mono
with contextlib.suppress(Exception):
scheduler.remove_job(self._job_id)
- if self._config.use_idle_window_planner:
- plan = self._planner.plan(penalty)
- logger.info(
- f"[Subconscious] Rescheduled: penalty=#{penalty}, {describe(plan)}"
- )
- delay_seconds = int(
- max(60.0, (plan.start - datetime.now(timezone.utc)).total_seconds())
- )
- else:
- base = self._config.rethink_base_delay_minutes
+
+ delay_seconds = self._compute_delay(penalty, had_history)
+ self._schedule_once(delay_seconds)
+ # 惩罚计数与热度图落盘，避免重启后丢掉已累积的退避
+ await self._save_state_logged()
+
+ def _compute_delay(self, penalty: int, had_history: bool) -> int:
+ """算下一次运行要等多久（秒）。规划失败时退到保底窗口。"""
+ base = self._config.rethink_base_delay_minutes
+ cap = self._config.rethink_max_delay_minutes
+ if not self._config.use_idle_window_planner:
multiplier = self._config.rethink_penalty_multiplier
- cap = self._config.rethink_max_delay_minutes
- raw = base * (multiplier ** (penalty - 1))
+ raw = base * (multiplier ** min(max(penalty - 1, 0), 32))
delay_minutes = min(raw, cap)
logger.info(
f"[Subconscious] Rescheduled(legacy): penalty=#{penalty}, "
- f"delay={delay_minutes:.0f}min (raw={raw:.0f}min, cap={cap}min)"
+ f"delay={delay_minutes:.0f}min (cap={cap}min)"
)
- delay_seconds = int(delay_minutes * 60)
- self._schedule_once(delay_seconds)
- # 惩罚计数与热度图落盘，避免重启后丢掉已累积的退避
- with contextlib.suppress(Exception):
+ return int(delay_minutes * 60)
+
+ try:
+ if had_history:
+ plan = self._planner.plan(penalty)
+ else:
+ # 冷启动只攒到一条样本，不足以判断作息，先走保底窗口
+ plan = self._planner.fallback_plan(penalty)
+ except Exception as e:
+ logger.opt(exception=e, colors=True, raw=True).error(
+ f"[Subconscious] Plan failed, fall back to fixed delay: {e}"
+ )
+ return int(min(base, cap) * 60)
+ logger.info(f"[Subconscious] Rescheduled: penalty=#{penalty}, {describe(plan)}")
+ return int(max(60.0, (plan.start - datetime.now(timezone.utc)).total_seconds()))
+
+ def _defer_pending_job(self, penalty: int, had_history: bool) -> None:
+ """节流期内不重新规划，只把待执行任务顺延到新的候选窗口。"""
+ try:
+ job = scheduler.get_job(self._job_id)
+ if job is None:
+ # 本就没有待执行任务（可能在运行中），补一个
+ self._schedule_once(self._compute_delay(penalty, had_history))
+ return
+ delay = self._compute_delay(penalty, had_history)
+ run_date = datetime.now(timezone.utc) + timedelta(seconds=delay)
+ if job.next_run_time is not None and job.next_run_time < run_date:
+ scheduler.reschedule_job(
+ self._job_id, trigger="date", run_date=run_date
+ )
+ except Exception as e:
+ logger.warning(f"[Subconscious] Defer pending job failed: {e}")
+
+ def _abort_running_cycle(self) -> None:
+ """用户发了新消息，掐断正在跑的那一轮。"""
+ task = self._cycle_task
+ if task is None or task.done():
+ return
+ logger.info("[Subconscious] New chat arrived, aborting in-flight cycle")
+ task.cancel()
+
+ async def _save_state_logged(self) -> None:
+ """保存状态，失败要出声：写盘失败会让重启后的退避凭空消失。"""
+ try:
await self._save_state()
+ except Exception as e:
+ logger.warning(f"[Subconscious] Save state failed: {e}")
def _schedule_once(self, delay_seconds: int) -> None:
run_date = datetime.now(timezone.utc) + timedelta(seconds=delay_seconds)
@@ -202,6 +259,8 @@ async def _run(self) -> None:
logger.warning("[Subconscious] Already running, skip")
return
self._is_running = True
+ self._cycle_task = asyncio.current_task()
+ self._context_fetched = False
try:
# 窗口已实际占用：记一次观测，避免下一轮重复选同一段
with contextlib.suppress(Exception):
@@ -246,6 +305,11 @@ async def _run_cycle(self) -> None:
logger.info("[Subconscious] Cycle start")
chat_obj.begin()
await chat_obj
+ except asyncio.CancelledError:
+ # 用户发来新消息把本轮掐了：旧上下文作废，不写摘要也不动惩罚
+ logger.info("[Subconscious] Cycle aborted by new user message")
+ self._chat_obj = None
+ raise
except Exception as e:
logger.opt(exception=e, colors=True, raw=True).exception(
f"[Subconscious] Error: {e}"
@@ -280,7 +344,10 @@ async def _post_process(self, chat_obj: ChatObject) -> None:
self._total_runs += 1
_state.set_last_run_at(datetime.now(timezone.utc).timestamp())
- # 先持久化，任一失败则不重置惩罚（保留重试机会）
+ # 先持久化，任一失败则不重置惩罚（保留重试机会）。
+ # 记下快照时的计数：保存期间用户可能又发消息把惩罚顶上去，
+ # 那部分增量不能被这次 reset 抹掉。
+ penalty_before = _state.get_penalty_count()
save_ok = True
try:
await self._save_state()
@@ -288,7 +355,20 @@ async def _post_process(self, chat_obj: ChatObject) -> None:
logger.warning(f"[Subconscious] Save state failed: {e}")
save_ok = False
- if save_ok:
+ if not save_ok:
+ logger.warning(
+ f"[Subconscious] Cycle done but save failed, "
+ f"penalty NOT reset (penalty={_state.get_penalty_count()})"
+ )
+ elif not _state.is_cycle_valid():
+ # 空转周期（没取上下文就收工）不清惩罚，保留压力等下一轮
+ logger.warning("[Subconscious] Cycle invalid, penalty kept")
+ elif _state.get_penalty_count() != penalty_before:
+ # 保存期间用户又发了消息，新增的那部分惩罚不能被这次收尾抹掉
+ logger.info(
+ "[Subconscious] Penalty kept: new chat arrived during the cycle"
+ )
+ else:
_state.reset_penalty()
# 惩罚清零也要落盘，否则重启会加载旧退避
try:
@@ -299,11 +379,6 @@ async def _post_process(self, chat_obj: ChatObject) -> None:
f"[Subconscious] Cycle done (total_runs={self._total_runs}), "
f"penalty reset, waiting for next user chat"
)
- else:
- logger.warning(
- f"[Subconscious] Cycle done but save failed, "
- f"penalty NOT reset (penalty={_state.get_penalty_count()})"
- )
async def _update_usage(self, chat_obj: ChatObject) -> None:
"""更新全局 InsightsModel 与目标用户的 usage，复用 Bot 的 add_usage 逻辑。
@@ -403,11 +478,16 @@ async def _load_state(self) -> None:
if isinstance(suggestions, list):
_state.set_knowledge_suggestions(suggestions)
_state.set_last_iter_summary(str(payload.get("last_iter_summary", "")))
- _state.set_last_run_at(float(payload.get("last_run_ts", 0.0) or 0.0))
+ _state.set_last_run_at(self._read_last_run_ts(payload))
self._planner.load(payload)
# 旧 penalty 语义是"被推迟的次数"，与新语义（要求的冷段长度）不可换算，
- # 升级后清零一次，避免带着旧值把门槛顶到上限
- if "idle_planner" not in payload and _state.get_penalty_count() > 0:
+ # 升级后清零一次，避免带着旧值把门槛顶到上限。
+ # 仍走旧调度时数值是自洽的，别动它。
+ if (
+ self._config.use_idle_window_planner
+ and "idle_planner" not in payload
+ and _state.get_penalty_count() > 0
+ ):
logger.info(
"[Subconscious] Migrate: reset legacy penalty "
f"#{_state.get_penalty_count()} (old delay-based semantics)"
@@ -416,6 +496,23 @@ async def _load_state(self) -> None:
except Exception as e:
logger.warning(f"[Subconscious] Load state failed: {e}")
+ @staticmethod
+ def _read_last_run_ts(payload: dict[str, Any]) -> float:
+ """取上轮运行时间，兼容 0.7.1 只写 last_run_time 的旧快照。"""
+ raw = payload.get("last_run_ts")
+ try:
+ if raw is not None:
+ return float(raw or 0.0)
+ except (TypeError, ValueError):
+ pass
+ legacy = payload.get("last_run_time")
+ if isinstance(legacy, str) and legacy:
+ try:
+ return datetime.fromisoformat(legacy).timestamp()
+ except ValueError:
+ return 0.0
+ return 0.0
+
async def _migrate_legacy_state(self) -> dict[str, Any]:
"""迁移旧版本挪用的 Bot Memory 表元状态（extra_prompt/memory_json）。"""
payload: dict[str, Any] = {}
@@ -448,7 +545,15 @@ async def _migrate_legacy_state(self) -> dict[str, Any]:
return payload
async def _save_state(self) -> None:
- """保存元状态到 SubconsciousState 表。"""
+ """保存元状态到 SubconsciousState 表。
+
+ 加锁串行化：并发调用时若旧快照后提交，会把新的热度与惩罚盖掉。
+ """
+ async with self._state_lock:
+ await self._save_state_locked()
+
+ async def _save_state_locked(self) -> None:
+ """真正写盘的部分，调用方需已持有 _state_lock。"""
payload: dict[str, Any] = {
"total_runs": self._total_runs,
"last_run_time": datetime.now(timezone.utc).isoformat(),
@@ -459,7 +564,19 @@ async def _save_state(self) -> None:
"pending_messages": _state.get_pending(),
"knowledge_suggestions": _state.get_knowledge_suggestions(),
}
- payload.update(self._planner.dump())
+ # 只有启用空闲窗口调度时才写热度图：旧调度模式下写进去，
+ # 会让下次升级的迁移判断误以为已是新格式，跳过旧 penalty 清零
+ if self._config.use_idle_window_planner:
+ payload.update(self._planner.dump())
+ # 整行替换前先补齐老版本实例写过、本实例不认识的字段，
+ # 否则旧实例一存盘，热度图累积就被抹成"从没规划过"
+ try:
+ existing = await self._read_state_payload(_STATE_UID)
+ except Exception:
+ existing = {}
+ for key, value in existing.items():
+ if key not in payload:
+ payload[key] = value
await self._write_state_payload(_STATE_UID, payload)
async def _load_prompt(self) -> str:
diff --git a/amrita_plugin_memory/rethinking/schedule.py b/amrita_plugin_memory/rethinking/schedule.py
index 3398ffe..14f18ac 100644
--- a/amrita_plugin_memory/rethinking/schedule.py
+++ b/amrita_plugin_memory/rethinking/schedule.py
@@ -12,6 +12,7 @@
from __future__ import annotations
+import logging
import time
from collections.abc import Sequence
from dataclasses import dataclass, field
@@ -30,9 +31,24 @@
#: 活跃日期登记上限
_MAX_SEEN_DAYS = 240
+_LOG = logging.getLogger(__name__)
+
_WD_NAMES = ("周一", "周二", "周三", "周四", "周五", "周六", "周日")
+def resolve_timezone(name: str | None) -> tuple[tzinfo, str, bool]:
+ """把配置里的时区名解析成 tzinfo。
+
+ 返回 (tz, 实际生效的名字, 是否回退)。解析失败回退 UTC+08:00，
+ 会话侧与调度侧共用这一份，避免两边报出不同的钟点。
+ """
+ raw = (name or "").strip() or "Asia/Shanghai"
+ try:
+ return ZoneInfo(raw), raw, False
+ except Exception:
+ return timezone(timedelta(hours=8)), "UTC+08:00", True
+
+
def _cell(wd: int, slot: int) -> int:
return wd * SLOTS_PER_DAY + (slot % SLOTS_PER_DAY)
@@ -60,7 +76,9 @@ def merge_noise(ts: Sequence[float], gap_seconds: float) -> list[float]:
return out
-def free_gaps(ts: Sequence[float], start: float, end: float) -> list[tuple[float, float]]:
+def free_gaps(
+ ts: Sequence[float], start: float, end: float
+) -> list[tuple[float, float]]:
"""返回 [start, end) 内的全部无消息区间。"""
if end <= start:
return []
@@ -69,9 +87,7 @@ def free_gaps(ts: Sequence[float], start: float, end: float) -> list[tuple[float
return [(pts[i], pts[i + 1]) for i in range(len(pts) - 1)]
-def longest_free(
- ts: Sequence[float], start: float, end: float
-) -> tuple[float, float]:
+def longest_free(ts: Sequence[float], start: float, end: float) -> tuple[float, float]:
"""[start, end) 内最长的无消息区间。"""
gaps = free_gaps(ts, start, end)
if not gaps:
@@ -113,15 +129,34 @@ def _normalize(values: Sequence[float]) -> list[float]:
return [v / total for v in values]
-def _runs(mask: Sequence[bool], limit: int) -> list[tuple[int, int]]:
- """环形找连续 True 段，返回 (起点, 长度)，长度不超过 limit。"""
+def _runs(
+ mask: Sequence[bool], limit: int, circular: bool = True
+) -> list[tuple[int, int]]:
+ """找连续 True 段，返回 (起点, 长度)，长度不超过 limit。
+
+ circular=True 用于单日 12 槽（跨午夜是本日内的相邻），
+ 拼接两天的 24 槽必须传 False——否则首尾会被当成相邻，选出一周之外的窗。
+ """
n = len(mask)
if n == 0:
return []
if all(mask):
- return [(0, limit)]
+ return [(0, min(n, limit))] if not circular else [(0, limit)]
if not any(mask):
return []
+ if not circular:
+ out: list[tuple[int, int]] = []
+ i = 0
+ while i < n:
+ if not mask[i]:
+ i += 1
+ continue
+ j = i
+ while j < n and mask[j]:
+ j += 1
+ out.append((i, min(j - i, limit)))
+ i = j
+ return out
start = mask.index(False)
seq = list(mask[start:]) + list(mask[:start])
out: list[tuple[int, int]] = []
@@ -180,6 +215,9 @@ class PlannerState:
seen_days: list[str] = field(default_factory=list)
recent: list[float] = field(default_factory=list)
last_update: float = 0.0
+ #: 热度图对应的时区名与目标用户，任一变化都要重建，否则星期几/钟点对不上
+ tz_name: str = ""
+ uid: str = ""
def to_dict(self) -> dict[str, Any]:
return {
@@ -188,6 +226,8 @@ def to_dict(self) -> dict[str, Any]:
"seen_days": self.seen_days[-_MAX_SEEN_DAYS:],
"recent": self.recent[-_MAX_RECENT:],
"last_update": self.last_update,
+ "tz_name": self.tz_name,
+ "uid": self.uid,
}
@@ -199,12 +239,9 @@ class IdlePlanner:
def __init__(self, cfg: Any) -> None:
self._cfg = cfg
- self.tz: tzinfo
tz_name = getattr(cfg, "idle_timezone", "Asia/Shanghai")
- try:
- self.tz = ZoneInfo(tz_name)
- except Exception:
- self.tz = timezone(timedelta(hours=8))
+ self.tz, self.tz_name, self.tz_fallback = resolve_timezone(tz_name)
+ self.uid = str(getattr(cfg, "target_user_id", "") or "")
self.state = PlannerState()
self.noise_gap = getattr(cfg, "idle_noise_gap_minutes", 15) * 60
@@ -231,6 +268,19 @@ def load(self, payload: dict[str, Any]) -> None:
raw = payload.get("idle_planner") if isinstance(payload, dict) else None
if not isinstance(raw, dict):
return
+ saved_tz = str(raw.get("tz_name", "") or "")
+ saved_uid = str(raw.get("uid", "") or "")
+ # 时区或目标用户换了，旧星期几/钟点的计数就不能再用
+ if saved_tz and saved_tz != self.tz_name:
+ _LOG.info(
+ f"[Subconscious] Planner history dropped: tz {saved_tz} -> {self.tz_name}"
+ )
+ return
+ if saved_uid and self.uid and saved_uid != self.uid:
+ _LOG.info(
+ f"[Subconscious] Planner history dropped: uid {saved_uid} -> {self.uid}"
+ )
+ return
try:
fine = [float(v) for v in raw.get("fine", [])]
coarse = [float(v) for v in raw.get("coarse", [])]
@@ -245,6 +295,8 @@ def load(self, payload: dict[str, Any]) -> None:
if isinstance(recent, list):
self.state.recent = [float(v) for v in recent][-_MAX_RECENT:]
self.state.last_update = float(raw.get("last_update", 0.0))
+ self.state.tz_name = self.tz_name
+ self.state.uid = self.uid
except (TypeError, ValueError):
self.state = PlannerState()
@@ -278,6 +330,8 @@ def observe(self, ts: float | None = None) -> None:
self.state.seen_days.append(day_key)
if len(self.state.seen_days) > _MAX_SEEN_DAYS:
self.state.seen_days = self.state.seen_days[-_MAX_SEEN_DAYS:]
+ self.state.tz_name = self.tz_name
+ self.state.uid = self.uid
self.state.recent.append(now_ts)
if len(self.state.recent) > _MAX_RECENT:
self.state.recent = self.state.recent[-_MAX_RECENT:]
@@ -307,21 +361,36 @@ def note_window_taken(self, start_ts: float) -> None:
# ------------------------------------------------------------ 分层分值
def active_days_of(self, weekdays: Sequence[int]) -> int:
+ """统计近期活跃天数：超过粗层半衰期 4 倍的老日期不再计入。
+
+ 不设上限的话，用户换作息后旧日期会永久撑着 fine/mid 层，
+ 稀薄但"看起来够量"的样本反而比全局层更不可信。
+ """
+ horizon = max(1, int(self.tau_coarse * 4))
+ cutoff = datetime.now(self.tz) - timedelta(days=horizon)
cnt = 0
- keys = set(self.state.seen_days)
- for key in keys:
+ for key in set(self.state.seen_days):
try:
dt = datetime.strptime(key, "%Y-%m-%d")
except ValueError:
continue
+ if dt.replace(tzinfo=self.tz) < cutoff:
+ continue
if dt.weekday() in weekdays:
cnt += 1
return cnt
def _shrink(self, cells: Sequence[float], n_days: int) -> list[float]:
- alpha = n_days / (n_days + self.shrink_k) if (n_days + self.shrink_k) > 0 else 0.0
- base = self.state.coarse
- return [alpha * cells[i] + (1 - alpha) * base[i] for i in range(SLOTS_PER_DAY)]
+ """把细层分布按可信度向全局分布收缩。
+
+ 两层先各自归一化再混合：细层一天一次采样、全局层跨所有天累加，
+ 直接相加会让全局层凭绝对计数压过 alpha 想表达的信任度。
+ """
+ total = n_days + self.shrink_k
+ alpha = n_days / total if total > 0 else 0.0
+ base = _normalize(self.state.coarse)
+ cur = _normalize(cells)
+ return [alpha * cur[i] + (1 - alpha) * base[i] for i in range(SLOTS_PER_DAY)]
def tier_table(self, tier: str, wd: int) -> tuple[list[float], list[float], int]:
"""返回 (槽位权重, 槽位原始计数, 有效样本天数)。"""
@@ -352,9 +421,21 @@ def _tiers_for(self, wd: int) -> list[str]:
# ------------------------------------------------------------ 冷段
- def _cold_runs(
- self, tier: str, wd: int
- ) -> list[tuple[int, int, int]]:
+ def _tier_for_neighbor(self, tier: str, wd: int) -> str:
+ """跨午夜扫描时，给相邻星期几单独挑一档可用分层。
+
+ 直接沿用当天的 tier，会在相邻那天样本不足时拿 0 计数参与判定，
+ 那些槽位会被误判成冷槽。
+ """
+ if tier == "fine" and self.active_days_of([wd]) < self.min_weeks_fine:
+ return self._tier_for_neighbor("mid", wd)
+ if tier == "mid":
+ days = [0, 1, 2, 3, 4] if wd <= 4 else [5, 6]
+ if self.active_days_of(days) < self.min_weeks_mid:
+ return "coarse"
+ return tier
+
+ def _cold_runs(self, tier: str, wd: int) -> list[tuple[int, int, int]]:
"""返回 [(环形起点槽, 长度, 归属星期几)]，长度上限一天。"""
weights, counts, _ = self.tier_table(tier, wd)
pcts = _normalize(weights)
@@ -363,19 +444,28 @@ def _cold_runs(
for s in range(SLOTS_PER_DAY)
]
# 跨午夜：接上下一星期几的表，否则 22:00-08:00 的安静段会被切断
- w2, c2, _ = self.tier_table(tier, (wd + 1) % DAYS)
+ wd2 = (wd + 1) % DAYS
+ tier2 = self._tier_for_neighbor(tier, wd2)
+ w2, c2, _ = self.tier_table(tier2, wd2)
p2 = _normalize(w2)
mask2 = [
p2[s] <= self.cold_pct_max and c2[s] <= self.cold_abs_max
for s in range(SLOTS_PER_DAY)
]
+ # 拼接表首尾在真实时间上相隔整整一天，不能按环形再接回去
ring = mask + mask2
- return [(start, length, wd) for start, length in _runs(ring, SLOTS_PER_DAY)]
+ return [
+ (start, length, wd) for start, length in _runs(ring, SLOTS_PER_DAY, False)
+ ]
def _lightest_run(
- self, tier: str, wd: int, length_slots: int
+ self, tier: str, wd: int, length_slots: int, now: datetime
) -> tuple[int, int] | None:
- """环形滑动窗口找总权重最低的连续 length_slots 槽段。"""
+ """环形滑动窗口找总权重最低的连续 length_slots 槽段。
+
+ 权重相同时取最近一次将要到来的那段，避免全天打平时
+ 因为起点槽已过而把窗口推到下周。
+ """
weights, _, _ = self.tier_table(tier, wd)
pcts = _normalize(weights)
if sum(pcts) <= 1e-9:
@@ -383,11 +473,14 @@ def _lightest_run(
n = SLOTS_PER_DAY
ring = pcts + pcts
best: tuple[int, int] | None = None
- best_sum = float("inf")
+ best_key: tuple[float, float] | None = None
for start in range(n):
total = sum(ring[start : start + length_slots])
- if total < best_sum - 1e-12:
- best_sum = total
+ seg_start, _ = self._seg_bounds(start, length_slots, wd, now)
+ delay = max(0.0, (seg_start - now.astimezone(self.tz)).total_seconds())
+ key = (total, delay)
+ if best_key is None or key < best_key:
+ best_key = key
best = (start, length_slots)
return best
@@ -402,6 +495,19 @@ def _next_occurrence(self, wd: int, now: datetime) -> datetime:
cand += timedelta(days=DAYS)
return cand
+ def _seg_bounds(
+ self, start_slot: int, length_slots: int, wd: int, now: datetime
+ ) -> tuple[datetime, datetime]:
+ """冷段对应的绝对时间区间；已过去的按下一个同类星期几算。"""
+ day0 = self._next_occurrence(wd, now)
+ seg_start = day0 + timedelta(minutes=start_slot * SLOT_MINUTES)
+ seg_end = seg_start + timedelta(minutes=length_slots * SLOT_MINUTES)
+ now_local = now.astimezone(self.tz)
+ if seg_end <= now_local:
+ seg_start += timedelta(days=DAYS)
+ seg_end += timedelta(days=DAYS)
+ return seg_start, seg_end
+
def _refine(
self,
start_slot: int,
@@ -412,23 +518,23 @@ def _refine(
penalty: int,
relaxed: int,
forced: bool,
- ) -> Plan:
- """把冷段换成具体时间窗：段内找最长空闲，二分校核，扣 buffer。"""
- day0 = self._next_occurrence(wd, now)
- seg_start = day0 + timedelta(minutes=start_slot * SLOT_MINUTES)
- seg_end = seg_start + timedelta(minutes=length_slots * SLOT_MINUTES)
+ ) -> Plan | None:
+ """把冷段换成具体时间窗：段内找最长空闲，二分校核，扣 buffer。
+
+ 段内连一次运行都放不下时返回 None，交给调用方换下一段，
+ 而不是报出一个实际用不了的时间窗。
+ """
+ seg_start, seg_end = self._seg_bounds(start_slot, length_slots, wd, now)
now_local = now.astimezone(self.tz)
- if seg_end <= now_local:
- seg_start += timedelta(days=DAYS)
- seg_end += timedelta(days=DAYS)
if seg_start < now_local:
seg_start = now_local + timedelta(minutes=1)
+ if seg_end <= seg_start:
+ return None
# 只信原始时间戳：连发消息先合并，再取段内最长无消息区间
ts = merge_noise(self.state.recent, self.noise_gap)
s_ts, e_ts = seg_start.timestamp(), seg_end.timestamp()
f_start, f_end = longest_free(ts, s_ts, e_ts)
- # 二分只为确认这段空闲的上界（段内可能有更细的分布）
capacity = bisect_free_minutes(ts, s_ts, e_ts)
gap_minutes = int((f_end - f_start) // 60)
# 二分给出理论上界，实际可用不得超出真实无消息区间
@@ -445,22 +551,23 @@ def _refine(
start_ts = min(max(f_start, target.timestamp()), latest_start)
start_ts = max(start_ts, f_start, now.timestamp() + 60)
+ note = ""
if f_end - start_ts < need * 60:
- # 段内剩余空间放不下一次运行，标记但不放弃（保底语义）
- length_minutes = max(1, int((f_end - start_ts) // 60))
- note = "insufficient_free_space"
+ # 段内剩余空间放不下一次运行，退回段首再算一次
start_ts = f_start
- length_minutes = max(1, int((f_end - f_start) // 60))
- else:
- # 结束时间严格受空闲段与冷段双重约束
- length_minutes = int((f_end - start_ts) // 60)
- if free_minutes < length_minutes: # 保守取真实空闲长度
- length_minutes = free_minutes
- note = ""
+ note = "insufficient_free_space"
+ if f_end - start_ts < 60:
+ return None
start = datetime.fromtimestamp(start_ts, self.tz)
- end = min(start + timedelta(minutes=length_minutes),
- datetime.fromtimestamp(f_end, self.tz))
+ # 尾部 buffer 不纳入回报窗口，否则 end 会把安全余量也算成可用时间
+ usable_end = datetime.fromtimestamp(f_end, self.tz) - timedelta(
+ minutes=self.buffer_minutes
+ )
+ end = min(start + timedelta(minutes=free_minutes), usable_end)
+ length_minutes = int((end - start).total_seconds() // 60)
+ if length_minutes < 1:
+ return None
return Plan(
start=start,
end=end,
@@ -476,9 +583,7 @@ def _refine(
def _fallback(self, now: datetime, penalty: int, note: str) -> Plan:
local = now.astimezone(self.tz)
- target = local.replace(
- hour=self.target_hour, minute=0, second=0, microsecond=0
- )
+ target = local.replace(hour=self.target_hour, minute=0, second=0, microsecond=0)
if target <= local + timedelta(minutes=1):
target += timedelta(days=1)
return Plan(
@@ -491,11 +596,49 @@ def _fallback(self, now: datetime, penalty: int, note: str) -> Plan:
note=note,
)
+ def has_history(self) -> bool:
+ """热度图是否有可用样本。"""
+ return sum(self.state.coarse) > 1e-6
+
+ def fallback_plan(
+ self, penalty: int = 1, now: datetime | None = None, note: str = "no_history"
+ ) -> Plan:
+ """直接给出保底窗口，不做冷段扫描。"""
+ now = now or datetime.now(timezone.utc)
+ return self._fallback(now.astimezone(self.tz), penalty, note)
+
def plan(self, penalty: int = 1, now: datetime | None = None) -> Plan:
result = self._plan(penalty, now)
self._last_plan = result
return result
+ def _want_slots(self, penalty: int) -> int:
+ """按惩罚值算本次要求的冷段槽数，指数增长但先封顶再取整。"""
+ exp = max(0, penalty - 1)
+ try:
+ want = self.base_slots * (self.multiplier**exp)
+ except OverflowError:
+ # 惩罚值离谱时直接给上限，不要让一次计算失败把调度整条打断
+ return self.max_slots
+ if want != want: # NaN 保护
+ return self.base_slots
+ return max(1, min(round(want), self.max_slots))
+
+ def _pick_run(
+ self, runs: Sequence[tuple[int, int, int]], now: datetime
+ ) -> tuple[int, int, int] | None:
+ """在满足门槛的冷段里挑一段：优先最近的、尚未来临的、较长的。"""
+ best: tuple[int, int, int] | None = None
+ best_key: tuple[int, float, int] | None = None
+ for start_slot, length_slots, owner in runs:
+ seg_start, _ = self._seg_bounds(start_slot, length_slots, owner, now)
+ delay = max(0.0, (seg_start - now.astimezone(self.tz)).total_seconds())
+ key = (0 if delay > 0 else 1, delay, -length_slots)
+ if best_key is None or key < best_key:
+ best_key = key
+ best = (start_slot, length_slots, owner)
+ return best
+
def _plan(self, penalty: int = 1, now: datetime | None = None) -> Plan:
now = now or datetime.now(timezone.utc)
local = now.astimezone(self.tz)
@@ -504,62 +647,66 @@ def _plan(self, penalty: int = 1, now: datetime | None = None) -> Plan:
if sum(self.state.coarse) <= 1e-6:
return self._fallback(local, penalty, "no_history")
- want = self.base_slots * (self.multiplier ** max(0, penalty - 1))
- want = max(1, min(round(want), self.max_slots))
+ want = self._want_slots(penalty)
tiers = self._tiers_for(wd)
for step in range(self.max_relax + 1):
need_slots = max(1, want >> step)
for tier in tiers:
- runs = [
- r for r in self._cold_runs(tier, wd) if r[1] >= need_slots
- ]
- if not runs:
+ runs = [r for r in self._cold_runs(tier, wd) if r[1] >= need_slots]
+ picked = self._pick_run(runs, now)
+ if picked is None:
continue
- start_slot, length_slots, owner = max(
- runs, key=lambda r: (r[1], -r[0])
- )
- return self._refine(
- start_slot,
- length_slots,
- owner,
+ plan = self._refine(
+ picked[0],
+ picked[1],
+ picked[2],
now,
tier,
penalty,
step,
forced=False,
)
+ if plan is not None:
+ return plan
# 全部放宽仍不足：强制取最长冷段
- best: tuple[int, int, int, str] | None = None
for tier in tiers:
- for run in self._cold_runs(tier, wd):
- if best is None or run[1] > best[1]:
- best = (run[0], run[1], run[2], tier)
- if best is not None:
- return self._refine(
- best[0], best[1], best[2], now, best[3], penalty,
- self.max_relax, forced=True,
+ runs = sorted(self._cold_runs(tier, wd), key=lambda r: -r[1])
+ picked = self._pick_run(runs, now)
+ if picked is None:
+ continue
+ plan = self._refine(
+ picked[0],
+ picked[1],
+ picked[2],
+ now,
+ tier,
+ penalty,
+ self.max_relax,
+ forced=True,
)
- # 完全没有冷槽（用户几乎全天活跃）：退到热度最低的连续段
+ if plan is not None:
+ return plan
+
+ # 完全没有可用冷段（用户几乎全天活跃）：退到热度最低的连续段
for tier in tiers:
- floor_run = self._lightest_run(tier, wd, 1)
- if floor_run is not None:
- return self._refine(
- floor_run[0], floor_run[1], wd, now, tier, penalty,
- self.max_relax, forced=True,
- )
+ floor_run = self._lightest_run(tier, wd, 1, now)
+ if floor_run is None:
+ continue
+ plan = self._refine(
+ floor_run[0],
+ floor_run[1],
+ wd,
+ now,
+ tier,
+ penalty,
+ self.max_relax,
+ forced=True,
+ )
+ if plan is not None:
+ return plan
return self._fallback(local, penalty, "no_cold_segment")
- return self._refine(
- best[0],
- best[1],
- best[2],
- now,
- best[3],
- penalty,
- self.max_relax,
- forced=True,
- )
def describe(plan: Plan) -> str:
diff --git a/amrita_plugin_memory/rethinking/tools.py b/amrita_plugin_memory/rethinking/tools.py
index dc7a5e4..e026504 100644
--- a/amrita_plugin_memory/rethinking/tools.py
+++ b/amrita_plugin_memory/rethinking/tools.py
@@ -29,6 +29,7 @@
from ..vector import AsyncUserMemory, MemoryMetadata, get_db_conn
from . import _state
from .consts import DEFAULT_SEND_PROMPT, ensure_prompt_file, load_character_prompt
+from .schedule import resolve_timezone
from .schemas import (
DELETE_MEMORY_SCHEMA,
DUPLICATE_HELPER_SCHEMA,
@@ -64,6 +65,9 @@
_DUPLICATE_DEFAULT_LIMIT = 40
_DUPLICATE_MAX_LIMIT = 200
+#: get_context 里向量库操作的超时上限（秒），超时只标 stale 不阻塞循环
+_CONTEXT_DB_TIMEOUT = 8.0
+
_SUBCONSCIOUS_TOOLS = _state.get_tools_manager()
@@ -249,6 +253,15 @@ async def subconscious_list_memory(data: dict[str, Any]) -> str:
@on_tools(ITER_STOP_SCHEMA, strict=True, bound_to=_SUBCONSCIOUS_TOOLS)
async def subconscious_iter_stop(data: dict[str, Any]) -> str:
summary = str(data.get("summary", ""))
+ if not _state.is_context_fetched():
+ # 没取环境就收工：周期算无效，不写摘要。惩罚保持不清零，
+ # 下一次调度仍带着这轮攒的量，避免"空转一轮还被当成完成"
+ logger.warning("[Subconscious] iter_stop without get_context, cycle invalid")
+ _state.mark_cycle_invalid()
+ return _tools_err(
+ "本轮未调用 subconscious_get_context，周期无效。"
+ "请先取当前环境再决定是否整理记忆"
+ )
_state.set_last_iter_summary(summary)
logger.info(f"[Subconscious] iter_stop: summary={summary[:80]}")
return json.dumps(
@@ -594,13 +607,10 @@ async def subconscious_get_context(data: dict[str, Any]) -> str:
runner = _state.get_runner()
cfg = runner._config if runner is not None else None
- tz_name = getattr(cfg, "idle_timezone", "Asia/Shanghai") if cfg else "Asia/Shanghai"
- try:
- from zoneinfo import ZoneInfo
-
- tz = ZoneInfo(tz_name)
- except Exception:
- tz = timezone.utc
+ tz_cfg = getattr(cfg, "idle_timezone", "Asia/Shanghai") if cfg else "Asia/Shanghai"
+ # 与调度侧共用同一份解析，避免规划窗口与回报钟点分处两个时区
+ tz, tz_name, tz_fell_back = resolve_timezone(tz_cfg)
+ if tz_fell_back:
stale.append("timezone")
now = datetime.now(timezone.utc)
@@ -622,12 +632,17 @@ async def subconscious_get_context(data: dict[str, Any]) -> str:
memory_hint = ""
try:
ope = _make_operator()
- await ope.init()
+ await asyncio.wait_for(ope.init(), timeout=_CONTEXT_DB_TIMEOUT)
pid = _get_partition_id()
- total = await ope.count_user_notes(pid)
+ total = await asyncio.wait_for(
+ ope.count_user_notes(pid), timeout=_CONTEXT_DB_TIMEOUT
+ )
threshold = getattr(cfg, "memory_warn_threshold", 100) if cfg else 100
if total > threshold:
- result = await ope.get_all_notes(pid, include=["metadatas"])
+ result = await asyncio.wait_for(
+ ope.get_all_notes(pid, include=["metadatas"]),
+ timeout=_CONTEXT_DB_TIMEOUT,
+ )
low_count = sum(
1
for m in (result.get("metadatas") or [])
@@ -648,11 +663,13 @@ async def subconscious_get_context(data: dict[str, Any]) -> str:
except Exception:
stale.append("idle_plan")
+ _state.mark_context_fetched()
return json.dumps(
{
"status": "success",
"now": local.strftime("%Y-%m-%d %H:%M:%S"),
"timezone": tz_name,
+ "timezone_fallback": tz_fell_back,
"weekday": local.strftime("%A"),
"last_run_at": last_run_iso,
"minutes_since_last_run": gap_minutes,
From 8fb54bc67f2f48b5fb47dc8cc3ef4d027e151554 Mon Sep 17 00:00:00 2001
From: AmriaLin <325560502+AmriaLin@users.noreply.github.com>
Date: Tue, 20 Oct 2026 07:44:55 +0000
Subject: [PATCH 6/6] =?UTF-8?q?style(schedule):=20=E5=8E=BB=E6=8E=89?=
=?UTF-8?q?=E5=88=86=E8=8A=82=E6=A8=AA=E7=BA=BF=E6=B3=A8=E9=87=8A?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
按仓库约定不写 # ---- 小节名 这类分隔线，改由空行与函数拆分表达结构。
模块 docstring 里便于独立单测措辞同步调整（本仓库不建 tests 目录）。
---
amrita_plugin_memory/rethinking/schedule.py | 21 +--------------------
1 file changed, 1 insertion(+), 20 deletions(-)
diff --git a/amrita_plugin_memory/rethinking/schedule.py b/amrita_plugin_memory/rethinking/schedule.py
index 14f18ac..b1b7006 100644
--- a/amrita_plugin_memory/rethinking/schedule.py
+++ b/amrita_plugin_memory/rethinking/schedule.py
@@ -7,7 +7,7 @@
- 规划：环形找连续冷槽 -> 段内用原始时间戳做分钟级二分 -> 扣 buffer 得可用窗口。
- 保底：冷段不足门槛时逐次减半放宽，最后强制取最长冷段。
-本模块不依赖 nonebot，纯数据 + 纯函数，便于独立单测。
+本模块不依赖 nonebot，纯数据 + 纯函数，可脱离运行时单独验证。
"""
from __future__ import annotations
@@ -61,9 +61,6 @@ def slot_of(ts: float, tz: tzinfo) -> tuple[int, int]:
return wd, slot
-# ---------------------------------------------------------------- 纯函数工具
-
-
def merge_noise(ts: Sequence[float], gap_seconds: float) -> list[float]:
"""把一波连发的消息压成单点：间距小于 gap 的后来者直接并掉。"""
out: list[float] = []
@@ -173,9 +170,6 @@ def _runs(
return out
-# ---------------------------------------------------------------- 规划结果
-
-
@dataclass
class Plan:
"""一次调度规划的结果。"""
@@ -231,9 +225,6 @@ def to_dict(self) -> dict[str, Any]:
}
-# ---------------------------------------------------------------- 规划器
-
-
class IdlePlanner:
"""基于热度图的空闲窗口规划器。"""
@@ -262,8 +253,6 @@ def __init__(self, cfg: Any) -> None:
self.fallback_minutes = getattr(cfg, "idle_fallback_minutes", 120)
self._last_plan: Plan | None = None
- # ------------------------------------------------------------ 持久化
-
def load(self, payload: dict[str, Any]) -> None:
raw = payload.get("idle_planner") if isinstance(payload, dict) else None
if not isinstance(raw, dict):
@@ -303,8 +292,6 @@ def load(self, payload: dict[str, Any]) -> None:
def dump(self) -> dict[str, Any]:
return {"idle_planner": self.state.to_dict()}
- # ------------------------------------------------------------ 观测
-
def _decay(self, now_ts: float) -> None:
if self.state.last_update <= 0:
self.state.last_update = now_ts
@@ -358,8 +345,6 @@ def note_window_taken(self, start_ts: float) -> None:
"""窗口实际被占用后，把该时段计入一次观测（避免反复选同一段）。"""
self.observe(start_ts + 60)
- # ------------------------------------------------------------ 分层分值
-
def active_days_of(self, weekdays: Sequence[int]) -> int:
"""统计近期活跃天数：超过粗层半衰期 4 倍的老日期不再计入。
@@ -419,8 +404,6 @@ def _tiers_for(self, wd: int) -> list[str]:
tiers.append("coarse")
return tiers
- # ------------------------------------------------------------ 冷段
-
def _tier_for_neighbor(self, tier: str, wd: int) -> str:
"""跨午夜扫描时，给相邻星期几单独挑一档可用分层。
@@ -579,8 +562,6 @@ def _refine(
note=note,
)
- # ------------------------------------------------------------ 主流程
-
def _fallback(self, now: datetime, penalty: int, note: str) -> Plan:
local = now.astimezone(self.tz)
target = local.replace(hour=self.target_hour, minute=0, second=0, microsecond=0)