refractor(proactive): upgrade proactive feature with disentangled job and steps

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

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

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:

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",

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."""
 

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 = ""

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",
 ]

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

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",
 ]

reme/steps/evolve/dream/proactive.py

@@ -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/<date>/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

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",
+]

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

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 <kept id>`` 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/<date>/...` stays parseable.
+
+  ## Output format
+
+  Return only one YAML fenced block with this exact shape:
+  ```yaml
+  follow_ups:
+    - title: <specific unresolved matter>
+      reason: <why it is still open>
+      confidence: 0.7
+      evidence: <path or path#anchor>
+      keywords: [<keyword>, ...]
+      paths: [<changed path>, ...]
+  [extends]extends:
+  [extends]  - title: <related-but-unfocused topic>
+  [extends]    reason: <why it matters, grounded in the material>
+  [extends]    confidence: 0.5
+  [extends]    evidence: <path or path#anchor>
+  [extends]    keywords: [<keyword>, ...]
+  [extends]    paths: [<changed path>, ...]
+  updates:
+    - id: <echoed carry-forward id>
+      action: keep|update|resolve
+      evidence: <path or path#anchor>
+      reason: <short justification>
+      confidence: <re-scored per rubric; emit for update, omit otherwise>
+  ```
+
+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/<date>/...` 解析失败。
+
+  ## 输出格式
+
+  只返回一个 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。记住：空列表是正常
+  输出，绝不编造主题。

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

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/<date>/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

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 '<empty>'!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 '<unknown>'!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)

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/<day>/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-<ts>.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

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)

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