fix: round-four findings — the deadlock behind the desync, clamped at the source

The round-three drain fix traded the pipe desync for a deadlock: once a
mid-run terminal error went out, the drain suppressed every later event —
including a permission prompt whose stdio wait is unbounded — so the worker
parked forever and the process never served another turn. Terminal now
cancels the run (the handler's cancelled event breaks both the confirm and
input waits), and the invariant itself moved into CanonicalTranslator:
nothing is emitted after the first terminal event, making every consumer
loop — stdio, HTTP, email — correct by construction, with tests.

The walk bound became a cap instead of a crash: a legitimate repo with a
giant assets/ subtree now indexes what was found (loudly), instead of a
RuntimeError advising the user to do what they had already done. The budget
counts entries after pruning, lives in CodeIndexConfig with the other
limits, and has tests. expanduser now applies to the STORED index root too,
so a '~'-style allowed_paths config can't produce a root that rejects the
very paths the refusal message recommends.

Ctrl+V: a transiently busy Windows clipboard no longer fails a paste whose
caption text was already in hand (a typed busy sentinel distinguishes it
from a real decode failure), and a Linux box with neither wl-paste nor
xclip finally says so instead of a silent no-op. Turns whose final carried
an empty answer keep the user's question in history. SKILL.md now
distinguishes the /query capacity 503 from the init 503 it previously
taught was the only kind. The cancel path uses the shared dead-child reset
helper instead of hand-rolling two-thirds of it.

Author: Ovtcharov

hub/agents/gaia/npm/SKILL.md

@@ -463,6 +463,11 @@ A healthy run streams `status` / `token` events and ends with one `final`. If
 `/v1/gaia/init` is 503, fix what its `hint` names and retry — the rest of your
 integration is fine.
 
+**A 503 from `/query` itself is a different condition**: every retained
+session slot is busy and none is idle enough to evict (SPEC §5.2). Do NOT
+loop on `/v1/gaia/init` — it will report ready. Wait for a running turn to
+finish (or close an idle session) and retry the same `/query`.
+
 For the full wire contract, lock schema, exit codes, and timeout table, see
 [`SPEC.md`](./SPEC.md). For the user-facing overview, see [`README.md`](./README.md)
 and <https://amd-gaia.ai/docs/guides/gaia>.

hub/agents/gaia/python/gaia_agent/stdio.py

@@ -883,6 +883,12 @@ def _run() -> None:
                     streamed_answer = str(canonical.get("answer") or "")
                 if canonical.get("type") in TERMINAL_TYPES:
                     terminated = True
+                    # The reader is gone the moment a terminal event goes
+                    # out, so cancel the run: a worker that parks on a
+                    # confirmation nobody can ever see would otherwise wait
+                    # forever and this drain — which must outlive the worker
+                    # — would never return, wedging the whole process.
+                    handler.cancelled.set()
 
         if not terminated:
             for canonical in translator.flush():
@@ -897,10 +903,11 @@ def _run() -> None:
         if terminated:
             # The normal exit. A turn that ended in an error event is not
             # recorded — replaying a failure as if it were an answer teaches
-            # the model that the failure is what it said — and an empty
-            # streamed answer is skipped for the same reason the fallback
-            # path below skips one.
-            if streamed_answer:
+            # the model that the failure is what it said. An EMPTY final is
+            # recorded (with its empty answer): dropping it would also drop
+            # the user's question, and "try answering my last question
+            # again" must not reach a model with no record it was asked.
+            if streamed_answer is not None:
                 _record_turn(agent, query, streamed_answer)
             return
         if "error" in result:
@@ -919,10 +926,9 @@ def _run() -> None:
                     break
         elif isinstance(value, str):
             answer = value
-        if answer:
-            # An empty answer is not a turn worth replaying into every later
-            # prompt — same rule as the streamed branch above.
-            _record_turn(agent, query, answer)
+        # Recorded even when empty — same reasoning as the streamed branch:
+        # the question half of the pair must survive.
+        _record_turn(agent, query, answer)
         _write({"type": "final", "answer": answer}, out)
     finally:
         # Every exit path, including the early returns above: leaving a dead

src/gaia/agents/tools/code_index_tools.py

@@ -68,7 +68,10 @@ def _init_code_index_state(
             repo_path: Repository root (absolute or relative, resolved here).
             code_index_config: Optional pre-built ``CodeIndexConfig``.
         """
-        self._repo_path = os.path.abspath(repo_path)
+        # expanduser everywhere a root is stored, or a '~'-style
+        # allowed_paths config produces a "<cwd>/~" root that rejects the
+        # very paths it was meant to allow.
+        self._repo_path = os.path.abspath(os.path.expanduser(repo_path))
         self._code_index_config = code_index_config
         self._code_index_sdk: Optional[Any] = None
 

src/gaia/code_index/sdk.py

@@ -64,6 +64,10 @@ class CodeIndexConfig:
 
     repo_path: str
     max_files: int = 5000
+    #: Ceiling on directory entries the discovery walk may enumerate (counted
+    #: after pruning). max_files caps what gets INDEXED; without this, walking
+    #: a huge non-repo root could burn the whole tool timeout just listing.
+    max_walk_entries: int = 200_000
     max_file_size_mb: float = 1
     chunk_overlap: int = 50
     embedding_model: str = DEFAULT_EMBEDDING_MODEL
@@ -604,19 +608,12 @@ def _discover_files(self) -> List[str]:
         # Bound the WALK, not just the result list: max_files caps how many
         # files are indexed, but enumerating a huge non-repo root (a home
         # directory, a whole drive) could burn the entire tool timeout just
-        # listing entries before a single one was embedded.
-        max_entries = max(self.config.max_files * 40, 200_000)
+        # listing entries. A cap, not an error — a legitimate repo with a
+        # giant assets/ subtree indexes what was found, exactly like the
+        # max_files cap below, and the truncation is logged loudly.
         entries_seen = 0
 
         for root, dirs, files in os.walk(str(self._repo_root)):
-            entries_seen += len(dirs) + len(files)
-            if entries_seen > max_entries:
-                raise RuntimeError(
-                    f"stopped after scanning {entries_seen} directory entries "
-                    f"under {self._repo_root} without finishing — this does "
-                    "not look like a code repository. Point repo_path at the "
-                    "repository root itself."
-                )
             rel_root = Path(root).relative_to(self._repo_root)
 
             # Filter out skipped directories in-place
@@ -629,6 +626,13 @@ def _discover_files(self) -> List[str]:
                 and not any(fnmatch.fnmatch(d, p) for p in ignore_patterns)
             ]
 
+            # Counted AFTER pruning, so node_modules/.git and gitignored
+            # trees cost nothing against the budget. Checked at the END of
+            # the loop body (files from this directory are kept), so one
+            # directory can overshoot by its own listing — which os.walk
+            # already paid for — but partial results always survive.
+            entries_seen += len(dirs) + len(files)
+
             if len(result) >= self.config.max_files:
                 break
 
@@ -662,6 +666,17 @@ def _discover_files(self) -> List[str]:
 
                 result.append(abs_path)
 
+            if entries_seen > self.config.max_walk_entries:
+                self.log.warning(
+                    f"discovery stopped after {entries_seen} directory "
+                    f"entries under {self._repo_root} (max_walk_entries="
+                    f"{self.config.max_walk_entries}) — indexing the "
+                    f"{len(result)} files found so far. If this is a code "
+                    "repository, raise max_walk_entries; if not, point "
+                    "repo_path at the repository root."
+                )
+                break
+
         return result
 
     def _read_gitignore_patterns(self) -> List[str]:

src/gaia/ui/sse_translation.py

@@ -156,12 +156,23 @@ def __init__(
         # Last user-facing status message, cleared as soon as any other event
         # is emitted — see ``_user_status``.
         self._last_status: Optional[str] = None
+        # Latched once a terminal event goes out — see ``translate``.
+        self._terminal_emitted = False
 
     # -- public API --------------------------------------------------------
 
     def translate(self, event: Dict[str, Any]) -> List[Dict[str, Any]]:
-        """Map one source event to zero or more canonical events."""
-        return self._track_status(self._translate(event))
+        """Map one source event to zero or more canonical events.
+
+        Nothing is emitted after the first terminal event: the contract is
+        "exactly one ``final`` or ``error`` ends every turn", and a reader
+        stops at it — anything written afterwards sits unread in the pipe and
+        gets consumed as the opening events of the NEXT turn. Clamping here
+        makes every consumer loop (stdio, HTTP, email) trivially correct.
+        """
+        if self._terminal_emitted:
+            return []
+        return self._clamp_terminal(self._track_status(self._translate(event)))
 
     def _track_status(self, out: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
         """Forget the last status once anything else reaches the wire.
@@ -208,7 +219,17 @@ def _translate(self, event: Dict[str, Any]) -> List[Dict[str, Any]]:
 
     def flush(self) -> List[Dict[str, Any]]:
         """Release any buffered ``tool_call`` at stream close."""
-        return self._track_status(self._flush_pending())
+        if self._terminal_emitted:
+            return []
+        return self._clamp_terminal(self._track_status(self._flush_pending()))
+
+    def _clamp_terminal(self, out: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """Truncate a batch at its first terminal event and latch the clamp."""
+        for i, e in enumerate(out):
+            if e.get("type") in TERMINAL_TYPES:
+                self._terminal_emitted = True
+                return out[: i + 1]
+        return out
 
     # -- tool_call buffering (spec §6.3) -----------------------------------
 

tests/unit/test_code_index_sdk.py

@@ -741,3 +741,45 @@ def fake_encode(texts, chunks):
             "hole.py" not in hashes
         ), "a file that lost chunks must be re-tried on the next index"
         assert result.chunks_dropped >= 1
+
+
+class TestWalkBound:
+    """max_walk_entries caps ENUMERATION (max_files caps what gets indexed).
+
+    A cap, not an error: a legitimate repo with a giant subtree indexes the
+    files found before the budget ran out, exactly like the max_files cap —
+    the truncation is logged, never raised.
+    """
+
+    def test_walk_stops_at_the_entry_budget_and_keeps_partial_results(
+        self, tmp_path, caplog
+    ):
+        skip_if_unavailable()
+        repo = tmp_path / "repo"
+        repo.mkdir()
+        # Nested, one file per subdir: the budget is checked per directory,
+        # so truncation lands mid-walk and partial results must survive.
+        for i in range(30):
+            sub = repo / f"sub{i:02d}"
+            sub.mkdir()
+            (sub / "f.py").write_text(f"def f{i}(): pass\n", encoding="utf-8")
+
+        sdk = make_sdk(tmp_path)
+        sdk.config.max_walk_entries = 35  # 30 subdir entries + a few files
+
+        import logging
+
+        with caplog.at_level(logging.WARNING):
+            files = sdk._discover_files()
+
+        assert 0 < len(files) < 30
+        assert any("max_walk_entries" in r.message for r in caplog.records)
+
+    def test_default_budget_does_not_touch_a_normal_repo(self, tmp_path):
+        skip_if_unavailable()
+        repo = tmp_path / "repo"
+        repo.mkdir()
+        for i in range(5):
+            (repo / f"f{i}.py").write_text("x = 1\n", encoding="utf-8")
+        sdk = make_sdk(tmp_path)
+        assert len(sdk._discover_files()) == 5

tests/unit/test_final_usage_completeness.py

@@ -83,3 +83,32 @@ def test_a_turn_with_no_ttft_reports_none(self):
     def test_a_bare_answer_carries_no_usage_at_all(self):
         final = translate({"type": "answer", "content": "hi"})
         assert "usage" not in final
+
+
+class TestTerminalClamp:
+    """Nothing is emitted after the first terminal event.
+
+    The contract is one final|error per turn, and every reader stops at it —
+    an event written afterwards sits unread in the pipe and opens the NEXT
+    turn (the stdio pipe-poisoning bug). The clamp lives in the translator so
+    every consumer loop is trivially correct.
+    """
+
+    def test_events_after_a_terminal_error_are_suppressed(self):
+        t = CanonicalTranslator(run_id=None, agent_id="gaia", debug=False)
+        first = t.translate({"type": "policy_alert", "reason": "blocked"})
+        assert any(e.get("type") == "error" for e in first)
+
+        assert t.translate({"type": "status", "message": "still going"}) == []
+        assert t.translate({"type": "answer", "answer": "late answer"}) == []
+        assert t.flush() == []
+
+    def test_a_batch_is_truncated_at_its_terminal(self):
+        t = CanonicalTranslator(run_id=None, agent_id="gaia", debug=False)
+        out = t.translate({"type": "answer", "answer": "done"})
+        # whatever preceded it, nothing FOLLOWS the terminal in the batch
+        terminal_indices = [
+            i for i, e in enumerate(out) if e.get("type") in ("final", "error")
+        ]
+        assert terminal_indices, "an answer event must produce a terminal"
+        assert terminal_indices[0] == len(out) - 1

tui/internal/client/subprocess.go

@@ -237,8 +237,11 @@ func (s *SubprocessClient) Send(ctx context.Context, query string) (<-chan inter
 		// and reads nothing.
 		defer func() {
 			if ctx.Err() != nil {
-				st.proc.reap()
-				s.discard(st.proc)
+				// resetDeadChild, not a hand-rolled subset: kill() is
+				// idempotent on the already-killed child, and one shared
+				// sequence means a future reset change cannot miss the
+				// cancellation path.
+				s.resetDeadChild(st.proc)
 			}
 		}()
 

tui/internal/ui/chat/clipboard.go

@@ -4,6 +4,7 @@
 package chat
 
 import (
+	"errors"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -171,10 +172,15 @@ func pasteFromClipboardOrImage() tea.Cmd {
 		}
 		png, ok, err := readClipboardImagePNG()
 		if ok {
-			// The image is the payload here. A decode failure surfaces —
-			// silently pasting the caption URL instead would hide that the
-			// image copy failed (the platform readers' documented contract).
 			if err != nil {
+				// A transient failure to OPEN the clipboard is not a broken
+				// image — deliver the caption text already in hand rather
+				// than failing a paste that has a usable payload.
+				if errors.Is(err, errClipboardBusy) && hasText {
+					return pasteClipboardMsg{text: text, err: nil}
+				}
+				// A real decode failure surfaces — silently pasting the
+				// caption URL would hide that the image copy failed.
 				return pasteImageMsg{err: err}
 			}
 			path, werr := writeClipboardImageToTemp(png)
@@ -184,6 +190,11 @@ func pasteFromClipboardOrImage() tea.Cmd {
 			// No image at all — the caption-ish text is all there is.
 			return pasteClipboardMsg{text: text, err: nil}
 		}
+		if err != nil {
+			// Nothing to paste AND the platform reader knows why (e.g. no
+			// wl-paste/xclip installed) — say so instead of a silent no-op.
+			return pasteImageMsg{err: err}
+		}
 		return pasteClipboardMsg{text: text, err: terr}
 	}
 }
@@ -221,6 +232,11 @@ func writeClipboardImageToTemp(png []byte) (string, error) {
 	return f.Name(), nil
 }
 
+// errClipboardBusy marks a clipboard that could not be OPENED (another app
+// holds it) — a transient condition distinct from an image that cannot be
+// decoded. Platform readers wrap their open failures with it.
+var errClipboardBusy = errors.New("clipboard is held by another application")
+
 // looksLikeImageCaption reports whether clipboard text reads as the metadata
 // a "Copy image" action leaves beside the bitmap — a single URL (or data:
 // URI) — rather than content someone copied for its own sake.

tui/internal/ui/chat/clipboardimage_linux.go

@@ -5,6 +5,7 @@ package chat
 
 import (
 	"bytes"
+	"fmt"
 	"os/exec"
 )
 
@@ -27,7 +28,12 @@ var linuxImageClipboardTools = []struct {
 // same reasoning atotto/clipboard's own Linux build already relies on for
 // text, via xclip/xsel).
 func readClipboardImagePNG() (data []byte, ok bool, err error) {
+	anyTool := false
 	for _, tool := range linuxImageClipboardTools {
+		if _, lookErr := exec.LookPath(tool.name); lookErr != nil {
+			continue
+		}
+		anyTool = true
 		out, runErr := exec.Command(tool.name, tool.args...).Output()
 		if runErr != nil {
 			// Not installed, or nothing image/png-shaped on the clipboard
@@ -41,5 +47,13 @@ func readClipboardImagePNG() (data []byte, ok bool, err error) {
 		}
 		return out, true, nil
 	}
+	if !anyTool {
+		// ok=false (no image was detected) but with the WHY: the caller
+		// surfaces this only when there is nothing else to paste, so a
+		// user without wl-paste/xclip learns what to install instead of
+		// getting a silent no-op.
+		return nil, false, fmt.Errorf(
+			"no clipboard image tool found — install wl-paste (Wayland) or xclip (X11) to paste screenshots")
+	}
 	return nil, false, nil
 }