[Enhancement] Long-horizon tasks enhancement

src/leapflow/cli/context.py

@@ -2505,6 +2505,95 @@ def _on_insight(insight: Any) -> None:
 
         return _on_insight
 
+    async def _generate_session_summary(self) -> str | None:
+        """Generate a structured task summary from the session's conversation."""
+        store = getattr(self, '_conversation_store', None)
+        if not store:
+            return None
+        session_id = getattr(self.engine, '_current_session_id', None) or ""
+        if not session_id:
+            return None
+        try:
+            # Fetch all messages (up to 200) to find both first user goal and final outcome.
+            # get_messages returns ASC order; we use a larger limit to capture session endpoints.
+            messages = store.get_messages(session_id, limit=200)
+        except Exception:
+            return None
+        if not messages or len(messages) < 2:
+            return None
+
+        first_user = ""
+        for m in messages:
+            if getattr(m, 'role', '') == "user" and getattr(m, 'content', ''):
+                first_user = m.content[:150]
+                break
+
+        tool_names = sorted(set(
+            getattr(m, 'tool_name', '') or ''
+            for m in messages
+            if getattr(m, 'tool_name', '')
+        ))[:8]
+
+        last_assistant = ""
+        for m in reversed(messages):
+            content = getattr(m, 'content', '') or ''
+            if getattr(m, 'role', '') == "assistant" and len(content.strip()) > 20:
+                last_assistant = content[:200]
+                break
+
+        if not first_user:
+            return None
+
+        parts = [f"Goal: {first_user}"]
+        if tool_names:
+            parts.append(f"Tools: {', '.join(tool_names)}")
+        if last_assistant:
+            parts.append(f"Outcome: {last_assistant}")
+        return "\n".join(parts)
+
+    async def _persist_session_summary(self) -> None:
+        """Generate and persist session summary at end of session."""
+        try:
+            summary = await self._generate_session_summary()
+            if not summary:
+                return
+
+            # Persist to memory via SemanticMemoryProvider
+            from leapflow.memory.protocol import MemoryEntry, MemoryKind, SignalDomain
+            session_id = getattr(self.engine, '_current_session_id', None) or ""
+            entry = MemoryEntry(
+                kind=MemoryKind.SESSION_SUMMARY,
+                domain=SignalDomain.SYSTEM,
+                content=summary,
+                metadata={
+                    "_session_id": session_id,
+                    "workspace": str(self.settings.workspace_root),
+                },
+            )
+            if hasattr(self, 'memory') and self.memory:
+                await self.memory.insert(entry, session_id=session_id)
+                # MemoryManager routes SESSION_SUMMARY to narrative (MEMORY.md) first,
+                # but query_recent_summaries() reads from DuckDB (semantic provider).
+                # Explicitly persist to semantic to enable cross-session querying.
+                semantic = self.memory.get_provider("semantic")
+                if semantic is not None:
+                    try:
+                        await semantic.insert(entry, session_id=session_id)
+                    except Exception:
+                        logger.debug("semantic insert for session summary failed", exc_info=True)
+                logger.debug("Session summary persisted for session=%s", session_id[:8])
+
+            # Update session title with the goal line
+            goal_line = summary.split("\n")[0].removeprefix("Goal: ").strip()
+            conv_store = getattr(self, '_conversation_store', None)
+            if conv_store and session_id and goal_line:
+                try:
+                    conv_store.end_session(session_id, title=goal_line)
+                except Exception:
+                    logger.debug("session title update failed", exc_info=True)
+        except Exception:
+            logger.debug("session summary persistence failed", exc_info=True)
+
     async def _on_session_end_learning(self) -> None:
         """End-of-session OPD learning pipeline (8 phases) with full observability.
 
@@ -2893,6 +2982,8 @@ async def cleanup(self) -> None:
         # OPD end-of-session learning pipeline
         if self.settings.replay_on_session_end:
             await self._on_session_end_learning()
+        # Persist session summary before memory shutdown
+        await self._persist_session_summary()
         # Shutdown all memory providers (stops GC, closes DB)
         await self.memory.shutdown_all()
         if isinstance(self.rpc, CuaDriverClient):

src/leapflow/config.py

@@ -321,6 +321,7 @@ class Settings:
     agent_iter_hard_cap: int = 500
     agent_iter_extension_step: int = 25
     agent_stall_rounds: int = 6
+    agent_checkpoint_interval: int = 15  # Force checkpoint every N rounds in research posture
     agent_cost_ceiling_context_multiple: float = 0.0
     agent_subagent_max_depth: int = 2
     agent_subagent_max_concurrent: int = 3
@@ -860,6 +861,7 @@ def _build_settings_from_env(
     agent_iter_hard_cap = int(os.getenv("LEAPFLOW_AGENT_ITER_HARD_CAP", "500"))
     agent_iter_extension_step = int(os.getenv("LEAPFLOW_AGENT_ITER_EXTENSION_STEP", "25"))
     agent_stall_rounds = int(os.getenv("LEAPFLOW_AGENT_STALL_ROUNDS", "6"))
+    agent_checkpoint_interval = int(os.getenv("LEAPFLOW_AGENT_CHECKPOINT_INTERVAL", "15"))
     agent_cost_ceiling_context_multiple = float(os.getenv("LEAPFLOW_AGENT_COST_CEILING_CONTEXT_MULTIPLE", "0.0"))
     agent_subagent_max_depth = int(os.getenv("LEAPFLOW_AGENT_SUBAGENT_MAX_DEPTH", "2"))
     agent_subagent_max_concurrent = int(os.getenv("LEAPFLOW_AGENT_SUBAGENT_MAX_CONCURRENT", "3"))
@@ -1217,6 +1219,7 @@ def _tuple_env(key: str, default: tuple) -> tuple:
         agent_iter_hard_cap=agent_iter_hard_cap,
         agent_iter_extension_step=agent_iter_extension_step,
         agent_stall_rounds=agent_stall_rounds,
+        agent_checkpoint_interval=agent_checkpoint_interval,
         agent_cost_ceiling_context_multiple=agent_cost_ceiling_context_multiple,
         agent_subagent_max_depth=agent_subagent_max_depth,
         agent_subagent_max_concurrent=agent_subagent_max_concurrent,

src/leapflow/daemon/approval_coordinator.py

@@ -213,6 +213,7 @@ def _normalize_decision(decision: str) -> str:
             "allow",
             "allow_once",
             "allow_session",
+            "allow_all_session",
             "allow_always",
             "deny",
             "deny_always",

src/leapflow/engine/context_control.py

@@ -756,6 +756,7 @@ class ContextGovernanceController:
     # exploration rounds; the ceiling prevents truly stuck tasks from running forever.
     convergence_round_ceiling: int = 40
     convergence_scale: float = 2.0
+    checkpoint_interval: int = 15
     posture_config: ContextPostureConfig = field(default_factory=ContextPostureConfig)
     difficulty_config: DifficultyConfig = field(default_factory=DifficultyConfig)
     evidence_tools: frozenset[str] = _EVIDENCE_TOOLS
@@ -784,6 +785,8 @@ def __post_init__(self) -> None:
         self._difficulty_ema_round = -1
         self._evidence_by_round: dict[int, int] = {}
         self._evidence_round_hwm = -1
+        self._last_tool_name: str = ""
+        self._last_tool_path: str = ""
 
     def reset_turn_scope(self) -> None:
         """Clear per-turn exploration state so posture never leaks across tasks."""
@@ -796,6 +799,8 @@ def reset_turn_scope(self) -> None:
         self._difficulty_ema_round = -1
         self._evidence_by_round.clear()
         self._evidence_round_hwm = -1
+        self._last_tool_name = ""
+        self._last_tool_path = ""
 
     reset_task_scope = reset_turn_scope
 
@@ -811,28 +816,81 @@ def _effective_convergence_round(self, difficulty: float) -> int:
         extension = round(self.convergence_round * max(0.0, min(1.0, difficulty)) * self.convergence_scale)
         return min(self.convergence_round_ceiling, self.convergence_round + extension)
 
+    _WRITE_TOOLS = frozenset({
+        "file_write", "gp_file_write",
+        "text_replace", "gp_text_replace",
+        "file_edit", "gp_file_edit",
+    })
+
     def compact_tool_result(self, tool_name: str, arguments: Dict[str, Any] | None, result: Any) -> Any:
-        """Return evidence and update the session exploration ledger."""
+        """Return evidence and update the session exploration ledger.
+
+        P0 convergence fixes:
+        - evidence_count only increments for genuinely new sources (not re-reads)
+        - Repeated reads beyond the limit are hard-gated: tool result is replaced
+          with an instruction to use research_note, unless the immediately preceding
+          tool call was a write/edit to the same path (post-edit verification is OK).
+        """
         self._tool_counts[tool_name] = self._tool_counts.get(tool_name, 0) + 1
         if isinstance(result, dict) and result.get("ok") is False:
             self._tool_failures += 1
-        if tool_name in self.evidence_tools:
-            self._evidence_count += 1
+
+        gated_result: Any = None
+
         if tool_name in {"file_read", "gp_file_read"}:
             path = str((arguments or {}).get("path") or (result.get("path") if isinstance(result, dict) else ""))
             if path:
                 key = str(Path(path).expanduser())
+                # Reset read count if the previous tool was a write/edit to this path
+                if self._last_tool_name in self._WRITE_TOOLS and self._last_tool_path == key:
+                    self._reads[key] = 0
+                is_new_source = key not in self._sources_seen
                 self._reads[key] = self._reads.get(key, 0) + 1
                 self._sources_seen.add(key)
+                # Only count genuinely new sources as progress
+                if is_new_source:
+                    self._evidence_count += 1
+                # Hard gate: block re-reads beyond the limit
+                if self._reads[key] > self.repeated_read_limit:
+                    gated_result = (
+                        f"[REPEATED READ \u2014 this file was already read {self._reads[key]} times in this turn. "
+                        f"Content was processed in earlier rounds but compressed. "
+                        f"Use research_note to record your findings instead of re-reading. "
+                        f"File: {path}]"
+                    )
         elif tool_name in {"file_list", "gp_file_list"}:
             path = str((arguments or {}).get("path") or (result.get("path") if isinstance(result, dict) else ""))
             if path:
                 key = str(Path(path).expanduser())
-                # Track repeated directory listings alongside repeated file reads:
-                # both are evidence of an agent struggling to make progress, and
-                # both should push the posture toward converging.
+                is_new_source = key not in self._sources_seen
                 self._reads[key] = self._reads.get(key, 0) + 1
                 self._sources_seen.add(key)
+                if is_new_source:
+                    self._evidence_count += 1
+                if self._reads[key] > self.repeated_read_limit:
+                    gated_result = (
+                        f"[REPEATED READ \u2014 this directory was already listed {self._reads[key]} times in this turn. "
+                        f"Content was processed in earlier rounds but compressed. "
+                        f"Use research_note to record your findings instead of re-reading. "
+                        f"Path: {path}]"
+                    )
+        elif tool_name in self.evidence_tools:
+            # shell_run and other non-path evidence tools: always new evidence
+            self._evidence_count += 1
+
+        # Track last tool for write-then-read exception
+        self._last_tool_name = tool_name
+        if tool_name in self._WRITE_TOOLS:
+            write_path = str((arguments or {}).get("path") or "")
+            self._last_tool_path = str(Path(write_path).expanduser()) if write_path else ""
+        elif tool_name in {"file_read", "gp_file_read", "file_list", "gp_file_list"}:
+            read_path = str((arguments or {}).get("path") or (result.get("path") if isinstance(result, dict) else ""))
+            self._last_tool_path = str(Path(read_path).expanduser()) if read_path else ""
+        else:
+            self._last_tool_path = ""
+
+        if gated_result is not None:
+            return gated_result
         return self.evidence_builder.build(tool_name, arguments, result)
 
     def tool_metadata(self, tool_name: str, arguments: Dict[str, Any] | None, result: Any) -> Dict[str, Any]:
@@ -1052,5 +1110,25 @@ def convergence_notice(self, round_number: int, *, open_questions: int | None =
             "prefer targeted reads, and synthesize the final answer."
         )
 
+    def checkpoint_notice(self, round_number: int) -> str | None:
+        """Emit checkpoint instruction at configurable intervals during research."""
+        interval = self.checkpoint_interval
+        if interval <= 0 or round_number < interval:
+            return None
+        if round_number % interval != 0:
+            return None
+        # Only checkpoint in research/expanding posture, not simple Q&A
+        snapshot = self.snapshot(round_number=round_number)
+        if snapshot.posture in (_POSTURE_BASELINE, _POSTURE_EXPANDED):
+            return None
+        return (
+            f"SYSTEM CHECKPOINT (round {round_number}): "
+            "Persist your current findings NOW. "
+            "(1) Call research_note to record key conclusions that should survive context compression. "
+            "(2) Write intermediate results to a file if the task requires a deliverable. "
+            "(3) If you can answer partially, provide it now. "
+            "Remaining work should be declared as open_questions via research_note."
+        )
+
 
 LongTaskContextController = ContextGovernanceController

src/leapflow/engine/context_disclosure.py

@@ -234,7 +234,7 @@ def plan(
             level=level,
             tool_definitions=tuple(expanded_defs),
             catalog_definitions=tuple(tool_definitions),
-            memory=MemoryDisclosure.QUERY_RETRIEVAL if expanded_categories else MemoryDisclosure.NONE,
+            memory=MemoryDisclosure.QUERY_RETRIEVAL if expanded_categories else MemoryDisclosure.SESSION_SUMMARY,
             history=HistoryDisclosure.RECENT if expanded_categories else HistoryDisclosure.SHORT,
             # At the CORE floor (no Tier 1 category opened) skip reasoning entirely: a
             # turn that only needs the static low-risk whitelist is, by construction, not