Merge remote-tracking branch 'origin/V1.4.2' into improvement/llm-providers-upgrade

Author: CraftBot

.gitignore

@@ -59,4 +59,6 @@ agent_file_system/ACTIONS.md
 agent_bundle/
 **/.craftbot/
 app/data/.file_index/
-.playwright-mcp
+.playwright-mcp
+# Sidecar Node runtime (install.py downloads it when the system Node is too old for Living UI)
+runtime/

agent_core/core/event_stream/event.py

@@ -142,6 +142,12 @@ class Event:
             uses it to keep the run's "Working…" indicator up across the
             bubble instead of treating every agent bubble as a run-ending
             reply. None/False for final replies and non-chat events.
+        question: For AGENT_MESSAGE events only: set when the message is a
+            question to the user with suggested responses (send_message with
+            suggested_responses). Shape:
+            ``{"options": ["Yes", "No"], "allow_free_text": true}``. The UI
+            renders it as answer chips plus a pinned question box above the
+            chat composer. None for ordinary messages.
     """
 
     message: str
@@ -157,6 +163,7 @@ class Event:
     action_output: Optional[Dict[str, Any]] = None
     platform: Optional[str] = None
     continue_work: Optional[bool] = None
+    question: Optional[Dict[str, Any]] = None
 
     def display_text(self) -> Optional[str]:
         """
@@ -189,6 +196,7 @@ def to_dict(self) -> Dict[str, Any]:
             "action_output": self.action_output,
             "platform": self.platform,
             "continue_work": self.continue_work,
+            "question": self.question,
         }
 
     @classmethod
@@ -228,6 +236,7 @@ def from_dict(cls, data: Dict[str, Any]) -> "Event":
             action_output=data.get("action_output"),
             platform=data.get("platform"),
             continue_work=data.get("continue_work"),
+            question=data.get("question"),
         )
 
     @property

agent_core/core/impl/action/context.py

@@ -0,0 +1,41 @@
+"""Execution-scoped context for in-process actions.
+
+``current_input_data`` holds the full ``input_data`` dict of the action
+currently executing in this context. It exists so cross-cutting helpers
+deep inside an action's call tree (e.g. multi-account routing reading the
+``account`` hint) can see routing keys without threading them through
+every action function signature.
+
+Scope rules:
+  - Set only by the internal executors (``_atomic_action_internal*``),
+    reset in a ``finally`` — never leaks across actions.
+  - Sync actions run in a thread pool where the caller's context does NOT
+    propagate, so the executor wraps the call and sets the var inside the
+    worker thread (see ``run_with_input_context``).
+  - Sandboxed (subprocess) actions cannot see it at all — helpers must
+    treat a ``None`` value as "no context available".
+"""
+
+from __future__ import annotations
+
+from contextvars import ContextVar
+from typing import Any, Callable, Dict, Optional
+
+current_input_data: ContextVar[Optional[Dict[str, Any]]] = ContextVar(
+    "current_input_data", default=None
+)
+
+
+def run_with_input_context(
+    function_to_call: Callable[[dict], dict], input_data: dict
+) -> dict:
+    """Call a sync action with ``current_input_data`` set for its duration.
+
+    Used as the thread-pool target: the worker thread has its own context,
+    so the var must be set (and reset) inside the thread, not the caller.
+    """
+    token = current_input_data.set(input_data)
+    try:
+        return function_to_call(input_data)
+    finally:
+        current_input_data.reset(token)

agent_core/core/impl/action/executor.py

@@ -571,7 +571,9 @@ def _atomic_action_internal(
                 "The action_code string did not define a callable Python function."
             )
 
-        execution_result = function_to_call(input_data)
+        from agent_core.core.impl.action.context import run_with_input_context
+
+        execution_result = run_with_input_context(function_to_call, input_data)
         return execution_result
 
     except Exception as e:
@@ -618,16 +620,29 @@ async def _atomic_action_internal_async(
                 "The action_code string did not define a callable Python function."
             )
 
+        from agent_core.core.impl.action.context import (
+            current_input_data,
+            run_with_input_context,
+        )
+
         # Check if the function is async (coroutine function)
         if inspect.iscoroutinefunction(function_to_call):
             logger.debug(f"[ASYNC] Action '{action_name}' is async, awaiting directly")
-            execution_result = await function_to_call(input_data)
+            ctx_token = current_input_data.set(input_data)
+            try:
+                execution_result = await function_to_call(input_data)
+            finally:
+                current_input_data.reset(ctx_token)
         else:
-            # Sync function - run in thread pool to avoid blocking
+            # Sync function - run in thread pool to avoid blocking. The
+            # worker thread doesn't inherit this context, so the wrapper
+            # sets current_input_data inside the thread.
             logger.debug(
                 f"[SYNC] Action '{action_name}' is sync, running in thread pool"
             )
-            thread_future = THREAD_POOL.submit(function_to_call, input_data)
+            thread_future = THREAD_POOL.submit(
+                run_with_input_context, function_to_call, input_data
+            )
             try:
                 execution_result = await asyncio.wrap_future(thread_future)
             except asyncio.CancelledError:

agent_core/core/impl/action/manager.py

@@ -99,6 +99,43 @@ async def _compat_wait_for(fut, timeout):
 
 nest_asyncio.apply()
 
+# ============================================================================
+# Second half of the nest_asyncio/3.14 shim: heal asyncio.current_task().
+# nest_asyncio forces the PURE-PYTHON asyncio.Task class, whose tasks
+# register in the Python-side registry (asyncio.tasks._py_current_task) —
+# but asyncio.current_task stays bound to the C-accelerated registry, so it
+# returns None inside EVERY task, on EVERY loop, process-wide. Everything
+# built on `async with asyncio.timeout(...)` then dies with "Timeout
+# (context manager) should be used inside a task" — most visibly the entire
+# aiohttp CLIENT (every request enters a timeout context), which is what
+# broke the external A2App adapter self-check on 2026-08-24 while the
+# aiohttp SERVER (no timeout context on the request path) kept working.
+# Rebinding current_task to the Python registry fixes timeout/aiohttp under
+# both plain awaits and nested re-entry (verified on 3.14.7 + aiohttp
+# 3.14.3). The wait_for replacement above stays: its explicit
+# cancellation-wait semantics are load-bearing for force-stop (PR #410).
+try:
+    import _asyncio as _compat_c_asyncio
+
+    if asyncio.Task is not getattr(_compat_c_asyncio, "Task", None) and hasattr(
+        asyncio.tasks, "_py_current_task"
+    ):
+        asyncio.current_task = asyncio.tasks._py_current_task
+        asyncio.tasks.current_task = asyncio.tasks._py_current_task
+        try:
+            _compat_sys.stderr.write(
+                "[compat-shim] asyncio.current_task routed to the Python "
+                "task registry (action/manager)\n"
+            )
+            _compat_sys.stderr.flush()
+        except Exception:
+            pass
+except Exception as _compat_ct_exc:
+    logger.warning(
+        f"[compat-shim] current_task rebinding skipped: {_compat_ct_exc!r}"
+    )
+# ============================================================================
+
 
 def _to_pretty_json(value: Any) -> str:
     """Serialize a value to pretty-printed JSON for readable logs and event streams."""
@@ -247,10 +284,7 @@ async def execute_action(
         # re-execute work the ledger shows as already completed (or as
         # interrupted mid-flight, where the effect may have happened).
         idem_key = None
-        # if getattr(action, "irreversible", False) and self._idempotency_guard:
-
-        # TODO: Temporary turning idempotency guard off.
-        if 1 == 0:
+        if getattr(action, "irreversible", False) and self._idempotency_guard:
             try:
                 decision = self._idempotency_guard.begin(
                     action.name, input_data, session_id

agent_core/core/impl/event_stream/event_stream.py

@@ -41,6 +41,11 @@
 # leaving the action displayed as "running" forever.
 MIN_KEEP_RECENT_EVENTS = 2
 
+# Smallest fold worth an LLM call. Summarization is a blocking ~15s round trip;
+# collapsing a couple of hundred tokens with one is a straight loss and the
+# threshold is breached again on the very next event, so we prune instead.
+MIN_FOLD_TOKENS = 2000
+
 # Event kinds that summarization must NEVER collapse — they are kept verbatim in
 # tail_events forever, so the contract they carry survives any number of
 # summarization passes. `requirements` (from set_requirement) defines the task's
@@ -217,6 +222,7 @@ def log(
         action_output: Optional[dict] = None,
         platform: Optional[str] = None,
         continue_work: Optional[bool] = None,
+        question: Optional[dict] = None,
     ) -> int:
         """
         Append a new event to the stream and trigger summarization if needed.
@@ -249,6 +255,9 @@ def log(
             continue_work: For AGENT_MESSAGE events: True when this is a
                 mid-run progress update and the agent keeps working after
                 sending it (drives the UI's persistent "Working…" row).
+            question: For AGENT_MESSAGE events: suggested-response payload
+                (``{"options": [...], "allow_free_text": bool}``) when the
+                message is a question the UI should pin above the composer.
 
         Returns:
             The zero-based index of the event within ``tail_events``.
@@ -270,6 +279,7 @@ def log(
             action_output=action_output,
             platform=platform,
             continue_work=continue_work,
+            question=question,
         )
         rec = EventRecord(event=ev)
 
@@ -298,9 +308,19 @@ def log_action_end(self, name: str, status: str, extra: str = "") -> int:
     # ───────────────────── summarization & pruning ───────────────────────
 
     def _externalize_message(
-        self, message: str, *, action_name: str | None = None
+        self,
+        message: str,
+        *,
+        action_name: str | None = None,
+        force: bool = False,
     ) -> str:
-        """Persist overly long messages to a temp file and return a pointer event."""
+        """Persist overly long messages to a temp file and return a pointer event.
+
+        `force` overrides the retrieval-action exemption below. It is used by
+        `_shrink_pinned_oversize`, where the agent has already consumed the
+        content in its own turn and the only thing left to do with an oversized
+        event is stop paying for it every prompt.
+        """
         if len(message) <= MAX_EVENT_INLINE_CHARS or self.temp_dir is None:
             return message
 
@@ -309,7 +329,12 @@ def _externalize_message(
         # send the agent chasing a pointer to a pointer. ("grep" / "stream
         # read" are legacy names kept for safety; the live actions are
         # grep_files / read_file.)
-        if action_name in ("grep_files", "read_file", "grep", "stream read"):
+        if not force and action_name in (
+            "grep_files",
+            "read_file",
+            "grep",
+            "stream read",
+        ):
             return message
 
         try:
@@ -388,6 +413,53 @@ def _find_token_cutoff(self, events: List[EventRecord], keep_tokens: int) -> int
         )
         return cutoff
 
+    def _shrink_pinned_oversize(self, cutoff: int) -> int:
+        """Externalize oversized events in the surviving tail, in place.
+
+        MIN_KEEP_RECENT_EVENTS pins the newest events so the UI (which mirrors
+        `tail_events`) never loses an `action_end` in the tick it arrives — an
+        action purged that early renders as "running" forever. But the pin is
+        blind to size: when a retrieval action returns a huge payload (grep_files
+        and read_file are exempt from log-time externalization, because they ARE
+        how the agent reads externalized content back), the pin holds tens of
+        thousands of tokens verbatim and a summarization pass cannot get under
+        the threshold. The next event re-triggers it and the SAME chunk gets
+        folded on the second try — one entirely wasted blocking LLM call per
+        oversized event.
+
+        Shrinking in place satisfies both constraints: the record survives with
+        its `action_id` intact so the UI still pairs start↔end, and its message
+        becomes a pointer the agent can re-read on demand. Caller holds the lock.
+
+        Returns the number of tokens reclaimed.
+        """
+        if self.temp_dir is None:
+            return 0
+
+        reclaimed = 0
+        for rec in self.tail_events[cutoff:]:
+            message = rec.event.message
+            if len(message) <= MAX_EVENT_INLINE_CHARS:
+                continue
+            pointer = self._externalize_message(
+                message, action_name=rec.event.action_name, force=True
+            )
+            if pointer is message:
+                # Externalization failed (already logged); leave the event alone.
+                continue
+            before = get_cached_token_count(rec)
+            rec.event.message = pointer
+            rec._cached_tokens = None
+            reclaimed += before - get_cached_token_count(rec)
+
+        if reclaimed:
+            self._total_tokens -= reclaimed
+            logger.info(
+                f"[EventStream] Collapsed oversized pinned event(s) in place, "
+                f"reclaiming {reclaimed} tokens (now {self._total_tokens})"
+            )
+        return reclaimed
+
     def summarize_by_LLM(self) -> None:
         """
         Summarize the oldest tail events using the language model.
@@ -406,6 +478,17 @@ def summarize_by_LLM(self) -> None:
             self.tail_events, self.tail_keep_after_summarize_tokens
         )
 
+        # Collapse anything oversized that the recent-event pin is holding
+        # verbatim BEFORE deciding whether an LLM call is warranted — that alone
+        # often drops the stream back under the threshold for free.
+        if self._shrink_pinned_oversize(cutoff):
+            if self._total_tokens < self.summarize_at_tokens:
+                return
+            # Budget changed; the fold boundary moves with it.
+            cutoff = self._find_token_cutoff(
+                self.tail_events, self.tail_keep_after_summarize_tokens
+            )
+
         if cutoff <= 0:
             # Nothing old enough to summarize
             return
@@ -419,6 +502,29 @@ def summarize_by_LLM(self) -> None:
             # Everything old enough to summarize is protected — nothing to collapse.
             return
 
+        chunk_tokens = sum(get_cached_token_count(r) for r in chunk)
+        if chunk_tokens < MIN_FOLD_TOKENS:
+            # The foldable region is smaller than the LLM call is worth — the tail
+            # is dominated by events we're required to keep (protected kinds, or
+            # the recent-event pin). Prune the chunk without a summary rather than
+            # burn ~15s and a full prompt to reclaim a rounding error. Losing this
+            # little detail is cheaper than the alternative, which is re-triggering
+            # on every subsequent log() call.
+            logger.warning(
+                f"[EventStream] Foldable region is only {chunk_tokens} tokens "
+                f"(< {MIN_FOLD_TOKENS}); pruning {len(chunk)} event(s) without an "
+                f"LLM call. Tail is dominated by pinned/protected events."
+            )
+            self._total_tokens -= chunk_tokens
+            self.tail_events = protected + self.tail_events[cutoff:]
+            self._append_summarization_notice(
+                folded_events=len(chunk),
+                folded_tokens=chunk_tokens,
+                summary=None,
+            )
+            self._session_sync_points.clear()
+            return
+
         first_ts = chunk[0].ts
         last_ts = chunk[-1].ts
         window = f"{first_ts.isoformat()} to {last_ts.isoformat()}"
@@ -448,8 +554,13 @@ def summarize_by_LLM(self) -> None:
             logger.info(
                 f"[EventStream] Running synchronous summarization ({self._total_tokens} tokens)"
             )
+            # json_mode=False: this prompt asks for a prose summary, and
+            # forcing a provider's JSON mode onto it degenerates (DeepSeek
+            # returns whitespace-only output that reads as empty).
             llm_output = self.llm.generate_response(
-                user_prompt=prompt, prompt_name="EVENT_STREAM_SUMMARIZATION"
+                user_prompt=prompt,
+                prompt_name="EVENT_STREAM_SUMMARIZATION",
+                json_mode=False,
             )
             new_summary = (llm_output or "").strip()
 
@@ -465,8 +576,8 @@ def summarize_by_LLM(self) -> None:
 
             # Apply summary and prune events
             self.head_summary = new_summary
-            # Calculate tokens being removed from the snapshotted chunk
-            removed_tokens = sum(get_cached_token_count(r) for r in chunk)
+            # Tokens being removed from the snapshotted chunk (measured above).
+            removed_tokens = chunk_tokens
             self._total_tokens -= removed_tokens
             # Keep protected events verbatim at the front of the surviving tail.
             self.tail_events = protected + self.tail_events[cutoff:]
@@ -492,7 +603,7 @@ def summarize_by_LLM(self) -> None:
             # Fallback: drop the oldest chunk without generating a summary so that
             # _total_tokens falls below the threshold.  Without this, every subsequent
             # log() call would immediately re-trigger summarization and flood the logs.
-            removed_tokens = sum(get_cached_token_count(r) for r in chunk)
+            removed_tokens = chunk_tokens
             self._total_tokens -= removed_tokens
             # Keep protected events verbatim even on the no-LLM prune fallback.
             self.tail_events = protected + self.tail_events[cutoff:]