deploy Nous v21 update

api/browser_investigator.py

@@ -7,10 +7,11 @@ CAPTCHAs, rotate proxies, or continue after an access challenge.
 from __future__ import annotations
 
 import asyncio
+import inspect
 import json
 import os
 import re
-from typing import Any
+from typing import Any, Callable
 from urllib.parse import urlparse
 
 
@@ -96,6 +97,7 @@ def investigate_websites(
     api_key: str,
     model: str,
     user_agent: str,
+    log_callback: Callable[[str, str, dict[str, Any]], Any] | None = None,
 ) -> list[dict[str, Any]]:
     """Run one bounded Browser Use investigation and return candidates."""
     if os.environ.get("BROWSER_USE_ENABLED", "1").strip().lower() in {"0", "false", "no", "off"}:
@@ -132,6 +134,52 @@ Return ONLY JSON in this shape:
 Confidence must be an integer from 0 to 100. Include only plausible official website URLs and never invent a URL.
 """.strip()
 
+    async def emit(kind: str, event: str, details: dict[str, Any]) -> None:
+        if not log_callback:
+            return
+        result = log_callback(kind, event, details)
+        if inspect.isawaitable(result):
+            await result
+
+    def output_actions(model_output: Any) -> list[Any]:
+        try:
+            actions = getattr(model_output, "action", []) or []
+            values: list[Any] = []
+            for action in list(actions)[:8]:
+                if hasattr(action, "model_dump"):
+                    values.append(action.model_dump(mode="json"))
+                else:
+                    values.append(str(action)[:2_000])
+            return values
+        except Exception:
+            return [str(model_output)[:2_000]] if model_output is not None else []
+
+    async def on_step(browser_state: Any, model_output: Any, step_number: int) -> None:
+        details = {
+            "step": step_number,
+            "url": str(getattr(browser_state, "url", "") or "")[:700],
+            "title": str(getattr(browser_state, "title", "") or "")[:300],
+            "actions": output_actions(model_output),
+            "model": model,
+        }
+        await emit("browser", "browser_use_step", details)
+        await emit("ai", "browser_use_model_step", {"step": step_number, "model": model, "actions": details["actions"]})
+
+    class LoggedChatOpenAI(ChatOpenAI):
+        """Capture Browser Use LLM turns without recording provider secrets."""
+
+        async def ainvoke(self, messages: list[Any], output_format: Any = None, **kwargs: Any) -> Any:
+            started = asyncio.get_running_loop().time()
+            serialised = [{"role": getattr(message, "type", "message"), "content": str(getattr(message, "content", ""))[:4_000]} for message in list(messages)[-20:]]
+            await emit("ai", "browser_use_llm_request", {"model": self.model, "messages": serialised, "structured_output": bool(output_format)})
+            try:
+                result = await super().ainvoke(messages, output_format=output_format, **kwargs)
+                await emit("ai", "browser_use_llm_response", {"model": self.model, "completion": str(getattr(result, "completion", result))[:4_000], "duration_ms": int((asyncio.get_running_loop().time() - started) * 1000)})
+                return result
+            except Exception as exc:
+                await emit("ai", "browser_use_llm_error", {"model": self.model, "duration_ms": int((asyncio.get_running_loop().time() - started) * 1000)},)
+                raise
+
     async def run() -> Any:
         browser_profile = BrowserProfile(
             headless=True,
@@ -141,7 +189,7 @@ Confidence must be an integer from 0 to 100. Include only plausible official web
             enable_default_extensions=False,
             captcha_solver=False,
         )
-        llm = ChatOpenAI(
+        llm = LoggedChatOpenAI(
             model=model,
             api_key=api_key,
             base_url=base_url,
@@ -161,10 +209,19 @@ Confidence must be an integer from 0 to 100. Include only plausible official web
             use_judge=False,
             enable_planning=False,
             enable_signal_handler=False,
+            register_new_step_callback=on_step,
         )
-        return await agent.run(max_steps=max(3, min(15, int(os.environ.get("BROWSER_USE_MAX_STEPS", "10")))))
+        try:
+            max_steps = int(os.environ.get("BROWSER_USE_MAX_STEPS", "10"))
+        except (TypeError, ValueError):
+            max_steps = 10
+        return await agent.run(max_steps=max(3, min(15, max_steps)))
 
-    timeout = max(30, min(300, int(os.environ.get("BROWSER_USE_TIMEOUT_SECONDS", "120"))))
+    try:
+        timeout = int(os.environ.get("BROWSER_USE_TIMEOUT_SECONDS", "120"))
+    except (TypeError, ValueError):
+        timeout = 120
+    timeout = max(30, min(300, timeout))
     try:
         history = asyncio.run(asyncio.wait_for(run(), timeout=timeout))
     except asyncio.TimeoutError as exc:

api/server.py

@@ -90,6 +90,63 @@ def now() -> str:
     return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
 
 
+try:
+    LOG_RETENTION = max(2_000, min(50_000, int(os.environ.get("LOG_RETENTION", "20000"))))
+except (TypeError, ValueError):
+    LOG_RETENTION = 20_000
+_LOG_SECRET_KEYS = {"api_key", "authorization", "password", "secret", "token", "access_token", "refresh_token"}
+
+
+def _redact_log_value(value: Any, depth: int = 0) -> Any:
+    """Keep logs useful without persisting provider credentials."""
+    if depth > 5:
+        return "[truncated]"
+    if isinstance(value, dict):
+        output: dict[str, Any] = {}
+        for key, item in list(value.items())[:80]:
+            key_text = str(key)
+            if key_text.lower() in _LOG_SECRET_KEYS or key_text.lower().endswith("_api_key"):
+                output[key_text] = "[redacted]"
+            else:
+                output[key_text] = _redact_log_value(item, depth + 1)
+        return output
+    if isinstance(value, (list, tuple, set)):
+        return [_redact_log_value(item, depth + 1) for item in list(value)[:100]]
+    if isinstance(value, str):
+        return value[:4_000] + ("…" if len(value) > 4_000 else "")
+    if isinstance(value, (int, float, bool)) or value is None:
+        return value
+    return str(value)[:4_000]
+
+
+def record_log(kind: str, event: str, *, level: str = "info", method: str = "", path: str = "",
+               status: int | None = None, duration_ms: int | None = None, details: Any = None,
+               error: str = "", run_id: int | None = None, prospect_id: int | None = None) -> None:
+    """Persist one bounded operational event; logging must never break a request."""
+    try:
+        safe_kind = str(kind or "system").strip().lower()[:30] or "system"
+        safe_level = str(level or "info").strip().lower()[:12] or "info"
+        safe_details = _redact_log_value(details if details is not None else {})
+        encoded = json.dumps(safe_details, ensure_ascii=False, default=str)[:20_000]
+        conn = db()
+        try:
+            cur = conn.execute(
+                """INSERT INTO logs(created_at,kind,level,event,method,path,status,duration_ms,run_id,prospect_id,details,error)
+                   VALUES(?,?,?,?,?,?,?,?,?,?,?,?)""",
+                (now(), safe_kind, safe_level, str(event or "event")[:240], str(method or "")[:12],
+                 str(path or "")[:500], status, duration_ms, run_id, prospect_id, encoded, str(error or "")[:2_000]),
+            )
+            # Retain enough history for investigation without allowing a busy
+            # browser session to grow the SQLite volume indefinitely.
+            if int(cur.lastrowid or 0) % 100 == 0:
+                conn.execute("DELETE FROM logs WHERE id NOT IN (SELECT id FROM logs ORDER BY id DESC LIMIT ?)", (LOG_RETENTION,))
+            conn.commit()
+        finally:
+            conn.close()
+    except Exception:
+        return
+
+
 def backend_mode() -> str:
     configured = os.environ.get("DISCOVERY_BACKEND", "").strip().lower()
     if configured in {"gmaps", "browser"}:
@@ -205,6 +262,21 @@ def db() -> sqlite3.Connection:
         raw_json TEXT NOT NULL DEFAULT '{}',
         UNIQUE(prospect_id, domain)
     )""")
+    conn.execute("""CREATE TABLE IF NOT EXISTS logs (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        created_at TEXT NOT NULL,
+        kind TEXT NOT NULL DEFAULT 'system',
+        level TEXT NOT NULL DEFAULT 'info',
+        event TEXT NOT NULL,
+        method TEXT NOT NULL DEFAULT '',
+        path TEXT NOT NULL DEFAULT '',
+        status INTEGER,
+        duration_ms INTEGER,
+        run_id INTEGER,
+        prospect_id INTEGER,
+        details TEXT NOT NULL DEFAULT '{}',
+        error TEXT NOT NULL DEFAULT ''
+    )""")
 
     # Migrate databases created by v1-v5 without destroying the user's
     # catalogue. SQLite permits adding nullable/defaulted columns in place.
@@ -268,6 +340,9 @@ def db() -> sqlite3.Connection:
     conn.execute("CREATE INDEX IF NOT EXISTS idx_prospects_score ON prospects(opportunity_score DESC)")
     conn.execute("CREATE INDEX IF NOT EXISTS idx_runs_created ON runs(id DESC)")
     conn.execute("CREATE INDEX IF NOT EXISTS idx_website_candidates_prospect ON website_candidates(prospect_id, confidence DESC)")
+    conn.execute("CREATE INDEX IF NOT EXISTS idx_logs_created ON logs(id DESC)")
+    conn.execute("CREATE INDEX IF NOT EXISTS idx_logs_kind ON logs(kind, id DESC)")
+    conn.execute("CREATE INDEX IF NOT EXISTS idx_logs_level ON logs(level, id DESC)")
     conn.commit()
     _schema_path = DB_PATH
     _schema_lock.release()
@@ -507,7 +582,19 @@ def _upstream_error(prefix: str, code: int, detail: str) -> str:
     return f"{prefix}_http_{code}:{detail}"
 
 
+def _log_http_kind(error_prefix: str) -> str:
+    return "ai" if error_prefix == "ai" else "search" if error_prefix == "gmaps" else "api"
+
+
+def _safe_log_url(url: str) -> str:
+    parsed = urlparse(url)
+    return f"{parsed.scheme}://{parsed.netloc}{parsed.path}"[:700]
+
+
 def http_json(method: str, url: str, payload: Any = None, headers: dict[str, str] | None = None, timeout: float = 20, error_prefix: str = "upstream") -> Any:
+    started = time.monotonic()
+    log_kind = _log_http_kind(error_prefix)
+    log_details = {"url": _safe_log_url(url), "timeout_seconds": timeout, "request": payload}
     data = None if payload is None else json.dumps(payload).encode("utf-8")
     request_headers = {"Accept": "application/json", "User-Agent": OUTBOUND_USER_AGENT, **(headers or {})}
     if data is not None:
@@ -516,19 +603,32 @@ def http_json(method: str, url: str, payload: Any = None, headers: dict[str, str
     try:
         with urllib.request.urlopen(request, timeout=timeout) as response:
             body = response.read()
+            response_status = int(getattr(response, "status", 200) or 200)
     except urllib.error.HTTPError as exc:
         detail = exc.read().decode("utf-8", "replace")[:300]
+        record_log(log_kind, "upstream_http_error", level="error", method=method, path=_safe_log_url(url), status=int(exc.code),
+                   duration_ms=int((time.monotonic() - started) * 1000), details={**log_details, "response": detail}, error=_upstream_error(error_prefix, int(exc.code), detail))
         raise RuntimeError(_upstream_error(error_prefix, int(exc.code), detail)) from exc
     except urllib.error.URLError as exc:
+        record_log(log_kind, "upstream_unreachable", level="error", method=method, path=_safe_log_url(url),
+                   duration_ms=int((time.monotonic() - started) * 1000), details=log_details, error=str(exc.reason)[:500])
         raise RuntimeError(f"{error_prefix}_unreachable:{exc.reason}") from exc
     try:
-        return json.loads(body or b"{}")
+        parsed = json.loads(body or b"{}")
+        record_log(log_kind, "upstream_request", method=method, path=_safe_log_url(url), status=response_status,
+                   duration_ms=int((time.monotonic() - started) * 1000), details={**log_details, "response": parsed})
+        return parsed
     except json.JSONDecodeError as exc:
-        raise RuntimeError("gmaps_invalid_json") from exc
+        record_log(log_kind, "upstream_invalid_json", level="error", method=method, path=_safe_log_url(url), status=response_status,
+                   duration_ms=int((time.monotonic() - started) * 1000), details={**log_details, "response": body[:2_000].decode("utf-8", "replace")}, error=str(exc))
+        raise RuntimeError(f"{error_prefix}_invalid_json") from exc
 
 
 def http_bytes(method: str, url: str, payload: Any = None, headers: dict[str, str] | None = None, timeout: float = 20, error_prefix: str = "upstream") -> bytes:
     """Make a small HTTP request where the response is not JSON (CSV download)."""
+    started = time.monotonic()
+    log_kind = _log_http_kind(error_prefix)
+    log_details = {"url": _safe_log_url(url), "timeout_seconds": timeout, "request": payload}
     data = None if payload is None else json.dumps(payload).encode("utf-8")
     request_headers = {"Accept": "*/*", "User-Agent": OUTBOUND_USER_AGENT, **(headers or {})}
     if data is not None:
@@ -536,11 +636,19 @@ def http_bytes(method: str, url: str, payload: Any = None, headers: dict[str, st
     request = urllib.request.Request(url, data=data, headers=request_headers, method=method.upper())
     try:
         with urllib.request.urlopen(request, timeout=timeout) as response:
-            return response.read()
+            body = response.read()
+            response_status = int(getattr(response, "status", 200) or 200)
+            record_log(log_kind, "upstream_request", method=method, path=_safe_log_url(url), status=response_status,
+                       duration_ms=int((time.monotonic() - started) * 1000), details={**log_details, "response_bytes": len(body)})
+            return body
     except urllib.error.HTTPError as exc:
         detail = exc.read().decode("utf-8", "replace")[:300]
+        record_log(log_kind, "upstream_http_error", level="error", method=method, path=_safe_log_url(url), status=int(exc.code),
+                   duration_ms=int((time.monotonic() - started) * 1000), details={**log_details, "response": detail}, error=_upstream_error(error_prefix, int(exc.code), detail))
         raise RuntimeError(_upstream_error(error_prefix, int(exc.code), detail)) from exc
     except urllib.error.URLError as exc:
+        record_log(log_kind, "upstream_unreachable", level="error", method=method, path=_safe_log_url(url),
+                   duration_ms=int((time.monotonic() - started) * 1000), details=log_details, error=str(exc.reason)[:500])
         raise RuntimeError(f"{error_prefix}_unreachable:{exc.reason}") from exc
 
 
@@ -765,7 +873,7 @@ def extract_search_results(page: Any, selector: str, title_selector: str | None,
     return output
 
 
-def google_search(query: str, requested: int) -> list[dict[str, Any]]:
+def google_search(query: str, requested: int, run_id: int | None = None) -> list[dict[str, Any]]:
     try:
         from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeout
     except ImportError as exc:
@@ -793,6 +901,7 @@ def google_search(query: str, requested: int) -> list[dict[str, Any]]:
                     break
                 context = None
                 try:
+                    record_log("search", "search_engine_started", run_id=run_id, details={"engine": engine, "query": query, "requested": requested})
                     context = browser.new_context(locale="en-ZA", java_script_enabled=True, service_workers="block", viewport={"width": 1280, "height": 900})
                     page = context.new_page()
                     page.set_default_timeout(ENGINE_TIMEOUT_MS)
@@ -806,15 +915,20 @@ def google_search(query: str, requested: int) -> list[dict[str, Any]]:
                     html = page.content()
                     if blocked(body) or blocked(html):
                         errors.append(engine + "_blocked")
+                        record_log("search", "search_engine_blocked", level="warn", run_id=run_id, details={"engine": engine, "query": query})
                         continue
                     output = extract_search_results(page, selector, title_selector, excluded_hosts, requested)
                     if output:
+                        record_log("search", "search_engine_results", run_id=run_id, details={"engine": engine, "query": query, "count": len(output), "results": output})
                         return output
                     errors.append(engine + ("_timeout" if navigation_timed_out else "_empty"))
+                    record_log("search", "search_engine_empty", level="warn", run_id=run_id, details={"engine": engine, "query": query, "navigation_timed_out": navigation_timed_out})
                 except PlaywrightTimeout:
                     errors.append(engine + "_timeout")
-                except Exception:
+                    record_log("search", "search_engine_timeout", level="warn", run_id=run_id, details={"engine": engine, "query": query})
+                except Exception as exc:
                     errors.append(engine + "_error")
+                    record_log("search", "search_engine_error", level="error", run_id=run_id, details={"engine": engine, "query": query}, error=str(exc)[:500])
                 finally:
                     if context is not None:
                         try:
@@ -823,7 +937,9 @@ def google_search(query: str, requested: int) -> list[dict[str, Any]]:
                             pass
         finally:
             browser.close()
-    raise RuntimeError("search_engines_unavailable:" + ",".join(errors))
+    message = "search_engines_unavailable:" + ",".join(errors)
+    record_log("search", "search_failed", level="error", run_id=run_id, details={"query": query, "errors": errors}, error=message)
+    raise RuntimeError(message)
 
 
 def validate_payload(payload: dict[str, Any]) -> tuple[str, str, str, int, bool, bool]:
@@ -859,6 +975,10 @@ def update_run(run_id: int, **fields: Any) -> None:
         conn.commit()
     finally:
         conn.close()
+    if any(key in safe for key in ("stage", "status", "error")):
+        record_log("system", "run_state_updated", level="error" if safe.get("status") == "failed" else "info",
+                   run_id=run_id, details={key: value for key, value in safe.items() if key in {"stage", "status", "progress", "found", "added", "error", "external_job_id"}},
+                   error=str(safe.get("error") or ""))
 
 
 def _json_list(value: Any) -> list[Any]:
@@ -873,10 +993,19 @@ def enrich_place(item: dict[str, Any]) -> dict[str, Any]:
     """Add website health, public email evidence, and explainable scoring."""
     prior_evidence = _json_list(item.get("email_evidence"))
     if item.get("website"):
+        scan_started = time.monotonic()
+        record_log("search", "website_scan_started", prospect_id=as_int(item.get("id")) or None,
+                   details={"business": item.get("name"), "url": _safe_log_url(str(item.get("website")))})
         try:
             item.update(scan_website(str(item["website"])))
+            record_log("search", "website_scan_completed", prospect_id=as_int(item.get("id")) or None,
+                       duration_ms=int((time.monotonic() - scan_started) * 1000),
+                       details={"business": item.get("name"), "url": _safe_log_url(str(item.get("website"))), "status": item.get("website_status"), "emails_found": len(item.get("website_emails") or [])})
         except Exception as exc:  # a single hostile site must not abort a run
             item.update({"website_status": "unknown", "website_checked_at": now(), "website_error": str(exc)[:300], "website_emails": [], "email_evidence": []})
+            record_log("search", "website_scan_failed", level="error", prospect_id=as_int(item.get("id")) or None,
+                       duration_ms=int((time.monotonic() - scan_started) * 1000),
+                       details={"business": item.get("name"), "url": _safe_log_url(str(item.get("website")))}, error=str(exc)[:500])
     else:
         item.setdefault("website_status", "no_detected_website")
         item.setdefault("website_emails", [])
@@ -935,6 +1064,7 @@ def insert_catalogue(run_id: int, query: str, raw_results: list[dict[str, Any]],
                 progress(index, total)
             emails = set(normalize_public_emails(item.get("emails")))
             if item["domain"].lower() in suppressed_domains or emails.intersection(suppressed_emails):
+                record_log("system", "prospect_suppressed", run_id=run_id, details={"name": item.get("name"), "domain": item.get("domain"), "email_match": bool(emails.intersection(suppressed_emails))})
                 continue
             try:
                 cur = conn.execute("""INSERT OR IGNORE INTO prospects(
@@ -961,6 +1091,11 @@ def insert_catalogue(run_id: int, query: str, raw_results: list[dict[str, Any]],
                 continue
             if cur.rowcount:
                 added += 1
+                conn.commit()
+                record_log("system", "prospect_added", run_id=run_id, details={"name": item.get("name"), "domain": item.get("domain"), "score": item.get("opportunity_score")})
+            else:
+                conn.commit()
+                record_log("system", "prospect_duplicate", run_id=run_id, details={"name": item.get("name"), "domain": item.get("domain")})
         conn.commit()
     finally:
         conn.close()
@@ -972,6 +1107,7 @@ def run_discovery(run_id: int, payload: dict[str, Any]) -> None:
     # Serialize local runs so two personal browser jobs do not fight over the
     # same machine's Chromium/RAM. The Maps worker itself remains async.
     with _discover_lock:
+        record_log("search", "discovery_started", run_id=run_id, details={"query": query, "location": location, "requested": requested, "backend": backend_mode(), "email_collection": include_email})
         update_run(run_id, status="running", stage="connecting", progress=5, source=backend_mode())
         try:
             if backend_mode() == "gmaps":
@@ -985,7 +1121,7 @@ def run_discovery(run_id: int, payload: dict[str, Any]) -> None:
                 raw_results = poll_gmaps_job(external_job_id, on_progress)
             else:
                 update_run(run_id, stage="browser_search", progress=20)
-                raw_results = google_search(query, requested)
+                raw_results = google_search(query, requested, run_id=run_id)
             update_run(run_id, stage="website_analysis", progress=76, found=len(raw_results))
 
             def enrichment_progress(index: int, total: int) -> None:
@@ -994,9 +1130,11 @@ def run_discovery(run_id: int, payload: dict[str, Any]) -> None:
 
             results, added = insert_catalogue(run_id, query, raw_results, requested, enrichment_progress, search_category=category)
             update_run(run_id, found=len(results), added=added, status="succeeded", stage="complete", progress=100, completed_at=now(), error="")
+            record_log("search", "discovery_completed", run_id=run_id, details={"found": len(results), "added": added, "backend": backend_mode()})
         except Exception as exc:
             message = str(exc).strip()[:500] or "discovery_failed"
             update_run(run_id, status="failed", stage="failed", progress=100, error=message, completed_at=now())
+            record_log("search", "discovery_failed", level="error", run_id=run_id, details={"backend": backend_mode()}, error=message)
 
 
 def create_discovery(payload: dict[str, Any]) -> dict[str, Any]:
@@ -1205,6 +1343,10 @@ def ai_assess_prospect(prospect: dict[str, Any]) -> dict[str, Any]:
 class Handler(BaseHTTPRequestHandler):
     server_version = "PersonalProspectOS/2.0"
 
+    def handle_one_request(self) -> None:
+        self._request_started_at = time.monotonic()
+        super().handle_one_request()
+
     def send_json(self, status: int, value: Any) -> None:
         body = json_bytes(value)
         self.send_response(status)
@@ -1213,6 +1355,15 @@ class Handler(BaseHTTPRequestHandler):
         self.send_header("Content-Length", str(len(body)))
         self.end_headers()
         self.wfile.write(body)
+        if str(getattr(self, "path", "")).startswith("/api/"):
+            error_value = value.get("error") if isinstance(value, dict) else ""
+            record_log(
+                "api", f"{getattr(self, 'command', 'HTTP')} {urlparse(self.path).path}",
+                level="error" if status >= 400 else "info", method=getattr(self, "command", "HTTP"),
+                path=urlparse(self.path).path, status=status,
+                duration_ms=int((time.monotonic() - getattr(self, "_request_started_at", time.monotonic())) * 1000),
+                details={"response_status": status}, error=str(error_value or "")[:500],
+            )
 
     def read_json(self) -> dict[str, Any]:
         length = min(int(self.headers.get("Content-Length", "0") or 0), 100_000)
@@ -1255,6 +1406,39 @@ class Handler(BaseHTTPRequestHandler):
                 if provider == "nous":
                     return self.send_json(200, {"provider": "nous", "base_url": NOUS_BASE_URL, "models": nous_models()})
                 return self.send_json(200, {"provider": provider or "custom", "models": []})
+            if parsed.path == "/api/logs":
+                params = parse_qs(parsed.query)
+                kind = (params.get("kind", [""])[0] or "").strip().lower()
+                level = (params.get("level", [""])[0] or "").strip().lower()
+                query = (params.get("q", [""])[0] or "").strip().lower()
+                try:
+                    limit = max(1, min(500, int(params.get("limit", ["200"])[0])))
+                except (TypeError, ValueError):
+                    limit = 200
+                clauses: list[str] = []
+                values: list[Any] = []
+                if kind and kind != "all":
+                    clauses.append("kind=?"); values.append(kind)
+                if level and level != "all":
+                    clauses.append("level=?"); values.append(level)
+                if query:
+                    clauses.append("(lower(event) LIKE ? OR lower(path) LIKE ? OR lower(error) LIKE ? OR lower(details) LIKE ?)")
+                    needle = "%" + query[:120] + "%"; values.extend([needle, needle, needle, needle])
+                where = " WHERE " + " AND ".join(clauses) if clauses else ""
+                rows = conn.execute("SELECT * FROM logs" + where + " ORDER BY id DESC LIMIT ?", (*values, limit)).fetchall()
+                items: list[dict[str, Any]] = []
+                for row in rows:
+                    item = dict(row)
+                    try:
+                        item["details"] = json.loads(item.get("details") or "{}")
+                    except (TypeError, ValueError, json.JSONDecodeError):
+                        item["details"] = {"raw": str(item.get("details") or "")[:20_000]}
+                    items.append(item)
+                total = int(conn.execute("SELECT COUNT(*) FROM logs" + where, tuple(values)).fetchone()[0])
+                counts = {str(row[0]): int(row[1]) for row in conn.execute("SELECT kind,COUNT(*) FROM logs GROUP BY kind")}
+                error_count = int(conn.execute("SELECT COUNT(*) FROM logs WHERE level='error'").fetchone()[0])
+                counts["errors"] = error_count
+                return self.send_json(200, {"items": items, "total": total, "counts": counts})
             if parsed.path == "/api/settings":
                 ai_runtime = ai_runtime_settings()
                 models = nous_models() if ai_runtime["provider"] == "nous" else []
@@ -1345,6 +1529,21 @@ class Handler(BaseHTTPRequestHandler):
         self.wfile.write(content)
 
     def do_POST(self) -> None:
+        if urlparse(self.path).path == "/api/logs":
+            try:
+                payload = self.read_json()
+                action = str(payload.get("action") or "").strip()
+                if not action:
+                    return self.send_json(400, {"error": "action_required"})
+                level = str(payload.get("level") or "info").strip().lower()
+                if level not in {"info", "warn", "error"}:
+                    level = "info"
+                details = payload.get("details") if isinstance(payload.get("details"), (dict, list, str, int, float, bool)) else {}
+                record_log("action", action, level=level, method="CLIENT", path=str(payload.get("path") or "")[:500], details=details,
+                           error=str(payload.get("error") or "")[:500])
+                return self.send_json(202, {"logged": True})
+            except (ValueError, TypeError, json.JSONDecodeError) as exc:
+                return self.send_json(400, {"error": str(exc)})
         if self.path == "/api/settings":
             try:
                 payload = self.read_json(); ai = payload.get("ai") if isinstance(payload.get("ai"), dict) else {}
@@ -1389,10 +1588,18 @@ class Handler(BaseHTTPRequestHandler):
                 return self.send_json(503, {"error": "browser_use_disabled"})
             try:
                 runtime = ai_runtime_settings()
+                record_log("browser", "website_investigation_started", prospect_id=prospect_id,
+                           details={"business": row["name"], "model": runtime["model"], "public_only": True})
+
+                def browser_log(kind: str, event: str, details: dict[str, Any]) -> None:
+                    record_log(kind, event, prospect_id=prospect_id, details=details)
+
                 candidates = investigate_websites(
                     dict(row), base_url=runtime["base_url"], api_key=runtime["api_key"],
-                    model=runtime["model"], user_agent=OUTBOUND_USER_AGENT,
+                    model=runtime["model"], user_agent=OUTBOUND_USER_AGENT, log_callback=browser_log,
                 )
+                record_log("browser", "website_investigation_completed", prospect_id=prospect_id,
+                           details={"candidate_count": len(candidates), "model": runtime["model"]})
                 conn = db()
                 for candidate in candidates:
                     encoded_evidence = json.dumps(candidate.get("evidence") or [], ensure_ascii=False)
@@ -1411,8 +1618,10 @@ class Handler(BaseHTTPRequestHandler):
                 rows = conn.execute("SELECT * FROM website_candidates WHERE prospect_id=? ORDER BY confidence DESC, id DESC", (prospect_id,)).fetchall()
                 return self.send_json(200, {"items": [dict(item) for item in rows], "count": len(rows)})
             except RuntimeError as exc:
+                record_log("browser", "website_investigation_failed", level="error", prospect_id=prospect_id, error=str(exc))
                 return self.send_json(503, {"error": str(exc)})
             except Exception as exc:
+                record_log("browser", "website_investigation_failed", level="error", prospect_id=prospect_id, error=str(exc)[:500])
                 return self.send_json(502, {"error": "browser_use_failed:" + str(exc)[:240]})
             finally:
                 try: