diff --git a/reme/components/job/base_job.py b/reme/components/job/base_job.py index 5133111d..26938748 100644 --- a/reme/components/job/base_job.py +++ b/reme/components/job/base_job.py @@ -1,5 +1,6 @@ """Base job component for sequential step execution.""" +import time from typing import TYPE_CHECKING from ..base_component import BaseComponent @@ -59,10 +60,27 @@ class BaseJob(BaseComponent): return [step_cls(**dict(params)) for step_cls, params in self.step_specs] def _record_call(self) -> None: - """Increment this job's application-lifetime call counter.""" + """Increment this job's application-lifetime call counter and mark it running. + + ``__job_last_run`` feeds the idle gate (``wait_for_idle_step``): metadata + is process-local, so a restart always leaves every job looking idle. + """ metadata = getattr(self.app_context, "metadata", None) if isinstance(metadata, dict): global_counter_inc(metadata, ["__job_counter", self.name]) + last_run = metadata.setdefault("__job_last_run", {}) + entry = dict(last_run.get(self.name) or {}) + entry.update(running=True, last_start=time.monotonic()) + last_run[self.name] = entry + + def _finish_call(self) -> None: + """Mark this job's run as finished for the idle gate.""" + metadata = getattr(self.app_context, "metadata", None) + if isinstance(metadata, dict): + last_run = metadata.setdefault("__job_last_run", {}) + entry = dict(last_run.get(self.name) or {}) + entry.update(running=False, last_end=time.monotonic()) + last_run[self.name] = entry async def __call__(self, **kwargs) -> Response: """Run all steps in order, capturing any failure into the response.""" @@ -76,4 +94,6 @@ class BaseJob(BaseComponent): self.logger.exception(f"Failed to execute job: {e}") context.response.success = False context.response.answer = str(e) + finally: + self._finish_call() return context.response diff --git a/reme/components/job/cron_job.py b/reme/components/job/cron_job.py index a72a7878..6a615a92 100644 --- a/reme/components/job/cron_job.py +++ b/reme/components/job/cron_job.py @@ -35,7 +35,7 @@ class CronJob(BackgroundJob): return max(0.0, (nxt - now).total_seconds()) async def _execute_steps(self) -> Response: - context = RuntimeContext(**self.kwargs) + context = RuntimeContext(stop_event=self._stop_event, **self.kwargs) for step in self._build_steps(): await step(context) return context.response @@ -46,11 +46,13 @@ class CronJob(BackgroundJob): await self._wait_or_stop(self._next_fire_delay()) if self._stop_event.is_set(): break + self._record_call() try: - self._record_call() await self._execute_steps() except Exception as exc: self.logger.exception(f"Cron job '{self.name}' failed: {exc}") + finally: + self._finish_call() response = Response() response.success = True return response diff --git a/reme/config/default.yaml b/reme/config/default.yaml index a5f61f5a..792a76a8 100644 --- a/reme/config/default.yaml +++ b/reme/config/default.yaml @@ -74,6 +74,30 @@ jobs: steps: - backend: optimize_index_step + proactive_refresh_cron: + backend: cron + cron: "0 3,9,15,21 * * *" # avoids 23:00 dream_cron and 02:00 optimize_index_cron + steps: + - backend: wait_for_idle_step + skip_key: proactive_skip + - backend: proactive_extract_step + file_catalog: proactive + agent_wrapper: proactive + scan_days: 2 + resource_lookback_days: 7 + max_resource_files: 20 + carry_forward_days: 14 + max_carry_forward_topics: 20 + llm_timeout_seconds: 300 + max_chars_per_file: 60000 + - backend: proactive_topics_step + known_threshold: 0.85 + known_threshold_calibrated_for: text-embedding-v4@1024 + min_push_confidence: 0.5 + max_topics: 10 + - backend: proactive_finish_step + file_catalog: proactive + auto_dream: backend: base description: "Auto-dream: scan today's day-index and daily notes, globally extract merged units/topics, integrate digest units, write interests.yaml, and persist the dream catalog." @@ -199,8 +223,17 @@ jobs: type: boolean description: "whether to include the raw YAML content in the response answer and metadata" default: true + horizon_days: + type: integer + description: "merge interests.yaml across this many recent days (1 = single-day legacy behaviour)" + default: 1 + min_confidence: + type: number + description: "minimum topic confidence to return (default 0.4 sits safely below the 0.5 fallback; v1 topics fall back to 0.5)" + default: 0.4 steps: - backend: proactive_step + min_confidence: 0.4 version: backend: base @@ -735,6 +768,7 @@ components: # base_url: ${EMBEDDING_BASE_URL:-https://dashscope.aliyuncs.com/compatible-mode/v1} # parameters: { } # + # embedding_store: # default: # backend: local @@ -767,6 +801,14 @@ components: tool_result_limit: 50000 model_config: max_retries: 1 + proactive: + backend: agentscope + as_llm: default + permission_mode: bypass + react_config: + max_iters: 5 + model_config: + max_retries: 1 claude_code: backend: claude_code model: ${CLAUDE_CODE_MODEL_NAME:-glm-5.1} @@ -802,6 +844,8 @@ components: backend: local dream: backend: local + proactive: + backend: local file_chunker: markdown: diff --git a/reme/schema/__init__.py b/reme/schema/__init__.py index e69e831b..e50fb705 100644 --- a/reme/schema/__init__.py +++ b/reme/schema/__init__.py @@ -26,9 +26,16 @@ from .dream import ( DreamTopic, DreamUnit, IntegrateOutcome, - ProactiveResult, TopicSelectionOutput, ) +from .proactive import ( + ProactiveExtractOutput, + ProactiveResult, + ProactiveState, + ProactiveStateFile, + ProactiveTopic, + TopicUpdate, +) from .emb_node import EmbNode from .file_chunk import FileChunk from .file_front_matter import FileFrontMatter @@ -73,12 +80,17 @@ __all__ = [ "PaperInfo", "PaperPick", "PaperPickList", + "ProactiveExtractOutput", "ProactiveResult", + "ProactiveState", + "ProactiveStateFile", + "ProactiveTopic", "Request", "Response", "StreamChunk", "TokenUsage", "TopicSelectionOutput", + "TopicUpdate", "TraverseGraph", "TraverseGraphEdge", "TraverseGraphNode", diff --git a/reme/schema/dream.py b/reme/schema/dream.py index e3398651..6eea81c6 100644 --- a/reme/schema/dream.py +++ b/reme/schema/dream.py @@ -6,6 +6,9 @@ from pydantic import BaseModel, Field from ..enumeration import DreamBucketEnum +# ProactiveResult moved to reme.schema.proactive; re-exported for import compatibility. +from .proactive import ProactiveResult # noqa: F401 # pylint: disable=unused-import + class DreamUnit(BaseModel): """One cross-file memory unit emitted by global extract.""" @@ -47,18 +50,6 @@ class TopicSelectionOutput(BaseModel): topics: list[DreamTopic] = Field(default_factory=list) -class ProactiveResult(BaseModel): - """Result of reading daily interest topics.""" - - date: str = "" - path: str = "" - topics: list[dict] = Field(default_factory=list) - content: str = "" - skipped: bool = False - error: str = "" - summary: str = "" - - class DreamState(BaseModel): """Shared state passed across the dream steps.""" diff --git a/reme/schema/proactive.py b/reme/schema/proactive.py new file mode 100644 index 00000000..7dc8d0d3 --- /dev/null +++ b/reme/schema/proactive.py @@ -0,0 +1,158 @@ +"""Proactive refresh schemas. + +Defines the topic model (v2), the LLM extract contract, the chain-shared +context state, and the ``daily/_proactive.yaml`` truth-source file model. +See ``PROACTIVE_SPEC.md`` sections F1/A2/A3 for the full contracts. +""" + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +TOPIC_KINDS = ("follow_up", "interest_extend") +UPDATE_ACTIONS = ("keep", "update", "resolve") + + +def clamp_confidence(value) -> float: + """Coerce confidence into [0, 1]; any conversion failure falls back to 0.5.""" + try: + return min(1.0, max(0.0, float(value))) + except (TypeError, ValueError): + return 0.5 + + +class ProactiveTopic(BaseModel): + """One proactive topic; every field has a default so v1 files parse seamlessly. + + Fallback rules (A2): invalid ``kind`` -> ``interest_extend``; unparseable + ``confidence`` -> 0.5. Missing ``id``, ``first_seen`` and + ``last_evidence_at`` are context-dependent and therefore resolved by the + loaders, not here. + """ + + id: str = "" + title: str = "" + reason: str = "" + kind: str = "interest_extend" + confidence: float = 0.5 + first_seen: str = "" + last_evidence_at: str = "" + evidence: str = "" + keywords: list[str] = Field(default_factory=list) + paths: list[str] = Field(default_factory=list) + + @field_validator("kind", mode="before") + @classmethod + def _fallback_kind(cls, value): + text = str(value or "").strip() + return text if text in TOPIC_KINDS else "interest_extend" + + @field_validator("confidence", mode="before") + @classmethod + def _fallback_confidence(cls, value): + return clamp_confidence(value) + + @field_validator("keywords", "paths", mode="before") + @classmethod + def _clean_str_list(cls, value): + if not isinstance(value, list): + return [] + return [str(item).strip() for item in value if str(item).strip()] + + +class TopicUpdate(BaseModel): + """LLM verdict for one carried-forward topic (F2.3).""" + + id: str = "" + action: str = "keep" + evidence: str = "" + reason: str = "" + confidence: float | None = None + + @field_validator("action", mode="before") + @classmethod + def _fallback_action(cls, value): + text = str(value or "").strip() + return text if text in UPDATE_ACTIONS else "keep" + + @field_validator("confidence", mode="before") + @classmethod + def _clean_confidence(cls, value): + if value is None or str(value).strip() == "": + return None + return clamp_confidence(value) + + +class ProactiveExtractOutput(BaseModel): + """Structured output contract for ``proactive_extract_step`` (A3).""" + + follow_ups: list[ProactiveTopic] = Field(default_factory=list) + extends: list[ProactiveTopic] = Field(default_factory=list) + updates: list[TopicUpdate] = Field(default_factory=list) + + +class ProactiveState(BaseModel): + """Chain-shared proactive context state (``context['proactive']``). + + Extract fills material/carry-forward/LLM output fields; topics fills the + filtering/rendering fields; finish records the catalog checkpoint. + ``file_skip_reason`` is metadata/log only and never persisted to + interests.yaml (v5 simplification R7). + """ + + date: str = "" + daily_dir: str = "daily" + workspace: str = "" + scan_days: int = 2 + carry_forward_days: int = 14 + material_paths: list[str] = Field(default_factory=list) + changed_paths: list[str] = Field(default_factory=list) + resource_paths: list[str] = Field(default_factory=list) + carry_forward_all: list[ProactiveTopic] = Field(default_factory=list) + carry_forward_prompt: list[ProactiveTopic] = Field(default_factory=list) + llm_calls: int = 0 + follow_ups: list[dict] = Field(default_factory=list) + extends: list[dict] = Field(default_factory=list) + updates: list[dict] = Field(default_factory=list) + early_exit: str = "" + updates_applied: int = 0 + updates_resolved: int = 0 + candidates_in: int = 0 + candidates: list[dict] = Field(default_factory=list) + dropped_missing: int = 0 + dropped_duplicate: int = 0 + dropped_known: int = 0 + topics_out: list[dict] = Field(default_factory=list) + push: bool = False + file_skip_reason: str = "" + interests_path: str = "" + interests_written: bool = False + checkpoint_paths: list[str] = Field(default_factory=list) + duration_ms: int = 0 + errors: list[str] = Field(default_factory=list) + + +class ProactiveStateFile(BaseModel): + """On-disk truth-source ``daily/_proactive.yaml`` (F1.3, v5: 3 sections). + + ``resolved`` tombstones carry ``first_seen`` so a resurrected topic can + keep its original age anchor (F2.4 reopen channel). + """ + + model_config = ConfigDict(extra="ignore") + + version: int = 1 + open_topics: list[ProactiveTopic] = Field(default_factory=list) + resolved: list[dict] = Field(default_factory=list) + + +class ProactiveResult(BaseModel): + """Result of reading daily interest topics (F5).""" + + date: str = "" + path: str = "" + topics: list[dict] = Field(default_factory=list) + content: str = "" + skipped: bool = False + error: str = "" + summary: str = "" + push: bool | None = None + generated_at: str = "" diff --git a/reme/steps/evolve/__init__.py b/reme/steps/evolve/__init__.py index 423b7598..e8fba860 100644 --- a/reme/steps/evolve/__init__.py +++ b/reme/steps/evolve/__init__.py @@ -1,14 +1,17 @@ """Evolve steps.""" -from ._evolve import now +from ._evolve import now, passthrough_response from .auto_memory import AutoMemoryStep from .auto_memory_cc import AutoMemoryCCStep from .auto_resource import AutoResourceStep from .compressor import CompressorStep -from .dream import DreamExtractStep, DreamFinishStep, DreamIntegrateStep, DreamTopicsStep, ProactiveStep +from .dream import DreamExtractStep, DreamFinishStep, DreamIntegrateStep, DreamTopicsStep +from .proactive import ProactiveExtractStep, ProactiveFinishStep, ProactiveStep, ProactiveTopicsStep +from .wait_for_idle import WaitForIdleStep __all__ = [ "now", + "passthrough_response", "AutoMemoryStep", "AutoMemoryCCStep", "AutoResourceStep", @@ -17,5 +20,9 @@ __all__ = [ "DreamFinishStep", "DreamIntegrateStep", "DreamTopicsStep", + "ProactiveExtractStep", + "ProactiveFinishStep", "ProactiveStep", + "ProactiveTopicsStep", + "WaitForIdleStep", ] diff --git a/reme/steps/evolve/_evolve.py b/reme/steps/evolve/_evolve.py index 6a2e5514..73a0148b 100644 --- a/reme/steps/evolve/_evolve.py +++ b/reme/steps/evolve/_evolve.py @@ -5,6 +5,8 @@ import zoneinfo from agentscope.message import Msg +from ...schema import Response + def now(timezone: str | None = None) -> datetime.datetime: """Return current datetime in the given IANA timezone, falling back to local.""" @@ -40,3 +42,22 @@ def agent_reply_result_text(reply_result: dict) -> str: if text: return text return str(reply_result.get("result") or "").strip() + + +def passthrough_response(step, skip_key: str) -> Response: + """Return a success response when a short-circuit flag is set (INV-7). + + Short-circuited rounds never write interests.yaml or checkpoint catalogs; + the job still reports success so the skipped round is not counted as a failure. + """ + assert step.context is not None + response = step.context.response + response.success = True + flag = step.context.get(skip_key) + if isinstance(flag, dict): + reason = str(flag.get("reason") or "skipped") + else: + reason = str(flag or "skipped") + response.answer = f"Skipped: {reason}" + step.logger.info(f"[{step.name}] short-circuit via {skip_key!r} reason={reason}") + return response diff --git a/reme/steps/evolve/dream/__init__.py b/reme/steps/evolve/dream/__init__.py index fa6291a7..fa0afb49 100644 --- a/reme/steps/evolve/dream/__init__.py +++ b/reme/steps/evolve/dream/__init__.py @@ -3,7 +3,6 @@ from .extract import DreamExtractStep from .finish import DreamFinishStep from .integrate import DreamIntegrateStep -from .proactive import ProactiveStep from .topics import DreamTopicsStep __all__ = [ @@ -11,5 +10,4 @@ __all__ = [ "DreamFinishStep", "DreamIntegrateStep", "DreamTopicsStep", - "ProactiveStep", ] diff --git a/reme/steps/evolve/dream/proactive.py b/reme/steps/evolve/dream/proactive.py deleted file mode 100644 index 7463970f..00000000 --- a/reme/steps/evolve/dream/proactive.py +++ /dev/null @@ -1,60 +0,0 @@ -"""Read daily interests.yaml for proactive use.""" - -from ...base_step import BaseStep -from ....components import R -from ....schema import ProactiveResult -from .utils import load_yaml_topics, today, workspace_dir - - -@R.register("proactive_step") -class ProactiveStep(BaseStep): - """Read ``daily//interests.yaml``.""" - - def __init__(self, include_content: bool = True, **kwargs): - super().__init__(**kwargs) - self.include_content = include_content - - async def execute(self): - assert self.context is not None - day = today(self, str(self.context.get("date", "") or "")) - include_content = bool(self.context.get("include_content", self.include_content)) - daily = self.config_value("daily_dir") - rel_path, abs_path = f"{daily}/{day}/interests.yaml", workspace_dir(self) / daily / day / "interests.yaml" - result = ProactiveResult(date=day, path=rel_path) - self.logger.info(f"[{self.name}] start date={day} path={rel_path} include_content={include_content}") - - if not abs_path.is_file(): - result.skipped, result.summary = True, f"Skipped: interests file not found at {rel_path}" - self.logger.info(f"[{self.name}] skip missing path={rel_path}") - return self._finish(True, result, include_content=include_content) - try: - self.logger.info(f"[{self.name}] read start path={rel_path}") - result.content = abs_path.read_text(encoding="utf-8") if include_content else "" - result.topics = load_yaml_topics(abs_path) - self.logger.info( - f"[{self.name}] read done path={rel_path} topics={len(result.topics)} chars={len(result.content)}", - ) - except Exception as e: # noqa: BLE001 - result.error, result.summary = f"{type(e).__name__}: {e}", "" - self.logger.error(f"[{self.name}] read failed path={rel_path}: {result.error}") - return self._finish(False, result, include_content=include_content) - - result.summary = f"Read {len(result.topics)} proactive topic(s) from {rel_path}" - return self._finish(True, result, include_content=include_content) - - def _finish(self, success: bool, result: ProactiveResult, *, include_content: bool): - assert self.context is not None - self.context.response.success = success - if not success: - self.context.response.answer = f"Error: {result.error}" - elif result.skipped: - self.context.response.answer = result.summary - else: - self.context.response.answer = { - "summary": result.summary, - "topics": result.topics, - **({"content": result.content} if include_content else {}), - } - self.context.response.metadata.update(result.model_dump()) - self.logger.info(f"[{self.name}] finish success={success} answer={self.context.response.answer!r}") - return self.context.response diff --git a/reme/steps/evolve/proactive/__init__.py b/reme/steps/evolve/proactive/__init__.py new file mode 100644 index 00000000..c6af9126 --- /dev/null +++ b/reme/steps/evolve/proactive/__init__.py @@ -0,0 +1,13 @@ +"""Proactive refresh steps: independent of the nightly dream chain.""" + +from .extract import ProactiveExtractStep +from .finish import ProactiveFinishStep +from .proactive import ProactiveStep +from .topics import ProactiveTopicsStep + +__all__ = [ + "ProactiveExtractStep", + "ProactiveFinishStep", + "ProactiveStep", + "ProactiveTopicsStep", +] diff --git a/reme/steps/evolve/proactive/extract.py b/reme/steps/evolve/proactive/extract.py new file mode 100644 index 00000000..addb5674 --- /dev/null +++ b/reme/steps/evolve/proactive/extract.py @@ -0,0 +1,255 @@ +"""Proactive refresh extract step: material scan + follow_ups/extends/updates (F2.0-F2.3).""" + +import asyncio +import json +import time + +from ...base_step import BaseStep +from .._evolve import agent_reply_result_text, passthrough_response +from ....components import R +from ....schema import ProactiveState +from ..dream.utils import daily_dir, pack_paths, today, workspace_dir +from .utils import ( + clean_candidate, + current_now, + load_carry_forward, + load_state, + parse_extract_reply, + resolve_agent_wrapper, + scan_material_daily, + scan_material_resource, +) + +_EXTRACT_SECTIONS = ("follow_ups", "extends", "updates") + + +@R.register("proactive_extract_step") +class ProactiveExtractStep(BaseStep): + """Scan the proactive material set and extract follow-ups, extends and updates. + + Owns the ``proactive`` catalog watermark (never touches the dream catalog) + and the daily LLM budget. Short-circuits via ``context[skip_key]`` on + busy/budget/timeout per F4.3; early-exits with zero LLM calls when no new + evidence exists (F2.0). + """ + + def __init__( + self, + scan_days: int = 2, + resource_lookback_days: int = 7, + max_resource_files: int = 20, + carry_forward_days: int = 14, + max_carry_forward_topics: int = 20, + llm_timeout_seconds: float = 300, + max_chars_per_file: int = 60000, + extends_enabled: bool = True, + skip_key: str = "proactive_skip", + **kwargs, + ): + super().__init__(**kwargs) + self.scan_days = max(int(scan_days), 1) + self.resource_lookback_days = max(int(resource_lookback_days), 0) + self.max_resource_files = max(int(max_resource_files), 0) + self.carry_forward_days = max(int(carry_forward_days), 1) + self.max_carry_forward_topics = max(int(max_carry_forward_topics), 0) + self.llm_timeout_seconds = float(llm_timeout_seconds) + self.max_chars_per_file = max(int(max_chars_per_file), 1000) + self.extends_enabled = bool(extends_enabled) + self.skip_key = skip_key + + async def execute(self): + assert self.context is not None + if self.context.get(self.skip_key): + return passthrough_response(self, self.skip_key) + started = time.monotonic() + day = today(self, str(self.context.get("date", "") or "")) + now_dt = current_now(self) + ws = workspace_dir(self) + daily = daily_dir(self) + if self.file_catalog is None: + raise RuntimeError("proactive_extract_step requires file_catalog") + state = ProactiveState( + date=day, + daily_dir=daily, + workspace=str(ws), + scan_days=self.scan_days, + carry_forward_days=self.carry_forward_days, + ) + self.logger.info(f"[{self.name}] start date={day} scan_days={self.scan_days} extends={self.extends_enabled}") + + # 1) Material set M (F2.0) - all cheap, before any LLM work. + m_daily = scan_material_daily(ws, day, daily, self.scan_days) + m_resource = scan_material_resource(ws, self.resource_lookback_days, self.max_resource_files, now_dt) + state.material_paths = list(dict.fromkeys(m_daily + m_resource)) + + # 2) Truth source + carry-forward (F1.3/F1.4). + state_file, needs_bootstrap = load_state(ws, daily) + carry_all, carry_prompt = await load_carry_forward( + ws, + state_file, + day, + self.carry_forward_days, + self.max_carry_forward_topics, + daily, + needs_bootstrap, + ) + state.carry_forward_all = carry_all + state.carry_forward_prompt = carry_prompt + + # 3) Change detection against the proactive catalog watermark. + # Resources share the same watermark (v5.1): an upload only TRIGGERS in + # the first round that sees it, so an unchanged resource cannot keep + # firing LLM rounds for the whole lookback. + existing = {} + for rel in dict.fromkeys(m_daily + m_resource): + try: + existing[rel] = (ws / rel).stat().st_mtime + except OSError as e: + self.logger.error(f"[{self.name}] stat failed on {rel}: {e}") + nodes = await self.file_catalog.get_nodes() + indexed = {n.path: n.st_mtime for n in nodes} + state.changed_paths = [rel for rel, mt in existing.items() if indexed.get(rel) != mt] + # Trigger vs context (v5.2): the watermark diff alone decides whether the + # round fires (a resource consumed once cannot keep firing rounds), but a + # fired round keeps the full recent-resource window in the blob so delayed + # associations (survey uploaded today, discussion next week) remain + # discoverable by the extends branch. + state.resource_paths = m_resource + self.logger.info( + f"[{self.name}] material daily={len(m_daily)} resource={len(m_resource)} " + f"changed={len(state.changed_paths)} carry_forward={len(carry_all)}", + ) + + # 4) Zero-consumption early exit (A4 row 1). + if not state.changed_paths: + state.early_exit = "no_new_evidence" + return self._finish(state, started, "Skipped: no new evidence; 0 LLM calls") + + # 5) LLM channel (F4.4): structurally one reply per round plus at most + # one parse-failure retry, so no persistent budget is needed (v5 R5). + wrapper = resolve_agent_wrapper(self) + if wrapper is None: + state.early_exit = "no_agent_wrapper" + self.logger.warning(f"[{self.name}] no agent_wrapper available; skipping round") + return self._finish(state, started, "Skipped: no agent_wrapper configured") + + # 6) One LLM call with sectioned output (A3); retry once on parse failure. + meta = await self._extract_with_retry(wrapper, state, ws, day) + if meta is None: + self.context[self.skip_key] = {"reason": "llm_timeout"} + self._store(state) + return passthrough_response(self, self.skip_key) + self._clean_output(state, meta, set(state.material_paths), day) + answer = ( + f"Extracted {len(state.follow_ups)} follow_up(s), {len(state.extends)} extend(s), " + f"{len(state.updates)} update(s) from {len(state.changed_paths)} changed file(s)" + ) + return self._finish(state, started, answer) + + def _build_messages(self, state: ProactiveState, ws, day: str, material_blob: str) -> tuple[str, str]: + carry_forward_json = json.dumps( + [ + { + "id": t.id, + "title": t.title, + "kind": t.kind, + "confidence": t.confidence, + "reason": t.reason, + "last_evidence_at": t.last_evidence_at, + "evidence": t.evidence, + } + for t in state.carry_forward_prompt + ], + ensure_ascii=False, + ) + changed = list(dict.fromkeys(state.changed_paths + state.resource_paths)) + user_message = self.prompt_format( + "extract_user_message", + date=day, + changed_paths_json=json.dumps(changed, ensure_ascii=False), + carry_forward_json=carry_forward_json, + material_blob=material_blob, + extends=self.extends_enabled, + ) + system_prompt = self.prompt_format( + "extract_system_prompt", + workspace_dir=str(ws), + extends=self.extends_enabled, + ) + return user_message, system_prompt + + async def _extract_with_retry(self, wrapper, state, ws, day: str) -> dict | None: + """Returns parsed meta; None means LLM timeout.""" + material_blob = pack_paths( + ws, + list(dict.fromkeys(state.changed_paths + state.resource_paths)), + limit_per_file=self.max_chars_per_file, + ) + user_message, system_prompt = self._build_messages(state, ws, day, material_blob) + for attempt in (1, 2): + raw = await self._reply(wrapper, state, user_message, system_prompt) + if raw is None: + return None + meta = parse_extract_reply(raw) + if meta: + return meta + self.logger.warning(f"[{self.name}] parse failed on attempt {attempt}; raw={raw[:200]!r}") + return {} + + async def _reply(self, wrapper, state, user_message, system_prompt): + """One timeout-wrapped reply; returns raw text or None on timeout.""" + state.llm_calls += 1 + try: + result = await asyncio.wait_for( + wrapper.reply(user_message, system_prompt=system_prompt), + timeout=self.llm_timeout_seconds, + ) + except asyncio.TimeoutError: + self.logger.warning(f"[{self.name}] LLM reply timed out after {self.llm_timeout_seconds}s") + return None + return agent_reply_result_text(result) + + def _clean_output(self, state: ProactiveState, meta: dict, allowed: set[str], day: str) -> None: + for raw in meta.get("follow_ups") or []: + if candidate := clean_candidate(raw, allowed, "follow_up", day): + state.follow_ups.append(candidate) + else: + state.dropped_missing += 1 + if self.extends_enabled: + for raw in meta.get("extends") or []: + if candidate := clean_candidate(raw, allowed, "interest_extend", day): + state.extends.append(candidate) + else: + state.dropped_missing += 1 + for raw in meta.get("updates") or []: + if not isinstance(raw, dict): + continue + topic_id = str(raw.get("id") or "").strip() + if not topic_id: + continue + action = str(raw.get("action") or "keep").strip() + if action not in ("keep", "update", "resolve"): + action = "keep" + state.updates.append( + { + "id": topic_id, + "action": action, + "evidence": str(raw.get("evidence") or "").strip()[:120], + "reason": str(raw.get("reason") or "").strip(), + "confidence": raw.get("confidence"), + }, + ) + + def _store(self, state: ProactiveState) -> None: + assert self.context is not None + data = state.model_dump() + self.context["proactive"] = data + self.context.response.metadata["proactive"] = data + + def _finish(self, state: ProactiveState, started: float, answer: str): + state.duration_ms = int((time.monotonic() - started) * 1000) + self._store(state) + self.context.response.success = True + self.context.response.answer = answer + self.logger.info(f"[{self.name}] finish answer={answer!r}") + return self.context.response diff --git a/reme/steps/evolve/proactive/extract.yaml b/reme/steps/evolve/proactive/extract.yaml new file mode 100644 index 00000000..be0bf157 --- /dev/null +++ b/reme/steps/evolve/proactive/extract.yaml @@ -0,0 +1,209 @@ +extract_system_prompt: | + You are the proactive discovery agent for a personal memory workspace. + You read recently changed conversation notes (and recent resource uploads) + and maintain a small set of user-interest topics. You never browse the web, + never write files, and never collect material yourself. + + workspace_dir: {workspace_dir} + + ## Branch A - follow_ups (open loops) + + Find unresolved matters the user may still want to close: + - questions that were asked but never answered; + - tasks that were started, interrupted, or postponed; + - commitments or plans mentioned without any follow-through yet. + Emit each as a `follow_ups` entry. + + [extends]## Branch B - extends (interest boundary) + [extends] + [extends]Infer topics the user has NOT focused on yet, but that are plausibly + [extends]relevant to their recent work, using the conversation notes plus the + [extends]recently uploaded resource files. These guide future knowledge-source + [extends]expansion; you only describe them, you never collect anything. + [extends]Emit each as an `extends` entry. + + ## updates (carried-forward topics) + + For every topic in carry_forward_topics, decide exactly one action: + - `keep`: still open but no new evidence today; + - `update`: new evidence appeared today; refresh evidence and confidence + (emit the re-scored confidence value); + - `resolve`: the conversation shows the matter is settled or done. + You must echo the given `id` unchanged. Never invent ids, and never emit an + update for a topic that is not in carry_forward_topics. + If two carry_forward topics turn out to be the same matter in different + words, keep the one with stronger evidence and `resolve` the other one + (note ``merged into `` in reason). + Only `action=update` when the fresh evidence lives in a file listed in this + round's changed_paths; otherwise emit `keep` (updates citing anything else + are rejected downstream). + + ## Confidence rubric (fixed; do not freestyle scores) + + - 0.9: explicit unfinished action with a time or commitment; + - 0.7: the user stated intent explicitly but no concrete action yet; + - 0.5: inferential link from several weak signals; + - 0.3: weak association only; + - below 0.3: do not output the topic at all. + + ## Hard rules + + - Empty lists are a normal, expected output. Never invent topics to fill + space, and never promote a one-off weak mention into a topic. + - Every topic must be traceable to this round's material: `paths` may only + contain values from changed_paths; entries violating this are discarded. + - Do not restate topics already listed in carry_forward_topics; use + `updates` to change their status instead. + - Same-matter rule: if a finding is really a carry_forward topic in + different words, or new progress/angle on it, emit `updates` with + action=update for that id - never open it as a new topic. + Example: carry_forward_topics has "explainability evaluation of memory + search" and the material says the evaluation dataset is now ready -> + update that topic; do NOT open "run the benchmark" as a new topic. Open + a new entry only for matters no carry_forward topic covers. + - Quote YAML strings containing punctuation such as `:`; write paths as + block lists so `daily//...` stays parseable. + + ## Output format + + Return only one YAML fenced block with this exact shape: + ```yaml + follow_ups: + - title: + reason: + confidence: 0.7 + evidence: + keywords: [, ...] + paths: [, ...] + [extends]extends: + [extends] - title: + [extends] reason: + [extends] confidence: 0.5 + [extends] evidence: + [extends] keywords: [, ...] + [extends] paths: [, ...] + updates: + - id: + action: keep|update|resolve + evidence: + reason: + confidence: + ``` + +extract_system_prompt_zh: | + 你是个人记忆 workspace 的 proactive 发现 agent。你阅读近期变化的对话笔记 + (以及最近上传的 resource 文件),维护一小组用户兴趣主题。你绝不联网、 + 绝不写文件、绝不自行收集资料。 + + workspace_dir: {workspace_dir} + + ## 分支 A - follow_ups(未决事项 / open loop) + + 找出用户可能仍想关闭的未决事项: + - 提出过但没有得到回答的问题; + - 开始过但被中断或搁置的任务; + - 提到过但没有后续动作的承诺或计划。 + 每条作为一个 `follow_ups` 条目输出。 + + [extends]## 分支 B - extends(兴趣边界扩展) + [extends] + [extends]推断用户尚未关注、但与其近期工作大概率相关的主题,依据是对话笔记 + [extends]加上最近上传的 resource 文件。它们用于指引未来的知识源扩展; + [extends]你只描述主题,不收集任何资料。每条作为一个 `extends` 条目输出。 + + ## updates(carry-forward 主题处置) + + 对 carry_forward_topics 中的每个主题,恰好选择一个动作: + - `keep`:仍然 open,但今天没有新证据; + - `update`:今天出现新证据;刷新 evidence 与 confidence(输出按 rubric + 重新打分的 confidence 值); + - `resolve`:对话显示事项已解决或完成。 + 必须原样回显给定的 `id`。禁止伪造 id,禁止对不在 carry_forward_topics + 中的主题输出 update。 + 若 carry_forward_topics 中有两个主题实为同一件事的不同表述,保留证据较强 + 的一个,对另一个输出 `resolve`(reason 注明 merged into <保留的 id>)。 + 仅当新证据出自本轮 changed_paths 列表中的文件时才输出 action=update; + 否则输出 keep(引用其他文件的 update 会在下游被拒绝并降级为 keep)。 + + ## confidence 评分规则(固化,禁止自由打分) + + - 0.9:有明确未完成动作且带时间/承诺; + - 0.7:用户显式表达意图但无明确动作; + - 0.5:推断性关联(多条弱信号); + - 0.3:弱联想; + - 低于 0.3:不要输出该主题。 + + ## 硬约束 + + - 空列表是正常且预期的输出。不得为凑数编造主题,不得把一次性弱提及 + 拔高为主题。 + - 每个主题必须能追溯到本轮素材:`paths` 只能使用 changed_paths 中的值, + 越界条目会被丢弃。 + - 不要复述 carry_forward_topics 中已 open 的主题;要改变其状态请用 + `updates`。 + - 同一件事规则:若新发现其实只是某个 carry_forward 主题的换种说法, + 或是它的新进展/新角度,必须对该 id 输出 action=update 的 `updates` + 条目,禁止把它当作新主题另开条目。例:carry_forward_topics 已有 + "记忆检索的可解释性评估",素材提到评测数据集已就绪 -> 更新该主题, + 不得新开"运行 benchmark"主题。仅当 carry_forward 主题均未覆盖时, + 才可新开主题。 + - 包含 `:` 等标点的 YAML 字符串请加引号;paths 使用 block list, + 避免 `daily//...` 解析失败。 + + ## 输出格式 + + 只返回一个 YAML fenced block,结构必须严格如下: + ```yaml + follow_ups: + - title: <具体的未决事项> + reason: <为什么仍未关闭> + confidence: 0.7 + evidence: <路径或 路径#锚点> + keywords: [<关键词>, ...] + paths: [<素材路径>, ...] + [extends]extends: + [extends] - title: <相关但未被关注的主题> + [extends] reason: <为什么重要,基于素材> + [extends] confidence: 0.5 + [extends] evidence: <路径或 路径#锚点> + [extends] keywords: [<关键词>, ...] + [extends] paths: [<素材路径>, ...] + updates: + - id: <回显的 carry-forward id> + action: keep|update|resolve + evidence: <路径或 路径#锚点> + reason: <简短理由> + confidence: <按 rubric 重新打分;action=update 时给出,其余可省略> + ``` + +extract_user_message: | + date: {date} + + changed_paths (the ONLY allowed values for `paths`): + {changed_paths_json} + + carry_forward_topics (already open; only act on them via updates): + {carry_forward_json} + + # Material + + {material_blob} + + Extract follow_ups[extends] / extends / updates per the contract. Remember: + empty lists are fine; never invent topics. + +extract_user_message_zh: | + 日期:{date} + + changed_paths(`paths` 唯一允许的取值): + {changed_paths_json} + + carry_forward_topics(已 open 的主题;只能通过 updates 处置): + {carry_forward_json} + + # 本轮素材 + + {material_blob} + + 按契约抽取 follow_ups[extends] / extends / updates。记住:空列表是正常 + 输出,绝不编造主题。 diff --git a/reme/steps/evolve/proactive/finish.py b/reme/steps/evolve/proactive/finish.py new file mode 100644 index 00000000..746d45a7 --- /dev/null +++ b/reme/steps/evolve/proactive/finish.py @@ -0,0 +1,65 @@ +"""Proactive finish step: checkpoint only the proactive catalog (F3).""" + +from ...base_step import BaseStep +from .._evolve import passthrough_response +from ....components import R +from ....schema import FileNode, ProactiveState +from ..dream.utils import workspace_dir + + +@R.register("proactive_finish_step") +class ProactiveFinishStep(BaseStep): + """Upsert this round's changed material into the proactive catalog. + + interests.yaml is deliberately NOT checkpointed (v5 R6): the catalog only + serves change-detection watermarking and interests.yaml is excluded from + the material set (INV-11), so no consumer would ever read it back. + Never calls ``refresh_day_index`` and never touches the dream catalog + (INV-2). When the short-circuit flag is set, no checkpoint happens so the + same material stays "changed" for the next round (INV-7). + """ + + def __init__(self, persist: bool = True, skip_key: str = "proactive_skip", **kwargs): + super().__init__(**kwargs) + self.persist = persist + self.skip_key = skip_key + + async def execute(self): + assert self.context is not None + if self.context.get(self.skip_key): + return passthrough_response(self, self.skip_key) + if self.file_catalog is None: + raise RuntimeError("proactive_finish_step requires file_catalog") + raw_state = self.context.get("proactive") + if not raw_state: + self.context.response.success = True + self.context.response.answer = "Skipped finish: no proactive extract state in context" + return self.context.response + state = ProactiveState.model_validate(raw_state) + if state.early_exit: + self.context.response.success = True + self.context.response.answer = f"Skipped finish: {state.early_exit}" + return self.context.response + ws = workspace_dir(self) + + checkpoint = [rel for rel in state.changed_paths if (ws / rel).is_file()] + nodes: list[FileNode] = [] + for rel in dict.fromkeys(checkpoint): + try: + nodes.append(FileNode(path=rel, st_mtime=(ws / rel).stat().st_mtime)) + except OSError: + continue + self.logger.info(f"[{self.name}] start checkpoint={len(nodes)} persist={self.persist}") + if nodes: + await self.file_catalog.upsert(nodes) + if self.persist and nodes: + await self.file_catalog.dump() + + state.checkpoint_paths = [n.path for n in nodes] + data = state.model_dump() + self.context["proactive"] = data + self.context.response.metadata["proactive"] = data + self.context.response.success = True + self.context.response.answer = f"Proactive finished: checkpointed {len(nodes)} path(s)" + self.logger.info(f"[{self.name}] finish checkpointed={len(nodes)}") + return self.context.response diff --git a/reme/steps/evolve/proactive/proactive.py b/reme/steps/evolve/proactive/proactive.py new file mode 100644 index 00000000..051f59c0 --- /dev/null +++ b/reme/steps/evolve/proactive/proactive.py @@ -0,0 +1,180 @@ +"""Read interests.yaml for proactive consumption (F5). + +Migrated from ``dream/proactive.py`` (INV-9 mechanical migration) and extended +with ``min_confidence`` filtering (default 0.4, safely below the 0.5 confidence +fallback) and ``horizon_days``: horizon=1 keeps the legacy single-day file +read (v1 compatible); horizon>1 reads the truth source directly, filtering by +``last_evidence_at`` recency (v5 R4 - the truth source already carries the +cross-day merge, so the reader no longer re-scans exposure products). +""" + +import datetime as dt + +import yaml + +from ...base_step import BaseStep +from ....components import R +from ....schema import ProactiveResult +from ..dream.utils import load_yaml_topics, today, workspace_dir +from .utils import ( + dump_topic, + load_state, + parse_interests_topics, + quarantine_interests, + sort_topics, + topic_id, +) + + +@R.register("proactive_step") +class ProactiveStep(BaseStep): + """Read ``daily//interests.yaml`` (optionally merged over a horizon). + + - v1 files (nightly) are read as ``push=true`` and never rewritten; + - ``push=false`` days contribute nothing; + - resolved ids (truth source registry) are suppressed; + - ``horizon_days>1`` reads the truth source and filters by evidence recency. + """ + + def __init__(self, include_content: bool = True, horizon_days: int = 1, min_confidence: float = 0.4, **kwargs): + super().__init__(**kwargs) + self.include_content = include_content + self.horizon_days = max(int(horizon_days), 1) + self.min_confidence = float(min_confidence) + + async def execute(self): + assert self.context is not None + day = today(self, str(self.context.get("date", "") or "")) + include_content = bool(self.context.get("include_content", self.include_content)) + horizon = int(self.context.get("horizon_days", self.horizon_days) or self.horizon_days) + horizon = max(horizon, 1) + raw_min_confidence = self.context.get("min_confidence", self.min_confidence) + try: + min_confidence = float(raw_min_confidence) + except (TypeError, ValueError): + min_confidence = self.min_confidence + daily = self.config_value("daily_dir") + ws = workspace_dir(self) + default_rel = f"{daily}/{day}/interests.yaml" + result = ProactiveResult(date=day, path=default_rel) + self.logger.info( + f"[{self.name}] start date={day} path={default_rel} include_content={include_content} " + f"horizon_days={horizon} min_confidence={min_confidence}", + ) + + if horizon == 1: + outcome = self._read_single(ws, daily, day, include_content, min_confidence, result) + if outcome is not None: + return outcome + + return self._read_horizon(ws, daily, day, horizon, include_content, min_confidence, result) + + # ------------------------------------------------------------------ + # Single-day read (legacy-compatible path) + # ------------------------------------------------------------------ + + def _read_single(self, ws, daily, day, include_content, min_confidence, result: ProactiveResult): + rel_path = f"{daily}/{day}/interests.yaml" + abs_path = ws / daily / day / "interests.yaml" + if not abs_path.is_file(): + result.skipped, result.summary = True, f"Skipped: interests file not found at {rel_path}" + self.logger.info(f"[{self.name}] skip missing path={rel_path}") + return self._finish(True, result, include_content=include_content) + try: + raw_text = abs_path.read_text(encoding="utf-8") + try: + data = yaml.safe_load(raw_text) + if not isinstance(data, dict): + raise ValueError("interests.yaml is not a mapping") + except Exception as e: # noqa: BLE001 + quarantine_interests(abs_path, e) + data = {} # A2: treat corrupt file as empty + topics, is_v1, push = parse_interests_topics(data, day) + if push is False: + result.skipped = True + result.summary = f"Skipped: interests file at {rel_path} has push=false" + result.push = False + self.logger.info(f"[{self.name}] skip push=false path={rel_path}") + return self._finish(True, result, include_content=include_content) + + result.push = True + result.content = raw_text if include_content else "" + if is_v1: + # Legacy shape (title/reason/evidence/keywords/paths) for v1 files; + # v1 topics carry no confidence and fall back to 0.5 for filtering. + resolved_ids = self._resolved_ids(ws, daily) + legacy_topics = load_yaml_topics(abs_path) + kept = [t for t in legacy_topics if topic_id(str(t.get("title") or "")) not in resolved_ids] + if min_confidence > 0.5 + 1e-9: # v1 topics carry no confidence; fallback is 0.5 (F1.1) + kept = [] + result.topics = kept + else: + result.generated_at = str(data.get("generated_at") or "") + resolved_ids = self._resolved_ids(ws, daily) + kept = [ + dump_topic(t) + for t in sort_topics(topics) + if t.id not in resolved_ids and t.confidence >= min_confidence - 1e-9 + ] + result.topics = kept + result.summary = f"Read {len(result.topics)} proactive topic(s) from {rel_path}" + self.logger.info(f"[{self.name}] read done path={rel_path} topics={len(result.topics)}") + return self._finish(True, result, include_content=include_content) + except Exception as e: # noqa: BLE001 + result.error, result.summary = f"{type(e).__name__}: {e}", "" + self.logger.error(f"[{self.name}] read failed path={rel_path}: {result.error}") + return self._finish(False, result, include_content=include_content) + + # ------------------------------------------------------------------ + # Multi-day merge (horizon_days > 1, or fallback target) + # ------------------------------------------------------------------ + + def _read_horizon(self, ws, daily, day, horizon, include_content, min_confidence, result: ProactiveResult): + """Truth-source view (v5 R4): open topics with evidence inside the horizon. + + The truth source already carries the cross-day merge (carry-forward), + so the reader no longer re-scans N days of exposure products. + """ + state_file, _needs_bootstrap = load_state(ws, daily) + if not state_file.open_topics: + result.skipped = True + result.summary = f"Skipped: truth source has no open topics (horizon_days={horizon})" + self.logger.info(f"[{self.name}] skip empty truth source horizon={horizon}") + return self._finish(True, result, include_content=include_content) + try: + base = dt.date.fromisoformat(day) + cutoff = (base - dt.timedelta(days=max(horizon - 1, 0))).isoformat() + except ValueError: + cutoff = "" + kept = [ + dump_topic(t) + for t in state_file.open_topics + if (not cutoff or str(t.last_evidence_at or "") >= cutoff) and t.confidence >= min_confidence - 1e-9 + ] + kept = sort_topics(kept) + result.path = f"{daily}/_proactive.yaml" + result.topics = kept + result.summary = f"Read {len(kept)} proactive topic(s) from the truth source (horizon_days={horizon})" + self.logger.info(f"[{self.name}] truth-source read done horizon={horizon} topics={len(kept)}") + return self._finish(True, result, include_content=include_content) + + def _resolved_ids(self, ws, daily: str) -> set[str]: + state_file, _needs_bootstrap = load_state(ws, daily) + return {str(r.get("id") or "") for r in state_file.resolved if isinstance(r, dict) and r.get("id")} + + def _finish(self, success: bool, result: ProactiveResult, *, include_content: bool): + assert self.context is not None + self.context.response.success = success + if not success: + self.context.response.answer = f"Error: {result.error}" + elif result.skipped: + self.context.response.answer = result.summary + else: + self.context.response.answer = { + "summary": result.summary, + "topics": result.topics, + **({"content": result.content} if include_content else {}), + } + self.context.response.metadata.update(result.model_dump()) + self.logger.info(f"[{self.name}] finish success={success} answer={self.context.response.answer!r}") + return self.context.response diff --git a/reme/steps/evolve/proactive/topics.py b/reme/steps/evolve/proactive/topics.py new file mode 100644 index 00000000..c71c4f8e --- /dev/null +++ b/reme/steps/evolve/proactive/topics.py @@ -0,0 +1,462 @@ +"""Proactive topics step: dedup, truth-source update, derived push, render (F2.4). + +Pure computation step (v5 R2): no LLM calls, no budget, no agent_wrapper. +Semantic dedup keeps only the ``>= known_threshold`` "known" drop; everything +below is kept (loose-not-leaky). The main dedup defense lives upstream in the +extract prompt's same-matter rule plus same-id merging here. + +``known_threshold`` is bound to the embedding model, dimensions AND vector-text +scheme it was calibrated for (v5.2: ``title。reason`` texts, threshold 0.85, +34-pair measurement: DUP band 0.773-0.943 vs KEEP band <=0.772). Cosine +magnitudes are not comparable across models, so a fingerprint mismatch degrades +the gate to exact normalize comparison instead of silently misfiring. Without +any configured embedder the step skips the semantic gate entirely and the +workflow continues on exact comparison (BM25-only deployments). +""" + +import datetime as dt +import re +import time + +from ...base_step import BaseStep +from .._evolve import passthrough_response +from ....components import R +from ....enumeration import ComponentEnum +from ....schema import ProactiveState, ProactiveStateFile, ProactiveTopic +from ....schema.proactive import clamp_confidence +from ..dream.utils import previous_dates, today, workspace_dir +from .utils import ( + current_now, + dump_topic, + interests_path_for, + load_state, + norm_path, + normalize_topic, + parse_interests_topics, + read_interests_data, + render_interests, + save_state, + sort_topics, + trim_state_file, + write_interests_if_changed, +) + + +@R.register("proactive_topics_step") +class ProactiveTopicsStep(BaseStep): + """Filter candidates, update the truth source, and render interests.yaml. + + Without ``as_embedding`` the exact ``normalize_topic`` comparison applies. + With embeddings, ``sim >= known_threshold`` drops as known; below the + threshold candidates are kept. Candidates whose id matches a resolved + tombstone are resurrected (tombstone removed, original ``first_seen`` + kept) instead of being silently suppressed. + """ + + def __init__( + self, + known_threshold: float = 0.85, + min_push_confidence: float = 0.5, + max_topics: int = 10, + dedup_lookback_days: int = 7, + digest_compare_limit: int = 500, + known_threshold_calibrated_for: str = "text-embedding-v4@1024", + skip_key: str = "proactive_skip", + **kwargs, + ): + super().__init__(**kwargs) + self.known_threshold = float(known_threshold) + self.known_threshold_calibrated_for = str(known_threshold_calibrated_for or "") + self.min_push_confidence = float(min_push_confidence) + self.max_topics = max(int(max_topics), 1) + self.dedup_lookback_days = max(int(dedup_lookback_days), 0) + self.digest_compare_limit = max(int(digest_compare_limit), 0) + self.skip_key = skip_key + + # pylint: disable=too-many-statements + async def execute(self): + assert self.context is not None + if self.context.get(self.skip_key): + return passthrough_response(self, self.skip_key) + started = time.monotonic() + if not self.context.get("proactive"): + self.context.response.success = True + self.context.response.answer = "Skipped topics: no proactive extract state in context" + return self.context.response + state = ProactiveState.model_validate(self.context.get("proactive")) + if state.early_exit: + self.context.response.success = True + self.context.response.answer = f"Skipped topics: {state.early_exit}" + return self.context.response + day = state.date or today(self, str(self.context.get("date", "") or "")) + ws = workspace_dir(self) + daily = state.daily_dir or "daily" + self.logger.info( + f"[{self.name}] start date={day} follow_ups={len(state.follow_ups)} extends={len(state.extends)} " + f"updates={len(state.updates)} carry_forward={len(state.carry_forward_all)}", + ) + + state_file, _needs_bootstrap = load_state(ws, daily) + open_by_id = {t.id: t for t in state_file.open_topics if t.id} + expiry_cutoff = self._expiry_cutoff(day, max(state.carry_forward_days, 1)) + + # 1) Apply updates to the truth source (F2.3). An action=update is only + # legal when its evidence anchors this round's new material (v5.2 hard + # check, same shape as INV-8); otherwise it degrades to keep so stale + # re-reads cannot rewrite evidence or refresh the freshness sort key. + new_material = {norm_path(str(path).split("#", 1)[0]) for path in state.changed_paths} + for update in state.updates: + topic = open_by_id.get(str(update.get("id") or "")) + if topic is None: + continue + action = str(update.get("action") or "keep") + if action == "update": + evidence_raw = str(update.get("evidence") or "").strip() + evidence_path = norm_path(evidence_raw.split("#", 1)[0]) + if evidence_path in new_material: + topic.last_evidence_at = self._evidence_date(evidence_path) or day + topic.evidence = evidence_raw[:120] + if update.get("confidence") is not None: + topic.confidence = clamp_confidence(update.get("confidence")) + state.updates_applied += 1 + else: + self.logger.info( + f"[{self.name}] update for {topic.id} downgraded to keep: evidence " + f"{evidence_path or ''!r} not in this round's new material", + ) + elif action == "resolve": + state_file.open_topics = [t for t in state_file.open_topics if t.id != topic.id] + del open_by_id[topic.id] + state_file.resolved.append( + { + "id": topic.id, + "title": topic.title, + "resolved_at": day, + "first_seen": topic.first_seen, + "evidence": str(update.get("evidence") or "")[:120], + }, + ) + state.updates_resolved += 1 + + # 2) Candidates: same-id merge, tombstone resurrect, dedup the rest. + raw_candidates = list(state.follow_ups) + list(state.extends) + state.candidates_in = len(raw_candidates) + merged: list[dict] = [] + resurrected: list[dict] = [] + fresh: list[dict] = [] + tombstone_by_id = { + str(r.get("id") or ""): r for r in state_file.resolved if isinstance(r, dict) and r.get("id") + } + for candidate in raw_candidates: + cid = str(candidate.get("id") or "") + existing = open_by_id.get(cid) + if existing is not None: + if expiry_cutoff and str(existing.first_seen or "") <= expiry_cutoff: + # Over-age re-mention: trim would prune it this round, so a + # fresh mention restarts the lifetime (unfinished business + # must keep being re-executed, v5.1). + self.logger.info( + f"[{self.name}] over-age topic {cid} re-mentioned; restarting first_seen " + f"({existing.first_seen} -> {day})", + ) + existing.first_seen = day + self._merge_into(existing, candidate, day) + merged.append(candidate) + continue + tombstone = tombstone_by_id.get(cid) + if tombstone is not None: + topic = self._resurrect(tombstone, candidate, day) + state_file.resolved = [r for r in state_file.resolved if r is not tombstone] + del tombstone_by_id[cid] + state_file.open_topics.append(topic) + open_by_id[cid] = topic + resurrected.append(candidate) + self.logger.info(f"[{self.name}] resurrected resolved topic {cid} ({topic.title!r})") + continue + fresh.append(candidate) + + dropped_duplicate = dropped_known = 0 + survivors: list[dict] = [] + seen_normalized: set[str] = set() + if fresh: + comparison = await self._comparison_texts(ws, daily, day, state_file, expiry_cutoff) + embedder = self._resolve_embedding() + if embedder is not None and self.known_threshold_calibrated_for: + fingerprint = self._embedding_fingerprint(embedder) + if fingerprint != self.known_threshold_calibrated_for: + self.logger.warning( + f"[{self.name}] embedding fingerprint {fingerprint or ''!r} != " + f"known_threshold calibration {self.known_threshold_calibrated_for!r}; " + f"cosine is not comparable across models, degrading to exact normalize comparison", + ) + embedder = None + comparison_norms = {normalize_topic(title) for title, _ in comparison} + for candidate in fresh: + normalized = normalize_topic(str(candidate.get("title") or "")) + if normalized in seen_normalized: + dropped_duplicate += 1 + continue + if embedder is None: + if normalized in comparison_norms: + dropped_duplicate += 1 + continue + else: + verdict, matched, similarity = await self._semantic_verdict( + embedder, + candidate, + comparison, + comparison_norms, + ) + if verdict == "known": + dropped_known += 1 + self.logger.info( + f"[{self.name}] candidate {candidate.get('title')!r} dropped as known " + f"(sim={similarity:.3f}, matched={matched!r})", + ) + continue + if verdict == "duplicate": # embedding failure fallback + dropped_duplicate += 1 + continue + seen_normalized.add(normalized) + survivors.append(candidate) + state.dropped_duplicate = dropped_duplicate + state.dropped_known = dropped_known + state.candidates = merged + resurrected + survivors + self.logger.info( + f"[{self.name}] candidates in={state.candidates_in} merged={len(merged)} " + f"resurrected={len(resurrected)} new={len(survivors)} " + f"dropped_duplicate={dropped_duplicate} dropped_known={dropped_known}", + ) + + # 3) Truth source: add new topics, prune, single atomic write (F1.3). + for candidate in survivors: + state_file.open_topics.append(ProactiveTopic.model_validate(candidate)) + trim_state_file(state_file, day, max(state.carry_forward_days, 1)) + await save_state(ws, state_file, daily) + + # 4) Push derived from the cumulative truth source (v5 R1): a topic + # discovered today with sufficient confidence. Monotonic across same-day + # rounds because such topics persist in the truth source once added. + push = any(t.first_seen == day and t.confidence >= self.min_push_confidence for t in state_file.open_topics) + if push: + file_skip_reason = "" + elif merged or resurrected or survivors: + file_skip_reason = "low_confidence" + else: + file_skip_reason = "all_duplicates" + + # 5) Render from the truth source and write idempotently (A4). + topics_out = sort_topics(list(state_file.open_topics))[: self.max_topics] + now_dt = current_now(self) + rendered = render_interests(day, topics_out, push, now_dt) + interests_path = interests_path_for(ws, daily, day) + written = write_interests_if_changed(ws, interests_path, rendered) + rel_path = norm_path(interests_path.relative_to(ws).as_posix()) + + state.topics_out = [dump_topic(t) for t in topics_out] + state.push = push + state.file_skip_reason = file_skip_reason + state.interests_path = rel_path + state.interests_written = written + state.duration_ms = int((time.monotonic() - started) * 1000) + self._store(state) + answer = ( + f"Topics: {len(topics_out)} rendered, push={push}, " + f"skip_reason={file_skip_reason or '-'}, written={written} to {rel_path}" + ) + self.context.response.success = True + self.context.response.answer = answer + self.logger.info(f"[{self.name}] finish {answer}") + return self.context.response + + @staticmethod + def _merge_into(existing: ProactiveTopic, candidate: dict, day: str) -> None: + """Same-id candidate refreshes evidence in place; first_seen is kept.""" + existing.last_evidence_at = day + if candidate.get("reason"): + existing.reason = str(candidate["reason"]) + if candidate.get("evidence"): + existing.evidence = str(candidate["evidence"])[:120] + existing.confidence = clamp_confidence(candidate.get("confidence")) + if candidate.get("keywords"): + existing.keywords = list(candidate["keywords"]) + if candidate.get("paths"): + existing.paths = list(candidate["paths"]) + + @staticmethod + def _resurrect(tombstone: dict, candidate: dict, day: str) -> ProactiveTopic: + """Reopen a resolved topic: original first_seen kept, evidence refreshed. + + The over-age trim still applies to the original ``first_seen``, so a + resurrection only extends a lifetime that has not fully elapsed. + """ + return ProactiveTopic( + id=str(tombstone.get("id") or candidate.get("id") or ""), + title=str(candidate.get("title") or tombstone.get("title") or ""), + reason=str(candidate.get("reason") or ""), + kind=str(candidate.get("kind") or "interest_extend"), + confidence=clamp_confidence(candidate.get("confidence")), + first_seen=str(tombstone.get("first_seen") or day), + last_evidence_at=day, + evidence=str(candidate.get("evidence") or "")[:120], + keywords=candidate.get("keywords") or [], + paths=candidate.get("paths") or [], + ) + + async def _comparison_texts( + self, + ws, + daily: str, + day: str, + state_file: ProactiveStateFile, + expiry_cutoff: str = "", + ) -> list[tuple[str, str]]: + """(title, embed_text) pairs from recent interests, open topics, digest nodes. + + Embed text carries the reason when available (v5.2 calibration: + title+reason separates the DUP/KEEP bands; bare titles do not). + Over-age open topics are excluded (v5.1): they are pruned by trim this + round, so using them as "known" would silently swallow a re-mention of + a matter that is about to restart. + """ + pairs: list[tuple[str, str]] = [] + for previous_day in previous_dates(day, self.dedup_lookback_days): + data = read_interests_data(interests_path_for(ws, daily, previous_day)) + if not data: + continue + topics, _is_v1, _push = parse_interests_topics(data, previous_day) + pairs.extend((t.title, _known_text(t.title, getattr(t, "reason", ""))) for t in topics) + pairs.extend( + (t.title, _known_text(t.title, t.reason)) + for t in state_file.open_topics + if not expiry_cutoff or str(t.first_seen or "") > expiry_cutoff + ) + pairs.extend((title, title) for title in await self._digest_titles()) + return pairs + + async def _digest_titles(self) -> list[str]: + if self.app_context is None or self.digest_compare_limit <= 0: + return [] + catalog = self.app_context.components.get(ComponentEnum.FILE_CATALOG, {}).get("digest") + if catalog is None: + return [] + try: + nodes = await catalog.get_nodes() + except Exception: # noqa: BLE001 + return [] + ordered = sorted(nodes, key=lambda n: float(getattr(n, "st_mtime", 0.0) or 0.0), reverse=True) + return [str(n.path).rsplit("/", 1)[-1].rsplit(".", 1)[0] for n in ordered[: self.digest_compare_limit]] + + def _resolve_embedding(self): + if self.context is not None: + candidate = self.context.get("as_embedding") + if candidate is not None: + return candidate + name = self.kwargs.get("as_embedding", "default") + if self.app_context is None: + return None + return self.app_context.components.get(ComponentEnum.AS_EMBEDDING, {}).get(name) + + async def _semantic_verdict( + self, + embedder, + candidate: dict, + comparison: list[tuple[str, str]], + comparison_norms: set[str], + ) -> tuple[str, str, float]: + """Semantic gate: known (>= known_threshold) | keep; duplicate on embedding failure.""" + candidate_text = _embed_text(candidate) + texts = [candidate_text] + [embed_text for _, embed_text in comparison] + try: + vectors = await embedder(texts) + except Exception as e: # noqa: BLE001 + self.logger.warning(f"[{self.name}] embedding failed, falling back to exact dedup: {e}") + normalized = normalize_topic(str(candidate.get("title") or "")) + return ("duplicate" if normalized in comparison_norms else "keep"), "", 0.0 + if not vectors or len(vectors) != len(texts): + return "keep", "", 0.0 + best, best_idx = 0.0, -1 + for idx, other in enumerate(vectors[1:]): + similarity = _cosine(vectors[0], other) + if similarity > best: + best, best_idx = similarity, idx + matched = comparison[best_idx][0] if 0 <= best_idx < len(comparison) else "" + if best >= self.known_threshold: + return "known", matched, best + return "keep", matched, best + + @staticmethod + def _expiry_cutoff(day: str, carry_forward_days: int) -> str: + """ISO cutoff matching trim_state_file: first_seen <= cutoff is over-age.""" + try: + base = dt.date.fromisoformat(day) + except ValueError: + return "" + return (base - dt.timedelta(days=max(int(carry_forward_days), 0))).isoformat() + + @staticmethod + def _evidence_date(path: str) -> str: + """Date embedded in a daily evidence path; '' when unparseable (v5.2). + + Lets last_evidence_at reflect when the evidence actually happened + (daily/2026-08-12/x.md -> 2026-08-12) instead of always today, so an + update anchored on an older file cannot game the freshness sort key. + """ + match = _EVIDENCE_DATE_RE.search(path or "") + if match: + try: + return dt.date.fromisoformat(match.group(1)).isoformat() + except ValueError: + return "" + return "" + + @staticmethod + def _embedding_fingerprint(embedder) -> str: + """`model@dimensions` of the resolved embedder; '' when not introspectable.""" + model = "" + kwargs = getattr(embedder, "kwargs", None) + if isinstance(kwargs, dict): + model = str(kwargs.get("model") or "") + if not model: + model = str(getattr(getattr(embedder, "model", None), "model", "") or "") + try: + dimensions = int(getattr(embedder, "dimensions", 0) or 0) + except Exception: # noqa: BLE001 - property may raise RuntimeError pre-init + dimensions = 0 + return f"{model}@{dimensions}" if model and dimensions else "" + + def _store(self, state: ProactiveState) -> None: + assert self.context is not None + data = state.model_dump() + self.context["proactive"] = data + self.context.response.metadata["proactive"] = data + + +def _embed_text(candidate: dict) -> str: + """Vector text for a candidate (v5.2 calibration: title+reason).""" + title = str(candidate.get("title") or "") + reason = str(candidate.get("reason") or "").strip() + if reason: + return f"{title}。{reason}" + keywords = ", ".join(str(k) for k in candidate.get("keywords") or []) + return f"{title} | {keywords}" if keywords else title + + +def _known_text(title: str, reason: str) -> str: + """Vector text for a known topic: title+reason when available, else title.""" + reason = str(reason or "").strip() + return f"{title}。{reason}" if reason else title + + +_EVIDENCE_DATE_RE = re.compile(r"(?:^|/)(\d{4}-\d{2}-\d{2})/") + + +def _cosine(a: list[float], b: list[float]) -> float: + if not a or not b or len(a) != len(b): + return 0.0 + dot = sum(x * y for x, y in zip(a, b)) + norm_a = sum(x * x for x in a) ** 0.5 + norm_b = sum(x * x for x in b) ** 0.5 + if norm_a == 0.0 or norm_b == 0.0: + return 0.0 + return dot / (norm_a * norm_b) diff --git a/reme/steps/evolve/proactive/utils.py b/reme/steps/evolve/proactive/utils.py new file mode 100644 index 00000000..95ff6d42 --- /dev/null +++ b/reme/steps/evolve/proactive/utils.py @@ -0,0 +1,473 @@ +"""Shared proactive helpers: frozen topic identity, truth-source state, rendering. + +Implements the A7 skeleton from PROACTIVE_SPEC.md. ``normalize_topic`` is a +frozen contract (INV-4): any change to it drifts every historical topic id. +""" + +import contextlib +import datetime as dt +import hashlib +import os +import re +import tempfile +import time +import unicodedata +from pathlib import Path + +import yaml + +from ....enumeration import ComponentEnum +from ....schema import ProactiveStateFile, ProactiveTopic +from ....schema.proactive import clamp_confidence +from ....utils import get_logger +from ...file_io._file_io import get_path_lock +from .._evolve import now +from ..dream.utils import clean_paths, recent_dates, scan_day_files + +logger = get_logger(log_to_file=False) + +PROACTIVE_STATE_NAME = "_proactive.yaml" +INTERESTS_NAME = "interests.yaml" +_RESOURCE_EXTS = {".md", ".txt", ".json", ".jsonl", ".csv", ".yaml", ".yml", ".html"} + + +# --------------------------------------------------------------------------- +# Frozen identity contract (A7 / INV-4) +# --------------------------------------------------------------------------- + + +def normalize_topic(title: str) -> str: + """NFKC -> casefold -> keep only chars whose category starts with L/N. + + Frozen contract (INV-4): removing all whitespace/punctuation means any + modification would drift every historical topic id. + """ + text = unicodedata.normalize("NFKC", title or "").casefold() + return "".join(ch for ch in text if unicodedata.category(ch)[0] in ("L", "N")) + + +def topic_id(title: str) -> str: + """Stable topic identity: ``sha1(normalize_topic(title))[:12]``.""" + return hashlib.sha1(normalize_topic(title).encode("utf-8")).hexdigest()[:12] + + +# --------------------------------------------------------------------------- +# Paths and material set M (F2.0) +# --------------------------------------------------------------------------- + + +def state_file_path(ws: Path, daily: str = "daily") -> Path: + """Truth-source path ``daily/_proactive.yaml``.""" + return ws / daily / PROACTIVE_STATE_NAME + + +def interests_path_for(ws: Path, daily: str, day: str) -> Path: + """Exposure-product path ``daily//interests.yaml``.""" + return ws / daily / day / INTERESTS_NAME + + +def norm_path(rel) -> str: + """Normalize a workspace-relative path (posix, no leading ./).""" + text = str(rel or "").strip().replace("\\", "/") + while text.startswith("./"): + text = text[2:] + return text + + +def scan_material_daily(ws: Path, day: str, daily: str, scan_days: int) -> list[str]: + """M_daily: chunk notes in the scan window, minus day indexes and ``_*`` files (INV-11).""" + out: list[str] = [] + for scan_day in recent_dates(day, scan_days): + day_index = f"{daily}/{scan_day}.md" + for rel in scan_day_files(ws, scan_day, daily, INTERESTS_NAME): + rel = norm_path(rel) + base = rel.rsplit("/", 1)[-1] + if rel == day_index or base.startswith("_"): + continue + if rel not in out: + out.append(rel) + return sorted(out) + + +def scan_material_resource(ws: Path, lookback_days: int, max_files: int, now_dt: dt.datetime) -> list[str]: + """M_resource: recent uploads under ``resource/``, newest mtime first.""" + root = ws / "resource" + if not root.is_dir(): + return [] + cutoff = now_dt.timestamp() - max(int(lookback_days), 0) * 86400 + found: list[tuple[float, str]] = [] + for path in root.rglob("*"): + if not path.is_file() or path.suffix.lower() not in _RESOURCE_EXTS: + continue + try: + mtime = path.stat().st_mtime + except OSError: + continue + if mtime >= cutoff: + found.append((mtime, norm_path(path.relative_to(ws).as_posix()))) + found.sort(key=lambda item: (-item[0], item[1])) + return [rel for _, rel in found[: max(int(max_files), 0)]] + + +# --------------------------------------------------------------------------- +# Truth-source state file daily/_proactive.yaml (F1.3) +# --------------------------------------------------------------------------- + + +def load_state(ws: Path, daily: str = "daily") -> tuple[ProactiveStateFile, bool]: + """Load the truth source; returns ``(state_file, needs_bootstrap)``. + + A missing file means first run (fresh workspace or upgrade) and triggers + the one-time F1.4 bootstrap from interests.yaml history. A corrupt or + invalid file rebuilds empty WITHOUT bootstrap (spec F1.3/A2/A5), as does + an existing file that already carries the ``open_topics`` key (an empty + list is a normal state, not a trigger). + """ + path = state_file_path(ws, daily) + if not path.is_file(): + logger.info(f"proactive state file missing, first-run bootstrap scheduled: {path}") + return ProactiveStateFile(), True + try: + data = yaml.safe_load(path.read_text(encoding="utf-8")) + if not isinstance(data, dict): + raise ValueError("state file is not a mapping") + except Exception as e: # noqa: BLE001 + logger.warning(f"proactive state file corrupt, rebuilding empty: {path} ({e})") + return ProactiveStateFile(), False + needs_bootstrap = "open_topics" not in data + try: + state = ProactiveStateFile.model_validate(data) + except Exception as e: # noqa: BLE001 + logger.warning(f"proactive state file invalid, rebuilding empty: {path} ({e})") + return ProactiveStateFile(), False + return state, needs_bootstrap + + +async def save_state(ws: Path, state_file: ProactiveStateFile, daily: str = "daily") -> None: + """Atomically persist the truth source (path lock + tmp file + os.replace).""" + path = state_file_path(ws, daily) + lock = await get_path_lock(path) + async with lock: + path.parent.mkdir(parents=True, exist_ok=True) + rendered = yaml.safe_dump(state_file.model_dump(), allow_unicode=True, sort_keys=False) + fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp") + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(rendered if rendered.endswith("\n") else f"{rendered}\n") + os.replace(tmp, path) + except Exception: + with contextlib.suppress(OSError): + os.unlink(tmp) + raise + + +def _safe_date(text: str) -> dt.date | None: + try: + return dt.date.fromisoformat(str(text or "").strip()) + except ValueError: + return None + + +async def load_carry_forward( + ws: Path, + state_file: ProactiveStateFile, + day: str, + days: int, + top_k: int, + daily: str = "daily", + needs_bootstrap: bool = False, +) -> tuple[list[ProactiveTopic], list[ProactiveTopic]]: + """Return ``(carry_forward_all, carry_forward_prompt)`` sorted per A4 rule 1. + + Bootstraps the truth source from interests.yaml history exactly once on + first run (missing state file) or when an existing file lacks the + ``open_topics`` key (F1.4). Over-age topics are dropped here with a log; + resolved ids are suppressed. + """ + if needs_bootstrap: + state_file.open_topics = _bootstrap_from_history(ws, day, days, daily) + await save_state(ws, state_file, daily) + resolved_ids = {str(r.get("id") or "") for r in state_file.resolved if isinstance(r, dict)} + base = _safe_date(day) + open_topics: list[ProactiveTopic] = [] + expired = 0 + for topic in state_file.open_topics: + if topic.id and topic.id in resolved_ids: + continue + first_seen = _safe_date(topic.first_seen) + if base is not None and first_seen is not None and (base - first_seen).days > int(days): + expired += 1 + continue + open_topics.append(topic) + if expired: + logger.info(f"proactive carry-forward dropped {expired} over-age topic(s) (window={days}d)") + ordered = sort_topics(open_topics) + return ordered, ordered[: max(int(top_k), 0)] + + +def _bootstrap_from_history(ws: Path, day: str, days: int, daily: str) -> list[ProactiveTopic]: + """One-time bootstrap: newest record per id wins, first_seen takes the min (F1.4).""" + if _safe_date(day) is None: + return [] + records: dict[str, ProactiveTopic] = {} + first_seen: dict[str, str] = {} + for file_date in reversed(recent_dates(day, days)): # newest -> oldest + data = read_interests_data(interests_path_for(ws, daily, file_date)) + if not data: + continue + topics, _is_v1, _push = parse_interests_topics(data, file_date) + for topic in topics: + anchor = topic.first_seen or file_date + if topic.id not in first_seen or anchor < first_seen[topic.id]: + first_seen[topic.id] = anchor + records.setdefault(topic.id, topic) + out: list[ProactiveTopic] = [] + for tid, topic in records.items(): + topic.first_seen = first_seen.get(tid) or topic.first_seen or day + out.append(topic) + logger.info(f"proactive bootstrap built {len(out)} open topic(s) from interests.yaml history") + return out + + +def trim_state_file(state_file: ProactiveStateFile, day: str, days: int) -> None: + """Prune budget/exposure/resolved windows and over-age/resolved open topics.""" + base = _safe_date(day) + if base is None: + return + cutoff = (base - dt.timedelta(days=max(int(days), 0))).isoformat() + state_file.resolved = [ + r for r in state_file.resolved if isinstance(r, dict) and str(r.get("resolved_at") or "") > cutoff + ] + resolved_ids = {str(r.get("id") or "") for r in state_file.resolved} + kept: list[ProactiveTopic] = [] + for topic in state_file.open_topics: + if topic.id and topic.id in resolved_ids: + continue + first_seen = _safe_date(topic.first_seen) + if first_seen is not None and first_seen.isoformat() <= cutoff: + continue + kept.append(topic) + state_file.open_topics = kept + + +# --------------------------------------------------------------------------- +# interests.yaml read/render (F1.2 / A2 / A4) +# --------------------------------------------------------------------------- + + +def quarantine_interests(path: Path, error: Exception) -> None: + """Rename a corrupt interests.yaml aside (A2): ``interests.corrupt-.yaml``.""" + stamp = int(time.time()) + corrupt = path.with_name(f"interests.corrupt-{stamp}.yaml") + try: + path.rename(corrupt) + logger.warning(f"quarantined corrupt interests file {path} -> {corrupt.name}: {error}") + except OSError: + logger.warning(f"corrupt interests file {path}: {error}") + + +def read_interests_file(path: Path) -> tuple[str, dict] | None: + """Read interests.yaml returning ``(raw_text, data)``; quarantine corrupt files (A2).""" + if not path.is_file(): + return None + try: + raw_text = path.read_text(encoding="utf-8") + data = yaml.safe_load(raw_text) + if not isinstance(data, dict): + raise ValueError("interests.yaml is not a mapping") + return raw_text, data + except Exception as e: # noqa: BLE001 + quarantine_interests(path, e) + return None + + +def read_interests_data(path: Path) -> dict | None: + """Parse interests.yaml; quarantine corrupt files (A2) and return None.""" + loaded = read_interests_file(path) + return loaded[1] if loaded else None + + +def parse_interests_topics(data: dict, file_date: str) -> tuple[list[ProactiveTopic], bool, bool]: + """Return ``(topics, is_v1, push)`` with A2 fallbacks applied. + + Missing ``first_seen``/``last_evidence_at`` fall back to the file date + (not today); missing ids are derived from the frozen title hash. + """ + is_v1 = data.get("version") is None + push = data.get("push", True) + if not isinstance(push, bool): + push = True + raw_topics = data.get("topics") or [] + topics: list[ProactiveTopic] = [] + for raw in raw_topics if isinstance(raw_topics, list) else []: + if not isinstance(raw, dict): + continue + title = str(raw.get("title") or "").strip() + reason = str(raw.get("reason") or "").strip() + if not title or not reason: + continue + topics.append( + ProactiveTopic( + id=str(raw.get("id") or "").strip() or topic_id(title), + title=title, + reason=reason, + kind=raw.get("kind", "interest_extend"), + confidence=raw.get("confidence", 0.5), + first_seen=str(raw.get("first_seen") or "").strip() or file_date, + last_evidence_at=str(raw.get("last_evidence_at") or "").strip() or file_date, + evidence=str(raw.get("evidence") or "").strip()[:120], + keywords=raw.get("keywords") or [], + paths=raw.get("paths") or [], + ), + ) + return topics, is_v1, push + + +def sort_topics(topics: list) -> list: + """Order: last_evidence_at desc -> follow_up first -> confidence desc -> id asc. + + Freshness is the primary key (v5 aging fix): stale topics sink below newly + evidenced ones of any kind, so long-lived follow_ups cannot permanently + crowd out new discoveries. + """ + + def get(topic, key): + return getattr(topic, key) if not isinstance(topic, dict) else topic.get(key) + + out = sorted(topics, key=lambda t: str(get(t, "id") or "")) + out.sort(key=lambda t: clamp_confidence(get(t, "confidence")), reverse=True) + out.sort(key=lambda t: 0 if get(t, "kind") == "follow_up" else 1) + out.sort(key=lambda t: str(get(t, "last_evidence_at") or ""), reverse=True) + return out + + +def dump_topic(topic) -> dict: + """Render one topic as an A2-ordered dict for interests.yaml v2.""" + get = (lambda k: getattr(topic, k)) if not isinstance(topic, dict) else topic.get + return { + "id": str(get("id") or ""), + "title": str(get("title") or ""), + "reason": str(get("reason") or ""), + "kind": str(get("kind") or "interest_extend"), + "confidence": clamp_confidence(get("confidence")), + "first_seen": str(get("first_seen") or ""), + "last_evidence_at": str(get("last_evidence_at") or ""), + "evidence": str(get("evidence") or "")[:120], + "keywords": [str(k) for k in (get("keywords") or [])], + "paths": [str(p) for p in (get("paths") or [])], + } + + +def render_interests(day: str, topics: list, push: bool, now_dt: dt.datetime) -> dict: + """Render the full v2 file content from the truth source (INV-6). + + v5: ``skip_reason`` is no longer persisted (no consumer); it survives as + structured log/metadata on ``ProactiveState.file_skip_reason`` (R7). + """ + return { + "version": 2, + "date": day, + "generated_at": now_dt.isoformat(timespec="seconds"), + "push": bool(push), + "topics": [dump_topic(t) for t in topics], + } + + +def write_interests_if_changed(ws: Path, path: Path, rendered: dict) -> bool: # pylint: disable=unused-argument + """Apply A4 render-write rules (idempotent skip + atomic replace). + + v5 (R1): the "push=false never overwrites nightly v1" special case is + gone; ``push`` is derived from the cumulative truth source, so re-renders + are monotonic and need no guard. + """ + existing: dict | None = None + if path.is_file(): + existing = read_interests_data(path) + if existing is not None: + existing_push = existing.get("push", True) + if not isinstance(existing_push, bool): + existing_push = True + if existing_push == bool(rendered.get("push")) and existing.get("topics") == rendered.get("topics"): + return False + path.parent.mkdir(parents=True, exist_ok=True) + payload = yaml.safe_dump(rendered, allow_unicode=True, sort_keys=False) + fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp") + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(payload if payload.endswith("\n") else f"{payload}\n") + os.replace(tmp, path) + except Exception: + with contextlib.suppress(OSError): + os.unlink(tmp) + raise + return True + + +# --------------------------------------------------------------------------- +# Candidate cleaning (extract side, A3) +# --------------------------------------------------------------------------- + + +def clean_candidate(raw, allowed_paths: set[str], kind: str, day: str) -> dict: + """Clean one LLM candidate; drop entries whose paths escape M (INV-8).""" + if not isinstance(raw, dict): + return {} + title = str(raw.get("title") or "").strip() + reason = str(raw.get("reason") or "").strip() + paths = clean_paths(raw.get("paths"), allowed_paths) + if not title or not reason or not paths: + return {} + keywords = raw.get("keywords") or [] + keywords = [str(k).strip() for k in keywords if str(k).strip()] if isinstance(keywords, list) else [] + return { + "id": topic_id(title), + "title": title, + "reason": reason, + "kind": kind, + "confidence": clamp_confidence(raw.get("confidence")), + "first_seen": day, + "last_evidence_at": day, + "evidence": str(raw.get("evidence") or "").strip()[:120], + "keywords": keywords, + "paths": paths, + } + + +def current_now(step) -> dt.datetime: + """Business-time access per INV-3 (timezone-aware; never datetime.now()).""" + tz = step.app_context.app_config.timezone if step.app_context is not None else None + return now(tz) + + +def parse_extract_reply(text: str) -> dict: + """Parse the A3 fenced YAML/JSON output; fenced blocks take priority. + + Unlike the dream parser there is no scalar-mapping fallback: proactive + output is sectioned lists, and a partial fallback would corrupt updates. + """ + candidates = [m.group(1).strip() for m in re.finditer(r"```(?:json|ya?ml)?\s*(.*?)```", text, re.S | re.I)] + candidates.append((text or "").strip()) + for raw in candidates: + if not raw: + continue + try: + data = yaml.safe_load(raw) + except yaml.YAMLError: + continue + if isinstance(data, dict) and data: + return data + return {} + + +def resolve_agent_wrapper(step): + """Return the step's agent_wrapper, falling back to the app default (F4.4).""" + wrapper = step.agent_wrapper + if wrapper is not None: + return wrapper + if step.app_context is not None: + fallback = step.app_context.components.get(ComponentEnum.AGENT_WRAPPER, {}).get("default") + if fallback is not None: + configured = step.kwargs.get("agent_wrapper", "default") + step.logger.warning(f"[{step.name}] agent_wrapper '{configured}' missing; using default") + return fallback + return None diff --git a/reme/steps/evolve/wait_for_idle.py b/reme/steps/evolve/wait_for_idle.py new file mode 100644 index 00000000..460825a6 --- /dev/null +++ b/reme/steps/evolve/wait_for_idle.py @@ -0,0 +1,94 @@ +"""Idle gate: wait until trunk jobs are quiet before running low-priority work (F4.2).""" + +import asyncio +import fnmatch +import time + +from ..base_step import BaseStep +from ...components import R + +DEFAULT_BUSY_JOB_PATTERNS = ["auto_memory*", "auto_resource*", "dream_cron", "optimize_index_cron"] + + +@R.register("wait_for_idle_step") +class WaitForIdleStep(BaseStep): + """Hold until all busy-trunk jobs are idle; give up the round after ``max_wait``. + + Giving up is not a failure: ``response.success`` stays True and the + short-circuit flag ``context[skip_key] = {"reason": "busy"}`` lets the + downstream steps of this job pass through (F4.3). + """ + + def __init__( + self, + busy_job_patterns: list[str] | None = None, + quiet_window: float = 120, + poll_interval: float = 10, + max_wait: float = 600, + skip_key: str = "proactive_skip", + **kwargs, + ): + super().__init__(**kwargs) + self.busy_job_patterns = list(busy_job_patterns or DEFAULT_BUSY_JOB_PATTERNS) + self.quiet_window = float(quiet_window) + self.poll_interval = max(float(poll_interval), 0.1) + self.max_wait = float(max_wait) + self.skip_key = skip_key + + async def execute(self): + assert self.context is not None + metadata = getattr(self.app_context, "metadata", None) + metadata = metadata if isinstance(metadata, dict) else {} + stop_event = getattr(self.context, "stop_event", None) + deadline = time.monotonic() + self.max_wait + self.logger.info( + f"[{self.name}] start patterns={self.busy_job_patterns} quiet_window={self.quiet_window}s " + f"max_wait={self.max_wait}s skip_key={self.skip_key}", + ) + while True: + busy = self._busy_jobs(metadata) + if not busy: + self.logger.info(f"[{self.name}] trunk idle; proceeding") + self.context.response.success = True + self.context.response.answer = "Trunk idle; proceeding" + return self.context.response + if stop_event is not None and stop_event.is_set(): + self.logger.info(f"[{self.name}] stop requested while waiting; giving up this round") + return self._give_up(busy, "stop_requested") + if time.monotonic() >= deadline: + self.logger.info(f"[{self.name}] still busy after max_wait={self.max_wait}s: {busy}") + return self._give_up(busy, "busy") + self.logger.info(f"[{self.name}] trunk busy {busy}; polling again in {self.poll_interval}s") + await self._sleep_or_stop(stop_event, self.poll_interval) + + def _give_up(self, busy: list[str], reason: str): + assert self.context is not None + self.context[self.skip_key] = {"reason": reason, "busy_jobs": busy} + self.context.response.success = True + self.context.response.answer = f"Skipped: trunk busy ({', '.join(busy)}); giving up this round" + return self.context.response + + async def _sleep_or_stop(self, stop_event, delay: float) -> None: + if isinstance(stop_event, asyncio.Event): + try: + await asyncio.wait_for(stop_event.wait(), timeout=delay) + except asyncio.TimeoutError: + pass + return + await asyncio.sleep(delay) + + def _busy_jobs(self, metadata: dict) -> list[str]: + """Return names of matching jobs that are running or inside the quiet window.""" + last_run = metadata.get("__job_last_run") or {} + now = time.monotonic() + busy: list[str] = [] + for name, info in last_run.items(): + if not isinstance(info, dict): + continue + if not any(fnmatch.fnmatch(name, pattern) for pattern in self.busy_job_patterns): + continue + last_end = info.get("last_end") + recently_active = isinstance(last_end, (int, float)) and now - float(last_end) <= self.quiet_window + if info.get("running") or recently_active: + busy.append(name) + return sorted(busy) diff --git a/tests/unit/test_auto_dream.py b/tests/unit/test_auto_dream.py index 9509d67d..cfe8a6eb 100644 --- a/tests/unit/test_auto_dream.py +++ b/tests/unit/test_auto_dream.py @@ -14,7 +14,7 @@ from reme.components.runtime_context import RuntimeContext from reme.schema import DreamState, FileNode from reme.steps.evolve.dream.extract import DreamExtractStep from reme.steps.evolve.dream.finish import DreamFinishStep -from reme.steps.evolve.dream.proactive import ProactiveStep +from reme.steps.evolve.proactive.proactive import ProactiveStep from reme.steps.evolve.dream.topics import DreamTopicsStep from reme.steps.evolve.dream.utils import parse_structured_reply, recent_dates, scan_day_files @@ -332,7 +332,7 @@ def test_proactive_keeps_skipped_and_error_answers_explicit(tmp_path): assert skipped.metadata["skipped"] is True _touch(tmp_path / "daily" / "2026-05-28" / "interests.yaml", "topics: []\n") - with patch("reme.steps.evolve.dream.proactive.load_yaml_topics", side_effect=ValueError("bad topics")): + with patch("reme.steps.evolve.proactive.proactive.load_yaml_topics", side_effect=ValueError("bad topics")): failed = await step(RuntimeContext(date="2026-05-28", file_store=_FileStore(tmp_path))) assert failed.success is False diff --git a/tests/unit/test_proactive_refresh.py b/tests/unit/test_proactive_refresh.py new file mode 100644 index 00000000..62994dd7 --- /dev/null +++ b/tests/unit/test_proactive_refresh.py @@ -0,0 +1,1479 @@ +"""Unit tests for the proactive refresh chain (PROACTIVE_SPEC.md A8 mapping). + +Coverage: F1 contracts, F2 discovery, F3 isolation, F4 idle gate/budget/timeout, +F5 read extensions, plus the M2 extends branch and semantic dedup. +""" + +import asyncio +import os +import tempfile +import time +from pathlib import Path +from unittest.mock import patch + +import yaml +from agentscope.model import ChatModelBase + +from reme.components.agent_wrapper.base_agent_wrapper import BaseAgentWrapper +from reme.components.application_context import ApplicationContext +from reme.components.file_catalog import BaseFileCatalog +from reme.components.file_store import BaseFileStore +from reme.components.runtime_context import RuntimeContext +from reme.schema import FileNode, ProactiveTopic, TopicUpdate +from reme.steps.evolve.dream.extract import DreamExtractStep +from reme.steps.evolve.proactive.extract import ProactiveExtractStep +from reme.steps.evolve.proactive.finish import ProactiveFinishStep +from reme.steps.evolve.proactive.proactive import ProactiveStep +from reme.steps.evolve.proactive.topics import ProactiveTopicsStep +from reme.steps.evolve.proactive.utils import ( + load_carry_forward, + load_state, + normalize_topic, + parse_interests_topics, + scan_material_daily, + topic_id, +) +from reme.steps.evolve.wait_for_idle import WaitForIdleStep + +DAY = "2026-08-13" + + +def _touch(path: Path, text: str = "x") -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return path + + +class _Catalog(BaseFileCatalog): + """In-memory catalog stub recording every mutation.""" + + def __init__(self, nodes=None): + super().__init__() + self.nodes = list(nodes or []) + self.upserts = [] + self.dumps = 0 + self.deleted = [] + + async def upsert(self, nodes): + self.upserts.extend(nodes) + known = {n.path: n for n in self.nodes} + for node in nodes: + known[node.path] = node + self.nodes = list(known.values()) + + async def delete(self, path): + paths = path if isinstance(path, list) else [path] + self.deleted.extend(paths) + drop = set(paths) + self.nodes = [n for n in self.nodes if n.path not in drop] + + async def get_nodes(self, paths=None): + if paths is None: + return list(self.nodes) + wanted = set(paths) + return [n for n in self.nodes if n.path in wanted] + + async def dump(self): + self.dumps += 1 + + +class _FileStore(BaseFileStore): + def __init__(self, workspace: Path): + super().__init__() + self._workspace_path = workspace + + @property + def workspace_path(self) -> Path: + return self._workspace_path + + async def upsert(self, files): + return None + + async def delete(self, path): + return None + + async def clear(self): + return None + + async def get_nodes(self, paths=None): + return [] + + async def get_outlinks(self, path, scope=None): + return [] + + async def get_inlinks(self, path, scope=None): + return [] + + async def vector_search(self, query, limit, search_filter): + return [] + + async def keyword_search(self, query, limit, search_filter): + return [] + + +class _AgentWrapper(BaseAgentWrapper): + """Deterministic LLM stub with scripted replies and call accounting.""" + + def __init__(self, replies=None, delay: float = 0.0): + super().__init__() + self.replies = list(replies or []) + self.delay = delay + self.calls = 0 + self.last_inputs = [] + self.last_system_prompts = [] + + async def reply(self, inputs, **kwargs): + self.calls += 1 + self.last_inputs.append(inputs) + self.last_system_prompts.append(str(kwargs.get("system_prompt") or "")) + item = self.replies.pop(0) if self.replies else "" + if isinstance(item, tuple): + text, delay = item + else: + text, delay = item, self.delay + if delay: + await asyncio.sleep(delay) + return { + "session_id": "stub", + "result": text, + "last_message": {"content": [{"type": "text", "text": text}]}, + } + + +class _DummyModel(ChatModelBase): + """Placeholder model satisfying dream extract's llm_available check.""" + + def __init__(self): + pass + + +class _FakeEmbedding: + """Scripted embedding oracle: exact text -> vector, else orthogonal default. + + Declares ``kwargs``/``dimensions`` so the topics step can fingerprint the + "model" against ``known_threshold_calibrated_for`` (defaults match). + """ + + def __init__(self, vectors: dict[str, list[float]], model: str = "text-embedding-v4", dimensions: int = 1024): + self.vectors = vectors + self.calls = 0 + self.kwargs = {"model": model} + self.dimensions = dimensions + + async def __call__(self, inputs, **kwargs): + self.calls += 1 + return [list(self.vectors.get(text, [0.0, 1.0])) for text in inputs] + + +def _reply(follow_ups=None, extends=None, updates=None) -> str: + doc = {} + if follow_ups is not None: + doc["follow_ups"] = follow_ups + if extends is not None: + doc["extends"] = extends + if updates is not None: + doc["updates"] = updates + body = yaml.safe_dump(doc, allow_unicode=True, sort_keys=False) + return f"```yaml\n{body}```" + + +def _topic(title, reason="because", confidence=0.7, paths=None, evidence=None, keywords=None): + return { + "title": title, + "reason": reason, + "confidence": confidence, + "evidence": evidence or (paths[0] if paths else ""), + "keywords": keywords or [], + "paths": paths or [], + } + + +def _state_topic(title, day=DAY, kind="follow_up", confidence=0.8, topic_id_value=None, status="open"): + return { + "id": topic_id_value or topic_id(title), + "title": title, + "reason": "seeded", + "kind": kind, + "confidence": confidence, + "status": status, + "first_seen": day, + "last_evidence_at": day, + "evidence": "", + "keywords": [], + "paths": [], + } + + +def _write_state(ws, budget=None, exposure=None, open_topics=None, resolved=None, omit_open_topics=False): + data = {"version": 1} + if budget is not None: + data["budget"] = budget + if exposure is not None: + data["exposure"] = exposure + if not omit_open_topics: + data["open_topics"] = open_topics or [] + if resolved is not None: + data["resolved"] = resolved + _touch(ws / "daily" / "_proactive.yaml", yaml.safe_dump(data, allow_unicode=True, sort_keys=False)) + + +def _read_state(ws) -> dict: + return yaml.safe_load((ws / "daily" / "_proactive.yaml").read_text(encoding="utf-8")) + + +def _interests(ws, day=DAY) -> Path: + return ws / "daily" / day / "interests.yaml" + + +def _write_interests_v1(ws, day, topics_yaml: list[dict]): + payload = {"date": day, "topic_count": len(topics_yaml), "topics": topics_yaml} + _touch(_interests(ws, day), yaml.safe_dump(payload, allow_unicode=True, sort_keys=False)) + + +def _write_interests_v2(ws, day, topics: list[dict], push=True, skip_reason=""): + payload = { + "version": 2, + "date": day, + "generated_at": f"{day}T09:00:00+08:00", + "push": push, + "skip_reason": skip_reason, + "topics": topics, + } + _touch(_interests(ws, day), yaml.safe_dump(payload, allow_unicode=True, sort_keys=False)) + + +async def _run_chain( + ws, + replies=None, + *, + wrapper=None, + catalog=None, + app=None, + extract_kwargs=None, + topics_kwargs=None, + extra_ctx=None, + context=None, +): + app = app or ApplicationContext(workspace_dir=str(ws)) + wrapper = wrapper or _AgentWrapper(replies or []) + catalog = catalog or _Catalog() + context = context or RuntimeContext( + date=DAY, + file_catalog=catalog, + file_store=_FileStore(ws), + agent_wrapper=wrapper, + **(extra_ctx or {}), + ) + extract = ProactiveExtractStep(app_context=app, **(extract_kwargs or {})) + resp_extract = await extract(context) + topics = ProactiveTopicsStep(app_context=app, **(topics_kwargs or {})) + resp_topics = await topics(context) + finish = ProactiveFinishStep(app_context=app) + resp_finish = await finish(context) + return app, wrapper, catalog, context, (resp_extract, resp_topics, resp_finish) + + +def test_known_threshold_model_binding(tmp_path): + """known_threshold is model-bound (v5.1): fingerprint mismatch degrades. + + The oracle maps every text to the same vector (sim 1.0 >= known_threshold), + but the "model" differs from the calibration, so the semantic gate must be + disabled: a paraphrase survives while an exact normalize duplicate of a + historical interests title still drops. + """ + + async def run(): + ws = tmp_path + _touch(ws / "daily" / DAY / "session.md", "notes") + _write_state(ws, open_topics=[_state_topic("检索引用质量评测")]) + _write_interests_v1(ws, "2026-08-12", [{"title": "向量数据库选型", "reason": "done"}]) + embedder = _FakeEmbedding({}, model="some-other-model-v9") + reply = _reply( + follow_ups=[ + _topic("检索引用可信度的评测方法", paths=[f"daily/{DAY}/session.md"]), + _topic("向量数据库选型", paths=[f"daily/{DAY}/session.md"]), + ], + ) + _, _, _, context, _ = await _run_chain(ws, [reply], extra_ctx={"as_embedding": embedder}) + state = context.get("proactive") + assert state["dropped_known"] == 0 # semantic gate disabled by mismatch + assert state["dropped_duplicate"] == 1 # exact compare still active + titles = [t["title"] for t in _read_state(ws)["open_topics"]] + assert "检索引用可信度的评测方法" in titles + assert "向量数据库选型" not in titles + + asyncio.run(run()) + + +def test_overage_remention_restarts(tmp_path): + """Over-age open topics re-mentioned today restart instead of vanishing. + + Same-id candidate refreshes first_seen so trim keeps the topic (unfinished + business must keep being re-executed). After the restart the revived title + is back in the comparison set, so a paraphrase of it drops as known. + A dying topic with NO same-id candidate stays excluded from comparison: + its paraphrase re-mention survives and opens a fresh topic. + """ + + async def run(): + ws = tmp_path + _touch(ws / "daily" / DAY / "session.md", "notes") + old_day = "2026-07-28" # 16 days before DAY, over-age at carry_forward_days=14 + _write_state( + ws, + open_topics=[_state_topic("旧任务确认", day=old_day), _state_topic("遗留事项", day=old_day)], + ) + # Discriminative oracle: only the legacy-task / its-supplement pair scores sim 1.0; + # every other pair is orthogonal. If the dying title leaked into the + # comparison set, the paraphrase below would drop as known. + embedder = _FakeEmbedding({"遗留事项的补充。because": [1.0, 0.0], "遗留事项。seeded": [1.0, 0.0]}) + reply = _reply( + follow_ups=[ + _topic("旧任务确认", paths=[f"daily/{DAY}/session.md"]), + _topic("旧任务确认的后续安排", paths=[f"daily/{DAY}/session.md"]), + _topic("遗留事项的补充", paths=[f"daily/{DAY}/session.md"]), + ], + ) + _, _, _, context, _ = await _run_chain(ws, [reply], extra_ctx={"as_embedding": embedder}) + state = context.get("proactive") + assert state["dropped_known"] == 1 # paraphrase of the REVIVED topic (default sim 1.0 vs it) + topics = {t["id"]: t for t in _read_state(ws)["open_topics"]} + assert topics[topic_id("旧任务确认")]["first_seen"] == DAY # restarted, not trimmed + assert topic_id("旧任务确认的后续安排") not in topics # known vs revived topic + assert topic_id("遗留事项的补充") in topics # dying title excluded from comparison + assert topic_id("遗留事项") not in topics # over-age, never re-mentioned -> trimmed + + asyncio.run(run()) + + +def test_update_evidence_anchor(tmp_path): + """v5.2: action=update must anchor this round's new material (hard check). + + An update citing a file outside the round's changed material degrades to + keep (no evidence rewrite, no freshness refresh); an update citing a + changed file applies, and last_evidence_at is parsed from the evidence + path date instead of defaulting to today. + """ + + async def run(): + ws = tmp_path + _touch(ws / "daily" / DAY / "session.md", "today notes") + _touch(ws / "daily" / "2026-08-12" / "old-note.md", "yesterday notes") + _write_state( + ws, + open_topics=[_state_topic("证据改挂主题"), _state_topic("昨日证据主题")], + ) + reply = _reply( + updates=[ + {"id": topic_id("证据改挂主题"), "action": "update", "evidence": "daily/2026-08-11/absent.md"}, + { + "id": topic_id("昨日证据主题"), + "action": "update", + "evidence": "daily/2026-08-12/old-note.md#L5", + "confidence": 0.9, + }, + ], + ) + _, _, _, context, _ = await _run_chain(ws, [reply]) + state = context.get("proactive") + assert state["updates_applied"] == 1 + topics = {t["id"]: t for t in _read_state(ws)["open_topics"]} + rejected = topics[topic_id("证据改挂主题")] + assert rejected["last_evidence_at"] == DAY # untouched, degraded to keep + assert rejected["evidence"] != "daily/2026-08-11/absent.md" + applied = topics[topic_id("昨日证据主题")] + assert applied["last_evidence_at"] == "2026-08-12" # parsed from evidence path + assert applied["evidence"] == "daily/2026-08-12/old-note.md#L5" + assert applied["confidence"] == 0.9 + + asyncio.run(run()) + + +def test_delayed_association_context_window(tmp_path): + """v5.2 trigger/context split: consumed resources re-enter fired rounds. + + Round 1 consumes the resource (checkpointed). Round 2 is triggered by a new + daily note; the blob must carry the unchanged resource again so the extends + branch can associate the new discussion with the earlier upload. + """ + + async def run(): + ws = tmp_path + _touch(ws / "daily" / DAY / "session.md", "initial discussion") + resource = _touch(ws / "resource" / "survey.md", "survey on citation credibility") + resource_rel = resource.relative_to(ws).as_posix() + reply1 = _reply(follow_ups=[_topic("初始话题", paths=[f"daily/{DAY}/session.md"])]) + reply2 = _reply( + extends=[_topic("延迟关联主题", confidence=0.5, paths=[f"daily/{DAY}/1500-extra.md", resource_rel])], + ) + wrapper = _AgentWrapper([reply1, reply2]) + catalog = _Catalog() + + await _run_chain(ws, wrapper=wrapper, catalog=catalog) + assert wrapper.calls == 1 + + _touch(ws / "daily" / DAY / "1500-extra.md", "now discussing citation credibility") + _, _, _, context2, _ = await _run_chain(ws, wrapper=wrapper, catalog=catalog) + assert wrapper.calls == 2 + assert resource_rel in wrapper.last_inputs[1] # resource back in the blob as context + state = context2.get("proactive") + assert state["early_exit"] in (None, "") + extends = state["extends"] + assert len(extends) == 1 and resource_rel in extends[0]["paths"] + + asyncio.run(run()) + + +def test_no_embedding_configured(tmp_path): + """No as_embedding anywhere: the chain runs and dedups exactly (BM25-only). + + Paraphrase-level candidates survive (no semantic gate), exact normalize + duplicates of historical exposure still drop; nothing crashes or skips. + """ + + async def run(): + ws = tmp_path + _touch(ws / "daily" / DAY / "session.md", "notes") + _write_state(ws, open_topics=[_state_topic("已知主题")]) + _write_interests_v1(ws, "2026-08-12", [{"title": "历史曝光主题", "reason": "x"}]) + reply = _reply( + follow_ups=[ + _topic("已知主题的不同说法", paths=[f"daily/{DAY}/session.md"]), + _topic("历史曝光主题", paths=[f"daily/{DAY}/session.md"]), + ], + ) + _, _, _, context, _ = await _run_chain(ws, [reply]) # no as_embedding in context/app + state = context.get("proactive") + assert state["dropped_known"] == 0 + assert state["dropped_duplicate"] == 1 + titles = [t["title"] for t in _read_state(ws)["open_topics"]] + assert "已知主题的不同说法" in titles + assert "历史曝光主题" not in titles + + asyncio.run(run()) + + +# --------------------------------------------------------------------------- +# F1 contracts +# --------------------------------------------------------------------------- + + +def test_normalize_and_topic_id_frozen(): + """A7 frozen contract: NFKC+casefold, keep L/N only; id is a 12-hex sha1 prefix.""" + assert normalize_topic("Memory Search: 可解释性评估!") == "memorysearch可解释性评估" + assert topic_id("A") == topic_id("a") == topic_id(" A ") + assert topic_id("A") != topic_id("B") + assert len(topic_id("anything")) == 12 + + +def test_legacy_file_parse(tmp_path): + """v1 files parse without error; new fields take deterministic defaults (A2).""" + content = ( + 'date: "2026-08-13"\n' + "topic_count: 1\n" + "topics:\n" + " - title: Retrieval quality\n" + " reason: Search behavior changed.\n" + " evidence: daily/2026-08-13/session.md\n" + ) + _touch(_interests(tmp_path), content) + topics, is_v1, push = parse_interests_topics(yaml.safe_load(content), DAY) + assert is_v1 is True and push is True + topic = topics[0] + assert topic.id == topic_id("Retrieval quality") + assert topic.kind == "interest_extend" + assert topic.confidence == 0.5 + assert topic.first_seen == DAY + assert topic.last_evidence_at == DAY + + async def run(): + step = ProactiveStep(app_context=ApplicationContext(workspace_dir=str(tmp_path))) + response = await step(RuntimeContext(date=DAY, include_content=False, file_store=_FileStore(tmp_path))) + assert response.success is True + assert response.metadata["push"] is True + + asyncio.run(run()) + + +def test_field_fallback(): + """Illegal kind/status/confidence/action values fall back deterministically.""" + topic = ProactiveTopic(title="T", reason="R", kind="bogus", confidence="not-a-number") + assert topic.kind == "interest_extend" + assert topic.confidence == 0.5 + assert ProactiveTopic(confidence=1.7).confidence == 1.0 + assert ProactiveTopic(confidence=-3).confidence == 0.0 + assert TopicUpdate(id="x", action="???").action == "keep" + + +def test_resolved_registry(tmp_path): + """Resolved ids disappear from carry-forward and from F5 reads.""" + + async def run(): + ws = tmp_path + id_a, id_b = topic_id("Topic A"), topic_id("Topic B") + _write_state( + ws, + open_topics=[_state_topic("Topic A", topic_id_value=id_a), _state_topic("Topic B", topic_id_value=id_b)], + resolved=[{"id": id_b, "title": "Topic B", "resolved_at": DAY, "evidence": ""}], + ) + state_file, needs_bootstrap = load_state(ws) + assert needs_bootstrap is False + carry_all, _ = await load_carry_forward(ws, state_file, DAY, 14, 20) + assert [t.title for t in carry_all] == ["Topic A"] + + _write_interests_v2( + ws, + DAY, + [_state_topic("Topic A", topic_id_value=id_a), _state_topic("Topic B", topic_id_value=id_b)], + ) + step = ProactiveStep(app_context=ApplicationContext(workspace_dir=str(ws))) + response = await step(RuntimeContext(date=DAY, include_content=False, file_store=_FileStore(ws))) + assert [t["title"] for t in response.answer["topics"]] == ["Topic A"] + + asyncio.run(run()) + + +def test_carry_forward_bootstrap(tmp_path): + """First run (missing state file) bootstraps from interests.yaml history; first_seen = min.""" + + async def run(): + ws = tmp_path + _write_interests_v1(ws, "2026-08-11", [_topic("Topic T", paths=["daily/2026-08-11/a.md"])]) + _write_interests_v1( + ws, + "2026-08-12", + [_topic("Topic T", paths=["daily/2026-08-12/a.md"]), _topic("Topic U", paths=["daily/2026-08-12/b.md"])], + ) + + # Upgrade path: no _proactive.yaml yet -> one-time bootstrap from history. + state_file, needs_bootstrap = load_state(ws) + assert needs_bootstrap is True + carry_all, _ = await load_carry_forward(ws, state_file, DAY, 14, 20, needs_bootstrap=needs_bootstrap) + by_title = {t.title: t for t in carry_all} + assert set(by_title) == {"Topic T", "Topic U"} + assert by_title["Topic T"].first_seen == "2026-08-11" # min across days, not latest + assert by_title["Topic U"].first_seen == "2026-08-12" + + persisted = _read_state(ws) + assert len(persisted["open_topics"]) == 2 + _, needs_bootstrap2 = load_state(ws) + assert needs_bootstrap2 is False # no re-bootstrap once the key exists + + # Empty open_topics list is a normal state, not a bootstrap trigger. + _write_state(ws, open_topics=[]) + _, needs_bootstrap3 = load_state(ws) + assert needs_bootstrap3 is False + + # An existing file lacking the key (e.g. hand-edited) still triggers it. + _write_state(ws, budget={}, omit_open_topics=True) + _, needs_bootstrap4 = load_state(ws) + assert needs_bootstrap4 is True + + asyncio.run(run()) + + +def test_state_file_corrupt_no_bootstrap(tmp_path): + """Corrupt _proactive.yaml rebuilds empty WITHOUT bootstrap (spec F1.3/A2/A5).""" + + async def run(): + ws = tmp_path + _write_interests_v1(ws, "2026-08-12", [_topic("Topic U", paths=["daily/2026-08-12/b.md"])]) + _touch(ws / "daily" / "_proactive.yaml", "version: [unclosed") + + state_file, needs_bootstrap = load_state(ws) + assert needs_bootstrap is False + assert state_file.open_topics == [] + carry_all, _ = await load_carry_forward(ws, state_file, DAY, 14, 20, needs_bootstrap=needs_bootstrap) + assert carry_all == [] + + asyncio.run(run()) + + +def test_bootstrap_upgrade_chain(tmp_path): + """Upgrade path: the first refresh run seeds the truth source from interests history.""" + + async def run(): + ws = tmp_path + _write_interests_v1(ws, "2026-08-12", [_topic("历史主题", paths=["daily/2026-08-12/a.md"])]) + _touch(ws / "daily" / DAY / "s1.md", "new evidence") + reply = _reply(follow_ups=[_topic("新主题", confidence=0.9, paths=[f"daily/{DAY}/s1.md"])]) + _, _, _, context, responses = await _run_chain(ws, [reply]) + assert all(r.success for r in responses) + + carry = context.get("proactive")["carry_forward_all"] + assert [t["title"] for t in carry] == ["历史主题"] # history reached the chain context + + truth_titles = {t["title"] for t in _read_state(ws)["open_topics"]} + assert truth_titles == {"历史主题", "新主题"} + data = yaml.safe_load(_interests(ws).read_text(encoding="utf-8")) + assert {t["title"] for t in data["topics"]} == {"历史主题", "新主题"} + + asyncio.run(run()) + + +def test_carry_forward_expiry(tmp_path): + """Topics older than carry_forward_days drop from carry-forward and are pruned.""" + + async def run(): + ws = tmp_path + _write_state(ws, open_topics=[_state_topic("Old Topic", day="2026-07-01")]) + state_file, _ = load_state(ws) + carry_all, _ = await load_carry_forward(ws, state_file, DAY, 14, 20) + assert carry_all == [] + + _touch(ws / "daily" / DAY / "note.md", "new evidence") + reply = _reply(follow_ups=[], updates=[]) + _, _, _, context, responses = await _run_chain(ws, [reply]) + assert all(r.success for r in responses) + assert context.get("proactive")["carry_forward_all"] == [] + assert _read_state(ws)["open_topics"] == [] # pruned from the truth source + + asyncio.run(run()) + + +def test_material_filter(tmp_path): + """Day indexes, ``_*`` files and interests.yaml never enter the material set (INV-11).""" + ws = tmp_path + _touch(ws / "daily" / f"{DAY}.md", "day index") + _touch(ws / "daily" / DAY / "_hidden.md", "underscore file") + _touch(ws / "daily" / DAY / "interests.yaml", "topics: []\n") + _touch(ws / "daily" / DAY / "good.md", "material") + assert scan_material_daily(ws, DAY, "daily", 2) == [f"daily/{DAY}/good.md"] + + +# --------------------------------------------------------------------------- +# F2 discovery + chain behaviour +# --------------------------------------------------------------------------- + + +def test_refresh_chain_e2e(tmp_path): + """wait_for_idle -> extract -> topics -> finish runs end to end.""" + + async def run(): + ws = tmp_path + note = _touch(ws / "daily" / DAY / "session.md", "we discussed the eval plan; nothing landed") + reply = _reply( + follow_ups=[_topic("记忆检索的可解释性评估", confidence=0.9, paths=["daily/2026-08-13/session.md"])], + ) + app = ApplicationContext(workspace_dir=str(ws)) + wrapper = _AgentWrapper([reply]) + catalog = _Catalog() + context = RuntimeContext(date=DAY, file_catalog=catalog, file_store=_FileStore(ws), agent_wrapper=wrapper) + + wait = WaitForIdleStep(app_context=app, max_wait=5, poll_interval=0.05) + resp_wait = await wait(context) + assert resp_wait.success is True + assert "proactive_skip" not in context + + extract = ProactiveExtractStep(app_context=app) + resp_extract = await extract(context) + topics = ProactiveTopicsStep(app_context=app) + resp_topics = await topics(context) + finish = ProactiveFinishStep(app_context=app) + resp_finish = await finish(context) + + assert resp_extract.success and resp_topics.success and resp_finish.success + state = context.get("proactive") + assert state["llm_calls"] == 1 + assert state["push"] is True + assert state["file_skip_reason"] == "" + assert state["interests_written"] is True + + data = yaml.safe_load(_interests(ws).read_text(encoding="utf-8")) + assert data["version"] == 2 + assert data["push"] is True + assert data["topics"][0]["title"] == "记忆检索的可解释性评估" + assert data["topics"][0]["kind"] == "follow_up" + + truth = _read_state(ws) + assert len(truth["open_topics"]) == 1 + checkpoint_paths = {n.path for n in catalog.upserts} + assert checkpoint_paths == {note.relative_to(ws).as_posix()} # interests.yaml not checkpointed (R6) + assert catalog.dumps == 1 + + asyncio.run(run()) + + +def test_open_loop_follow_up(tmp_path): + """Branch A emits follow_ups with stable ids; paths restricted to M.""" + + async def run(): + ws = tmp_path + _touch(ws / "daily" / DAY / "session.md", "open loop content") + reply = _reply( + follow_ups=[ + _topic("未决事项甲", confidence=0.9, paths=[f"daily/{DAY}/session.md"]), + _topic("越界事项", confidence=0.9, paths=["daily/2026-08-01/absent.md"]), + ], + ) + _, _, _, context, _ = await _run_chain(ws, [reply]) + state = context.get("proactive") + assert len(state["follow_ups"]) == 1 + follow_up = state["follow_ups"][0] + assert follow_up["kind"] == "follow_up" + assert follow_up["id"] == topic_id("未决事项甲") + assert state["dropped_missing"] == 1 # out-of-M paths dropped, never repaired + data = yaml.safe_load(_interests(ws).read_text(encoding="utf-8")) + assert [t["title"] for t in data["topics"]] == ["未决事项甲"] + + asyncio.run(run()) + + +def test_updates_actions(tmp_path): + """keep/update/resolve act on the truth source; unknown ids ignored.""" + + async def run(): + ws = tmp_path + id_a, id_b, id_c = topic_id("Topic A"), topic_id("Topic B"), topic_id("Topic C") + _write_state( + ws, + open_topics=[ + {**_state_topic("Topic A", day="2026-08-12", topic_id_value=id_a)}, + {**_state_topic("Topic B", day="2026-08-12", topic_id_value=id_b)}, + {**_state_topic("Topic C", day="2026-08-12", topic_id_value=id_c)}, + ], + ) + _touch(ws / "daily" / DAY / "session.md", "new evidence today") + evidence = f"daily/{DAY}/session.md" + reply = _reply( + follow_ups=[], + updates=[ + {"id": id_a, "action": "keep"}, + {"id": id_b, "action": "update", "evidence": evidence}, + {"id": id_c, "action": "resolve", "evidence": evidence}, + {"id": "ffffffffffff", "action": "resolve"}, + ], + ) + _, _, _, context, responses = await _run_chain(ws, [reply]) + assert all(r.success for r in responses) + state = context.get("proactive") + assert state["updates_applied"] == 1 + assert state["updates_resolved"] == 1 + + truth = _read_state(ws) + by_id = {t["id"]: t for t in truth["open_topics"]} + assert set(by_id) == {id_a, id_b} + assert by_id[id_a]["last_evidence_at"] == "2026-08-12" # keep: untouched + assert by_id[id_b]["last_evidence_at"] == DAY + assert by_id[id_b]["evidence"] == evidence + resolved_ids = {r["id"] for r in truth["resolved"]} + assert resolved_ids == {id_c} + + data = yaml.safe_load(_interests(ws).read_text(encoding="utf-8")) + assert {t["title"] for t in data["topics"]} == {"Topic A", "Topic B"} + + asyncio.run(run()) + + +def test_resolved_resurrect(tmp_path): + """A candidate hitting a tombstone resurrects it: tombstone removed, first_seen kept.""" + + async def run(): + ws = tmp_path + tid = topic_id("Revived Topic") + _write_state( + ws, + resolved=[ + { + "id": tid, + "title": "Revived Topic", + "resolved_at": "2026-08-12", + "first_seen": "2026-08-08", + "evidence": "", + }, + ], + ) + _touch(ws / "daily" / DAY / "session.md", "the plan was reopened today") + reply = _reply(follow_ups=[_topic("Revived Topic", confidence=0.7, paths=[f"daily/{DAY}/session.md"])]) + _, _, _, context, _ = await _run_chain(ws, [reply]) + state = context.get("proactive") + assert [c["title"] for c in state["candidates"]] == ["Revived Topic"] + + truth = _read_state(ws) + assert truth["resolved"] == [] # tombstone removed + by_id = {t["id"]: t for t in truth["open_topics"]} + assert tid in by_id + assert by_id[tid]["first_seen"] == "2026-08-08" # original age anchor kept + assert by_id[tid]["last_evidence_at"] == DAY + + asyncio.run(run()) + + +def test_trim_freshness_first(tmp_path): + """max_topics trims by the freshness-first order, regardless of kind (v5 aging).""" + + async def run(): + ws = tmp_path + stale = [ + {**_state_topic(f"Stale Follow {i}", day="2026-08-05", kind="follow_up"), "last_evidence_at": "2026-08-05"} + for i in range(3) + ] + fresh = [_state_topic(f"Fresh Extend {i}", day=DAY, kind="interest_extend") for i in range(2)] + _write_state(ws, open_topics=stale + fresh) + _touch(ws / "daily" / DAY / "session.md", "material") + reply = _reply(follow_ups=[_topic("New Follow", confidence=0.9, paths=[f"daily/{DAY}/session.md"])]) + _, _, _, context, _ = await _run_chain(ws, [reply], topics_kwargs={"max_topics": 4}) + titles = [t["title"] for t in context.get("proactive")["topics_out"]] + assert len(titles) == 4 + assert titles[0] == "New Follow" # today's follow_up first + assert set(titles[1:3]) == {"Fresh Extend 0", "Fresh Extend 1"} # today's extends next + assert titles[3].startswith("Stale Follow") # only one stale topic survives the cap + + asyncio.run(run()) + + +def test_dedup_fallback(tmp_path): + """M1 fallback: normalize_topic exact match against recent interests drops duplicates.""" + + async def run(): + ws = tmp_path + _write_interests_v1(ws, "2026-08-12", [_topic("Retrieval Evaluation", paths=["daily/2026-08-12/a.md"])]) + # Seed an empty truth source so the history topic stays out of open_topics + # (otherwise first-run bootstrap would merge the same-id candidate instead + # of dropping it as a duplicate). + _write_state(ws, open_topics=[]) + _touch(ws / "daily" / DAY / "session.md", "material") + reply = _reply( + follow_ups=[_topic("retrieval evaluation!!", confidence=0.9, paths=[f"daily/{DAY}/session.md"])], + ) + _, _, _, context, _ = await _run_chain(ws, [reply]) + state = context.get("proactive") + assert state["dropped_duplicate"] == 1 + assert state["candidates"] == [] + assert state["file_skip_reason"] == "all_duplicates" # metadata-only since v5 (R7) + data = yaml.safe_load(_interests(ws).read_text(encoding="utf-8")) + assert data["push"] is False + assert data["topics"] == [] + + asyncio.run(run()) + + +def test_skip_low_confidence(tmp_path): + """Candidates under min_push_confidence persist but are not pushed.""" + + async def run(): + ws = tmp_path + _touch(ws / "daily" / DAY / "session.md", "weak signal") + reply = _reply(follow_ups=[_topic("弱信号主题", confidence=0.3, paths=[f"daily/{DAY}/session.md"])]) + _, _, _, context, _ = await _run_chain(ws, [reply]) + state = context.get("proactive") + assert state["push"] is False + assert state["file_skip_reason"] == "low_confidence" + data = yaml.safe_load(_interests(ws).read_text(encoding="utf-8")) + assert data["push"] is False + assert "skip_reason" not in data # metadata-only since v5 (R7) + assert [t["title"] for t in data["topics"]] == ["弱信号主题"] # rendered from truth source + assert [t["title"] for t in _read_state(ws)["open_topics"]] == ["弱信号主题"] # persisted + + asyncio.run(run()) + + +def test_no_new_evidence_zero_llm(tmp_path): + """No changed material and no resources -> early exit with zero LLM calls.""" + + async def run(): + ws = tmp_path + note = _touch(ws / "daily" / DAY / "session.md", "already seen") + rel = note.relative_to(ws).as_posix() + catalog = _Catalog([FileNode(path=rel, st_mtime=note.stat().st_mtime)]) + wrapper = _AgentWrapper(["should never be consumed"]) + _, wrapper, catalog, context, responses = await _run_chain( + ws, + wrapper=wrapper, + catalog=catalog, + ) + state = context.get("proactive") + assert state["early_exit"] == "no_new_evidence" + assert wrapper.calls == 0 + assert not _interests(ws).exists() + assert catalog.upserts == [] and catalog.dumps == 0 + assert all(r.success for r in responses) + + asyncio.run(run()) + + +def test_resource_watermark_consumed_once(tmp_path): + """v5.1: resources share the proactive watermark and are consumed once. + + Round 1 consumes a fresh upload (1 LLM call, checkpointed); round 2 sees + nothing changed anywhere and early-exits with zero LLM calls; round 3 + re-uploads the resource (new mtime) and consumes it again. + """ + + async def run(): + ws = tmp_path + _touch(ws / "daily" / DAY / "session.md", "discussion about attribution") + resource = _touch(ws / "resource" / "notes.md", "fresh upload") + reply1 = _reply(follow_ups=[_topic("未决问题", paths=[f"daily/{DAY}/session.md"])]) + reply2 = _reply(follow_ups=[_topic("资源后续", paths=["resource/notes.md"])]) + wrapper = _AgentWrapper([reply1, reply2]) + catalog = _Catalog() + + await _run_chain(ws, wrapper=wrapper, catalog=catalog) + assert wrapper.calls == 1 + assert {n.path for n in catalog.upserts} >= {f"daily/{DAY}/session.md", "resource/notes.md"} + + _, _, _, context2, _ = await _run_chain(ws, wrapper=wrapper, catalog=catalog) + assert wrapper.calls == 1 # early exit, no second LLM call + assert context2.get("proactive")["early_exit"] == "no_new_evidence" + + stat = resource.stat() + os.utime(resource, (stat.st_atime + 60, stat.st_mtime + 60)) + _, _, _, context3, _ = await _run_chain(ws, wrapper=wrapper, catalog=catalog) + assert wrapper.calls == 2 # changed resource is consumed again + assert not context3.get("proactive")["early_exit"] + + asyncio.run(run()) + + +def test_push_sticky(tmp_path): + """Within one day push only goes up: derived from the cumulative truth source (R1).""" + + async def run(): + ws = tmp_path + _touch(ws / "daily" / DAY / "s1.md", "strong evidence") + reply1 = _reply(follow_ups=[_topic("高置信主题", confidence=0.9, paths=[f"daily/{DAY}/s1.md"])]) + _, _, _, context1, _ = await _run_chain(ws, [reply1]) + assert context1.get("proactive")["push"] is True + assert yaml.safe_load(_interests(ws).read_text(encoding="utf-8"))["push"] is True + + _touch(ws / "daily" / DAY / "s2.md", "weak evidence") + reply2 = _reply(follow_ups=[_topic("低置信主题", confidence=0.3, paths=[f"daily/{DAY}/s2.md"])]) + _, _, _, context2, _ = await _run_chain(ws, [reply2]) + state2 = context2.get("proactive") + assert state2["push"] is True # sticky: the 0.9 topic discovered today persists in truth source + assert state2["file_skip_reason"] == "" + assert yaml.safe_load(_interests(ws).read_text(encoding="utf-8"))["push"] is True + + asyncio.run(run()) + + +def test_idempotent_write(tmp_path): + """Same render skips writing; nightly v1 is adopted via bootstrap (v5 R1).""" + + async def run(): + ws = tmp_path + _touch(ws / "daily" / DAY / "s1.md", "evidence one") + reply1 = _reply(follow_ups=[_topic("主题甲", confidence=0.9, paths=[f"daily/{DAY}/s1.md"])]) + _, _, _, context1, _ = await _run_chain(ws, [reply1]) + assert context1.get("proactive")["interests_written"] is True + mtime_before = _interests(ws).stat().st_mtime + + _touch(ws / "daily" / DAY / "s2.md", "evidence two, no new topics") + reply2 = _reply(follow_ups=[], updates=[]) + time.sleep(0.01) + _, _, _, context2, _ = await _run_chain(ws, [reply2]) + assert context2.get("proactive")["interests_written"] is False # identical render + assert _interests(ws).stat().st_mtime == mtime_before + + asyncio.run(run()) + + async def run_v1_nightly_adoption(): + """v5 R1: no v1 guard - first-run bootstrap adopts the nightly topic.""" + with tempfile.TemporaryDirectory() as tmp: + ws = Path(tmp) + nightly = [ + { + "title": "Nightly topic", + "reason": "written by dream", + "evidence": f"daily/{DAY}/n.md", + "keywords": [], + "paths": [f"daily/{DAY}/n.md"], + }, + ] + _write_interests_v1(ws, DAY, nightly) + _touch(ws / "daily" / DAY / "s1.md", "material") + reply = _reply(follow_ups=[], updates=[]) + _, _, _, context, _ = await _run_chain(ws, [reply]) + state = context.get("proactive") + # Bootstrap gives the nightly topic first_seen = file date (today); + # push is derived true because its fallback confidence 0.5 passes. + assert state["push"] is True + assert state["interests_written"] is True + data = yaml.safe_load(_interests(ws).read_text(encoding="utf-8")) + assert data["version"] == 2 + assert [t["title"] for t in data["topics"]] == ["Nightly topic"] + + asyncio.run(run_v1_nightly_adoption()) + + +# --------------------------------------------------------------------------- +# F3 isolation / interleave +# --------------------------------------------------------------------------- + + +def test_refresh_isolation(tmp_path): + """Refresh never writes digest, never touches the dream catalog or day index.""" + + async def run(): + ws = tmp_path + digest_file = _touch(ws / "digest" / "wiki" / "existing.md", "digest content") + day_index = _touch(ws / "daily" / f"{DAY}.md", "index content") + _touch(ws / "daily" / DAY / "session.md", "conversation") + reply = _reply(follow_ups=[_topic("隔离主题", confidence=0.9, paths=[f"daily/{DAY}/session.md"])]) + proactive_catalog = _Catalog() + dream_catalog = _Catalog() + _, _, _, _, responses = await _run_chain(ws, [reply], catalog=proactive_catalog) + assert all(r.success for r in responses) + + assert digest_file.read_text(encoding="utf-8") == "digest content" + assert day_index.read_text(encoding="utf-8") == "index content" + assert not dream_catalog.upserts and dream_catalog.dumps == 0 and not dream_catalog.deleted + assert proactive_catalog.dumps == 1 + assert {n.path for n in proactive_catalog.upserts} == {f"daily/{DAY}/session.md"} + + asyncio.run(run()) + + +def test_nightly_overwrite_continuity(tmp_path): + """Nightly v1 overwrite cannot break continuity: truth source restores exposure.""" + + async def run(): + ws = tmp_path + _touch(ws / "daily" / DAY / "s1.md", "round one") + reply1 = _reply(follow_ups=[_topic("持续主题", confidence=0.9, paths=[f"daily/{DAY}/s1.md"])]) + await _run_chain(ws, [reply1]) + truth_before = _read_state(ws)["open_topics"] + + # Nightly dream overwrites the exposure product with a v1 file. + _write_interests_v1(ws, DAY, [_topic("Nightly only", paths=[f"daily/{DAY}/s1.md"])]) + + _touch(ws / "daily" / DAY / "s2.md", "round two") + reply2 = _reply(follow_ups=[], updates=[]) + _, _, _, context, _ = await _run_chain(ws, [reply2]) + assert context.get("proactive")["interests_written"] is True + + assert _read_state(ws)["open_topics"] == truth_before # truth source intact + data = yaml.safe_load(_interests(ws).read_text(encoding="utf-8")) + assert data["version"] == 2 # exposure restored from the truth source + assert [t["title"] for t in data["topics"]] == ["持续主题"] + + asyncio.run(run()) + + +def test_refresh_nightly_interleave(tmp_path): + """Refresh and nightly writes interleave without corrupting state or continuity.""" + + async def run(): + ws = tmp_path + _touch(ws / "daily" / DAY / "s1.md", "round one") + reply1 = _reply(follow_ups=[_topic("连续主题", confidence=0.9, paths=[f"daily/{DAY}/s1.md"])]) + await _run_chain(ws, [reply1]) + + # Nightly dream overwrites interests.yaml with a v1 file. + _write_interests_v1(ws, DAY, [_topic("Nightly topic", paths=[f"daily/{DAY}/s1.md"])]) + nightly = yaml.safe_load(_interests(ws).read_text(encoding="utf-8")) + assert nightly.get("version") is None + + # Next refresh round re-renders from the intact truth source. + _touch(ws / "daily" / DAY / "s2.md", "round two") + reply2 = _reply(follow_ups=[], updates=[]) + _, _, _, _, responses = await _run_chain(ws, [reply2]) + assert all(r.success for r in responses) + data = yaml.safe_load(_interests(ws).read_text(encoding="utf-8")) + assert data["version"] == 2 + assert data["push"] is True + assert [t["title"] for t in data["topics"]] == ["连续主题"] + assert [t["title"] for t in _read_state(ws)["open_topics"]] == ["连续主题"] + + asyncio.run(run()) + + +def test_nightly_unit_set_invariant(): + """Dream extract sees identical units whether or not refresh ran first (INV-9).""" + + dream_reply = ( + "```yaml\n" + "units:\n" + " - name: eval-plan-gap\n" + " bucket: wiki\n" + " summary: The evaluation plan is discussed but never lands.\n" + f" paths: [daily/{DAY}/session.md]\n" + "topics: []\n" + "```" + ) + + async def dream_units(ws) -> list[dict]: + catalog = _Catalog() + wrapper = _AgentWrapper([dream_reply]) + step = DreamExtractStep(scan_days=1, app_context=ApplicationContext(workspace_dir=str(ws))) + with patch("reme.steps.evolve.dream.extract.refresh_day_index", return_value={}): + response = await step( + RuntimeContext( + date=DAY, + file_catalog=catalog, + file_store=_FileStore(ws), + agent_wrapper=wrapper, + as_llm=_DummyModel(), + ), + ) + assert response.success is True + return response.metadata["dream"]["units"] + + async def run(): + with tempfile.TemporaryDirectory() as tmp_a, tempfile.TemporaryDirectory() as tmp_b: + ws_a, ws_b = Path(tmp_a), Path(tmp_b) + for ws in (ws_a, ws_b): + _touch(ws / "daily" / DAY / "session.md", "same discussion content") + + # Run the full proactive chain on ws_a only. + reply = _reply(follow_ups=[_topic("主动主题", confidence=0.9, paths=[f"daily/{DAY}/session.md"])]) + await _run_chain(ws_a, [reply]) + + units_a = await dream_units(ws_a) + units_b = await dream_units(ws_b) + assert units_a == units_b + assert len(units_a) == 1 + + asyncio.run(run()) + + +# --------------------------------------------------------------------------- +# F4 idle gate, short-circuit, budget, timeout +# --------------------------------------------------------------------------- + + +def test_idle_gate(tmp_path): + """Running/quiet-window jobs block; idle trunk proceeds; fnmatch patterns apply.""" + + async def run(): + now_mono = time.monotonic() + app = ApplicationContext(workspace_dir=str(tmp_path)) + app.metadata["__job_last_run"] = { + "auto_memory_batch": {"running": True, "last_start": now_mono, "last_end": now_mono - 999}, + "unrelated_job": {"running": True, "last_start": now_mono, "last_end": now_mono}, + } + step = WaitForIdleStep(app_context=app, max_wait=0.3, poll_interval=0.05) + context = RuntimeContext() + response = await step(context) + assert response.success is True # giving up is not a failure + flag = context.get("proactive_skip") + assert flag["reason"] == "busy" + assert flag["busy_jobs"] == ["auto_memory_batch"] # unrelated_job not matched + + app.metadata["__job_last_run"] = { + "auto_memory_batch": {"running": False, "last_start": now_mono - 900, "last_end": now_mono - 900}, + } + context2 = RuntimeContext() + response2 = await WaitForIdleStep(app_context=app, quiet_window=120)(context2) + assert response2.success is True + assert "proactive_skip" not in context2 + + asyncio.run(run()) + + +def test_idle_timeout_skip(tmp_path): + """Waiting past max_wait gives up the round: success=True, skip flag set, no writes.""" + + async def run(): + app = ApplicationContext(workspace_dir=str(tmp_path)) + app.metadata["__job_last_run"] = { + "dream_cron": {"running": True, "last_start": time.monotonic(), "last_end": 0.0}, + } + step = WaitForIdleStep(app_context=app, max_wait=0.2, poll_interval=0.05) + context = RuntimeContext() + started = time.monotonic() + response = await step(context) + assert time.monotonic() - started >= 0.2 + assert response.success is True + assert "Skipped" in str(response.answer) + assert context.get("proactive_skip")["reason"] == "busy" + assert not _interests(tmp_path).exists() + + asyncio.run(run()) + + +def test_wait_for_idle_skip_key(tmp_path): + """The short-circuit flag key is parameterized, not hardcoded.""" + + async def run(): + app = ApplicationContext(workspace_dir=str(tmp_path)) + app.metadata["__job_last_run"] = { + "dream_cron": {"running": True, "last_start": time.monotonic(), "last_end": 0.0}, + } + step = WaitForIdleStep(app_context=app, max_wait=0.15, poll_interval=0.05, skip_key="custom_skip") + context = RuntimeContext() + await step(context) + assert context.get("custom_skip")["reason"] == "busy" + assert context.get("proactive_skip") is None + + asyncio.run(run()) + + +def test_llm_timeout(tmp_path): + """A hanging reply times out, short-circuits the round, and retries next round.""" + + async def run(): + ws = tmp_path + _touch(ws / "daily" / DAY / "session.md", "fresh material") + wrapper = _AgentWrapper(["never returned"], delay=0.5) + catalog = _Catalog() + _, wrapper, catalog, context, responses = await _run_chain( + ws, + wrapper=wrapper, + catalog=catalog, + extract_kwargs={"llm_timeout_seconds": 0.1}, + ) + assert context.get("proactive_skip")["reason"] == "llm_timeout" + assert wrapper.calls == 1 + assert not _interests(ws).exists() + assert catalog.upserts == [] and catalog.dumps == 0 + assert all(r.success for r in responses) + + asyncio.run(run()) + + +def test_parse_failure_retry_then_empty(tmp_path): + """Unparseable output retries once; still failing yields empty, not error.""" + + async def run(): + ws = tmp_path + _touch(ws / "daily" / DAY / "session.md", "material") + wrapper = _AgentWrapper(["this is not structured at all", _reply(follow_ups=[], updates=[])]) + _, wrapper, _, context, responses = await _run_chain(ws, wrapper=wrapper) + assert wrapper.calls == 2 + assert all(r.success for r in responses) + assert context.get("proactive")["follow_ups"] == [] + + asyncio.run(run()) + + +# --------------------------------------------------------------------------- +# F5 read interface +# --------------------------------------------------------------------------- + + +def test_proactive_backward_compat(tmp_path): + """Default params on a v1 file keep the legacy answer field-for-field.""" + + async def run(): + content = ( + "date: 2026-05-28\n" + "topics:\n" + " - title: Retrieval quality\n" + " reason: Search behavior changed repeatedly.\n" + " evidence: daily/2026-05-28/session.md\n" + ) + _touch(tmp_path / "daily" / "2026-05-28" / "interests.yaml", content) + step = ProactiveStep(app_context=ApplicationContext(workspace_dir=str(tmp_path))) + response = await step( + RuntimeContext(date="2026-05-28", include_content=True, file_store=_FileStore(tmp_path)), + ) + assert response.success is True + assert response.answer == { + "summary": "Read 1 proactive topic(s) from daily/2026-05-28/interests.yaml", + "topics": [ + { + "title": "Retrieval quality", + "reason": "Search behavior changed repeatedly.", + "evidence": "daily/2026-05-28/session.md", + "keywords": [], + "paths": [], + }, + ], + "content": content, + } + assert response.metadata["topics"] == response.answer["topics"] + assert response.metadata["content"] == content + assert response.metadata["push"] is True # additive keys only + + asyncio.run(run()) + + +def test_horizon_merge(tmp_path): + """Horizon>1 reads the truth source filtered by evidence recency (v5 R4).""" + + async def run(): + ws = tmp_path + id_x, id_y = topic_id("Topic X"), topic_id("Topic Y") + _write_state( + ws, + open_topics=[ + _state_topic("Topic X", day=DAY, topic_id_value=id_x, confidence=0.9), + _state_topic("Topic Y", day="2026-08-09", topic_id_value=id_y, kind="interest_extend", confidence=0.6), + ], + ) + step = ProactiveStep(app_context=ApplicationContext(workspace_dir=str(ws)), horizon_days=3) + response = await step(RuntimeContext(date=DAY, include_content=False, file_store=_FileStore(ws))) + assert response.success is True + assert [t["title"] for t in response.answer["topics"]] == ["Topic X"] # Y evidence too old + assert response.metadata["path"] == "daily/_proactive.yaml" + + # A wider horizon includes the older topic again. + response2 = await step( + RuntimeContext(date=DAY, include_content=False, file_store=_FileStore(ws), horizon_days=7), + ) + assert {t["title"] for t in response2.answer["topics"]} == {"Topic X", "Topic Y"} + + # Empty truth source -> skipped, not an error. + with tempfile.TemporaryDirectory() as tmp: + empty_ws = Path(tmp) + response3 = await step( + RuntimeContext(date=DAY, include_content=False, file_store=_FileStore(empty_ws), horizon_days=3), + ) + assert response3.success is True + assert response3.metadata["skipped"] is True + + asyncio.run(run()) + + +def test_min_confidence(tmp_path): + """min_confidence filters v2 topics; v1 topics fall back to 0.5.""" + + async def run(): + ws = tmp_path + _write_interests_v2( + ws, + DAY, + [ + _state_topic("High topic", confidence=0.7), + _state_topic("Low topic", kind="interest_extend", confidence=0.3), + ], + ) + step = ProactiveStep(app_context=ApplicationContext(workspace_dir=str(ws)), min_confidence=0.5) + response = await step(RuntimeContext(date=DAY, include_content=False, file_store=_FileStore(ws))) + assert [t["title"] for t in response.answer["topics"]] == ["High topic"] + + # Default min_confidence=0.4 sits below the 0.5 fallback but above weak 0.3. + step_default = ProactiveStep(app_context=ApplicationContext(workspace_dir=str(ws))) + default_response = await step_default( + RuntimeContext(date=DAY, include_content=False, file_store=_FileStore(ws)), + ) + assert [t["title"] for t in default_response.answer["topics"]] == ["High topic"] + + with tempfile.TemporaryDirectory() as tmp: + ws_v1 = Path(tmp) + _write_interests_v1(ws_v1, DAY, [_topic("Legacy topic", paths=[f"daily/{DAY}/a.md"])]) + step_v1 = ProactiveStep(app_context=ApplicationContext(workspace_dir=str(ws_v1))) + ok = await step_v1( + RuntimeContext(date=DAY, include_content=False, file_store=_FileStore(ws_v1), min_confidence=0.5), + ) + assert [t["title"] for t in ok.answer["topics"]] == ["Legacy topic"] # fallback 0.5 passes + filtered = await step_v1( + RuntimeContext(date=DAY, include_content=False, file_store=_FileStore(ws_v1), min_confidence=0.6), + ) + assert filtered.answer["topics"] == [] + + asyncio.run(run()) + + +# --------------------------------------------------------------------------- +# M2: extends branch + semantic dedup +# --------------------------------------------------------------------------- + + +def test_extend_from_resources(tmp_path): + """Branch B infers extends from recent uploads; out-of-M paths are dropped.""" + + async def run(): + ws = tmp_path + _touch(ws / "daily" / DAY / "session.md", "discussion about source attribution") + _touch(ws / "resource" / "citation-notes.md", "freshly uploaded notes") + resource_rel = "resource/citation-notes.md" + reply = _reply( + follow_ups=[], + extends=[ + _topic("引用质量评测", confidence=0.5, paths=[resource_rel, f"daily/{DAY}/session.md"]), + _topic("越界扩展", confidence=0.5, paths=["daily/2026-08-01/absent.md"]), + ], + ) + _, wrapper, _, context, _ = await _run_chain(ws, [reply]) + assert "extends" in wrapper.last_inputs[0] # extends section rendered + state = context.get("proactive") + assert len(state["extends"]) == 1 + extend = state["extends"][0] + assert extend["kind"] == "interest_extend" + assert resource_rel in extend["paths"] + assert state["dropped_missing"] == 1 + data = yaml.safe_load(_interests(ws).read_text(encoding="utf-8")) + assert [t["title"] for t in data["topics"]] == ["引用质量评测"] + + asyncio.run(run()) + + +def test_extends_disabled_m1_shape(): + """extends_enabled=False restores the M1 prompt shape and ignores extends output.""" + + async def run_once(extends_enabled: bool): + with tempfile.TemporaryDirectory() as tmp: + ws = Path(tmp) + _touch(ws / "daily" / DAY / "s1.md", "material") + reply = _reply( + follow_ups=[_topic("Follow Up Topic", confidence=0.9, paths=[f"daily/{DAY}/s1.md"])], + extends=[_topic("Extends Topic", confidence=0.9, paths=[f"daily/{DAY}/s1.md"])], + ) + _, wrapper, _, context, _ = await _run_chain( + ws, + [reply], + extract_kwargs={"extends_enabled": extends_enabled}, + ) + return wrapper, context.get("proactive") + + async def run(): + wrapper_on, state_on = await run_once(True) + assert "extends:" in wrapper_on.last_system_prompts[0] # M2 default renders branch B + assert [t["title"] for t in state_on["extends"]] == ["Extends Topic"] + + wrapper_off, state_off = await run_once(False) + system_off = wrapper_off.last_system_prompts[0] + assert "extends" not in system_off.replace("interest_extend", "") # M1 prompt shape + assert state_off["extends"] == [] # extends entries in the reply are ignored + assert [t["title"] for t in state_off["topics_out"]] == ["Follow Up Topic"] + + asyncio.run(run()) + + +def test_semantic_dedup(tmp_path): + """With embeddings, similarity below known_threshold is kept; no LLM tier (v5 R2).""" + + async def run(): + ws = tmp_path + _write_state(ws, open_topics=[_state_topic("Gray Comparison", kind="interest_extend")]) + _touch(ws / "daily" / DAY / "session.md", "material") + reply = _reply( + follow_ups=[ + _topic("Candidate Topic", confidence=0.8, paths=[f"daily/{DAY}/session.md"], keywords=["alpha"]), + ], + ) + fake = _FakeEmbedding( + { + # v5.2 scheme: candidate and known sides both embed "title + CJK period + reason" + "Candidate Topic。because": [1.0, 0.0], + "Gray Comparison。seeded": [0.8, 0.6], # cos = 0.800 < 0.85 -> kept, no second tier + }, + ) + _, wrapper, _, context, _ = await _run_chain(ws, [reply], extra_ctx={"as_embedding": fake}) + state = context.get("proactive") + assert state["dropped_known"] == 0 + assert state["dropped_duplicate"] == 0 + assert [c["title"] for c in state["candidates"]] == ["Candidate Topic"] + assert fake.calls >= 1 + assert wrapper.calls == 1 # topics step is pure computation now + + asyncio.run(run()) + + +def test_known_drop(tmp_path): + """sim >= known_threshold drops the candidate as already known.""" + + async def run(): + ws = tmp_path + _write_state(ws, open_topics=[_state_topic("Known Topic", kind="interest_extend")]) + _touch(ws / "daily" / DAY / "session.md", "material") + reply = _reply( + follow_ups=[_topic("Fresh Candidate", confidence=0.8, paths=[f"daily/{DAY}/session.md"])], + ) + fake = _FakeEmbedding( + { + "Fresh Candidate。because": [1.0, 0.0], + "Known Topic。seeded": [1.0, 0.0], # identical -> sim 1.0 >= 0.85 + }, + ) + _, _, _, context, _ = await _run_chain(ws, [reply], extra_ctx={"as_embedding": fake}) + state = context.get("proactive") + assert state["dropped_known"] == 1 + assert state["candidates"] == [] + data = yaml.safe_load(_interests(ws).read_text(encoding="utf-8")) + assert [t["title"] for t in data["topics"]] == ["Known Topic"] + + asyncio.run(run())