fix(acp): publish current context and compression depth

acp_adapter/server.py

@@ -844,9 +844,9 @@ def _build_usage_update(state: SessionState) -> UsageUpdate | None:
 
         Zed's circular context indicator is driven by ACP ``usage_update``
         session updates: ``size`` is the model context window and ``used`` is
-        the current request pressure.  Hermes estimates ``used`` from the same
-        buckets it sends to providers: system prompt, conversation history, and
-        tool schemas.
+        the current request pressure. Prefer the provider-reported prompt size
+        from the most recent model call; before any call has completed, fall
+        back to a rough estimate of the same request buckets.
         """
         agent = state.agent
         compressor = getattr(agent, "context_compressor", None)
@@ -855,21 +855,32 @@ def _build_usage_update(state: SessionState) -> UsageUpdate | None:
             return None
 
         try:
-            from agent.model_metadata import estimate_request_tokens_rough
-
-            used = estimate_request_tokens_rough(
-                state.history,
-                system_prompt=getattr(agent, "_cached_system_prompt", "") or "",
-                tools=getattr(agent, "tools", None) or None,
-            )
-        except Exception:
-            logger.debug("Could not estimate ACP native context usage", exc_info=True)
             used = int(getattr(compressor, "last_prompt_tokens", 0) or 0)
+        except (TypeError, ValueError):
+            used = 0
+        if used <= 0:
+            try:
+                from agent.model_metadata import estimate_request_tokens_rough
+
+                used = estimate_request_tokens_rough(
+                    state.history,
+                    system_prompt=getattr(agent, "_cached_system_prompt", "") or "",
+                    tools=getattr(agent, "tools", None) or None,
+                )
+            except Exception:
+                logger.debug("Could not estimate ACP native context usage", exc_info=True)
+                used = 0
+
+        try:
+            compression_count = max(0, int(getattr(compressor, "compression_count", 0) or 0))
+        except (TypeError, ValueError):
+            compression_count = 0
 
         return UsageUpdate(
             session_update="usage_update",
             size=max(size, 0),
             used=max(used, 0),
+            field_meta={"hermes": {"compressionCount": compression_count}},
         )
 
     async def _send_usage_update(self, state: SessionState) -> None:
@@ -966,6 +977,24 @@ def _schedule_usage_update(self, state: SessionState) -> None:
         loop = asyncio.get_running_loop()
         loop.call_soon(asyncio.create_task, self._send_usage_update(state))
 
+    def _schedule_usage_update_from_thread(
+        self,
+        loop: asyncio.AbstractEventLoop,
+        state: SessionState,
+    ) -> None:
+        """Publish current context pressure after a worker-thread model step."""
+
+        def _schedule_on_loop() -> None:
+            asyncio.create_task(self._send_usage_update(state))
+
+        try:
+            loop.call_soon_threadsafe(_schedule_on_loop)
+        except RuntimeError:
+            logger.debug(
+                "ACP event loop closed before usage update for %s",
+                state.session_id,
+            )
+
     async def _register_session_mcp_servers(
         self,
         state: SessionState,
@@ -1491,7 +1520,13 @@ async def load_session(
                 exc_info=True,
             )
         self._schedule_available_commands_update(session_id)
-        self._schedule_usage_update(state)
+        # Unlike session/new, session/load already has a client-side routing
+        # binding before the request starts so replay notifications can be
+        # consumed. Publish restored context/compression telemetry in the same
+        # request lifetime; deferring it until after return can lose the update
+        # during transport/task handoff and leaves a resumed client stale until
+        # its first prompt.
+        await self._send_usage_update(state)
         return LoadSessionResponse(
             models=self._build_model_state(state),
             modes=self._session_modes(state),
@@ -1527,7 +1562,11 @@ async def resume_session(
                 exc_info=True,
             )
         self._schedule_available_commands_update(state.session_id)
-        self._schedule_usage_update(state)
+        # `session/resume` has the same response-boundary contract as
+        # `session/load`: the client routing entry already exists, so restored
+        # context/compression telemetry must be delivered before this request
+        # returns rather than via a task that can lose the update on handoff.
+        await self._send_usage_update(state)
         return ResumeSessionResponse(
             models=self._build_model_state(state),
             modes=self._session_modes(state),
@@ -1791,9 +1830,15 @@ async def prompt(
                 edit_approval_policy_getter=lambda: self._edit_approval_policy_for_state(state),
             )
             reasoning_cb = make_thinking_cb(conn, session_id, loop)
-            step_cb = make_step_cb(conn, session_id, loop, tool_call_ids, tool_call_meta)
+            base_step_cb = make_step_cb(conn, session_id, loop, tool_call_ids, tool_call_meta)
             message_cb = make_message_cb(conn, session_id, loop)
 
+            def _step_cb(api_call_count: int, prev_tools: Any = None) -> None:
+                base_step_cb(api_call_count, prev_tools)
+                self._schedule_usage_update_from_thread(loop, state)
+
+            step_cb = _step_cb
+
             def stream_delta_cb(text: str) -> None:
                 nonlocal streamed_message
                 if text:
@@ -2336,6 +2381,15 @@ def _cmd_reset(self, args: str, state: SessionState) -> str:
             reset_session_state = getattr(state.agent, "reset_session_state", None)
             if callable(reset_session_state):
                 reset_session_state()
+            compressor = getattr(state.agent, "context_compressor", None)
+            persist_compression_count = getattr(
+                compressor, "_persist_compression_count", None
+            )
+            if callable(persist_compression_count):
+                # ACP /reset clears the existing stable session in place. Other
+                # hosts may call on_session_reset while switching to a new id,
+                # where persisting here would erase the old session's history.
+                persist_compression_count()
         except Exception:
             reset_failed = True
             logger.warning("ACP session state reset failed for %s", state.session_id, exc_info=True)

agent/context_compressor.py

@@ -1638,10 +1638,12 @@ def bind_session_state(self, session_db: Any = None, session_id: str = "") -> No
         self._fallback_compression_streak = 0
         self._ineffective_compression_count = 0
         self._prellm_skip_count = 0
+        self.compression_count = 0
         self._anti_thrash_recovery_deadline = 0.0
         self.get_active_compression_failure_cooldown()
         self._load_fallback_compression_streak()
         self._load_ineffective_compression_count()
+        self._load_compression_count()
 
     def on_session_start(self, session_id: str, **kwargs) -> None:
         """Bind session-scoped compression state for a new or resumed session."""
@@ -1651,6 +1653,7 @@ def on_session_start(self, session_id: str, **kwargs) -> None:
         session_db = kwargs.get("session_db", getattr(self, "_session_db", None))
         previous_fallback_streak = self._fallback_compression_streak
         previous_ineffective_count = self._ineffective_compression_count
+        previous_compression_count = self.compression_count
         if boundary_reason == "compression" and old_session_id:
             getter = getattr(session_db, "get_compression_fallback_streak", None)
             if callable(getter):
@@ -1682,12 +1685,35 @@ def on_session_start(self, session_id: str, **kwargs) -> None:
                         "compression parent ineffective count lookup failed (non-sqlite): %s",
                         exc,
                     )
+            compression_getter = getattr(session_db, "get_compression_count", None)
+            if callable(compression_getter):
+                try:
+                    stored_count = compression_getter(old_session_id)
+                    if isinstance(stored_count, (int, float, str)):
+                        previous_compression_count = max(
+                            previous_compression_count,
+                            max(0, int(stored_count)),
+                        )
+                except (TypeError, ValueError, sqlite3.Error) as exc:
+                    logger.debug("compression parent count lookup failed: %s", exc)
+                except Exception as exc:
+                    logger.debug("compression parent count lookup failed (non-sqlite): %s", exc)
         self.bind_session_state(session_db, session_id)
         if boundary_reason == "compression":
+            # A recovery path may re-adopt a child row that already has a
+            # newer durable count. Compression depth is monotonic within the
+            # logical conversation, so preserve the newest valid value from
+            # live state, the parent row, or the destination row.
+            previous_compression_count = max(
+                previous_compression_count,
+                self.compression_count,
+            )
             # Rotation creates a fresh child row before this callback. Preserve
             # the logical conversation's streak until boundary bookkeeping
             # persists the updated value onto the child row.
             self._fallback_compression_streak = previous_fallback_streak
+            self.compression_count = previous_compression_count
+            self._persist_compression_count()
             # Same for the anti-thrash strike counter — but unlike the streak,
             # no later boundary bookkeeping writes it, so persist the carried
             # value onto the (fresh) child row now. Otherwise a restart between
@@ -1716,6 +1742,38 @@ def _load_fallback_compression_streak(self) -> None:
         except Exception as exc:
             logger.debug("compression fallback streak lookup failed (non-sqlite): %s", exc)
 
+    def _load_compression_count(self) -> None:
+        session_db = getattr(self, "_session_db", None)
+        session_id = getattr(self, "_session_id", "")
+        getter = getattr(session_db, "get_compression_count", None)
+        if not session_id or not callable(getter):
+            return
+        try:
+            stored_count = getter(session_id)
+            self.compression_count = max(
+                0,
+                int(stored_count)
+                if isinstance(stored_count, (int, float, str))
+                else 0,
+            )
+        except (TypeError, ValueError, sqlite3.Error) as exc:
+            logger.debug("compression count lookup failed: %s", exc)
+        except Exception as exc:
+            logger.debug("compression count lookup failed (non-sqlite): %s", exc)
+
+    def _persist_compression_count(self) -> None:
+        session_db = getattr(self, "_session_db", None)
+        session_id = getattr(self, "_session_id", "")
+        setter = getattr(session_db, "set_compression_count", None)
+        if not session_id or not callable(setter):
+            return
+        try:
+            setter(session_id, self.compression_count)
+        except sqlite3.Error as exc:
+            logger.debug("compression count persist failed: %s", exc)
+        except Exception as exc:
+            logger.debug("compression count persist failed (non-sqlite): %s", exc)
+
     def _persist_fallback_compression_streak(self) -> None:
         session_db = getattr(self, "_session_db", None)
         session_id = getattr(self, "_session_id", "")
@@ -1796,6 +1854,7 @@ def record_completed_compaction(
         breaker exists for, and its recovery probe bounds the block.
         """
         self._verify_compaction_cleared_threshold = True
+        self._persist_compression_count()
         if feasibility_skip:
             # A deliberate pre-LLM feasibility skip (#60451) is not a
             # summary-quality verdict: it must neither extend a fallback

contributors/emails/stefan@noble-pro.com

@@ -0,0 +1,2 @@
+stefanpieter
+# PR #74032: ACP current-context and compression telemetry

hermes_state.py

@@ -3623,6 +3623,40 @@ def _do(conn):
 
         self._execute_write(_do)
 
+    def get_compression_count(self, session_id: str) -> int:
+        """Return the durable completed-compaction count for a session."""
+        if not session_id:
+            return 0
+        with self._lock:
+            conn = self._conn
+            if conn is None:
+                return 0
+            row = conn.execute(
+                "SELECT compression_count FROM sessions WHERE id = ?",
+                (session_id,),
+            ).fetchone()
+        if row is None:
+            return 0
+        value = row["compression_count"] if isinstance(row, sqlite3.Row) else row[0]
+        try:
+            return max(0, int(value or 0))
+        except (TypeError, ValueError):
+            return 0
+
+    def set_compression_count(self, session_id: str, count: int) -> None:
+        """Persist the completed-compaction count for a session."""
+        if not session_id:
+            return
+        normalized = max(0, int(count))
+
+        def _do(conn):
+            conn.execute(
+                "UPDATE sessions SET compression_count = ? WHERE id = ?",
+                (normalized, session_id),
+            )
+
+        self._execute_write(_do)
+
     def get_compression_ineffective_count(self, session_id: str) -> int:
         """Return the persisted ineffective-compaction strike count.
 

hermes_state_common.py

@@ -180,6 +180,7 @@ def _ephemeral_child_sql(alias: str = "s") -> str:
     handoff_error TEXT,
     compression_failure_cooldown_until REAL,
     compression_failure_error TEXT,
+    compression_count INTEGER NOT NULL DEFAULT 0,
     compression_fallback_streak INTEGER NOT NULL DEFAULT 0,
     compression_ineffective_count INTEGER NOT NULL DEFAULT 0,
     profile_name TEXT,