Add per-job run logs and separate vision/text model selection.

Capture pipeline stdout into job.log + API/UI so failed runs can be reviewed, and let users pick OpenRouter vision vs text models independently.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/job_log.py

@@ -0,0 +1,93 @@
+"""
+job_log.py - Capture pipeline stdout/stderr into a per-job log.
+
+The pipeline already prints stage progress via print(). For a reviewable
+post-run log we tee those lines into memory + outputs/<job_id>/job.log
+without rewriting every call site.
+"""
+
+import sys
+import time
+from contextlib import contextmanager
+from typing import Callable, Iterator, List, Optional, TextIO
+
+
+class _LineSplitter:
+    """Accumulate write() chunks and emit complete lines."""
+
+    def __init__(self, on_line: Callable[[str], None]):
+        self._buf = ""
+        self._on_line = on_line
+
+    def write(self, s: str) -> None:
+        if not s:
+            return
+        self._buf += s
+        while "\n" in self._buf:
+            line, self._buf = self._buf.split("\n", 1)
+            # Strip trailing CR from Windows-ish streams; keep content intact.
+            self._on_line(line.rstrip("\r"))
+
+    def flush_remainder(self) -> None:
+        if self._buf:
+            self._on_line(self._buf.rstrip("\r"))
+            self._buf = ""
+
+
+class _Tee:
+    """Mirror writes to the original stream and a line callback."""
+
+    def __init__(self, stream: TextIO, on_line: Callable[[str], None]):
+        self._stream = stream
+        self._lines = _LineSplitter(on_line)
+
+    def write(self, s: str) -> int:
+        n = self._stream.write(s)
+        self._stream.flush()
+        self._lines.write(s)
+        return n
+
+    def flush(self) -> None:
+        self._stream.flush()
+
+    def flush_remainder(self) -> None:
+        self._lines.flush_remainder()
+
+    def __getattr__(self, name: str):
+        return getattr(self._stream, name)
+
+
+def stamp_line(line: str, t: Optional[float] = None) -> str:
+    """Prefix a log line with HH:MM:SS."""
+    ts = time.strftime("%H:%M:%S", time.localtime(t if t is not None else time.time()))
+    return f"[{ts}] {line}"
+
+
+@contextmanager
+def capture_stdio(on_line: Callable[[str], None]) -> Iterator[None]:
+    """
+    Tee sys.stdout and sys.stderr into on_line(raw_line) for the duration.
+
+    Safe for the single-job-at-a-time usage of this app; overlapping jobs
+    would interleave (same limitation as the LLM cost counters).
+    """
+    old_out, old_err = sys.stdout, sys.stderr
+    tee_out = _Tee(old_out, on_line)
+    tee_err = _Tee(old_err, on_line)
+    sys.stdout = tee_out  # type: ignore[assignment]
+    sys.stderr = tee_err  # type: ignore[assignment]
+    try:
+        yield
+    finally:
+        tee_out.flush_remainder()
+        tee_err.flush_remainder()
+        sys.stdout = old_out
+        sys.stderr = old_err
+
+
+def read_log_file(path: str) -> List[str]:
+    try:
+        with open(path, encoding="utf-8") as f:
+            return [ln.rstrip("\n") for ln in f]
+    except OSError:
+        return []

backend/jobs.py

@@ -9,6 +9,9 @@ for the completion email.
 State is in-memory (fine for a single-user tool); the report is also persisted
 to outputs/<job_id>/ so results survive a restart even though live status does
 not. No external queue/DB.
+
+A teed stdout/stderr log is kept in memory and written to outputs/<job_id>/job.log
+so failed or suspicious runs can be reviewed after the fact.
 """
 
 import os
@@ -16,14 +19,16 @@ import time
 import uuid
 import shutil
 import threading
-from typing import Dict, Optional
+from typing import Dict, List, Optional
 
 from backend import config
+from backend.job_log import capture_stdio, read_log_file, stamp_line
 from backend.pipeline.runner import run_pipeline
 from backend.email_sender import send_conflict_report
 
 _jobs: Dict[str, Dict] = {}
 _lock = threading.Lock()
+_LOG_TAIL = 80
 
 
 def _set(job_id: str, **fields) -> None:
@@ -31,8 +36,32 @@ def _set(job_id: str, **fields) -> None:
         _jobs[job_id].update(fields)
 
 
-def create_job(pdf_path: str, source_filename: str, email: Optional[str] = None,
-               project_input: Optional[Dict] = None, text_local: bool = False) -> str:
+def _append_log(job_id: str, raw_line: str, log_path: str) -> None:
+    """Stamp, store, and append one captured stdout/stderr line."""
+    entry = stamp_line(raw_line)
+    with _lock:
+        job = _jobs.get(job_id)
+        if job is not None:
+            job.setdefault("log", []).append(entry)
+    try:
+        os.makedirs(os.path.dirname(log_path), exist_ok=True)
+        with open(log_path, "a", encoding="utf-8") as f:
+            f.write(entry + "\n")
+    except OSError:
+        # Don't fail the job over log I/O; avoid print() here — it would
+        # re-enter the stdio tee while a job is capturing.
+        pass
+
+
+def create_job(
+    pdf_path: str,
+    source_filename: str,
+    email: Optional[str] = None,
+    project_input: Optional[Dict] = None,
+    text_local: bool = False,
+    vision_model: Optional[str] = None,
+    text_model: Optional[str] = None,
+) -> str:
     """Register a job and kick off its background thread. Returns the job_id."""
     job_id = uuid.uuid4().hex[:12]
     with _lock:
@@ -43,36 +72,64 @@ def create_job(pdf_path: str, source_filename: str, email: Optional[str] = None,
             "email": email or None,
             "project_input": project_input or {},
             "text_local": text_local,
+            "vision_model": (vision_model or "").strip() or None,
+            "text_model": (text_model or "").strip() or None,
             "stage": None,
             "created_at": time.time(),
             "finished_at": None,
             "report": None,
             "error": None,
+            "log": [],
         }
-    threading.Thread(target=_run, args=(job_id, pdf_path, project_input, text_local),
-                     daemon=True).start()
+    threading.Thread(
+        target=_run,
+        args=(job_id, pdf_path, project_input, text_local, vision_model, text_model),
+        daemon=True,
+    ).start()
     return job_id
 
 
-def _run(job_id: str, pdf_path: str, project_input: Optional[Dict] = None,
-         text_local: bool = False) -> None:
+def _run(
+    job_id: str,
+    pdf_path: str,
+    project_input: Optional[Dict] = None,
+    text_local: bool = False,
+    vision_model: Optional[str] = None,
+    text_model: Optional[str] = None,
+) -> None:
     out_dir = os.path.join(config.OUTPUT_DIR, job_id)
+    log_path = os.path.join(out_dir, "job.log")
     try:
         _set(job_id, status="running")
         # Keep a copy of the source PDF so its sheets can be viewed later.
         os.makedirs(out_dir, exist_ok=True)
+        # Truncate any leftover log if job_id somehow collided (shouldn't).
+        with open(log_path, "w", encoding="utf-8"):
+            pass
         shutil.copy2(pdf_path, os.path.join(out_dir, "source.pdf"))
-        report = run_pipeline(
-            pdf_path,
-            out_dir=out_dir,
-            on_stage=lambda name: _set(job_id, stage=name),
-            project_input=project_input,
-            source_name=_jobs[job_id].get("source"),
-            text_local=text_local,
-        )
+
+        def on_line(raw: str) -> None:
+            _append_log(job_id, raw, log_path)
+
+        with capture_stdio(on_line):
+            report = run_pipeline(
+                pdf_path,
+                out_dir=out_dir,
+                on_stage=lambda name: _set(job_id, stage=name),
+                project_input=project_input,
+                source_name=_jobs[job_id].get("source"),
+                text_local=text_local,
+                vision_model=vision_model,
+                text_model=text_model,
+            )
         _set(job_id, status="done", report=report, finished_at=time.time(), stage=None)
         _notify(job_id, report, out_dir)
     except Exception as e:
+        # Also land in the job log via print under the tee when possible.
+        try:
+            _append_log(job_id, f"[Jobs] Job {job_id} failed: {e}", log_path)
+        except Exception:
+            pass
         print(f"[Jobs] Job {job_id} failed: {e}")
         _set(job_id, status="error", error=str(e), finished_at=time.time())
         _notify_error(job_id)
@@ -103,10 +160,6 @@ def _notify_error(job_id: str) -> None:
     if not email:
         return
     # Reuse the report mailer with a minimal error-shaped payload.
-    err_report = {
-        "source": job.get("source", ""),
-        "summary": {"conflicts_found": 0, "by_severity": {}, "disciplines": []},
-    }
     try:
         from backend.email_sender import _smtp_ready, _send
         from email.message import EmailMessage
@@ -121,6 +174,7 @@ def _notify_error(job_id: str) -> None:
             "Your conflict check did not complete.\n\n"
             f"Drawing set: {job.get('source','')}\n"
             f"Error: {job.get('error','unknown')}\n\n"
+            f"Review the run log at: {config.APP_BASE_URL.rstrip('/')}/?job={job_id}\n\n"
             "Generated by Conflict Checker"
         )
         _send(msg)
@@ -128,6 +182,25 @@ def _notify_error(job_id: str) -> None:
         print(f"[Email] Failed to send error notice: {e}")
 
 
+def _log_from_disk(job_id: str) -> List[str]:
+    return read_log_file(os.path.join(config.OUTPUT_DIR, job_id, "job.log"))
+
+
+def get_job_log(job_id: str) -> Optional[List[str]]:
+    """Full job log lines, from memory or disk. None if job unknown."""
+    with _lock:
+        job = _jobs.get(job_id)
+        if job is not None:
+            return list(job.get("log") or [])
+
+    log = _log_from_disk(job_id)
+    # Job exists on disk if we have a log or a report artifact.
+    report_path = os.path.join(config.OUTPUT_DIR, job_id, "conflicts.json")
+    if log or os.path.isfile(report_path):
+        return log
+    return None
+
+
 def get_job(job_id: str) -> Optional[Dict]:
     """Public job view. Includes the full report only when done.
 
@@ -137,29 +210,46 @@ def get_job(job_id: str) -> Optional[Dict]:
     with _lock:
         job = _jobs.get(job_id)
         if job:
-            return dict(job)
+            out = dict(job)
+            log = list(job.get("log") or [])
+            out["log_tail"] = log[-_LOG_TAIL:]
+            # Full log on terminal states so the UI can show it without a
+            # second fetch; keep polls light while running.
+            if out.get("status") in ("done", "error"):
+                out["log"] = log
+            else:
+                out.pop("log", None)
+            return out
 
     # Try loading from disk
     report_path = os.path.join(config.OUTPUT_DIR, job_id, "conflicts.json")
-    if not os.path.isfile(report_path):
+    log = _log_from_disk(job_id)
+    if not os.path.isfile(report_path) and not log:
         return None
     try:
         import json
-        with open(report_path, encoding="utf-8") as f:
-            report = json.load(f)
+        report = None
+        if os.path.isfile(report_path):
+            with open(report_path, encoding="utf-8") as f:
+                report = json.load(f)
         source_pdf = os.path.join(config.OUTPUT_DIR, job_id, "source.pdf")
+        status = "done" if report is not None else "error"
         return {
-            "job_id":       job_id,
-            "status":       "done",
-            "source":       report.get("source", os.path.basename(report_path)),
-            "email":        None,
-            "project_input": report.get("project_input", {}),
-            "text_local":   report.get("summary", {}).get("text_backend") == "local",
-            "stage":        None,
-            "created_at":   os.path.getmtime(source_pdf) if os.path.isfile(source_pdf) else None,
-            "finished_at":  os.path.getmtime(report_path),
-            "report":       report,
-            "error":        None,
+            "job_id":        job_id,
+            "status":        status,
+            "source":        (report or {}).get("source", os.path.basename(report_path)),
+            "email":         None,
+            "project_input": (report or {}).get("project_input", {}),
+            "text_local":    (report or {}).get("summary", {}).get("text_backend") == "local",
+            "vision_model":  None,
+            "text_model":    None,
+            "stage":         None,
+            "created_at":    os.path.getmtime(source_pdf) if os.path.isfile(source_pdf) else None,
+            "finished_at":   os.path.getmtime(report_path) if os.path.isfile(report_path) else None,
+            "report":        report,
+            "error":         None if report is not None else "Report missing; see job log",
+            "log":           log,
+            "log_tail":      log[-_LOG_TAIL:],
         }
     except Exception as e:
         print(f"[Jobs] Failed to load job {job_id} from disk: {e}")

backend/llm.py

@@ -23,6 +23,9 @@ _clients: Dict[str, OpenAI] = {}
 # call_json when routing a no-image (text) call. Module-global mirrors the
 # set_stage/cost pattern (single-user tool).
 _text_local = False
+# Optional per-run model overrides from the UI (empty = use config defaults).
+_vision_model_override: Optional[str] = None
+_text_model_override: Optional[str] = None
 
 
 def set_text_backend(local: bool) -> None:
@@ -30,6 +33,13 @@ def set_text_backend(local: bool) -> None:
     global _text_local
     _text_local = bool(local)
 
+
+def set_model_overrides(vision: Optional[str] = None, text: Optional[str] = None) -> None:
+    """Per-run OpenRouter model picks. None/blank clears back to config defaults."""
+    global _vision_model_override, _text_model_override
+    _vision_model_override = (vision or "").strip() or None
+    _text_model_override = (text or "").strip() or None
+
 # --- per-job cost accounting -------------------------------------------------
 # OpenRouter returns the real USD cost of each call when we request usage
 # accounting. We accumulate it in a module-level counter; the runner resets it
@@ -167,12 +177,16 @@ def _resolve_backend(has_images: bool, model_override: Optional[str]) -> Dict[st
         return {
             "base_url": config.LOCAL_BASE_URL,
             "api_key": config.LOCAL_API_KEY,
-            "model": model_override or config.LOCAL_TEXT_MODEL or config.TEXT_MODEL,
+            "model": (model_override or _text_model_override
+                      or config.LOCAL_TEXT_MODEL or config.TEXT_MODEL),
             "usage": False,          # local has no OpenRouter usage accounting
             "local": True,
         }
     # Vision, or text-on-OpenRouter (default / fallback).
-    default_model = config.MODEL if has_images else config.TEXT_MODEL
+    if has_images:
+        default_model = _vision_model_override or config.MODEL
+    else:
+        default_model = _text_model_override or config.TEXT_MODEL
     return {
         "base_url": config.AI_BASE_URL,
         "api_key": config.AI_API_KEY,
@@ -340,7 +354,8 @@ def call_json(
                     _models["text_local"].add(be["model"])
                     _models["fallback_count"] += 1
                 be = {"base_url": config.AI_BASE_URL, "api_key": config.AI_API_KEY,
-                      "model": config.TEXT_MODEL, "usage": True, "local": False}
+                      "model": _text_model_override or config.TEXT_MODEL,
+                      "usage": True, "local": False}
                 cache_key = None       # don't cache fallback under the local-model key
                 fell_back = True
                 continue

backend/main.py

@@ -14,11 +14,12 @@ import tempfile
 from typing import Optional
 
 from fastapi import FastAPI, UploadFile, File, Form, HTTPException
-from fastapi.responses import HTMLResponse, JSONResponse, Response
+from fastapi.responses import HTMLResponse, JSONResponse, Response, PlainTextResponse
 from fastapi.staticfiles import StaticFiles
 
 from backend import config
-from backend.jobs import create_job, get_job
+from backend.jobs import create_job, get_job, get_job_log
+from backend.models_catalog import list_models
 from backend.pipeline.pdf_processor import render_page_jpeg
 
 app = FastAPI(title=config.APP_TITLE, version=config.APP_VERSION)
@@ -29,10 +30,17 @@ _FRONTEND_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "
 @app.get("/health")
 def health():
     return {"status": "ok", "model": config.MODEL,
+            "text_model": config.TEXT_MODEL,
             "key_configured": bool(config.AI_API_KEY),
             "email_configured": bool(config.SMTP_HOST and config.SMTP_USER and config.SMTP_PASSWORD)}
 
 
+@app.get("/models")
+def models():
+    """Vision vs text OpenRouter model lists for the UI dropdowns."""
+    return JSONResponse(list_models())
+
+
 @app.post("/check")
 async def check(
     file: UploadFile = File(...),
@@ -42,6 +50,8 @@ async def check(
     occupancy: Optional[str] = Form(None),
     work_type: Optional[str] = Form(None),
     text_local: bool = Form(False),
+    vision_model: Optional[str] = Form(None),
+    text_model: Optional[str] = Form(None),
 ):
     """
     Accept a PDF, start a background conflict check, and return a job_id
@@ -50,6 +60,9 @@ async def check(
     Optional intake fields (project_name/address/occupancy/work_type) feed the
     Stage 0 jurisdiction profile; anything left blank is derived from the cover
     sheet.
+
+    vision_model / text_model override the configured defaults for this run
+    (vision always OpenRouter; text follows the OpenRouter vs hybrid choice).
     """
     if not file.filename.lower().endswith(".pdf"):
         raise HTTPException(status_code=400, detail="Please upload a PDF.")
@@ -69,8 +82,17 @@ async def check(
         }.items()
         if v and v.strip()
     }
-    job_id = create_job(tmp_path, source_filename=file.filename, email=email,
-                        project_input=project_input, text_local=text_local)
+    v_model = (vision_model or "").strip() or None
+    t_model = (text_model or "").strip() or None
+    job_id = create_job(
+        tmp_path,
+        source_filename=file.filename,
+        email=email,
+        project_input=project_input,
+        text_local=text_local,
+        vision_model=v_model,
+        text_model=t_model,
+    )
     return JSONResponse({"job_id": job_id, "status": "queued", "email": email})
 
 
@@ -82,6 +104,17 @@ def job_status(job_id: str):
     return JSONResponse(job)
 
 
+@app.get("/jobs/{job_id}/log")
+def job_log(job_id: str, plain: bool = False):
+    """Full captured run log (also on disk as outputs/<job_id>/job.log)."""
+    lines = get_job_log(job_id)
+    if lines is None:
+        raise HTTPException(status_code=404, detail="Job not found")
+    if plain:
+        return PlainTextResponse("\n".join(lines) + ("\n" if lines else ""))
+    return JSONResponse({"job_id": job_id, "lines": lines, "text": "\n".join(lines)})
+
+
 @app.get("/jobs/{job_id}/sheet-image/{page}")
 def sheet_image(job_id: str, page: int):
     """Render one page of a completed job's source PDF as JPEG (sheet viewer)."""

backend/models_catalog.py

@@ -0,0 +1,101 @@
+"""
+models_catalog.py - OpenRouter model list for the UI dropdowns.
+
+Fetches https://openrouter.ai/api/v1/models (cached ~1h) and splits into:
+  - vision: accepts image input and returns text
+  - text:   chat models that return text (may also be multimodal)
+"""
+
+import time
+from typing import Any, Dict, List
+
+from backend import config
+
+_TTL_SEC = 3600
+_cache: Dict[str, Any] = {"at": 0.0, "payload": None}
+
+
+def _entry(m: Dict[str, Any]) -> Dict[str, str]:
+    return {
+        "id": m.get("id") or "",
+        "name": m.get("name") or m.get("id") or "",
+    }
+
+
+def _ensure_default(items: List[Dict[str, str]], model_id: str) -> List[Dict[str, str]]:
+    """Prepend the configured default if OpenRouter didn't return it."""
+    if not model_id:
+        return items
+    if any(x["id"] == model_id for x in items):
+        return items
+    return [{"id": model_id, "name": model_id}] + items
+
+
+def _fetch_raw() -> List[Dict[str, Any]]:
+    import httpx  # local import so the app can start without httpx in odd envs
+    headers = {"Accept": "application/json"}
+    if config.AI_API_KEY:
+        headers["Authorization"] = f"Bearer {config.AI_API_KEY}"
+    url = f"{config.AI_BASE_URL.rstrip('/')}/models"
+    # Ask for text-output chat models (includes multimodal). "all" is huge.
+    with httpx.Client(timeout=30.0) as client:
+        r = client.get(url, headers=headers, params={"output_modalities": "text"})
+        r.raise_for_status()
+        data = r.json()
+    return data.get("data") or []
+
+
+def list_models() -> Dict[str, Any]:
+    """Return {vision, text, defaults} for the frontend selects."""
+    now = time.time()
+    if _cache["payload"] and (now - _cache["at"]) < _TTL_SEC:
+        return _cache["payload"]
+
+    try:
+        raw = _fetch_raw()
+    except Exception as e:
+        # Degrade to configured defaults so the UI still works offline.
+        print(f"[Models] OpenRouter catalog fetch failed: {e}")
+        vision = _ensure_default([], config.MODEL)
+        text = _ensure_default([], config.TEXT_MODEL)
+        payload = {
+            "vision": vision,
+            "text": text,
+            "defaults": {"vision": config.MODEL, "text": config.TEXT_MODEL},
+            "error": str(e),
+        }
+        return payload
+
+    vision: List[Dict[str, str]] = []
+    text: List[Dict[str, str]] = []
+    for m in raw:
+        mid = m.get("id") or ""
+        if not mid:
+            continue
+        arch = m.get("architecture") or {}
+        inputs = arch.get("input_modalities") or []
+        outputs = arch.get("output_modalities") or []
+        # Legacy string form: "text+image->text"
+        modality = (arch.get("modality") or "").lower()
+        has_image_in = ("image" in inputs) or ("image" in modality.split("->")[0])
+        has_text_out = ("text" in outputs) or ("->text" in modality) or (not outputs and not modality)
+        has_text_in = ("text" in inputs) or ("text" in modality) or not inputs
+
+        if has_image_in and has_text_out:
+            vision.append(_entry(m))
+        if has_text_in and has_text_out:
+            text.append(_entry(m))
+
+    vision.sort(key=lambda x: x["name"].lower())
+    text.sort(key=lambda x: x["name"].lower())
+    vision = _ensure_default(vision, config.MODEL)
+    text = _ensure_default(text, config.TEXT_MODEL)
+
+    payload = {
+        "vision": vision,
+        "text": text,
+        "defaults": {"vision": config.MODEL, "text": config.TEXT_MODEL},
+    }
+    _cache["at"] = now
+    _cache["payload"] = payload
+    return payload

backend/pipeline/runner.py

@@ -42,7 +42,9 @@ from backend.pipeline.risk import score_and_prioritize
 from backend.pipeline.rfi import generate_rfis
 from backend.pipeline.report import build_report, to_markdown
 from backend.pipeline._stage import validate_issue
-from backend.llm import reset_cost, get_cost, set_stage, set_text_backend
+from backend.llm import (
+    reset_cost, get_cost, set_stage, set_text_backend, set_model_overrides,
+)
 
 
 def run_pipeline(
@@ -52,6 +54,8 @@ def run_pipeline(
     project_input: Optional[Dict] = None,
     source_name: Optional[str] = None,
     text_local: bool = False,
+    vision_model: Optional[str] = None,
+    text_model: Optional[str] = None,
 ) -> Dict:
     """
     Run the full QAQC pipeline on one PDF and return the report dict.
@@ -59,6 +63,9 @@ def run_pipeline(
     project_input: optional intake fields (project_name, address, occupancy,
     work_type). Cover-sheet-derived values fill any gaps; intake fields win.
 
+    vision_model / text_model: optional per-run OpenRouter (or local text)
+    model overrides from the UI. Blank/None keeps config defaults.
+
     If out_dir is given, writes conflicts.json, report.md, and the intermediate
     artifacts (assertions.json, clusters.json, and one json per QAQC stage).
     """
@@ -70,7 +77,29 @@ def run_pipeline(
 
     reset_cost()
     set_text_backend(text_local)
+    set_model_overrides(vision_model, text_model)
+    if vision_model or text_model:
+        print(f"[Runner] model overrides: vision={vision_model or '(default)'} "
+              f"text={text_model or '(default)'}")
 
+    try:
+        return _run_stages(
+            pdf_path, out_dir, stage, project_input, source_name, text_local,
+        )
+    finally:
+        # Don't leak per-run picks into a later overlapping/CLI call.
+        set_model_overrides(None, None)
+        set_text_backend(False)
+
+
+def _run_stages(
+    pdf_path: str,
+    out_dir: Optional[str],
+    stage: Callable[[str], None],
+    project_input: Optional[Dict],
+    source_name: Optional[str],
+    text_local: bool,
+) -> Dict:
     stage("PDF -> images")
     pages = convert_pdf_to_images(pdf_path)