Explain: add annotation support

explain/agents.py

@@ -20,6 +20,7 @@ class BaseAgent(ABC):
     agent_bin: Path
     log_level: str = "CRITICAL"
     additional_flags: list[str] = field(default_factory=list)
+    agent_subprocess_buffer_limit = 1024 * 1024
 
     name: ClassVar[str]
     program_name: ClassVar[str]

explain/amp_agent.py

@@ -130,6 +130,7 @@ async def ask(self, question: str, port: int, tools: list[str]) -> str:
                 stdin=asyncio.subprocess.PIPE,
                 stdout=asyncio.subprocess.PIPE,
                 stderr=asyncio.subprocess.PIPE,
+                limit=self.agent_subprocess_buffer_limit,
             )
             assert amp.stdin and amp.stdout and amp.stderr
 

explain/claude_agent.py

@@ -132,6 +132,7 @@ async def ask(self, question: str, port: int, tools: list[str]) -> str:
                 *self.additional_flags,
                 stdout=asyncio.subprocess.PIPE,
                 stderr=asyncio.subprocess.PIPE,
+                limit=self.agent_subprocess_buffer_limit,
             )
             assert claude.stdout and claude.stderr
 

explain/codex_agent.py

@@ -107,6 +107,7 @@ async def ask(self, question: str, port: int, tools: list[str]) -> str:
                 *self.additional_flags,
                 stdout=asyncio.subprocess.PIPE,
                 stderr=asyncio.subprocess.PIPE,
+                limit=self.agent_subprocess_buffer_limit,
             )
             assert codex.stdout and codex.stderr
 

explain/copilot_cli_agent.py

@@ -132,6 +132,7 @@ async def ask(self, question: str, port: int, tools: list[str]) -> str:
                 stdin=asyncio.subprocess.PIPE,
                 stdout=asyncio.subprocess.PIPE,
                 stderr=asyncio.subprocess.PIPE,
+                limit=self.agent_subprocess_buffer_limit,
             )
             assert copilot.stdin and copilot.stdout and copilot.stderr
 

explain/explain.py

@@ -10,6 +10,7 @@
 """
 
 import asyncio
+import bisect
 import contextlib
 import functools
 import inspect
@@ -149,6 +150,9 @@ def wrapped(*args: Any, **kwargs: Any):
 SOURCE_CONTEXT_LINES = 5
 """The maximum size of source context to show either side of the current position."""
 
+ANNOTATIONS_PAGE_LIMIT = 200
+"""A cap on the number of annotations returned in a single call to annotations_list."""
+
 
 def get_context(fname: str, line: int) -> str:
     """
@@ -728,6 +732,136 @@ def tool_ugo_bookmark(self, name: str) -> None:
         """
         self.udb.bookmarks.goto(name)
 
+    @report
+    @chain_of_thought
+    def tool_annotations_list(
+        self,
+        name: str | None = None,
+        detail: str | None = None,
+        limit: int = ANNOTATIONS_PAGE_LIMIT,
+        offset: int = 0,
+    ) -> dict:
+        """
+        Returns a list of annotations in the recording.
+
+        Use `annotations_count` first if expecting a large result set, to determine whether to apply
+        filters or pagination with `annotations_list`.
+
+        Params:
+        - `name`: if provided, only return annotations with this name
+        - `detail`: if provided, only return annotations with this detail
+        - `limit`: maximum number of annotations to return (default and maximum value: 200)
+        - `offset`: number of annotations to skip from the start (default: 0)
+
+        Returns a dict with:
+        - `annotations`: list of annotations {"name", "detail", "content", "bbcount"} in time order
+        - `total`: total number of matching annotations (across all pages)
+        - `returned`: number of annotations in this response
+        """
+        if limit < 1 or limit > ANNOTATIONS_PAGE_LIMIT:
+            raise Exception(f"Limit must be in the range 1-{ANNOTATIONS_PAGE_LIMIT}.")
+
+        # Prior to UDB 9.2, `annotations.get()` required `name` to be `str` instead of `str | None`.
+        # For backwards compatibility, we convert name from None to "" here, which means do not
+        # filter by name. This also works with 9.2+.
+        name_str = name if name is not None else ""
+        results = self.udb.annotations.get(name_str, detail)
+        total = len(results)
+
+        page = results[offset : offset + limit]
+        returned = len(page)
+
+        response: dict = {
+            "annotations": [
+                {
+                    "name": r.name,
+                    "detail": r.detail,
+                    "content": r.get_content_as_printable_text(),
+                    "bbcount": r.bbcount,
+                }
+                for r in page
+            ],
+            "total": total,
+            "returned": returned,
+        }
+
+        return response
+
+    @report
+    @chain_of_thought
+    def tool_annotations_count(self, name: str | None = None, detail: str | None = None) -> int:
+        """
+        Find the total count of annotations matching the given filters, without returning content.
+
+        Use this before `annotations_list` when you expect a large number of annotations, to
+        determine whether to apply filters or pagination.
+
+        Params:
+        - `name`: if provided, only count annotations with this name.
+        - `detail`: if provided, only count annotations with this detail.
+
+        Returns the number of annotations.
+        """
+
+        # Prior to UDB 9.2, `annotations.get()` required `name` to be `str` instead of `str | None`.
+        # For backwards compatibility, we convert name from None to "" here, which means do not
+        # filter by name. This also works with 9.2+.
+        name_str = name if name is not None else ""
+        results = self.udb.annotations.get(name_str, detail)
+        return len(results)
+
+    @report
+    @source_context
+    @collect_output
+    @chain_of_thought
+    def tool_annotation_goto(
+        self,
+        name: str | None = None,
+        detail: str | None = None,
+        bbcount: int | None = None,
+    ) -> None:
+        """
+        Navigate to the point in recorded history marked by the specified annotation.
+
+        Both `name` and `detail` must together identify a unique annotation. If multiple
+        annotations match, you should also specify a bbcount time belonging to the required
+        annotation.
+
+        Use `annotations_list` first to discover available annotations and their exact names,
+        details and bbcount.
+        """
+        # Prior to UDB 9.2, `annotations.get()` required `name` to be `str` instead of `str | None`.
+        # For backwards compatibility, we convert name from None to "" here, which means do not
+        # filter by name. This also works with 9.2+.
+        name_str = name if name is not None else ""
+        results = self.udb.annotations.get(name_str, detail)
+        if not results:
+            raise Exception(
+                f"No annotation found with name={name_str!r}"
+                + (f", detail={detail!r}" if detail else "")
+                + ". Use annotations_list to see available annotations."
+            )
+
+        if bbcount is not None:
+            idx = bisect.bisect_left(results, bbcount, key=lambda r: r.bbcount)
+            if idx < len(results) and results[idx].bbcount == bbcount:
+                self.udb.time.goto(bbcount)
+                return
+            raise Exception(
+                f"No annotation found with name={name_str!r}"
+                + (f", detail={detail!r}" if detail else "")
+                + f", and bbcount={bbcount}."
+            )
+
+        if len(results) != 1:
+            raise Exception(
+                f"Multiple annotations match name={name_str!r}"
+                + (f", detail={detail!r}" if detail else "")
+                + ". To select a unique annotation specify a more precise name and detail, or"
+                + " provide a bbcount."
+            )
+        self.udb.time.goto(results[0].bbcount)
+
     @report
     @chain_of_thought
     def tool_gtest_get_tests(self) -> list[tuple[str, str]]:

explain/instructions.md

@@ -157,6 +157,29 @@ debug tool:
     last a OR last b (to investigate why the `if` statement was entered)
 ```
 
+## Annotations
+
+Annotations are markers embedded in the recording at specific interesting points in time. They
+should be retrieved early to help inform the investigation.
+
+`annotations_list` can be used to list annotations in a recording, with optional name and detail
+filters. A recording may contain thousands of annotations, so the output may be paginated. If the
+`total` field is larger than `returned`, further annotations can be requested by increasing the
+`offset` parameter.
+
+`annotations_count` can be used to count the total number of annotations matching the
+specified `name` and `detail` parameters, without returning them.
+
+Searching source code for the debugged program may help to identify useful annotation names and
+details. Annotations are added by the debugged program with the `undoex_annotation_add_raw_data`,`undoex_annotation_add_int` or `undoex_annotation_add_text` functions.
+
+Only fetch annotations you actually need. There may be thousands of annotations, so prefer targeted
+queries to `annotations_list` over iterating through all annotations.
+
+You can go to an annotation with `annotation_goto`. If there are multiple annotations with the
+required name and detail, provide the bbcount of the annotation instead. The bbcount for each
+annotation is also returned within `annotations_list`.
+
 ## Bookmarks
 
 Set bookmarks (using `ubookmark`) at interesting points in recorded history if they may require