fix(compression): preserve human intent and durable handoffs

agent/context_compressor.py

@@ -298,6 +298,7 @@ def _strip_persistence_markers(messages: List[Dict[str, Any]]) -> None:
 _AUTO_FOCUS_MAX_TURNS = 3
 _AUTO_FOCUS_TURN_MAX_CHARS = 260
 _AUTO_FOCUS_MAX_CHARS = 700
+_ACTIVE_TASK_MAX_CHARS = 1400
 # Keep a short run of recent messages verbatim even when the token budget is
 # already exhausted.  The public ``protect_last_n`` default is intentionally
 # high for small/light tails, but using all 20 as a hard floor here would bring
@@ -321,6 +322,9 @@ def _strip_persistence_markers(messages: List[Dict[str, Any]]) -> None:
 # the summary, the downstream model may re-emit it as an active directive on
 # the next turn, triggering bogus attachment sends (#14665).
 _MEDIA_DIRECTIVE_RE = re.compile(r"MEDIA:\S+")
+_HISTORICAL_TASK_SECTION_RE = re.compile(
+    rf"(?ms)^{re.escape(HISTORICAL_TASK_HEADING)}\s*\n.*?(?=^## |\Z)"
+)
 
 
 def _dedupe_append(items: list[str], value: str, *, limit: int) -> None:
@@ -2142,25 +2146,25 @@ def _generate_summary(
         _template_sections = f"""{HISTORICAL_TASK_HEADING}
 [THE SINGLE MOST IMPORTANT FIELD. Capture the user's most recent unfulfilled
 input verbatim — the exact words they used. This includes:
-- Explicit task assignments ("refactor the auth module")
-- Questions awaiting an answer ("waarom staat X op Y?", "wat zijn de volgende stappen?")
-- Decisions awaiting input ("optie A of B?")
+- Explicit task assignments ("<specific user task>")
+- Questions awaiting an answer ("<specific user question>")
+- Decisions awaiting input ("<option A or B?>")
 - Ongoing discussions where the assistant owes the next substantive reply
 A conversation where the user just asked a question IS an active task — the
 task is "answer that question with full context". Do NOT write "None" merely
 because the user did not issue an imperative command; reserve "None" for the
 rare case where the last exchange was fully resolved and the user said
 something like "thanks, that's all".
 If multiple items are outstanding, list only the ones NOT yet completed.
-Continuation should pick up exactly here. Examples:
-"User asked: 'Now refactor the auth module to use JWT instead of sessions'"
-"User asked: 'Waarom stond provider ineens op openrouter?' — needs investigation + answer"
-"User chose option A; awaiting implementation of step 2"
+This historical snapshot must identify the latest unresolved user input precisely. Examples:
+"User asked: '<exact latest user request>'"
+"User asked: '<exact latest user question>' — needs investigation + answer"
+"User chose <option>; awaiting implementation of <specific next step>"
 If the user's most recent message was a reverse signal (stop, undo, roll
 back, never mind, just verify, change of topic) that supersedes earlier
 work, write the reverse signal verbatim and DO NOT carry forward the
-cancelled task. Example: "User asked: 'Stop the i18n refactor and just
-verify the current diff' — earlier i18n in-flight work is cancelled."
+cancelled task. Example: "User asked: '<exact reverse signal>' — earlier
+in-flight work is cancelled."
 If no outstanding task exists, write "None."]
 
 ## Goal
@@ -2322,6 +2326,7 @@ def _generate_summary(
             # Redact the summary output as well — the summarizer LLM may
             # ignore prompt instructions and echo back secrets verbatim.
             summary = redact_sensitive_text(content.strip())
+            summary = self._ground_historical_task_snapshot(summary, turns_to_summarize)
             # Store for iterative updates on next compaction
             self._previous_summary = summary
             self._clear_compression_failure_cooldown()
@@ -2593,6 +2598,54 @@ def _derive_auto_focus_topic(
             focus = focus[: _AUTO_FOCUS_MAX_CHARS - 1].rstrip() + "…"
         return focus
 
+    @classmethod
+    def _latest_user_task_snapshot(
+        cls,
+        messages: List[Dict[str, Any]],
+    ) -> Optional[str]:
+        """Return a deterministic task-snapshot line from the newest real user turn.
+
+        The LLM summarizer is allowed to compress prose, but it must not invent
+        the "what is the active task?" anchor from a prompt example or stale
+        prior summary.  This helper extracts the anchor locally from the exact
+        compacted turns so the summary can be grounded before it becomes live
+        context.
+        """
+        for msg in reversed(messages):
+            if msg.get("role") != "user":
+                continue
+            content = msg.get("content")
+            if cls._is_context_summary_content(content):
+                continue
+            text = redact_sensitive_text(_content_text_for_contains(content).strip())
+            if not text:
+                continue
+            text = re.sub(r"\s+", " ", text)
+            if len(text) > _ACTIVE_TASK_MAX_CHARS:
+                text = text[: _ACTIVE_TASK_MAX_CHARS - 15].rstrip() + " ...[truncated]"
+            return (
+                f"User asked (deterministic, from compacted turns): {text!r}\n"
+                "Historical only; newer protected-tail messages after this summary win."
+            )
+        return None
+
+    @classmethod
+    def _ground_historical_task_snapshot(
+        cls,
+        summary: str,
+        messages: List[Dict[str, Any]],
+    ) -> str:
+        """Force the task snapshot section to match a real user turn when possible."""
+        snapshot = cls._latest_user_task_snapshot(messages)
+        if not snapshot:
+            return summary
+
+        body = cls._strip_summary_prefix(summary)
+        replacement = f"{HISTORICAL_TASK_HEADING}\n{snapshot}"
+        if _HISTORICAL_TASK_SECTION_RE.search(body):
+            return _HISTORICAL_TASK_SECTION_RE.sub(replacement, body, count=1)
+        return f"{replacement}\n\n{body}".strip()
+
     @classmethod
     def _find_latest_context_summary(
         cls,

agent/conversation_compression.py

@@ -415,43 +415,88 @@ def conversation_history_after_compression(agent: Any, messages: list) -> Option
     return None
 
 
+_SYNTHETIC_USER_PREFIXES = (
+    "[System: Your previous response was truncated",
+    "[System: The previous response was cut off",
+    "[System: Your previous tool call",
+    "[Your active task list was preserved across context compression]",
+    "[IMPORTANT: Background process ",
+)
+
+
+def _message_text(message: Any) -> str:
+    content = message.get("content") if isinstance(message, dict) else None
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        return "\n".join(
+            str(part.get("text") or part.get("content") or "")
+            for part in content
+            if isinstance(part, dict)
+        )
+    return ""
+
+
+def _is_real_user_message(message: Any) -> bool:
+    """Distinguish human intent from user-role runtime scaffolding."""
+    if not isinstance(message, dict) or message.get("role") != "user":
+        return False
+    if any(
+        message.get(flag)
+        for flag in (
+            "_length_continuation_synthetic",
+            "_todo_snapshot_synthetic",
+            "_empty_recovery_synthetic",
+            "_verification_stop_synthetic",
+            "_pre_verify_synthetic",
+        )
+    ):
+        return False
+    text = _message_text(message).strip()
+    if not text:
+        return False
+    return not text.startswith(_SYNTHETIC_USER_PREFIXES)
+
+
+def _insert_real_user_anchor(messages: list, anchor: dict) -> None:
+    """Insert the latest human turn at a valid summary boundary."""
+    for index, message in enumerate(messages):
+        if not isinstance(message, dict) or message.get("role") != "assistant":
+            continue
+        previous_role = (
+            messages[index - 1].get("role")
+            if index > 0 and isinstance(messages[index - 1], dict)
+            else None
+        )
+        if previous_role != "user":
+            messages.insert(index, anchor)
+            return
+    if not messages or not (
+        isinstance(messages[-1], dict) and messages[-1].get("role") == "user"
+    ):
+        messages.append(anchor)
+    else:
+        messages.insert(0, anchor)
+
+
 def _ensure_compressed_has_user_turn(original_messages: list, compressed: list) -> None:
-    """Preserve a real user turn when a compressor returns assistant/tool-only context.
-
-    On repeated compaction the protected head decays to the system prompt only,
-    the middle summary can land as ``role="assistant"``, and a tool-heavy tail
-    can be all assistant/tool — so the compacted transcript can legitimately
-    contain zero user messages. Strict chat templates (LM Studio / llama.cpp
-    Jinja) then fail with "No user query found in messages" (#55677).
-
-    The restored turn is appended at the END: the guard only runs when
-    ``compressed`` currently ends with an assistant/tool message (any existing
-    user turn — including a todo-snapshot append — short-circuits the
-    ``any()`` check), so appending a user message never creates consecutive
-    same-role messages. ``_fresh_compaction_message_copy`` copies the message
-    and strips the ``_db_persisted`` marker so the rotation/in-place flush
-    still persists the restored row to the new session (#57491).
-
-    If the pre-compression transcript itself carried no user turn at all
-    (near-impossible — every real conversation opens with a user request —
-    but kept as a defensive backstop), a minimal continuation marker is
-    appended instead so strict templates still see a user message.
-    """
-    if any(isinstance(msg, dict) and msg.get("role") == "user" for msg in compressed):
+    """Preserve human intent, not merely a synthetic user-role placeholder."""
+    if any(_is_real_user_message(message) for message in compressed):
         return
     from agent.context_compressor import _fresh_compaction_message_copy
 
-    for msg in reversed(original_messages):
-        if not isinstance(msg, dict) or msg.get("role") != "user":
-            continue
-        compressed.append(_fresh_compaction_message_copy(msg))
-        return
+    for message in reversed(original_messages):
+        if _is_real_user_message(message):
+            _insert_real_user_anchor(
+                compressed,
+                _fresh_compaction_message_copy(message),
+            )
+            return
     compressed.append({
         "role": "user",
         "content": (
             "Continue from the compressed conversation context above. "
-            "This marker exists because the compacted transcript contained "
-            "no preserved user turn."
+            "This marker exists because no human user turn was available."
         ),
     })
 
@@ -691,6 +736,36 @@ def compress_context(
             if not _existing_sp:
                 _existing_sp = agent._build_system_prompt(system_message)
             return messages, _existing_sp
+        try:
+            _compression_tip = _lock_db.get_compression_tip(_lock_sid)
+            _session_row = _lock_db.get_session(_lock_sid)
+        except Exception as _state_err:
+            logger.debug(
+                "compression post-lock lineage check failed for session=%s: %s",
+                _lock_sid, _state_err,
+            )
+            _compression_tip = _lock_sid
+            _session_row = None
+        if (
+            (_compression_tip and _compression_tip != _lock_sid)
+            or (
+                isinstance(_session_row, dict)
+                and _session_row.get("end_reason") == "compression"
+            )
+        ):
+            logger.warning(
+                "compression skipped: session=%s already rotated to %s",
+                _lock_sid, _compression_tip or "unknown",
+            )
+            _existing_sp = getattr(agent, "_cached_system_prompt", None)
+            if not _existing_sp:
+                _existing_sp = agent._build_system_prompt(system_message)
+            if _lock_db is not None and _lock_sid and _lock_holder:
+                try:
+                    _lock_db.release_compression_lock(_lock_sid, _lock_holder)
+                except Exception as _rel_err:
+                    logger.debug("compression lock release failed: %s", _rel_err)
+            return messages, _existing_sp
         if _lock_holder is not None:
             _lock_refresher = _CompressionLockLeaseRefresher(
                 _lock_db,
@@ -780,6 +855,25 @@ def _release_lock() -> None:
         _release_lock()
         return messages, _existing_sp
 
+    if not compressed:
+        logger.error(
+            "context compression returned an empty transcript; refusing to "
+            "rotate session=%s so the parent remains resumable",
+            agent.session_id or "none",
+        )
+        try:
+            agent._emit_warning(
+                "⚠ Compression returned an empty transcript. "
+                "No session split was performed; conversation continues unchanged."
+            )
+        except Exception:
+            pass
+        _existing_sp = getattr(agent, "_cached_system_prompt", None)
+        if not _existing_sp:
+            _existing_sp = agent._build_system_prompt(system_message)
+        _release_lock()
+        return messages, _existing_sp
+
     try:
         summary_error = getattr(agent.context_compressor, "_last_summary_error", None)
         if summary_error:
@@ -809,7 +903,11 @@ def _release_lock() -> None:
 
         todo_snapshot = agent._todo_store.format_for_injection()
         if todo_snapshot:
-            compressed.append({"role": "user", "content": todo_snapshot})
+            compressed.append({
+                "role": "user",
+                "content": todo_snapshot,
+                "_todo_snapshot_synthetic": True,
+            })
         _ensure_compressed_has_user_turn(messages, compressed)
 
         agent._invalidate_system_prompt()
@@ -961,7 +1059,20 @@ def _release_lock() -> None:
                 # refresh the stored system prompt and reset the flush cursor so the
                 # next turn re-bases its append diff.
                 agent._session_db.update_system_prompt(agent.session_id, new_system_prompt)
-                agent._last_flushed_db_idx = 0
+                if in_place:
+                    agent._last_flushed_db_idx = 0
+                else:
+                    # A headless turn can be killed before its finalizer. Persist
+                    # the rotated child's compacted handoff at the boundary so
+                    # the new session is immediately resumable.
+                    agent._session_db.replace_messages(agent.session_id, compressed)
+                    agent._last_flushed_db_idx = len(compressed)
+                    agent._flushed_db_message_session_id = agent.session_id
+                    agent._flushed_db_message_ids = {
+                        id(message)
+                        for message in compressed
+                        if isinstance(message, dict)
+                    }
             except Exception as e:
                 # If the rotation rolled back to the parent (orphan-avoidance
                 # above), agent.session_id is the still-indexed parent and