cost controls: concise mode, read-only-survivors, flash-lite extraction

server/app.py

@@ -34,6 +34,9 @@ DB_PATH = os.getenv("JOBS_DB", "jobs.db")
 ALLOWED_MODELS = [m.strip() for m in os.getenv(
     "ALLOWED_MODELS", "gemini-2.5-flash,gemini-2.5-pro"
 ).split(",") if m.strip()]
+DEFAULT_PAPERS = int(os.getenv("DEFAULT_PAPERS", "5"))
+# Cheaper model for the ~20 mechanical extraction calls (blank = use run model).
+EXTRACT_MODEL = os.getenv("EXTRACT_MODEL", "").strip()
 
 HERE = os.path.dirname(__file__)
 
@@ -46,6 +49,7 @@ class Job:
     depth: int = 1
     model: str = ""
     year_min: int = 0
+    style: str = "concise"
     status: str = "queued"  # queued | running | done | error
     log: list[str] = field(default_factory=list)
     review: str = ""
@@ -94,7 +98,7 @@ class JobStore:
         return Job(**json.loads(row[0])) if row else None
 
     def find_cached(
-        self, topic: str, max_papers: int, depth: int, model: str, year_min: int, ttl: int
+        self, topic, max_papers, depth, model, year_min, style, ttl
     ) -> "Job | None":
         """Most recent completed job with identical params, within ``ttl`` seconds."""
         cutoff = time.time() - ttl
@@ -112,6 +116,7 @@ class JobStore:
                 and d.get("depth") == depth
                 and (d.get("model") or "") == (model or "")
                 and (d.get("year_min") or 0) == year_min
+                and (d.get("style") or "concise") == style
             ):
                 return Job(**d)
         return None
@@ -130,12 +135,10 @@ class JobManager:
         self._pool = ThreadPoolExecutor(max_workers=MAX_CONCURRENT_JOBS)
         self._store = JobStore(DB_PATH)
 
-    def submit(
-        self, topic: str, max_papers: int, depth: int, model: str, year_min: int
-    ) -> Job:
+    def submit(self, topic, max_papers, depth, model, year_min, style) -> Job:
         # Return a recent identical completed review instead of re-spending.
         cached = self._store.find_cached(
-            topic, max_papers, depth, model, year_min, CACHE_TTL_SECONDS
+            topic, max_papers, depth, model, year_min, style, CACHE_TTL_SECONDS
         )
         if cached:
             cached.cached = True
@@ -148,6 +151,7 @@ class JobManager:
             depth=depth,
             model=model,
             year_min=year_min,
+            style=style,
         )
         with self._lock:
             self._jobs[job.id] = job
@@ -181,6 +185,8 @@ class JobManager:
                 meter=meter,
                 model=job.model or None,
                 year_min=job.year_min,
+                style=job.style,
+                extract_model=EXTRACT_MODEL or None,
             )
             result = agent.invoke(
                 {
@@ -267,11 +273,15 @@ async def create_review(request: Request) -> JSONResponse:
         return JSONResponse({"error": "Topic is too long."}, status_code=400)
 
     try:
-        max_papers = int(body.get("max_papers", 8))
+        max_papers = int(body.get("max_papers", DEFAULT_PAPERS))
     except (TypeError, ValueError):
-        max_papers = 8
+        max_papers = DEFAULT_PAPERS
     max_papers = max(3, min(max_papers, MAX_PAPERS_CAP))
 
+    style = (body.get("style") or "concise").strip().lower()
+    if style not in ("concise", "comprehensive"):
+        style = "concise"
+
     try:
         depth = int(body.get("depth", 1))
     except (TypeError, ValueError):
@@ -291,7 +301,7 @@ async def create_review(request: Request) -> JSONResponse:
 
     # Rate limit only counts when we actually start a new run (cache hits are free).
     cached = jobs._store.find_cached(
-        topic, max_papers, depth, model, year_min, CACHE_TTL_SECONDS
+        topic, max_papers, depth, model, year_min, style, CACHE_TTL_SECONDS
     )
     if not cached:
         allowed, retry_in = limiter.allow(_client_ip(request))
@@ -301,9 +311,10 @@ async def create_review(request: Request) -> JSONResponse:
                 status_code=429,
             )
 
-    job = jobs.submit(topic, max_papers, depth, model, year_min)
+    job = jobs.submit(topic, max_papers, depth, model, year_min, style)
     return JSONResponse(
-        {"job_id": job.id, "max_papers": max_papers, "depth": depth, "cached": job.cached}
+        {"job_id": job.id, "max_papers": max_papers, "depth": depth,
+         "style": style, "cached": job.cached}
     )
 
 
@@ -323,6 +334,7 @@ def get_review(job_id: str) -> JSONResponse:
             "papers": job.papers,
             "warnings": job.warnings,
             "model": job.model,
+            "style": job.style,
             "cached": job.cached,
             "error": job.error,
             "stats": job.stats,
@@ -332,7 +344,13 @@ def get_review(job_id: str) -> JSONResponse:
 
 @app.get("/api/config")
 def get_config() -> dict:
-    return {"models": ALLOWED_MODELS, "max_papers": MAX_PAPERS_CAP, "max_depth": MAX_DEPTH}
+    return {
+        "models": ALLOWED_MODELS,
+        "max_papers": MAX_PAPERS_CAP,
+        "default_papers": DEFAULT_PAPERS,
+        "max_depth": MAX_DEPTH,
+        "styles": ["concise", "comprehensive"],
+    }
 
 
 @app.get("/r/{job_id}", response_class=HTMLResponse)

server/static/index.html

@@ -216,8 +216,13 @@
     <details class="advanced">
       <summary>Options</summary>
       <div class="opts">
-        <div class="opt"><label for="n">Papers — <span id="nval">8</span></label>
-          <input id="n" type="range" min="3" max="12" value="8" /></div>
+        <div class="opt"><label for="style">Length</label>
+          <select id="style">
+            <option value="concise">Concise · cheaper</option>
+            <option value="comprehensive">Comprehensive</option>
+          </select></div>
+        <div class="opt"><label for="n">Papers — <span id="nval">5</span></label>
+          <input id="n" type="range" min="3" max="12" value="5" /></div>
         <div class="opt"><label for="depth">Depth</label>
           <select id="depth">
             <option value="1">1 — single pass</option>
@@ -322,7 +327,7 @@ form.addEventListener('submit', async (e) => {
     res = await fetch('/api/review', { method:'POST', headers:{'Content-Type':'application/json'},
       body: JSON.stringify({ topic, max_papers: parseInt(nInput.value,10),
         depth: parseInt($('depth').value,10), model: $('model').value,
-        year_min: parseInt($('since').value,10) }) });
+        year_min: parseInt($('since').value,10), style: $('style').value }) });
   } catch { return fail('Network error — is the server up?'); }
   const data = await res.json();
   if (!res.ok) return fail(data.error || 'Request failed.');

src/agent.py

@@ -95,15 +95,23 @@ def build_graph(
     meter: dict | None = None,
     model: str | None = None,
     year_min: int = 0,
+    style: str = "concise",
+    extract_model: str | None = None,
 ):
-    """Build and compile the research agent graph."""
+    """Build and compile the research agent graph.
+
+    ``style`` is "concise" (default — shorter review, far cheaper since output
+    tokens dominate cost) or "comprehensive" (longer, more detailed).
+    ``extract_model`` overrides the model for the ~20 mechanical extraction
+    calls (e.g. a cheaper flash-lite), keeping the run model for synthesis.
+    """
     from langgraph.graph import END, StateGraph
 
     say = progress or (lambda _msg: None)
     searcher = SearchTool(max_results=RESULTS_PER_QUERY)
     scholar = SemanticScholarTool(max_results=RESULTS_PER_QUERY, year_min=year_min)
     downloader = DownloadTool()
-    extractor = ExtractionTool(model=model, meter=meter)
+    extractor = ExtractionTool(model=extract_model or model, meter=meter)
     pool_cap = max_papers + POOL_BUFFER
 
     def _run_searches(queries: list[str], existing: list[Paper]) -> list[Paper]:
@@ -153,8 +161,10 @@ def build_graph(
                 "No papers found from arXiv or Semantic Scholar — both sources may "
                 "be rate-limiting right now. Please try again in a minute."
             )
-        todo = [p for p in papers if not p.full_text]
-        say(f"Reading {len(todo)} papers (parallel)...")
+        # Only read the top max_papers (already ranked by score_node); the rest
+        # were filtered/ranked on abstracts alone, so we never pay to read them.
+        todo = [p for p in papers[:max_papers] if not p.full_text]
+        say(f"Reading top {len(todo)} papers (parallel)...")
 
         def _read_one(paper: Paper) -> None:
             paper.full_text = downloader.get_text(paper)
@@ -190,12 +200,21 @@ def build_graph(
 
     def synthesize_node(state: AgentState) -> dict:
         top = state["papers"][:max_papers]
-        say(f"Synthesizing review from top {len(top)} papers...")
+        say(f"Synthesizing {style} review from top {len(top)} papers...")
         papers_block = utils.format_papers_for_synthesis(top)
+        if style == "comprehensive":
+            length_note = "Write a thorough, detailed review."
+            max_out = 16384
+        else:  # concise (default) — output tokens dominate cost, so keep it tight
+            length_note = (
+                "Write a CONCISE, focused review of about 600-900 words total — "
+                "short paragraphs, high signal, no filler or repetition."
+            )
+            max_out = 4096
         prompt = (
-            f"Write a structured literature review on: {state['topic']}\n\n"
+            f"{length_note}\n\nLiterature review on: {state['topic']}\n\n"
             f"Papers:\n{papers_block}\n\n"
-            "Structure the review with these Markdown sections:\n"
+            "Use these Markdown sections:\n"
             "1. ## Introduction\n"
             "2. ## Key Themes (group by methodology)\n"
             "3. ## Key Findings\n"
@@ -205,8 +224,7 @@ def build_graph(
             "its source as [Author, Year] using the first author's surname. Do not "
             "invent papers, authors, or findings not present above."
         )
-        # Generous ceiling so long, multi-paper reviews don't truncate mid-sentence.
-        review = utils.complete(prompt, max_tokens=16384, meter=meter, model=model)
+        review = utils.complete(prompt, max_tokens=max_out, meter=meter, model=model)
         review = review.rstrip() + "\n" + utils.references_markdown(top)
 
         gaps: list[str] = []
@@ -246,9 +264,9 @@ def build_graph(
     graph.add_node("synthesize", synthesize_node)
     graph.add_node("deepen", deepen_node)
     graph.set_entry_point("search")
-    graph.add_edge("search", "read")
-    graph.add_edge("read", "score")
-    graph.add_edge("score", "synthesize")
+    graph.add_edge("search", "score")   # rank on abstracts first...
+    graph.add_edge("score", "read")     # ...then read only the top max_papers
+    graph.add_edge("read", "synthesize")
     graph.add_conditional_edges("synthesize", should_deepen, {"deepen": "deepen", END: END})
-    graph.add_edge("deepen", "read")
+    graph.add_edge("deepen", "score")
     return graph.compile()

src/cli.py

@@ -23,6 +23,9 @@ def review(
     depth: int = typer.Option(1, "--depth", help="Iterative gap-filling passes."),
     model: str = typer.Option("", "--model", help="Override the model for this run."),
     since: int = typer.Option(0, "--since", help="Only papers from this year onward."),
+    comprehensive: bool = typer.Option(
+        False, "--comprehensive", help="Longer, more detailed review (costs more)."
+    ),
 ) -> None:
     """Search arXiv, read papers, and write a structured literature review."""
     from .agent import build_graph
@@ -41,6 +44,7 @@ def review(
             meter=meter,
             model=model or None,
             year_min=since,
+            style="comprehensive" if comprehensive else "concise",
         )
         result = agent.invoke(
             {