claude-code/ indexing runs 16 calls at a time and skips the unread cache write

pageindex/client.py

@@ -387,9 +387,8 @@ class PageIndexClient:
             own use — a product serving other people needs an API key).
             There a reasoning effort must be a level Claude Code lists for
             the model (one listing none drops it), checked before any
-            prompt goes out, and ``"none"``
-            turns thinking off only where the model allows it, the
-            model's default applying otherwise.
+            prompt goes out, and ``"none"`` asks for low effort and turns
+            thinking off where the model allows it.
         chat_model (str, optional): Your own model for the chat surfaces
             (``chat``, ``chat_completions``; a value you set also
             carries onto ``chat(protocol="messages")`` and the two

pageindex/integrations/claude_agent_sdk.py

@@ -168,9 +168,9 @@ def _session_message(payload: dict):
 
 CLAUDE_CODE_PREFIX = "claude-code/"
 
-# One CLI process per call, ~150 MB each (measured): indexing's 64-wide
-# summary lane must not start 64 of them.
-CLAUDE_CODE_CONCURRENCY = 8
+# One CLI process per call, ~150 MB each (measured): indexing's summary
+# lane (SUMMARY_CONCURRENCY wide) must not start that many.
+CLAUDE_CODE_CONCURRENCY = 16
 _GATES: dict = {}  # event loop -> [its cap, calls inside]
 
 # AssistantMessage.error kinds, for a CLI that reports no api_error_status.
@@ -221,8 +221,10 @@ def claude_code_options(model: str, reasoning_effort=None, **options):
         options["thinking"] = {"type": "disabled"}
     elif reasoning_effort:
         options["effort"] = reasoning_effort
-    # not a git repo, so the CLI adds no git status to the prompt
-    cwd = os.path.join(tempfile.gettempdir(), "pageindex-claude-code")
+    # not a git repo, so the CLI adds no git status to the prompt; resolved,
+    # as the CLI lists a symlinked cwd (macOS's /var) twice in every prompt
+    cwd = os.path.realpath(
+        os.path.join(tempfile.gettempdir(), "pageindex-claude-code"))
     os.makedirs(cwd, exist_ok=True)
     return ClaudeAgentOptions(
         model=wire, tools=[], setting_sources=[], strict_mcp_config=True,
@@ -343,16 +345,26 @@ async def claude_code_complete(model: str, prompt: str,
     """One indexing prompt on the claude-code/ route: no tools, one turn.
     Every failure is final — the CLI already retried what it could — and
     only a 400 stays per-prompt, for the indexing callers to absorb."""
-    from ..utils import LLMRetriesExhausted, _add_usage_counts
+    from ..utils import (LLMRetriesExhausted, _add_usage_counts, _llm_priority,
+                         _PriorityGate)
     loop = asyncio.get_running_loop()
-    # one cap per event loop; submits on separate threads each get their own
-    entry = _GATES.setdefault(loop, [asyncio.Semaphore(CLAUDE_CODE_CONCURRENCY), 0])
+    # one cap per event loop; submits on separate threads each get their own.
+    # The wider indexing lanes queue their calls here, highest priority first
+    entry = _GATES.setdefault(loop, [_PriorityGate(CLAUDE_CODE_CONCURRENCY), 0])
     entry[1] += 1
     reply = ""
     try:
         options = claude_code_options(model, reasoning_effort, max_turns=1)
+        if reasoning_effort == "none":
+            # where Claude Code keeps thinking on anyway, it thinks least at low
+            options.effort = "low"
+        # one turn, never read back: a cache write would only cost more
+        options.env["DISABLE_PROMPT_CACHING"] = "1"
+        # a stream with no bytes, pings included, is dead: re-sent after 30 s,
+        # not the CLI's 3 minutes
+        options.env["CLAUDE_BYTE_STREAM_IDLE_TIMEOUT_MS"] = "30000"
         from claude_agent_sdk import ResultMessage
-        async with entry[0]:
+        async with entry[0].slot(_llm_priority.get()):
             async for message in claude_code_messages(prompt, options):
                 if isinstance(message, ResultMessage):
                     reply = message.result or ""

pageindex/tree_optimize.py

@@ -65,11 +65,12 @@
 
 from .utils import (SUMMARY_MAX_WORDS, TOKENS_PER_WORD, ConfigLoader, _is_unrecoverable,
                     count_tokens, intro_title, is_intro, llm_acompletion,
-                    strip_internal_keys)
+                    llm_priority, strip_internal_keys)
 
 TRIGGER_PAGES = 5        # only look ahead on nodes larger than this
 ROUTING_COST = 1         # R(v), in pages
 EXPAND_CONCURRENCY = 32  # measured plateau: the ready frontier is 21-28 wide on few-hundred-page PDFs
+EXPAND_PRIORITY = float("inf")  # ahead of every summary: none starts under a node expand may still change
 PAGE_CHARS = 6000        # per-page text handed to the model
 TITLE_MAX_CHARS = 200    # a union title longer than this falls back to a page label
 EMPTY_RETRIES = 0
@@ -789,7 +790,8 @@ async def proposals_for(node, own):
             attempts += 1
             try:
                 async with semaphore:
-                    proposed = await propose_children(node, pages, args)
+                    with llm_priority(EXPAND_PRIORITY):
+                        proposed = await propose_children(node, pages, args)
             except Exception as exc:
                 if _is_unrecoverable(exc):
                     raise  # every remaining node would fail identically

pageindex/utils.py

@@ -10,7 +10,7 @@
 import copy
 import asyncio
 import heapq
-from contextlib import asynccontextmanager
+from contextlib import asynccontextmanager, contextmanager
 from io import BytesIO
 from dotenv import find_dotenv, load_dotenv
 load_dotenv(find_dotenv(usecwd=True))
@@ -37,6 +37,20 @@
 _llm_usage: contextvars.ContextVar = contextvars.ContextVar(
     "pageindex_llm_usage", default=None)
 
+# Set by llm_priority: where a gate narrower than the indexing lanes queues
+# their model calls (the claude-code/ route's cap), the highest goes first.
+_llm_priority: contextvars.ContextVar = contextvars.ContextVar(
+    "pageindex_llm_priority", default=0)
+
+
+@contextmanager
+def llm_priority(prio):
+    token = _llm_priority.set(prio)
+    try:
+        yield
+    finally:
+        _llm_priority.reset(token)
+
 
 def _add_usage(response) -> None:
     usage, counted = _llm_usage.get(), getattr(response, "usage", None)
@@ -1140,7 +1154,8 @@ def _task(self, node, depth):
     async def _ask(self, prompt, prio):
         self._asked = True
         async with self._gate.slot(prio):
-            reply = await self._complete(prompt)
+            with llm_priority(prio):
+                reply = await self._complete(prompt)
         if parse_summary(reply):
             self._answered = True
         return reply

tests/test_claude_code.py

@@ -2,6 +2,7 @@
 user's subscription — over a faked CLI session: no subprocess, no login."""
 import asyncio
 import logging
+import os
 import sys
 import time
 from types import SimpleNamespace
@@ -205,13 +206,18 @@ def test_index_call_runs_isolated_on_the_subscription(monkeypatch):
     assert opts.model == "haiku" and opts.tools == [] and opts.setting_sources == []
     assert opts.strict_mcp_config and opts.max_turns == 1 and not opts.mcp_servers
     assert opts.thinking == {"type": "disabled"} and second.options.thinking is None
+    # Claude Code keeps thinking on for some models; it thinks least at low
+    assert opts.effort == "low" and second.options.effort is None
     # an @/path in document text must not make the CLI read that file
     assert opts.verbatim_prompts
     # a .env written for API models must not bill or reroute the run
     assert all(opts.env[var] == "" for var in (
         "ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN", "ANTHROPIC_BASE_URL"))
+    assert opts.env["DISABLE_PROMPT_CACHING"] == "1"
+    assert opts.env["CLAUDE_BYTE_STREAM_IDLE_TIMEOUT_MS"] == "30000"
     assert "no-session-persistence" in opts.extra_args
     assert opts.max_buffer_size > 2**20  # a page image's tool result is one line
+    assert opts.cwd == os.path.realpath(opts.cwd)  # a symlinked cwd is listed twice
     assert first.closed and second.closed
 
 
@@ -270,8 +276,8 @@ def test_an_effort_claude_code_does_not_list_is_refused_before_any_prompt(monkey
             asyncio.run(llm_acompletion(f"claude-code/{model}", "P", reasoning_effort=effort))
         assert _is_unrecoverable(err.value)  # fatal, not one prompt's miss
     assert all(s.prompt is None and s.closed for s in sessions)  # nothing billed
-    # no effort and "none" are never checked; an unlisted model takes any
-    # listed model's levels; a model without effort levels drops it
+    # no effort is never checked and "none" runs at low; an unlisted model
+    # takes any listed model's levels; a model without effort levels drops it
     for model, effort in (("haiku", None), ("haiku", "none"),
                           ("claude-haiku-5-5", "high"), ("claude-sonnet-4-5", "max"),
                           ("tiny", "high")):
@@ -315,6 +321,24 @@ async def many():
     assert cc._GATES == {}  # the cap goes with its last call, not kept per loop
 
 
+@needs_claude
+def test_queued_index_calls_go_highest_priority_first(monkeypatch):
+    from pageindex.utils import llm_priority
+    monkeypatch.setattr(cc, "CLAUDE_CODE_CONCURRENCY", 1)
+    sessions = fake_cli(monkeypatch, result("ok"), during=lambda: asyncio.sleep(0.01))
+
+    async def call(prompt, prio):
+        with llm_priority(prio):
+            return await llm_acompletion("claude-code/haiku", prompt)
+
+    async def run():
+        first = asyncio.create_task(call("first", 0))
+        await asyncio.sleep(0)  # it takes the one slot; the rest queue behind it
+        await asyncio.gather(first, call("low", 1), call("high", 3), call("mid", 2))
+    asyncio.run(run())
+    assert [s.prompt for s in sessions] == ["first", "high", "mid", "low"]
+
+
 @pytest.fixture
 def asyncio_log():
     """The asyncio logger without the probe filter an earlier test may have
@@ -426,6 +450,7 @@ def test_chat_sends_one_message_with_a_block_per_turn(monkeypatch, chat_client):
     assert list(opts.mcp_servers) == ["pageindex"]
     assert opts.allowed_tools == ["mcp__pageindex"] and opts.tools == []
     assert opts.max_turns == 10 and not opts.include_partial_messages
+    assert "DISABLE_PROMPT_CACHING" not in opts.env  # later turns read the cache
 
     envelope = chat_client.chat(history, doc_id="pi-doc1", protocol="chat_completions")
     assert envelope["model"] == "haiku"

tests/test_tree_format.py

@@ -79,6 +79,39 @@ async def run():
     assert all(n["summary"] == "ok" for n in utils._subtree(tree))
 
 
+def test_expand_and_summary_calls_carry_their_queue_priority(monkeypatch):
+    body = "body " * 250
+    pages = [body] * 12
+    pages[5] = "Sub One\n" + body
+    pages[8] = "Sub Two\n" + body
+    lines = [[line for line in page.splitlines() if line.strip()] for page in pages]
+    tree = [{"title": "R", "start_index": 1, "end_index": 12, "node_id": "0000", "nodes": [
+        {"title": "A", "start_index": 1, "end_index": 3, "node_id": "0001"},
+        {"title": "X", "start_index": 4, "end_index": 12, "node_id": "0002"}]}]
+    seen = []
+
+    async def propose(model, prompt):
+        seen.append(("expand", utils._llm_priority.get()))
+        return {"subsections": [{"title": "Sub One", "page": 6}, {"title": "Sub Two", "page": 9}]}
+
+    async def summarize(model, prompt):
+        seen.append(("summary", utils._llm_priority.get()))
+        return '{"summary": "ok"}'
+    monkeypatch.setattr(tree_optimize, "ask_model", propose)
+    monkeypatch.setattr(utils, "llm_acompletion", summarize)
+
+    async def run():
+        scheduler = utils.SummaryScheduler(tree, [(page, 0) for page in pages], model="m")
+        await tree_optimize.optimize(tree, pages, lines, model="m", do_expand=True,
+                                     on_final=scheduler.mark_final)
+        await scheduler.finish()
+    asyncio.run(run())
+
+    # where the claude-code/ route's cap queues them: expand first, then the deepest summaries
+    assert {prio for kind, prio in seen if kind == "expand"} == {tree_optimize.EXPAND_PRIORITY}
+    assert sorted(prio for kind, prio in seen if kind == "summary") == [1, 2, 2, 3, 3, 3]
+
+
 def test_a_running_header_or_a_word_prefix_does_not_open_the_page():
     page = ["4.1. Discriminant Functions 181", "opening words", "4.1. Discriminant Functions"]
     assert not tree_optimize.heading_at_page_start([page], 1, "4.1. Discriminant Functions")