From 80927a28723efb802a4dd1f7013cebe6006ebb9a Mon Sep 17 00:00:00 2001 From: "DESKTOP-RTLN3BA\\$punk" Date: Sun, 5 Jul 2026 17:08:01 -0700 Subject: [PATCH] feat(billing): meter platform scrapers per item; consolidate web scraping onto web.crawl Add per-item, per-platform billing for the platform-native connectors (Reddit, Google Search, Google Maps places/reviews, YouTube videos/comments) through the capability gate/charge seam. Rates are config-driven with a shared wallet-credit module (wallet_credit) and a dedicated PlatformScrapeCreditService; agent and REST capability runs now record cost_micros. Google Maps scrape dual-meters places and attached reviews. Remove the main-agent scrape_webpage tool now that the web.crawl capability covers single-page (maxCrawlDepth=0) and site crawling. The main agent now reaches crawling via task(web_crawler, ...). Update prompts, tool catalog, receipts, skills, proprietary docs, and tests; drop the obsolete chat-turn crawl fold path. Co-authored-by: Cursor --- docker/.env.example | 25 ++ surfsense_backend/.env.example | 21 +- .../skills/builtin/kb-research/SKILL.md | 2 +- .../skills/builtin/meeting-prep/SKILL.md | 2 +- .../system_prompt/prompts/kb_first.md | 5 +- .../prompts/providers/anthropic.md | 2 +- .../prompts/providers/openai_classic.md | 2 +- .../system_prompt/prompts/routing.md | 12 +- .../prompts/tools/scrape_webpage/__init__.py | 1 - .../tools/scrape_webpage/description.md | 11 - .../prompts/tools/scrape_webpage/example.md | 24 -- .../prompts/tools/task/description.md | 8 +- .../main_agent/tools/index.py | 1 - .../main_agent/tools/registry.py | 8 +- .../main_agent/tools/scrape_webpage.py | 372 ------------------ .../shared/receipts/receipt.py | 2 +- .../multi_agent_chat/shared/tools/catalog.py | 4 - .../app/capabilities/core/access/agent.py | 3 +- .../app/capabilities/core/access/rest.py | 3 +- .../app/capabilities/core/billing.py | 151 +++++-- .../app/capabilities/core/types.py | 11 +- .../google_maps/reviews/definition.py | 7 +- .../google_maps/reviews/schemas.py | 11 + .../google_maps/scrape/definition.py | 8 +- .../google_maps/scrape/schemas.py | 27 ++ .../google_search/scrape/definition.py | 7 +- .../google_search/scrape/schemas.py | 11 + .../capabilities/reddit/scrape/definition.py | 7 +- .../app/capabilities/reddit/scrape/schemas.py | 11 + .../youtube/comments/definition.py | 7 +- .../capabilities/youtube/comments/schemas.py | 11 + .../capabilities/youtube/scrape/definition.py | 7 +- .../capabilities/youtube/scrape/schemas.py | 16 + surfsense_backend/app/config/__init__.py | 39 +- surfsense_backend/app/proprietary/README.md | 2 +- .../app/proprietary/web_crawler/README.md | 5 +- .../platform_scrape_credit_service.py | 77 ++++ .../app/services/wallet_credit.py | 99 +++++ .../app/services/web_crawl_credit_service.py | 64 +-- .../handlers/tools/scrape_webpage/emission.py | 43 -- .../tools/scrape_webpage/shared/__init__.py | 0 .../tools/scrape_webpage/shared/tool_input.py | 9 - .../handlers/tools/scrape_webpage/thinking.py | 49 --- .../scripts/e2e_phase3_crawl_billing.py | 48 +-- .../multi_agent_chat/test_prompt_resources.py | 2 +- .../test_default_permissions_layering.py | 3 +- .../tests/unit/capabilities/test_billing.py | 169 ++++++++ .../services/test_web_crawl_credit_service.py | 81 +--- 48 files changed, 724 insertions(+), 766 deletions(-) delete mode 100644 surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/tools/scrape_webpage/__init__.py delete mode 100644 surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/tools/scrape_webpage/description.md delete mode 100644 surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/tools/scrape_webpage/example.md delete mode 100644 surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/tools/scrape_webpage.py create mode 100644 surfsense_backend/app/services/platform_scrape_credit_service.py create mode 100644 surfsense_backend/app/services/wallet_credit.py delete mode 100644 surfsense_backend/app/tasks/chat/streaming/handlers/tools/scrape_webpage/emission.py delete mode 100644 surfsense_backend/app/tasks/chat/streaming/handlers/tools/scrape_webpage/shared/__init__.py delete mode 100644 surfsense_backend/app/tasks/chat/streaming/handlers/tools/scrape_webpage/shared/tool_input.py delete mode 100644 surfsense_backend/app/tasks/chat/streaming/handlers/tools/scrape_webpage/thinking.py diff --git a/docker/.env.example b/docker/.env.example index a191c27c1..28f9356c8 100644 --- a/docker/.env.example +++ b/docker/.env.example @@ -425,6 +425,31 @@ SURFSENSE_ENABLE_DOOM_LOOP=true # ETL_CREDIT_BILLING_ENABLED=FALSE # MICROS_PER_PAGE=1000 +# Debit the credit wallet per *successful* web crawl. Default FALSE keeps +# crawling effectively free for self-hosted installs. Price is config-driven: +# WEB_CRAWL_MICROS_PER_SUCCESS = round(USD_per_1000_crawls * 1_000) +# 2000 == $2/1000 (default) | 1000 == $1/1000. Captcha solves bill as a +# separate per-attempt unit (independent flag). +# WEB_CRAWL_CREDIT_BILLING_ENABLED=FALSE +# WEB_CRAWL_MICROS_PER_SUCCESS=2000 +# WEB_CRAWL_CAPTCHA_BILLING_ENABLED=FALSE +# WEB_CRAWL_CAPTCHA_MICROS_PER_SOLVE=3000 + +# Debit the credit wallet per *item returned* by the platform-native scrapers +# (Reddit, Google Search, Google Maps, YouTube). Default FALSE keeps scraping +# effectively free for self-hosted installs. Each rate is micro-USD per item, +# config-driven: = round(USD_per_1000_items * 1_000). Defaults sit +# at/above Apify's first-party actor rates (we charge no subscription tiers, +# start fees, or separate proxy/compute billing). google_maps.scrape is +# dual-metered (places + attached reviews). +# PLATFORM_SCRAPE_BILLING_ENABLED=FALSE +# REDDIT_SCRAPE_MICROS_PER_ITEM=3500 +# GOOGLE_SEARCH_MICROS_PER_SERP=5500 +# GOOGLE_MAPS_MICROS_PER_PLACE=5000 +# GOOGLE_MAPS_MICROS_PER_REVIEW=2000 +# YOUTUBE_MICROS_PER_VIDEO=3500 +# YOUTUBE_MICROS_PER_COMMENT=3500 + # Safety ceiling on per-call premium reservation, in micro-USD ($1.00 default). # QUOTA_MAX_RESERVE_MICROS=1000000 diff --git a/surfsense_backend/.env.example b/surfsense_backend/.env.example index 65b60bdd1..32df59a0a 100644 --- a/surfsense_backend/.env.example +++ b/surfsense_backend/.env.example @@ -256,10 +256,10 @@ MICROS_PER_PAGE=1000 # crawling effectively free for self-hosted/OSS installs; hosted sets TRUE. # Price is fully config-driven (the only source of truth, no hardcoded rate): # WEB_CRAWL_MICROS_PER_SUCCESS = round(USD_per_1000_crawls * 1_000) -# 1000 == $1 / 1000 crawls (default) | 2000 == $2/1000 | 500 == $0.50/1000 +# 2000 == $2 / 1000 crawls (default) | 1000 == $1/1000 | 500 == $0.50/1000 # Retune anytime with just an env change + restart (no code/migration). # WEB_CRAWL_CREDIT_BILLING_ENABLED=FALSE -# WEB_CRAWL_MICROS_PER_SUCCESS=1000 +# WEB_CRAWL_MICROS_PER_SUCCESS=2000 # Phase 3d: bill captcha solves as a SEPARATE per-attempt unit (the solver # charges per attempt regardless of crawl success). Independent of the crawl @@ -270,6 +270,23 @@ MICROS_PER_PAGE=1000 # WEB_CRAWL_CAPTCHA_BILLING_ENABLED=FALSE # WEB_CRAWL_CAPTCHA_MICROS_PER_SOLVE=3000 +# Debit the credit wallet per *item returned* by the platform-native scrapers +# (Reddit, Google Search, Google Maps, YouTube). Default FALSE keeps scraping +# effectively free for self-hosted/OSS installs; hosted deployments set TRUE. +# Each rate is micro-USD per item, fully config-driven (no hardcoded rate): +# = round(USD_per_1000_items * 1_000) +# 3500 == $3.50/1000 | 5000 == $5/1000 | 2000 == $2/1000 +# Defaults sit at/above Apify's first-party actor rates (Jul 2026); justified +# because we charge no subscription tiers, no per-run start fees, and no +# separate proxy/compute/storage billing. Retune anytime via env + restart. +# PLATFORM_SCRAPE_BILLING_ENABLED=FALSE +# REDDIT_SCRAPE_MICROS_PER_ITEM=3500 +# GOOGLE_SEARCH_MICROS_PER_SERP=5500 +# GOOGLE_MAPS_MICROS_PER_PLACE=5000 +# GOOGLE_MAPS_MICROS_PER_REVIEW=2000 +# YOUTUBE_MICROS_PER_VIDEO=3500 +# YOUTUBE_MICROS_PER_COMMENT=3500 + # Low-balance warning threshold (micro-USD), surfaced to the UI. Default $0.50. CREDIT_LOW_BALANCE_WARNING_MICROS=500000 diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/skills/builtin/kb-research/SKILL.md b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/skills/builtin/kb-research/SKILL.md index 3f0b46d76..9ac3a0ad1 100644 --- a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/skills/builtin/kb-research/SKILL.md +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/skills/builtin/kb-research/SKILL.md @@ -1,7 +1,7 @@ --- name: kb-research description: Structured approach to finding and synthesizing information from the user's knowledge base -allowed-tools: scrape_webpage, read_file, ls_tree, grep +allowed-tools: read_file, ls_tree, grep --- # Knowledge-base research diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/skills/builtin/meeting-prep/SKILL.md b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/skills/builtin/meeting-prep/SKILL.md index c08819e8b..5a6a4cb7a 100644 --- a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/skills/builtin/meeting-prep/SKILL.md +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/skills/builtin/meeting-prep/SKILL.md @@ -1,7 +1,7 @@ --- name: meeting-prep description: Pull together briefing materials before a scheduled meeting -allowed-tools: task, scrape_webpage, read_file +allowed-tools: task, read_file --- # Meeting preparation diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/kb_first.md b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/kb_first.md index 3523447b9..d3b826d61 100644 --- a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/kb_first.md +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/kb_first.md @@ -7,8 +7,9 @@ CRITICAL — ground factual answers in what you actually receive this turn: - injected workspace context (see ``), - the user's connected apps via `task(mcp_discovery, ...)` (Slack, Jira, Notion, Gmail, Calendar, etc. — live data that is NOT in the knowledge base), -- results from your other tool calls (`scrape_webpage`) or the Google Search - specialist via `task(google_search, ...)`, +- results from your specialist calls — the web crawler via + `task(web_crawler, ...)` or the Google Search specialist via + `task(google_search, ...)`, - or substantive summaries returned by a `task` specialist you invoked. For questions about the user's own files and notes, dispatch diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/providers/anthropic.md b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/providers/anthropic.md index 918c94d7b..3fb8659be 100644 --- a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/providers/anthropic.md +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/providers/anthropic.md @@ -5,7 +5,7 @@ Structured reasoning: - For non-trivial work, `` / short `` before tool calls is fine. Professional objectivity: -- Accuracy over flattery; verify with **scrape_webpage** or **task** (e.g. `task(google_search, …)` for public facts) when unsure — don’t invent connector access. +- Accuracy over flattery; verify with **task** (e.g. `task(web_crawler, …)` to read a page, `task(google_search, …)` for public facts) when unsure — don’t invent connector access. Task management: - For 3+ steps, use todo tooling; update statuses promptly. diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/providers/openai_classic.md b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/providers/openai_classic.md index 7f0f69d70..9f4198e10 100644 --- a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/providers/openai_classic.md +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/providers/openai_classic.md @@ -16,6 +16,6 @@ Output style: Tool calls: - Parallelise independent calls in one turn. - For SurfSense-product questions, point the user to https://www.surfsense.com/docs; - use **task(google_search, …)** / **scrape_webpage** for fresh public facts; integrations and + use **task(google_search, …)** / **task(web_crawler, …)** for fresh public facts; integrations and heavy workflows → **task**. diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/routing.md b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/routing.md index 2d69d9b0e..a429b54e3 100644 --- a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/routing.md +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/routing.md @@ -3,7 +3,6 @@ You have two execution channels. Pick the one that owns the work — never simulate one with the other. ### 1. Direct tools (you call them yourself) -- `scrape_webpage` — fetch the body of a specific public URL. - `update_memory` — curate persistent memory (see ``). - `write_todos` — maintain a structured plan when the turn series spans multiple specialists or steps. Mark each item @@ -18,10 +17,11 @@ https://www.surfsense.com/docs. There is no docs-search tool; give the link. overviews, a specialist's summary of a SERP) are pointers, not sources. When the answer lives on a page — a team roster, a portfolio or directory listing, a pricing table, docs — fetch the page before answering: -- One or a few known URLs → `scrape_webpage` directly. +- One or a few known URLs → `task(web_crawler, …)` with those URLs (it + fetches only the seeds at `maxCrawlDepth=0`). - A site section or many pages (a whole team + portfolio, every pricing page of a list of companies, a paginated directory) → - `task(web_crawler, …)` with the seed URLs. + `task(web_crawler, …)` with the seed URLs and a higher depth. Never answer with "you can find it at " for public facts your tools can retrieve — retrieve them, then answer with the facts and cite the page. Large results are fine: extract and return them, don't ask permission for @@ -35,7 +35,8 @@ web search returns only snippets that need a second pass. Use the Search specialist for entities without a storefront (online-only companies, software vendors, publications), for facts and current events, and to enrich places Maps already found. When a lead list needs both, run Maps -discovery first, then scrape or search the found websites for contacts. +discovery first, then crawl (`task(web_crawler, …)`) or search the found +websites for contacts. **Requested-N lists count distinct entities that fit the ask.** When the user asks for N leads/items/results, every entry must be a distinct @@ -222,7 +223,8 @@ user: "Post the launch announcement to #general and let me know when it's up." task(mcp_discovery, "In Slack, post '' to #general. Return the message permalink.") Next turn (with the receipt's `verifiable_url` in hand): - scrape_webpage(url=) + task(web_crawler, "Crawl and confirm + the post is live; return what you find.") → confirm the post is live, then tell the user it's up with the URL. If the reply has NO Receipt with `status="success"`, treat it as a silent failure: surface the error verbatim, do not retry. diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/tools/scrape_webpage/__init__.py b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/tools/scrape_webpage/__init__.py deleted file mode 100644 index a101e7e1c..000000000 --- a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/tools/scrape_webpage/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""``scrape_webpage`` — description + few-shot examples.""" diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/tools/scrape_webpage/description.md b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/tools/scrape_webpage/description.md deleted file mode 100644 index d8f731359..000000000 --- a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/tools/scrape_webpage/description.md +++ /dev/null @@ -1,11 +0,0 @@ -- `scrape_webpage` — Fetch and extract readable content from a single URL. - - Use when the user wants the actual page body (article, table, dashboard - snapshot), not just search snippets. - - Try the tool when a URL is given or referenced; don't refuse without - attempting unless the URL is clearly unsafe or invalid. - - Public web only. For URLs behind a connector (Notion pages, Linear - issues, Confluence, anything that needs auth), use `task` with the - matching specialist instead. - - Args: `url`, `max_length` (default 50000). - - Returns title, metadata, and markdown-ish body. Summarise clearly and - link back with `[label](url)`. diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/tools/scrape_webpage/example.md b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/tools/scrape_webpage/example.md deleted file mode 100644 index 977d40b6d..000000000 --- a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/tools/scrape_webpage/example.md +++ /dev/null @@ -1,24 +0,0 @@ - -user: "Check out https://dev.to/some-article" -→ scrape_webpage(url="https://dev.to/some-article") -(Respond with a structured analysis — key points, takeaways.) - - - -user: "Read this article and summarize it for me: https://example.com/blog/ai-trends" -→ scrape_webpage(url="https://example.com/blog/ai-trends") -(Thorough summary using headings and bullets.) - - - -user: (after discussing https://example.com/stats) "Can you get the live data from that page?" -→ scrape_webpage(url="https://example.com/stats") -(Always attempt scraping first. Never refuse before trying.) - - - -user: "https://example.com/blog/weekend-recipes" -→ scrape_webpage(url="https://example.com/blog/weekend-recipes") -(When a user sends just a URL with no instructions, scrape it and provide -a concise summary.) - diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/tools/task/description.md b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/tools/task/description.md index d6a81d8d3..cfd47b7fd 100644 --- a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/tools/task/description.md +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/system_prompt/prompts/tools/task/description.md @@ -43,9 +43,9 @@ like podcasts/videos), the operation did not happen — treat as failure and surface that to the user verbatim, do not retry blindly. - 2. **`scrape_webpage`** — when a Receipt carries a `verifiable_url` + 2. **`task(web_crawler, …)`** — when a Receipt carries a `verifiable_url` (Notion page URL, Slack permalink, Jira issue URL, Linear identifier - URL, etc.), you can fetch that URL and confirm the operation + URL, etc.), you can crawl that URL and confirm the operation externally. Use this for high-stakes mutations the user explicitly called out (e.g. "send the launch email to the whole team") or when the subagent's self-report contradicts what the user expected. @@ -54,7 +54,7 @@ - `status="success"`: the mutation already committed in the backend. If a `verifiable_url` is present and the request was high-stakes, - you may `scrape_webpage` it to externally confirm. Otherwise trust + you may crawl it via `task(web_crawler, …)` to externally confirm. Otherwise trust the Receipt and tell the user it is done. Celery-backed deliverables (podcasts, video presentations) also land here — the subagent already waited for the worker to finish, so a `success` Receipt @@ -67,6 +67,6 @@ their backend before returning. If you ever do see a pending Receipt, tell the user the work has been **kicked off** (quote the `external_id` / `preview` so they can find it later), do not - `scrape_webpage` it, and do not re-dispatch the same + crawl it, and do not re-dispatch the same `task(...)` call hoping it will be done "this time". diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/tools/index.py b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/tools/index.py index 1a15884eb..fc87f0bb7 100644 --- a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/tools/index.py +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/tools/index.py @@ -6,7 +6,6 @@ Connector integrations, MCP, deliverables, etc. are delegated via ``task`` subag from __future__ import annotations MAIN_AGENT_SURFSENSE_TOOL_NAMES_ORDERED: tuple[str, ...] = ( - "scrape_webpage", "update_memory", "create_automation", ) diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/tools/registry.py b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/tools/registry.py index b87e7f11b..0d65e37fe 100644 --- a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/tools/registry.py +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/tools/registry.py @@ -23,17 +23,12 @@ from langchain_core.tools import BaseTool from app.db import ChatVisibility -from .scrape_webpage import create_scrape_webpage_tool from .update_memory import ( create_update_memory_tool, create_update_team_memory_tool, ) -def _build_scrape_webpage_tool(_deps: dict[str, Any]) -> BaseTool: - return create_scrape_webpage_tool() - - def _build_create_automation_tool(deps: dict[str, Any]) -> BaseTool: # Deferred import: the automation package is a sibling under ``main_agent`` # and is only needed at build time, mirroring the shared registry's @@ -63,12 +58,11 @@ def _build_update_memory_tool(deps: dict[str, Any]) -> BaseTool: # Ordered to match the historical main-agent binding order: -# scrape_webpage, create_automation, update_memory. +# create_automation, update_memory. # Each entry is ``(factory, required_dependency_names)``. _MAIN_AGENT_TOOL_FACTORIES: dict[ str, tuple[Callable[[dict[str, Any]], BaseTool], tuple[str, ...]] ] = { - "scrape_webpage": (_build_scrape_webpage_tool, ()), "create_automation": ( _build_create_automation_tool, ("workspace_id", "user_id", "llm"), diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/tools/scrape_webpage.py b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/tools/scrape_webpage.py deleted file mode 100644 index 5f2eaa913..000000000 --- a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/tools/scrape_webpage.py +++ /dev/null @@ -1,372 +0,0 @@ -""" -Web scraping tool for the SurfSense agent. - -This module provides a tool for scraping and extracting content from webpages -using the existing WebCrawlerConnector. For YouTube URLs, it fetches the -transcript directly via the YouTubeTranscriptApi instead of crawling the page. -""" - -import hashlib -import logging -import time -from typing import Any -from urllib.parse import urlparse - -from fake_useragent import UserAgent -from langchain_core.tools import tool -from requests import Session -from scrapling.fetchers import AsyncFetcher -from youtube_transcript_api import YouTubeTranscriptApi - -from app.proprietary.platforms.youtube.url_resolver import get_youtube_video_id -from app.proprietary.web_crawler import ( - CrawlOutcomeStatus, - WebCrawlerConnector, -) -from app.utils.proxy import get_proxy_url, get_requests_proxies - -logger = logging.getLogger(__name__) - - -def _bill_successful_scrape() -> None: - """Fold one successful crawl into the current chat turn's bill (Phase 3c). - - The cost rides the turn accumulator and settles at the premium - ``finalize_credit`` step — no separate wallet hit. Free / BYOK / anonymous - turns (which never reserve/finalize) still record the line in the - breakdown but are never debited. No-op when crawl billing is disabled or - there is no active turn (e.g. non-chat callers). - """ - from app.services.token_tracking_service import get_current_accumulator - from app.services.web_crawl_credit_service import WebCrawlCreditService - - if not WebCrawlCreditService.billing_enabled(): - return - acc = get_current_accumulator() - if acc is None: - return - acc.add( - model="web_crawl", - prompt_tokens=0, - completion_tokens=0, - total_tokens=0, - cost_micros=WebCrawlCreditService.successes_to_micros(1), - call_kind="web_crawl", - ) - - -def _bill_captcha_attempts(outcome) -> None: - """Fold captcha solve attempts (Phase 3d) into the current chat turn's bill. - - Per *attempt*, not per success: a solve that didn't rescue the crawl still - cost real solver money, so this runs before the success/failure branch. - Mirrors :func:`_bill_successful_scrape` (rides the turn accumulator, settles - at ``finalize_credit``; record-only on free/anonymous turns). No-op when - captcha billing is off, there were no attempts, or no turn is active. - """ - from app.services.token_tracking_service import get_current_accumulator - from app.services.web_crawl_credit_service import WebCrawlCreditService - - if not WebCrawlCreditService.captcha_billing_enabled(): - return - attempts = getattr(outcome, "captcha_attempts", 0) or 0 - if attempts <= 0: - return - acc = get_current_accumulator() - if acc is None: - return - acc.add( - model="web_crawl_captcha", - prompt_tokens=0, - completion_tokens=0, - total_tokens=0, - cost_micros=WebCrawlCreditService.captcha_solves_to_micros(attempts), - call_kind="web_crawl_captcha", - ) - - -def extract_domain(url: str) -> str: - """Extract the domain from a URL.""" - try: - parsed = urlparse(url) - domain = parsed.netloc - if domain.startswith("www."): - domain = domain[4:] - return domain - except Exception: - return "" - - -def generate_scrape_id(url: str) -> str: - """Generate a unique ID for a scraped webpage.""" - hash_val = hashlib.md5(url.encode()).hexdigest()[:12] - return f"scrape-{hash_val}" - - -def truncate_content(content: str, max_length: int = 50000) -> tuple[str, bool]: - """ - Truncate content to a maximum length. - - Returns: - Tuple of (truncated_content, was_truncated) - """ - if len(content) <= max_length: - return content, False - - # Prefer truncating at a sentence/paragraph boundary. - truncated = content[:max_length] - last_period = truncated.rfind(".") - last_newline = truncated.rfind("\n\n") - - boundary = max(last_period, last_newline) - if boundary > max_length * 0.8: # only if the boundary isn't too far back - truncated = content[: boundary + 1] - - return truncated + "\n\n[Content truncated...]", True - - -async def _scrape_youtube_video( - url: str, video_id: str, max_length: int -) -> dict[str, Any]: - """ - Fetch YouTube video metadata and transcript via the YouTubeTranscriptApi. - - Returns a result dict in the same shape as the regular scrape_webpage output. - """ - scrape_id = generate_scrape_id(url) - domain = "youtube.com" - - # --- Video metadata via oEmbed --- - residential_proxies = get_requests_proxies() - - params = { - "format": "json", - "url": f"https://www.youtube.com/watch?v={video_id}", - } - oembed_url = "https://www.youtube.com/oembed" - - try: - oembed_fetch_start = time.perf_counter() - oembed_page = await AsyncFetcher.get( - oembed_url, - params=params, - proxy=get_proxy_url(), - stealthy_headers=True, - ) - logger.info( - "[scrape_webpage][perf] source=oembed video=%s status=%s fetch_ms=%.1f", - video_id, - getattr(oembed_page, "status", None), - (time.perf_counter() - oembed_fetch_start) * 1000, - ) - video_data = oembed_page.json() - except Exception: - video_data = {} - - title = video_data.get("title", "YouTube Video") - author = video_data.get("author_name", "Unknown") - - # --- Transcript via YouTubeTranscriptApi --- - try: - transcript_fetch_start = time.perf_counter() - ua = UserAgent() - http_client = Session() - http_client.headers.update({"User-Agent": ua.random}) - if residential_proxies: - http_client.proxies.update(residential_proxies) - ytt_api = YouTubeTranscriptApi(http_client=http_client) - - # Pick the first transcript (video's primary language) rather than - # defaulting to English. - transcript_list = ytt_api.list(video_id) - transcript = next(iter(transcript_list)) - captions = transcript.fetch() - - logger.info( - "[scrape_webpage][perf] source=transcript video=%s fetch_ms=%.1f", - video_id, - (time.perf_counter() - transcript_fetch_start) * 1000, - ) - logger.info( - f"[scrape_webpage] Fetched transcript for {video_id} " - f"in {transcript.language} ({transcript.language_code})" - ) - - transcript_segments = [] - for line in captions: - start_time = line.start - duration = line.duration - text = line.text - timestamp = f"[{start_time:.2f}s-{start_time + duration:.2f}s]" - transcript_segments.append(f"{timestamp} {text}") - transcript_text = "\n".join(transcript_segments) - except Exception as e: - logger.warning(f"[scrape_webpage] No transcript for video {video_id}: {e}") - transcript_text = f"No captions available for this video. Error: {e!s}" - - content = f"# {title}\n\n**Author:** {author}\n**Video ID:** {video_id}\n\n## Transcript\n\n{transcript_text}" - - content, was_truncated = truncate_content(content, max_length) - word_count = len(content.split()) - - description = f"YouTube video by {author}" - - return { - "id": scrape_id, - "assetId": url, - "kind": "article", - "href": url, - "title": title, - "description": description, - "content": content, - "domain": domain, - "word_count": word_count, - "was_truncated": was_truncated, - "crawler_type": "youtube_transcript", - "author": author, - } - - -def create_scrape_webpage_tool(): - """ - Factory function to create the scrape_webpage tool. - - Returns: - A configured tool function for scraping webpages. - """ - - @tool - async def scrape_webpage( - url: str, - max_length: int = 50000, - ) -> dict[str, Any]: - """ - Scrape and extract the main content from a webpage. - - Use this tool when the user wants you to read, summarize, or answer - questions about a specific webpage's content, or to pull a site's - contact details (emails, phone numbers, social profiles) for lead or - competitive-intelligence work. This tool actually fetches and reads the - full page content; JS-rendered pages are loaded in a real browser and - auto-scrolled, so lazy-loaded listings (directories, infinite-scroll - feeds) are captured too. For YouTube video URLs it fetches the - transcript directly instead of crawling the page. - - For a single page this returns its content plus any contacts found. To - sweep a whole site (e.g. hunt a contact/privacy page for an email), use - the web.crawl capability with maxCrawlDepth > 0 instead. - - Common triggers: - - "Read this article and summarize it" - - "What does this page say about X?" - - "Summarize this blog post for me" - - "Find the contact email / socials for this company" - - "What's in this webpage?" - - Args: - url: The URL of the webpage to scrape (must be HTTP/HTTPS) - max_length: Maximum content length to return (default: 50000 chars) - - Returns: - A dictionary containing: - - id: Unique identifier for this scrape - - assetId: The URL (for deduplication) - - kind: "article" (type of content) - - href: The URL to open when clicked - - title: Page title - - description: Brief description or excerpt - - content: The extracted main content (markdown format) - - domain: The domain name - - word_count: Approximate word count - - was_truncated: Whether content was truncated - - contacts: {emails, phones, socials} harvested from the page - - links: every link on the page as {url, text, rel, kind} where - kind is internal/external/social/email/tel. Use the anchor text - to tie a link to an entity (e.g. which person a LinkedIn URL - belongs to) and to pick the next page to scrape. - - error: Error message (if scraping failed) - """ - scrape_id = generate_scrape_id(url) - domain = extract_domain(url) - - if not url.startswith(("http://", "https://")): - url = f"https://{url}" - - try: - # YouTube URLs use the transcript API instead of crawling. - video_id = get_youtube_video_id(url) - if video_id: - return await _scrape_youtube_video(url, video_id, max_length) - - connector = WebCrawlerConnector() - outcome = await connector.crawl_url(url) - - # 03d: bill any captcha attempts (even if the crawl ultimately failed). - _bill_captcha_attempts(outcome) - - if outcome.status is not CrawlOutcomeStatus.SUCCESS or not outcome.result: - return { - "id": scrape_id, - "assetId": url, - "kind": "article", - "href": url, - "title": domain or "Webpage", - "domain": domain, - "error": outcome.error or "No content returned from crawler", - } - - result = outcome.result - _bill_successful_scrape() - content = result.get("content", "") - metadata = result.get("metadata", {}) - - title = metadata.get("title", "") - if not title: - title = domain or url.split("/")[-1] or "Webpage" - - description = metadata.get("description", "") - if not description and content: - first_para = content.split("\n\n")[0] if content else "" - description = ( - first_para[:300] + "..." if len(first_para) > 300 else first_para - ) - - content, was_truncated = truncate_content(content, max_length) - word_count = len(content.split()) - - return { - "id": scrape_id, - "assetId": url, - "kind": "article", - "href": url, - "title": title, - "description": description, - "content": content, - "domain": domain, - "word_count": word_count, - "was_truncated": was_truncated, - "crawler_type": result.get("crawler_type", "unknown"), - "author": metadata.get("author"), - "date": metadata.get("date"), - # Contact/social signals from raw HTML (footer/legal boilerplate - # the markdown omits) — surfaced for lead/CI discovery. - "contacts": result.get("contacts"), - # Per-anchor inventory (url/text/rel/kind): anchor text is the - # raw material for tying targets to entities. - "links": result.get("link_records"), - } - - except Exception as e: - error_message = str(e) - logger.error(f"[scrape_webpage] Error scraping {url}: {error_message}") - return { - "id": scrape_id, - "assetId": url, - "kind": "article", - "href": url, - "title": domain or "Webpage", - "domain": domain, - "error": f"Failed to scrape: {error_message[:100]}", - } - - return scrape_webpage diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/shared/receipts/receipt.py b/surfsense_backend/app/agents/chat/multi_agent_chat/shared/receipts/receipt.py index db75ac707..569e710eb 100644 --- a/surfsense_backend/app/agents/chat/multi_agent_chat/shared/receipts/receipt.py +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/shared/receipts/receipt.py @@ -105,7 +105,7 @@ class Receipt(TypedDict, total=False): ``None`` only when the operation failed before the backend assigned one.""" verifiable_url: str | None - """URL the parent can pass to ``scrape_webpage`` to verify the + """URL the parent can crawl (via ``task(web_crawler, …)``) to verify the operation. ``None`` when no public URL exists (Gmail, KB, raw images stored in the DB).""" diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/shared/tools/catalog.py b/surfsense_backend/app/agents/chat/multi_agent_chat/shared/tools/catalog.py index e5585f296..1a11dd6d7 100644 --- a/surfsense_backend/app/agents/chat/multi_agent_chat/shared/tools/catalog.py +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/shared/tools/catalog.py @@ -64,10 +64,6 @@ TOOL_CATALOG: list[ToolMetadata] = [ name="search_knowledge_base", description="Search the user's knowledge base with hybrid semantic + keyword retrieval", ), - ToolMetadata( - name="scrape_webpage", - description="Scrape and extract the main content from a webpage", - ), ToolMetadata( name="create_automation", description="Draft an automation from an NL intent; user approves the card; tool saves", diff --git a/surfsense_backend/app/capabilities/core/access/agent.py b/surfsense_backend/app/capabilities/core/access/agent.py index 7ca446edb..e2ec39562 100644 --- a/surfsense_backend/app/capabilities/core/access/agent.py +++ b/surfsense_backend/app/capabilities/core/access/agent.py @@ -95,7 +95,7 @@ def _capability_tool(capability: Capability, workspace_id: int) -> BaseTool: raise duration_ms = int((time.perf_counter() - started) * 1000) - await charge_capability(output, unit, ctx) + cost_micros = await charge_capability(output, unit, ctx) serialized = serialize_output(output) async with async_session_maker() as rec_session: @@ -109,6 +109,7 @@ def _capability_tool(capability: Capability, workspace_id: int) -> BaseTool: input=input_dump, thread_id=thread_id, duration_ms=duration_ms, + cost_micros=cost_micros, ) if serialized.char_count <= RUN_OUTPUT_CHAR_CAP: diff --git a/surfsense_backend/app/capabilities/core/access/rest.py b/surfsense_backend/app/capabilities/core/access/rest.py index c6f50efcd..9494f1f0b 100644 --- a/surfsense_backend/app/capabilities/core/access/rest.py +++ b/surfsense_backend/app/capabilities/core/access/rest.py @@ -140,7 +140,7 @@ def _register_verb(router: APIRouter, capability: Capability) -> None: ) from exc duration_ms = int((time.perf_counter() - started) * 1000) - await charge_capability(output, unit, ctx) + cost_micros = await charge_capability(output, unit, ctx) serialized = serialize_output(output) run_id = await _record_rest_run( @@ -152,6 +152,7 @@ def _register_verb(router: APIRouter, capability: Capability) -> None: input=input_dump, user_id=user_id, duration_ms=duration_ms, + cost_micros=cost_micros, ) if run_id is not None: response.headers["X-Run-Id"] = f"run_{run_id}" diff --git a/surfsense_backend/app/capabilities/core/billing.py b/surfsense_backend/app/capabilities/core/billing.py index a1fda82b2..7445e4c18 100644 --- a/surfsense_backend/app/capabilities/core/billing.py +++ b/surfsense_backend/app/capabilities/core/billing.py @@ -13,6 +13,8 @@ from app.capabilities.core.types import ( CapabilityContext, ) from app.config import config +from app.services import wallet_credit +from app.services.platform_scrape_credit_service import PlatformScrapeCreditService from app.services.token_tracking_service import record_token_usage from app.services.web_crawl_credit_service import WebCrawlCreditService from app.utils.captcha import captcha_enabled @@ -23,6 +25,24 @@ if TYPE_CHECKING: from sqlalchemy.ext.asyncio import AsyncSession +# Each platform meter -> the config knob holding its micro-USD per-item rate. +# The rate is looked up live (not cached) so an env retune + restart takes +# effect without a code change, mirroring the crawl biller. +_PLATFORM_RATE_KEYS: dict[BillingUnit, str] = { + BillingUnit.REDDIT_ITEM: "REDDIT_SCRAPE_MICROS_PER_ITEM", + BillingUnit.GOOGLE_SEARCH_SERP: "GOOGLE_SEARCH_MICROS_PER_SERP", + BillingUnit.GOOGLE_MAPS_PLACE: "GOOGLE_MAPS_MICROS_PER_PLACE", + BillingUnit.GOOGLE_MAPS_REVIEW: "GOOGLE_MAPS_MICROS_PER_REVIEW", + BillingUnit.YOUTUBE_VIDEO: "YOUTUBE_MICROS_PER_VIDEO", + BillingUnit.YOUTUBE_COMMENT: "YOUTUBE_MICROS_PER_COMMENT", +} + + +def _platform_rate(unit: BillingUnit) -> int: + """Micro-USD per item for a platform meter, read live from config.""" + return int(getattr(config, _PLATFORM_RATE_KEYS[unit])) + + async def gate_capability( payload: BillableInput, unit: BillingUnit | None, ctx: CapabilityContext ) -> None: @@ -35,6 +55,8 @@ async def gate_capability( return if unit is BillingUnit.WEB_CRAWL: await _gate_web_crawl(ctx, payload.estimated_units) + return + await _gate_platform(payload, unit, ctx) async def _gate_web_crawl(ctx: CapabilityContext, estimated_successes: int) -> None: @@ -62,62 +84,145 @@ async def _gate_web_crawl(ctx: CapabilityContext, estimated_successes: int) -> N await service.check_balance(owner_user_id, required_micros) -async def charge_capability( - output: BillableOutput, unit: BillingUnit | None, ctx: CapabilityContext +async def _gate_platform( + payload: BillableInput, unit: BillingUnit, ctx: CapabilityContext ) -> None: - """Bill the workspace owner for this result's billable successes (03c). ``None`` = free. + """Reserve the worst-case per-item cost for a platform scraper verb. - For crawl-backed verbs this also bills any captcha *attempts* (Phase 3d) as a - separate per-attempt unit — the solver charges per attempt even when the crawl - ultimately failed, so it can't ride the per-success crawl meter. + ``google_maps.scrape`` is dual-metered: it can attach reviews per place, so + its gate also reserves ``estimated_review_units`` at the review rate — same + two-meters-one-verb shape as crawl + captcha. """ - if unit is None: - return - if unit is BillingUnit.WEB_CRAWL: - await _charge_web_crawl(ctx, output.billable_units) - await _charge_captcha(ctx, getattr(output, "captcha_attempts", 0)) - - -async def _charge_web_crawl(ctx: CapabilityContext, successes: int) -> None: - if successes <= 0: - return - service = WebCrawlCreditService(ctx.session) + service = PlatformScrapeCreditService(ctx.session) if not service.billing_enabled(): return owner_user_id = await _resolve_workspace_owner(ctx.session, ctx.workspace_id) if owner_user_id is None: return + + required_micros = service.items_to_micros( + payload.estimated_units, _platform_rate(unit) + ) + if unit is BillingUnit.GOOGLE_MAPS_PLACE: + review_units = getattr(payload, "estimated_review_units", 0) + required_micros += service.items_to_micros( + review_units, _platform_rate(BillingUnit.GOOGLE_MAPS_REVIEW) + ) + await wallet_credit.check_balance(ctx.session, owner_user_id, required_micros) + + +async def charge_capability( + output: BillableOutput, unit: BillingUnit | None, ctx: CapabilityContext +) -> int: + """Bill the workspace owner for this result and return the micros charged. + + For crawl-backed verbs this also bills any captcha *attempts* (Phase 3d) as a + separate per-attempt unit — the solver charges per attempt even when the crawl + ultimately failed, so it can't ride the per-success crawl meter. Platform + verbs bill per item returned; ``google_maps.scrape`` additionally bills its + attached reviews. ``None`` unit = free = returns 0. + + The returned total lets the doors persist a per-run ``cost_micros``. + """ + if unit is None: + return 0 + if unit is BillingUnit.WEB_CRAWL: + charged = await _charge_web_crawl(ctx, output.billable_units) + charged += await _charge_captcha(ctx, getattr(output, "captcha_attempts", 0)) + return charged + return await _charge_platform(output, unit, ctx) + + +async def _charge_web_crawl(ctx: CapabilityContext, successes: int) -> int: + if successes <= 0: + return 0 + service = WebCrawlCreditService(ctx.session) + if not service.billing_enabled(): + return 0 + owner_user_id = await _resolve_workspace_owner(ctx.session, ctx.workspace_id) + if owner_user_id is None: + return 0 + cost_micros = service.successes_to_micros(successes) # Stage the audit row before charge_credits' commit flushes both. await record_token_usage( ctx.session, usage_type="web_crawl", workspace_id=ctx.workspace_id, user_id=owner_user_id, - cost_micros=service.successes_to_micros(successes), + cost_micros=cost_micros, call_details={"successes": successes}, ) await service.charge_credits(owner_user_id, successes) + return cost_micros -async def _charge_captcha(ctx: CapabilityContext, attempts: int) -> None: +async def _charge_captcha(ctx: CapabilityContext, attempts: int) -> int: if attempts <= 0: - return + return 0 service = WebCrawlCreditService(ctx.session) if not service.captcha_billing_enabled(): - return + return 0 owner_user_id = await _resolve_workspace_owner(ctx.session, ctx.workspace_id) if owner_user_id is None: - return + return 0 + cost_micros = service.captcha_solves_to_micros(attempts) # Stage the audit row before charge_captcha's commit flushes both. await record_token_usage( ctx.session, usage_type="web_crawl_captcha", workspace_id=ctx.workspace_id, user_id=owner_user_id, - cost_micros=service.captcha_solves_to_micros(attempts), + cost_micros=cost_micros, call_details={"attempts": attempts}, ) await service.charge_captcha(owner_user_id, attempts) + return cost_micros + + +async def _charge_platform( + output: BillableOutput, unit: BillingUnit, ctx: CapabilityContext +) -> int: + """Charge a platform verb per item; dual-meter ``google_maps.scrape`` reviews.""" + service = PlatformScrapeCreditService(ctx.session) + if not service.billing_enabled(): + return 0 + owner_user_id = await _resolve_workspace_owner(ctx.session, ctx.workspace_id) + if owner_user_id is None: + return 0 + + charged = await _charge_platform_meter( + service, ctx, owner_user_id, unit, output.billable_units + ) + if unit is BillingUnit.GOOGLE_MAPS_PLACE: + reviews = getattr(output, "attached_review_count", 0) + charged += await _charge_platform_meter( + service, ctx, owner_user_id, BillingUnit.GOOGLE_MAPS_REVIEW, reviews + ) + return charged + + +async def _charge_platform_meter( + service: PlatformScrapeCreditService, + ctx: CapabilityContext, + owner_user_id: UUID, + unit: BillingUnit, + items: int, +) -> int: + if items <= 0: + return 0 + rate = _platform_rate(unit) + cost_micros = service.items_to_micros(items, rate) + # Stage the audit row before charge's commit flushes both. + await record_token_usage( + ctx.session, + usage_type=unit.value, + workspace_id=ctx.workspace_id, + user_id=owner_user_id, + cost_micros=cost_micros, + call_details={"items": items}, + ) + await service.charge(owner_user_id, items, rate) + return cost_micros async def _resolve_workspace_owner( diff --git a/surfsense_backend/app/capabilities/core/types.py b/surfsense_backend/app/capabilities/core/types.py index 14c1f17e7..ec44d29aa 100644 --- a/surfsense_backend/app/capabilities/core/types.py +++ b/surfsense_backend/app/capabilities/core/types.py @@ -13,9 +13,18 @@ if TYPE_CHECKING: class BillingUnit(StrEnum): - """The meter a verb charges on (priced by the billing service, 03c). ``None`` = free.""" + """The meter a verb charges on (priced by the billing service, 03c). ``None`` = free. + + Each value doubles as the ``TokenUsage.usage_type`` audit string for that meter. + """ WEB_CRAWL = "web_crawl" + REDDIT_ITEM = "reddit_item" + GOOGLE_SEARCH_SERP = "google_search_serp" + GOOGLE_MAPS_PLACE = "google_maps_place" + GOOGLE_MAPS_REVIEW = "google_maps_review" + YOUTUBE_VIDEO = "youtube_video" + YOUTUBE_COMMENT = "youtube_comment" class BillableInput(Protocol): diff --git a/surfsense_backend/app/capabilities/google_maps/reviews/definition.py b/surfsense_backend/app/capabilities/google_maps/reviews/definition.py index 384e01577..e96c6f906 100644 --- a/surfsense_backend/app/capabilities/google_maps/reviews/definition.py +++ b/surfsense_backend/app/capabilities/google_maps/reviews/definition.py @@ -1,8 +1,9 @@ -"""``google_maps.reviews`` capability registration (free — see 04-capabilities open item).""" +"""``google_maps.reviews`` capability registration (billed per review; see +config ``GOOGLE_MAPS_MICROS_PER_REVIEW``).""" from __future__ import annotations -from app.capabilities.core import Capability, register_capability +from app.capabilities.core import BillingUnit, Capability, register_capability from app.capabilities.google_maps.reviews.executor import build_reviews_executor from app.capabilities.google_maps.reviews.schemas import ReviewsInput, ReviewsOutput @@ -17,7 +18,7 @@ GOOGLE_MAPS_REVIEWS = Capability( input_schema=ReviewsInput, output_schema=ReviewsOutput, executor=build_reviews_executor(), - billing_unit=None, + billing_unit=BillingUnit.GOOGLE_MAPS_REVIEW, ) register_capability(GOOGLE_MAPS_REVIEWS) diff --git a/surfsense_backend/app/capabilities/google_maps/reviews/schemas.py b/surfsense_backend/app/capabilities/google_maps/reviews/schemas.py index 757ca2ffd..1868765e9 100644 --- a/surfsense_backend/app/capabilities/google_maps/reviews/schemas.py +++ b/surfsense_backend/app/capabilities/google_maps/reviews/schemas.py @@ -60,9 +60,20 @@ class ReviewsInput(BaseModel): raise ValueError("Provide at least one of 'urls' or 'place_ids'.") return self + @property + def estimated_units(self) -> int: + """Worst-case billable reviews for the pre-flight gate: up to + ``max_reviews`` per source place.""" + return (len(self.urls) + len(self.place_ids)) * self.max_reviews + class ReviewsOutput(BaseModel): items: list[ReviewItem] = Field( default_factory=list, description="One item per review, in the scraper's emission order.", ) + + @property + def billable_units(self) -> int: + """One returned review = one billable unit.""" + return len(self.items) diff --git a/surfsense_backend/app/capabilities/google_maps/scrape/definition.py b/surfsense_backend/app/capabilities/google_maps/scrape/definition.py index 9b6b0d2f1..035b7e00d 100644 --- a/surfsense_backend/app/capabilities/google_maps/scrape/definition.py +++ b/surfsense_backend/app/capabilities/google_maps/scrape/definition.py @@ -1,8 +1,10 @@ -"""``google_maps.scrape`` capability registration (free — see 04-capabilities open item).""" +"""``google_maps.scrape`` capability registration (dual-metered: billed per +place via ``GOOGLE_MAPS_MICROS_PER_PLACE`` plus per attached review via +``GOOGLE_MAPS_MICROS_PER_REVIEW``).""" from __future__ import annotations -from app.capabilities.core import Capability, register_capability +from app.capabilities.core import BillingUnit, Capability, register_capability from app.capabilities.google_maps.scrape.executor import build_scrape_executor from app.capabilities.google_maps.scrape.schemas import ScrapeInput, ScrapeOutput @@ -19,7 +21,7 @@ GOOGLE_MAPS_SCRAPE = Capability( input_schema=ScrapeInput, output_schema=ScrapeOutput, executor=build_scrape_executor(), - billing_unit=None, + billing_unit=BillingUnit.GOOGLE_MAPS_PLACE, ) register_capability(GOOGLE_MAPS_SCRAPE) diff --git a/surfsense_backend/app/capabilities/google_maps/scrape/schemas.py b/surfsense_backend/app/capabilities/google_maps/scrape/schemas.py index 49082a120..95833c8f7 100644 --- a/surfsense_backend/app/capabilities/google_maps/scrape/schemas.py +++ b/surfsense_backend/app/capabilities/google_maps/scrape/schemas.py @@ -83,9 +83,36 @@ class ScrapeInput(BaseModel): ) return self + @property + def estimated_units(self) -> int: + """Worst-case billable places: each search query yields up to + ``max_places``; direct URLs and place_ids yield one each.""" + return ( + len(self.search_queries) * self.max_places + + len(self.urls) + + len(self.place_ids) + ) + + @property + def estimated_review_units(self) -> int: + """Worst-case attached reviews for the dual-meter gate: up to + ``max_reviews`` per place across the worst-case place fan-out.""" + return self.estimated_units * self.max_reviews + class ScrapeOutput(BaseModel): items: list[PlaceItem] = Field( default_factory=list, description="One place item per result, in the scraper's emission order.", ) + + @property + def billable_units(self) -> int: + """One returned place = one billable place unit.""" + return len(self.items) + + @property + def attached_review_count(self) -> int: + """Reviews attached inline across all places — the second (review) + meter for this dual-metered verb (populated only when max_reviews > 0).""" + return sum(len(item.reviews) for item in self.items) diff --git a/surfsense_backend/app/capabilities/google_search/scrape/definition.py b/surfsense_backend/app/capabilities/google_search/scrape/definition.py index 9ef0f0dd6..2f2c9de03 100644 --- a/surfsense_backend/app/capabilities/google_search/scrape/definition.py +++ b/surfsense_backend/app/capabilities/google_search/scrape/definition.py @@ -1,8 +1,9 @@ -"""``google_search.scrape`` capability registration (free — see 04-capabilities open item).""" +"""``google_search.scrape`` capability registration (billed per SERP page; see +config ``GOOGLE_SEARCH_MICROS_PER_SERP``).""" from __future__ import annotations -from app.capabilities.core import Capability, register_capability +from app.capabilities.core import BillingUnit, Capability, register_capability from app.capabilities.google_search.scrape.executor import build_scrape_executor from app.capabilities.google_search.scrape.schemas import ScrapeInput, ScrapeOutput @@ -18,7 +19,7 @@ GOOGLE_SEARCH_SCRAPE = Capability( input_schema=ScrapeInput, output_schema=ScrapeOutput, executor=build_scrape_executor(), - billing_unit=None, + billing_unit=BillingUnit.GOOGLE_SEARCH_SERP, ) register_capability(GOOGLE_SEARCH_SCRAPE) diff --git a/surfsense_backend/app/capabilities/google_search/scrape/schemas.py b/surfsense_backend/app/capabilities/google_search/scrape/schemas.py index 5fdb52e24..f7224be9c 100644 --- a/surfsense_backend/app/capabilities/google_search/scrape/schemas.py +++ b/surfsense_backend/app/capabilities/google_search/scrape/schemas.py @@ -47,9 +47,20 @@ class ScrapeInput(BaseModel): description="Restrict results to a single domain, e.g. 'example.com'.", ) + @property + def estimated_units(self) -> int: + """Worst-case billable SERP pages for the pre-flight gate: one page per + query per requested result page.""" + return len(self.queries) * self.max_pages_per_query + class ScrapeOutput(BaseModel): items: list[SerpItem] = Field( default_factory=list, description="One item per fetched SERP page, in the scraper's emission order.", ) + + @property + def billable_units(self) -> int: + """One fetched SERP page = one billable unit.""" + return len(self.items) diff --git a/surfsense_backend/app/capabilities/reddit/scrape/definition.py b/surfsense_backend/app/capabilities/reddit/scrape/definition.py index f3e868744..01fb6db5f 100644 --- a/surfsense_backend/app/capabilities/reddit/scrape/definition.py +++ b/surfsense_backend/app/capabilities/reddit/scrape/definition.py @@ -1,8 +1,9 @@ -"""``reddit.scrape`` capability registration (free — see 04-capabilities open item).""" +"""``reddit.scrape`` capability registration (billed per item; see config +``REDDIT_SCRAPE_MICROS_PER_ITEM``).""" from __future__ import annotations -from app.capabilities.core import Capability, register_capability +from app.capabilities.core import BillingUnit, Capability, register_capability from app.capabilities.reddit.scrape.executor import build_scrape_executor from app.capabilities.reddit.scrape.schemas import ScrapeInput, ScrapeOutput @@ -18,7 +19,7 @@ REDDIT_SCRAPE = Capability( input_schema=ScrapeInput, output_schema=ScrapeOutput, executor=build_scrape_executor(), - billing_unit=None, + billing_unit=BillingUnit.REDDIT_ITEM, ) register_capability(REDDIT_SCRAPE) diff --git a/surfsense_backend/app/capabilities/reddit/scrape/schemas.py b/surfsense_backend/app/capabilities/reddit/scrape/schemas.py index 6fbc32fec..0c2da536e 100644 --- a/surfsense_backend/app/capabilities/reddit/scrape/schemas.py +++ b/surfsense_backend/app/capabilities/reddit/scrape/schemas.py @@ -94,9 +94,20 @@ class ScrapeInput(BaseModel): ) return self + @property + def estimated_units(self) -> int: + """Worst-case billable items for the pre-flight gate: ``max_items`` is a + hard cross-source ceiling (le=100), so no call can exceed it.""" + return self.max_items + class ScrapeOutput(BaseModel): items: list[RedditItem] = Field( default_factory=list, description="One item per result (post/comment/community/user), in emission order.", ) + + @property + def billable_units(self) -> int: + """One returned item = one billable unit.""" + return len(self.items) diff --git a/surfsense_backend/app/capabilities/youtube/comments/definition.py b/surfsense_backend/app/capabilities/youtube/comments/definition.py index 7ed3fb345..ec689a23c 100644 --- a/surfsense_backend/app/capabilities/youtube/comments/definition.py +++ b/surfsense_backend/app/capabilities/youtube/comments/definition.py @@ -1,8 +1,9 @@ -"""``youtube.comments`` capability registration (free — see 04-capabilities open item).""" +"""``youtube.comments`` capability registration (billed per comment; see config +``YOUTUBE_MICROS_PER_COMMENT``).""" from __future__ import annotations -from app.capabilities.core import Capability, register_capability +from app.capabilities.core import BillingUnit, Capability, register_capability from app.capabilities.youtube.comments.executor import build_comments_executor from app.capabilities.youtube.comments.schemas import CommentsInput, CommentsOutput @@ -17,7 +18,7 @@ YOUTUBE_COMMENTS = Capability( input_schema=CommentsInput, output_schema=CommentsOutput, executor=build_comments_executor(), - billing_unit=None, + billing_unit=BillingUnit.YOUTUBE_COMMENT, ) register_capability(YOUTUBE_COMMENTS) diff --git a/surfsense_backend/app/capabilities/youtube/comments/schemas.py b/surfsense_backend/app/capabilities/youtube/comments/schemas.py index e72181a32..a9da5b7a2 100644 --- a/surfsense_backend/app/capabilities/youtube/comments/schemas.py +++ b/surfsense_backend/app/capabilities/youtube/comments/schemas.py @@ -36,9 +36,20 @@ class CommentsInput(BaseModel): description="Comment ordering: most-liked first, or most-recent first.", ) + @property + def estimated_units(self) -> int: + """Worst-case billable comments for the pre-flight gate: up to + ``max_comments`` per video URL.""" + return len(self.urls) * self.max_comments + class CommentsOutput(BaseModel): items: list[CommentItem] = Field( default_factory=list, description="One item per comment or reply, in the scraper's emission order.", ) + + @property + def billable_units(self) -> int: + """One returned comment or reply = one billable unit.""" + return len(self.items) diff --git a/surfsense_backend/app/capabilities/youtube/scrape/definition.py b/surfsense_backend/app/capabilities/youtube/scrape/definition.py index a357f809b..43874e254 100644 --- a/surfsense_backend/app/capabilities/youtube/scrape/definition.py +++ b/surfsense_backend/app/capabilities/youtube/scrape/definition.py @@ -1,8 +1,9 @@ -"""``youtube.scrape`` capability registration (free — see 04-capabilities open item).""" +"""``youtube.scrape`` capability registration (billed per video; see config +``YOUTUBE_MICROS_PER_VIDEO``).""" from __future__ import annotations -from app.capabilities.core import Capability, register_capability +from app.capabilities.core import BillingUnit, Capability, register_capability from app.capabilities.youtube.scrape.executor import build_scrape_executor from app.capabilities.youtube.scrape.schemas import ScrapeInput, ScrapeOutput @@ -18,7 +19,7 @@ YOUTUBE_SCRAPE = Capability( input_schema=ScrapeInput, output_schema=ScrapeOutput, executor=build_scrape_executor(), - billing_unit=None, + billing_unit=BillingUnit.YOUTUBE_VIDEO, ) register_capability(YOUTUBE_SCRAPE) diff --git a/surfsense_backend/app/capabilities/youtube/scrape/schemas.py b/surfsense_backend/app/capabilities/youtube/scrape/schemas.py index 8947f601f..4d80769b6 100644 --- a/surfsense_backend/app/capabilities/youtube/scrape/schemas.py +++ b/surfsense_backend/app/capabilities/youtube/scrape/schemas.py @@ -58,9 +58,25 @@ class ScrapeInput(BaseModel): raise ValueError("Provide at least one of 'urls' or 'search_queries'.") return self + @property + def estimated_units(self) -> int: + """Worst-case billable videos for the pre-flight gate. + + The x3 is load-bearing: for a channel source ``max_results`` caps + videos, shorts, and streams *independently*, so one source can return + up to 3x ``max_results`` items. Over-reserving here keeps the + never-go-negative invariant a passing gate promises. + """ + return (len(self.urls) + len(self.search_queries)) * self.max_results * 3 + class ScrapeOutput(BaseModel): items: list[VideoItem] = Field( default_factory=list, description="One video item per result, in the scraper's emission order.", ) + + @property + def billable_units(self) -> int: + """One returned video/short/stream = one billable unit.""" + return len(self.items) diff --git a/surfsense_backend/app/config/__init__.py b/surfsense_backend/app/config/__init__.py index 7305f6dff..1c39627f1 100644 --- a/surfsense_backend/app/config/__init__.py +++ b/surfsense_backend/app/config/__init__.py @@ -659,12 +659,12 @@ class Config: # ``WEB_CRAWL_MICROS_PER_SUCCESS`` is the single source of truth; retune it # to any rate with just an env change + restart (no code/migration): # WEB_CRAWL_MICROS_PER_SUCCESS = round(USD_per_1000_crawls * 1_000) - # $1/1000 -> 1000 (default) | $2/1000 -> 2000 | $0.50/1000 -> 500 + # $2/1000 -> 2000 (default) | $1/1000 -> 1000 | $0.50/1000 -> 500 WEB_CRAWL_CREDIT_BILLING_ENABLED = ( os.getenv("WEB_CRAWL_CREDIT_BILLING_ENABLED", "FALSE").upper() == "TRUE" ) WEB_CRAWL_MICROS_PER_SUCCESS = int( - os.getenv("WEB_CRAWL_MICROS_PER_SUCCESS", "1000") + os.getenv("WEB_CRAWL_MICROS_PER_SUCCESS", "2000") ) # Phase 3d captcha-solve billing. Captcha can't ride the per-success crawl @@ -682,6 +682,41 @@ class Config: os.getenv("WEB_CRAWL_CAPTCHA_MICROS_PER_SOLVE", "3000") ) + # Platform-native scraper billing (Reddit, Google Search, Google Maps, + # YouTube). Debits the credit wallet per *item returned* — the same + # per-unit model as web crawl, one meter per verb. Off by default so + # self-hosted / OSS installs keep scraping effectively-free; hosted + # deployments set this TRUE. + # + # Rates are fully config-driven (no hardcoded price). Each is micro-USD + # per item; retune with an env change + restart (no code/migration): + # = round(USD_per_1000_items * 1_000) + # $3.50/1000 -> 3500 | $5.00/1000 -> 5000 | $2.00/1000 -> 2000 + # Defaults sit at/above Apify's first-party actor rates (Jul 2026), which + # is justified because SurfSense charges no subscription tiers, no + # per-run actor-start fees, and no separate proxy/compute/storage billing. + PLATFORM_SCRAPE_BILLING_ENABLED = ( + os.getenv("PLATFORM_SCRAPE_BILLING_ENABLED", "FALSE").upper() == "TRUE" + ) + REDDIT_SCRAPE_MICROS_PER_ITEM = int( + os.getenv("REDDIT_SCRAPE_MICROS_PER_ITEM", "3500") + ) + GOOGLE_SEARCH_MICROS_PER_SERP = int( + os.getenv("GOOGLE_SEARCH_MICROS_PER_SERP", "5500") + ) + GOOGLE_MAPS_MICROS_PER_PLACE = int( + os.getenv("GOOGLE_MAPS_MICROS_PER_PLACE", "5000") + ) + GOOGLE_MAPS_MICROS_PER_REVIEW = int( + os.getenv("GOOGLE_MAPS_MICROS_PER_REVIEW", "2000") + ) + YOUTUBE_MICROS_PER_VIDEO = int(os.getenv("YOUTUBE_MICROS_PER_VIDEO", "3500")) + # Kept separate from the video rate so comments can be re-tuned toward the + # cheaper per-comment market ($0.40-2.00/1k) without touching video pricing. + YOUTUBE_MICROS_PER_COMMENT = int( + os.getenv("YOUTUBE_MICROS_PER_COMMENT", "3500") + ) + # Low-balance WARNING threshold (micro-USD). Surfaced by the quota service # so the UI can nudge the user to top up / enable auto-reload. $0.50. CREDIT_LOW_BALANCE_WARNING_MICROS = int( diff --git a/surfsense_backend/app/proprietary/README.md b/surfsense_backend/app/proprietary/README.md index 580a34fbf..b3bed791c 100644 --- a/surfsense_backend/app/proprietary/README.md +++ b/surfsense_backend/app/proprietary/README.md @@ -31,7 +31,7 @@ Apache-2.0* — instead of per-file headers scattered across the tree. - **Do not** add Apache-2.0-intended code here. - Apache-2.0 code elsewhere **may import from** this package (the indexer and the - chat `scrape_webpage` tools do); that does not move them under this license. + `web.crawl` capability do); that does not move them under this license. - Depend only on the public API exported from each subpackage's `__init__`, not on internal modules, so the boundary stays clean and swappable. - **Boundary test:** put code here only if it is used *exclusively* by the moat. diff --git a/surfsense_backend/app/proprietary/web_crawler/README.md b/surfsense_backend/app/proprietary/web_crawler/README.md index 029c9e147..2637cca53 100644 --- a/surfsense_backend/app/proprietary/web_crawler/README.md +++ b/surfsense_backend/app/proprietary/web_crawler/README.md @@ -77,8 +77,9 @@ provenance (depth, referrer). ## Agent tooling layer (outside this package) -- The main chat agent has `scrape_webpage`; the `web_crawler` subagent has the - `web.crawl` capability (single URL or site mode). +- The `web_crawler` subagent exposes the `web.crawl` capability (single URL at + `maxCrawlDepth=0`, or site mode at higher depth); the main chat agent reaches + it by delegating via `task(web_crawler, …)`. - Tool outputs over the 40k-char cap (`RUN_OUTPUT_CHAR_CAP` in `app/capabilities/core/runs.py`) are stored as JSONL runs; agents page them with `read_run` (line paging + `char_offset` for giant single items), grep diff --git a/surfsense_backend/app/services/platform_scrape_credit_service.py b/surfsense_backend/app/services/platform_scrape_credit_service.py new file mode 100644 index 000000000..a461a7f44 --- /dev/null +++ b/surfsense_backend/app/services/platform_scrape_credit_service.py @@ -0,0 +1,77 @@ +"""Charge the credit wallet per *item returned* by a platform-native scraper. + +Deliberately mirrors :class:`app.services.web_crawl_credit_service.WebCrawlCreditService`: +a simple **gate -> pre-check -> post-charge** model (no reserve/finalize) — a +scrape has no LLM token accumulator to settle against. The billable unit is one +returned item (a Reddit post/comment, a SERP page, a Maps place/review, a +YouTube video/comment); the per-item rate is passed in by the caller from the +verb's config knob so this one service serves every platform meter. + +The price is **fully config-driven** — there is no hardcoded rate here. When +``config.PLATFORM_SCRAPE_BILLING_ENABLED`` is False (the default for +self-hosted / OSS installs) every check/charge is a no-op, preserving the prior +effectively-free scraping behaviour. + +Wallet math (spendable / check / debit) is shared with the crawl biller via +:mod:`app.services.wallet_credit`. +""" + +from uuid import UUID + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import config + +# One "out of credit" type across every per-unit biller; the capability doors +# already catch exactly this one. +from app.services.etl_credit_service import InsufficientCreditsError +from app.services import wallet_credit + +__all__ = ["InsufficientCreditsError", "PlatformScrapeCreditService"] + + +class PlatformScrapeCreditService: + """Checks and charges the credit wallet for platform scraper items.""" + + def __init__(self, session: AsyncSession): + self.session = session + + @staticmethod + def billing_enabled() -> bool: + return config.PLATFORM_SCRAPE_BILLING_ENABLED + + @staticmethod + def items_to_micros(items: int, rate_micros: int) -> int: + """Convert an item count to USD micro-credits at ``rate_micros``/item.""" + return int(items) * int(rate_micros) + + async def check_credits( + self, user_id: str | UUID, estimated_items: int, rate_micros: int + ) -> None: + """Raise :class:`InsufficientCreditsError` if the wallet can't afford + ``estimated_items`` at ``rate_micros`` each. + + No-op when platform billing is disabled. ``estimated_items`` is a safe + upper bound (the request's worst-case fan-out) so a passing pre-flight + guarantees the wallet can never go negative. + """ + if not config.PLATFORM_SCRAPE_BILLING_ENABLED: + return + required = self.items_to_micros(estimated_items, rate_micros) + await wallet_credit.check_balance(self.session, user_id, required) + + async def charge( + self, user_id: str | UUID, items: int, rate_micros: int + ) -> int | None: + """Debit the wallet for ``items`` returned at ``rate_micros`` each. + + No-op when platform billing is disabled or ``items <= 0``. Returns the + new balance in micros, or ``None`` when nothing was charged. + """ + if not config.PLATFORM_SCRAPE_BILLING_ENABLED: + return None + if items <= 0: + return None + return await wallet_credit.apply_debit( + self.session, user_id, self.items_to_micros(items, rate_micros) + ) diff --git a/surfsense_backend/app/services/wallet_credit.py b/surfsense_backend/app/services/wallet_credit.py new file mode 100644 index 000000000..218dd777f --- /dev/null +++ b/surfsense_backend/app/services/wallet_credit.py @@ -0,0 +1,99 @@ +"""Shared credit-wallet primitives for the flat-rate per-unit billers. + +Both :class:`app.services.web_crawl_credit_service.WebCrawlCreditService` and +:class:`app.services.platform_scrape_credit_service.PlatformScrapeCreditService` +follow the same gate -> pre-check -> post-charge model against the unified +``User.credit_micros_*`` wallet. The wallet math lives here once instead of +being copied per service: + +- :func:`spendable_micros` — ``balance - reserved`` (ungated by any flag) +- :func:`check_balance` — raise :class:`InsufficientCreditsError` if short +- :func:`apply_debit` — debit + commit + best-effort auto-reload + +``InsufficientCreditsError`` is re-exported from ``etl_credit_service`` so every +per-unit biller (ETL, crawl, platform scrape) shares one "out of credit" type — +the capability doors already catch exactly that one. +""" + +from uuid import UUID + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.services.etl_credit_service import InsufficientCreditsError + +__all__ = ["InsufficientCreditsError", "apply_debit", "check_balance", "spendable_micros"] + + +async def spendable_micros(session: AsyncSession, user_id: str | UUID) -> int: + """Raw ``balance - reserved`` read, **ungated** by any billing flag.""" + from app.db import User + + result = await session.execute( + select(User.credit_micros_balance, User.credit_micros_reserved).where( + User.id == user_id + ) + ) + row = result.first() + if not row: + raise ValueError(f"User with ID {user_id} not found") + + balance, reserved = row + return balance - reserved + + +async def check_balance( + session: AsyncSession, user_id: str | UUID, required_micros: int +) -> None: + """Raise :class:`InsufficientCreditsError` if the wallet can't cover + ``required_micros``. Generic and **ungated** — the caller decides when at + least one relevant biller is enabled. No-op for a non-positive requirement. + """ + if required_micros <= 0: + return + available = await spendable_micros(session, user_id) + if required_micros > available: + raise InsufficientCreditsError( + message=( + "This run would exceed your available credit. " + f"Available: ${available / 1_000_000:.2f}, " + f"estimated need: ${required_micros / 1_000_000:.2f}. " + "Add more credits to continue." + ), + balance_micros=available, + required_micros=required_micros, + ) + + +async def apply_debit( + session: AsyncSession, user_id: str | UUID, cost_micros: int +) -> int | None: + """Debit ``cost_micros`` from the wallet and commit. + + Flushes any audit row the caller staged before this, then fires a + best-effort auto-reload check. No-op for a non-positive cost; returns the + new balance in micros, or ``None`` when nothing was charged. + """ + if cost_micros <= 0: + return None + + from app.db import User + + result = await session.execute(select(User).where(User.id == user_id)) + user = result.unique().scalar_one_or_none() + if not user: + raise ValueError(f"User with ID {user_id} not found") + + user.credit_micros_balance -= cost_micros + await session.commit() + await session.refresh(user) + + # Best-effort: fire an auto-reload check if the balance dropped low. + try: + from app.services.auto_reload_service import maybe_trigger_auto_reload + + await maybe_trigger_auto_reload(user_id) + except Exception: + pass + + return user.credit_micros_balance diff --git a/surfsense_backend/app/services/web_crawl_credit_service.py b/surfsense_backend/app/services/web_crawl_credit_service.py index 368d8894d..4c198cd45 100644 --- a/surfsense_backend/app/services/web_crawl_credit_service.py +++ b/surfsense_backend/app/services/web_crawl_credit_service.py @@ -13,14 +13,13 @@ for self-hosted / OSS installs) every check/charge is a no-op, preserving the prior effectively-free crawl behaviour. ``billing_enabled()`` and ``successes_to_micros()`` are exposed as static -helpers so the chat ``scrape_webpage`` tools can share the flag/price math: -they fold a single success into the current chat turn's existing bill (via the -turn accumulator) instead of debiting the wallet directly. +helpers so the capability biller (``app/capabilities/core/billing.py``, the +``web.crawl`` verb) can share the flag/price math when gating and charging a +crawl run. """ from uuid import UUID -from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.config import config @@ -29,6 +28,10 @@ from app.config import config # to catch for "out of credit" across every per-unit wallet biller. from app.services.etl_credit_service import InsufficientCreditsError +# Wallet math (spendable / check / debit) is shared with the platform-scrape +# biller — see app/services/wallet_credit.py. +from app.services import wallet_credit + __all__ = ["InsufficientCreditsError", "WebCrawlCreditService"] @@ -76,19 +79,7 @@ class WebCrawlCreditService: Used by :meth:`check_balance` for combined (crawl + captcha) pre-flight, where the relevant gate is decided by the caller, not by a single flag. """ - from app.db import User - - result = await self.session.execute( - select(User.credit_micros_balance, User.credit_micros_reserved).where( - User.id == user_id - ) - ) - row = result.first() - if not row: - raise ValueError(f"User with ID {user_id} not found") - - balance, reserved = row - return balance - reserved + return await wallet_credit.spendable_micros(self.session, user_id) async def get_available_micros(self, user_id: str | UUID) -> int | None: """Return spendable credit in micro-USD (``balance - reserved``). @@ -108,20 +99,7 @@ class WebCrawlCreditService: whichever billers are enabled and only calls this when at least one is. No-op for a non-positive requirement. """ - if required_micros <= 0: - return - available = await self._spendable_micros(user_id) - if required_micros > available: - raise InsufficientCreditsError( - message=( - "This run would exceed your available credit. " - f"Available: ${available / 1_000_000:.2f}, " - f"estimated need: ${required_micros / 1_000_000:.2f}. " - "Add more credits to continue." - ), - balance_micros=available, - required_micros=required_micros, - ) + await wallet_credit.check_balance(self.session, user_id, required_micros) async def check_credits( self, user_id: str | UUID, estimated_successes: int = 1 @@ -160,29 +138,7 @@ class WebCrawlCreditService: Mirrors ``EtlCreditService.charge_credits``' commit-then-refresh + best-effort auto-reload. No-op for a non-positive cost. """ - if cost_micros <= 0: - return None - - from app.db import User - - result = await self.session.execute(select(User).where(User.id == user_id)) - user = result.unique().scalar_one_or_none() - if not user: - raise ValueError(f"User with ID {user_id} not found") - - user.credit_micros_balance -= cost_micros - await self.session.commit() - await self.session.refresh(user) - - # Best-effort: fire an auto-reload check if the balance dropped low. - try: - from app.services.auto_reload_service import maybe_trigger_auto_reload - - await maybe_trigger_auto_reload(user_id) - except Exception: - pass - - return user.credit_micros_balance + return await wallet_credit.apply_debit(self.session, user_id, cost_micros) async def charge_credits(self, user_id: str | UUID, successes: int) -> int | None: """Debit the wallet for ``successes`` successful crawls. diff --git a/surfsense_backend/app/tasks/chat/streaming/handlers/tools/scrape_webpage/emission.py b/surfsense_backend/app/tasks/chat/streaming/handlers/tools/scrape_webpage/emission.py deleted file mode 100644 index 293d2a1e9..000000000 --- a/surfsense_backend/app/tasks/chat/streaming/handlers/tools/scrape_webpage/emission.py +++ /dev/null @@ -1,43 +0,0 @@ -"""scrape_webpage: redacted payload + terminal summary.""" - -from __future__ import annotations - -from collections.abc import Iterator - -from app.tasks.chat.streaming.handlers.tools.emission_context import ( - ToolCompletionEmissionContext, -) - - -def iter_completion_emission_frames( - ctx: ToolCompletionEmissionContext, -) -> Iterator[str]: - out = ctx.tool_output - if isinstance(out, dict): - display_output = {k: v for k, v in out.items() if k != "content"} - if "content" in out: - content = out.get("content", "") - display_output["content_preview"] = ( - content[:500] + "..." if len(content) > 500 else content - ) - yield ctx.emit_tool_output_card(display_output) - else: - yield ctx.emit_tool_output_card({"result": out}) - - if isinstance(out, dict) and "error" not in out: - title = out.get("title", "Webpage") - word_count = out.get("word_count", 0) - yield ctx.streaming_service.format_terminal_info( - f"Scraped: {title[:40]}{'...' if len(title) > 40 else ''} ({word_count:,} words)", - "success", - ) - else: - error_msg = ( - out.get("error", "Failed to scrape") - if isinstance(out, dict) - else "Failed to scrape" - ) - yield ctx.streaming_service.format_terminal_info( - f"Scrape failed: {error_msg}", - "error", - ) diff --git a/surfsense_backend/app/tasks/chat/streaming/handlers/tools/scrape_webpage/shared/__init__.py b/surfsense_backend/app/tasks/chat/streaming/handlers/tools/scrape_webpage/shared/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/surfsense_backend/app/tasks/chat/streaming/handlers/tools/scrape_webpage/shared/tool_input.py b/surfsense_backend/app/tasks/chat/streaming/handlers/tools/scrape_webpage/shared/tool_input.py deleted file mode 100644 index 581f0e64a..000000000 --- a/surfsense_backend/app/tasks/chat/streaming/handlers/tools/scrape_webpage/shared/tool_input.py +++ /dev/null @@ -1,9 +0,0 @@ -"""Tool-call args for scrape_webpage thinking.""" - -from __future__ import annotations - -from typing import Any - - -def as_tool_input_dict(tool_input: Any) -> dict[str, Any]: - return tool_input if isinstance(tool_input, dict) else {} diff --git a/surfsense_backend/app/tasks/chat/streaming/handlers/tools/scrape_webpage/thinking.py b/surfsense_backend/app/tasks/chat/streaming/handlers/tools/scrape_webpage/thinking.py deleted file mode 100644 index 8a04acbe6..000000000 --- a/surfsense_backend/app/tasks/chat/streaming/handlers/tools/scrape_webpage/thinking.py +++ /dev/null @@ -1,49 +0,0 @@ -"""scrape_webpage: thinking-step copy.""" - -from __future__ import annotations - -from typing import Any - -from app.tasks.chat.streaming.handlers.tools.scrape_webpage.shared.tool_input import ( - as_tool_input_dict, -) -from app.tasks.chat.streaming.handlers.tools.shared.model import ( - ToolStartThinking, -) - - -def resolve_start_thinking(tool_name: str, tool_input: Any) -> ToolStartThinking: - del tool_name - d = as_tool_input_dict(tool_input) - url = d.get("url", "") if isinstance(tool_input, dict) else str(tool_input) - return ToolStartThinking( - title="Scraping webpage", - items=[f"URL: {url[:80]}{'...' if len(url) > 80 else ''}"], - ) - - -def resolve_completed_thinking( - tool_name: str, - tool_output: Any, - last_items: list[str], -) -> tuple[str, list[str]]: - del tool_name - items = last_items - if isinstance(tool_output, dict): - title = tool_output.get("title", "Webpage") - word_count = tool_output.get("word_count", 0) - has_error = "error" in tool_output - if has_error: - completed = [ - *items, - f"Error: {tool_output.get('error', 'Failed to scrape')[:50]}", - ] - else: - completed = [ - *items, - f"Title: {title[:50]}{'...' if len(title) > 50 else ''}", - f"Extracted: {word_count:,} words", - ] - else: - completed = [*items, "Content extracted"] - return ("Scraping webpage", completed) diff --git a/surfsense_backend/scripts/e2e_phase3_crawl_billing.py b/surfsense_backend/scripts/e2e_phase3_crawl_billing.py index 31b8c832b..c60cd0927 100644 --- a/surfsense_backend/scripts/e2e_phase3_crawl_billing.py +++ b/surfsense_backend/scripts/e2e_phase3_crawl_billing.py @@ -1,4 +1,4 @@ -"""Manual functional e2e for Phase 3 crawler + billing (3a / 3b / 3c). +"""Manual functional e2e for Phase 3 crawler core (3a / 3b). Run from the backend directory: cd surfsense_backend @@ -8,8 +8,10 @@ Run from the backend directory: What it exercises (everything REAL — live network, live proxy, live DB reads): Stage 1 (3a + 3b) — direct fetch + proxy egress-IP proof + crawl_url ladder. - Stage 2 (3c chat surface) — the scrape_webpage tool folds one successful - crawl into the current chat turn's accumulator (billed at finalize). + +Crawl billing now lives entirely in the ``web.crawl`` capability (charged +directly on the wallet via ``charge_capability``); there is no longer a +chat-turn "fold" surface to exercise here. This is NOT a pytest test (it needs a live stack + proxy creds + network). It is the manual functional counterpart to the unit suites; the undetectability / @@ -32,8 +34,6 @@ for _candidate in (_BACKEND_ROOT / ".env", _BACKEND_ROOT.parent / ".env"): break -from app.config import config # noqa: E402 - # Content-rich, generally crawl-friendly targets (real extraction expected). _ARTICLE_URLS = [ "https://en.wikipedia.org/wiki/Competitive_intelligence", @@ -117,44 +117,10 @@ async def stage1_crawl_and_proxy() -> bool: return ok -# =========================================================================== -# Stage 2 — chat scrape folds cost into the turn accumulator (3c surface 2) -# =========================================================================== -async def stage2_chat_fold() -> bool: - _hr("STAGE 2 — chat scrape_webpage folds crawl cost into turn (3c)") - config.WEB_CRAWL_CREDIT_BILLING_ENABLED = True - price = config.WEB_CRAWL_MICROS_PER_SUCCESS - - from app.agents.chat.multi_agent_chat.main_agent.tools.scrape_webpage import ( - create_scrape_webpage_tool, - ) - from app.services.token_tracking_service import start_turn - - acc = start_turn() - tool = create_scrape_webpage_tool() - result = await tool.ainvoke({"url": _ARTICLE_URLS[0]}) - crawled_ok = "error" not in result and bool(result.get("content")) - print(f" scrape error : {result.get('error', '')}") - print(f" turn cost_micros : {acc.total_cost_micros}") - print(f" call kinds : {[c.call_kind for c in acc.calls]}") - if not crawled_ok: - print(" [INFO] crawl did not succeed (site/proxy) — cannot assert fold") - return False - return _check( - "one web_crawl line folded at configured price", - acc.total_cost_micros == price - and any(c.call_kind == "web_crawl" for c in acc.calls), - f"expected={price} got={acc.total_cost_micros}", - ) - - async def main() -> int: - print("Phase 3 functional e2e (3a/3b/3c) — live network + proxy, DB rolled back") + print("Phase 3 functional e2e (3a/3b) — live network + proxy, DB rolled back") results: dict[str, bool] = {} - for name, coro in ( - ("Stage 1 crawl+proxy", stage1_crawl_and_proxy), - ("Stage 2 chat fold", stage2_chat_fold), - ): + for name, coro in (("Stage 1 crawl+proxy", stage1_crawl_and_proxy),): try: results[name] = await coro() except Exception as exc: diff --git a/surfsense_backend/tests/unit/agents/multi_agent_chat/test_prompt_resources.py b/surfsense_backend/tests/unit/agents/multi_agent_chat/test_prompt_resources.py index 65fbe0ec1..bbe64fad3 100644 --- a/surfsense_backend/tests/unit/agents/multi_agent_chat/test_prompt_resources.py +++ b/surfsense_backend/tests/unit/agents/multi_agent_chat/test_prompt_resources.py @@ -45,7 +45,7 @@ def test_every_subagent_has_description_md(name: str): [ "core_behavior.md", "routing.md", - "tools/scrape_webpage/description.md", + "tools/task/description.md", ], ) def test_main_agent_prompt_fragments_resolve(filename: str): diff --git a/surfsense_backend/tests/unit/agents/new_chat/test_default_permissions_layering.py b/surfsense_backend/tests/unit/agents/new_chat/test_default_permissions_layering.py index 59e687b82..8c559af4d 100644 --- a/surfsense_backend/tests/unit/agents/new_chat/test_default_permissions_layering.py +++ b/surfsense_backend/tests/unit/agents/new_chat/test_default_permissions_layering.py @@ -9,7 +9,7 @@ earliest -> latest: 3. (future) user-defined rules from the Agent Permissions UI Without #1 every read-only built-in (``ls``, ``read_file``, ``grep``, -``glob``, ``scrape_webpage`` …) defaulted to ``ask`` because +``glob`` …) defaulted to ``ask`` because ``permissions.evaluate`` returns ``ask`` when no rule matches. That caused two production-painful behaviors: @@ -58,7 +58,6 @@ class TestReadOnlyToolsAllowed: "read_file", "grep", "glob", - "scrape_webpage", "get_connected_accounts", "write_todos", "task", diff --git a/surfsense_backend/tests/unit/capabilities/test_billing.py b/surfsense_backend/tests/unit/capabilities/test_billing.py index ef5086587..28021af47 100644 --- a/surfsense_backend/tests/unit/capabilities/test_billing.py +++ b/surfsense_backend/tests/unit/capabilities/test_billing.py @@ -256,3 +256,172 @@ async def test_gate_is_noop_for_free_verb(monkeypatch): await gate_capability(CrawlInput(startUrls=["https://a.com"]), None, _ctx(session)) session.execute.assert_not_called() + + +# =================================================================== +# Platform scraper per-item billing (Reddit / Search / Maps / YouTube) +# =================================================================== + + +class _FakePlatformOutput: + """Stand-in for a verb output: only the billing-read properties matter.""" + + def __init__(self, items: int, attached_review_count: int = 0): + self._items = items + self._reviews = attached_review_count + + @property + def billable_units(self) -> int: + return self._items + + @property + def attached_review_count(self) -> int: + return self._reviews + + +class _FakePlatformInput: + """Stand-in for a verb input reporting its worst-case unit counts.""" + + def __init__(self, estimated_units: int, estimated_review_units: int = 0): + self._units = estimated_units + self._review_units = estimated_review_units + + @property + def estimated_units(self) -> int: + return self._units + + @property + def estimated_review_units(self) -> int: + return self._review_units + + +@pytest.fixture +def _enable_platform_billing(monkeypatch): + monkeypatch.setattr(config, "PLATFORM_SCRAPE_BILLING_ENABLED", True) + + +async def test_platform_charges_owner_per_item( + monkeypatch, record_usage, _enable_platform_billing +): + monkeypatch.setattr(config, "REDDIT_SCRAPE_MICROS_PER_ITEM", 3500) + session, user = _make_session(_OWNER, balance_micros=1_000_000) + + charged = await charge_capability( + _FakePlatformOutput(3), BillingUnit.REDDIT_ITEM, _ctx(session) + ) + + assert charged == 3 * 3500 + assert user.credit_micros_balance == 1_000_000 - 3 * 3500 + record_usage.assert_awaited_once() + kwargs = record_usage.await_args.kwargs + assert kwargs["usage_type"] == "reddit_item" + assert kwargs["user_id"] == _OWNER + assert kwargs["workspace_id"] == _WORKSPACE_ID + assert kwargs["cost_micros"] == 3 * 3500 + + +async def test_platform_maps_scrape_dual_meters_places_and_reviews( + monkeypatch, record_usage, _enable_platform_billing +): + monkeypatch.setattr(config, "GOOGLE_MAPS_MICROS_PER_PLACE", 5000) + monkeypatch.setattr(config, "GOOGLE_MAPS_MICROS_PER_REVIEW", 2000) + session, user = _make_session(_OWNER, balance_micros=1_000_000) + + # 2 places + 10 attached reviews -> 2*5000 + 10*2000 = 30000. + charged = await charge_capability( + _FakePlatformOutput(2, attached_review_count=10), + BillingUnit.GOOGLE_MAPS_PLACE, + _ctx(session), + ) + + assert charged == 2 * 5000 + 10 * 2000 + assert user.credit_micros_balance == 1_000_000 - 30_000 + assert record_usage.await_count == 2 + usage_types = {c.kwargs["usage_type"] for c in record_usage.await_args_list} + assert usage_types == {"google_maps_place", "google_maps_review"} + + +async def test_platform_charge_disabled_is_noop(monkeypatch, record_usage): + monkeypatch.setattr(config, "PLATFORM_SCRAPE_BILLING_ENABLED", False) + monkeypatch.setattr(config, "REDDIT_SCRAPE_MICROS_PER_ITEM", 3500) + session, user = _make_session(_OWNER, balance_micros=1_000_000) + + charged = await charge_capability( + _FakePlatformOutput(3), BillingUnit.REDDIT_ITEM, _ctx(session) + ) + + assert charged == 0 + record_usage.assert_not_awaited() + session.execute.assert_not_called() + assert user.credit_micros_balance == 1_000_000 + + +async def test_platform_no_items_is_free( + monkeypatch, record_usage, _enable_platform_billing +): + monkeypatch.setattr(config, "YOUTUBE_MICROS_PER_COMMENT", 3500) + session, user = _make_session(_OWNER, balance_micros=1_000_000) + + charged = await charge_capability( + _FakePlatformOutput(0), BillingUnit.YOUTUBE_COMMENT, _ctx(session) + ) + + assert charged == 0 + record_usage.assert_not_awaited() + assert user.credit_micros_balance == 1_000_000 + + +async def test_platform_gate_blocks_when_worst_case_exceeds_balance( + monkeypatch, _enable_platform_billing +): + monkeypatch.setattr(config, "GOOGLE_SEARCH_MICROS_PER_SERP", 5500) + session = _gate_session(_OWNER, balance_micros=6000) # affords 1 SERP, not 2 + + with pytest.raises(InsufficientCreditsError): + await gate_capability( + _FakePlatformInput(estimated_units=2), + BillingUnit.GOOGLE_SEARCH_SERP, + _ctx(session), + ) + + +async def test_platform_gate_maps_reserves_places_plus_reviews( + monkeypatch, _enable_platform_billing +): + monkeypatch.setattr(config, "GOOGLE_MAPS_MICROS_PER_PLACE", 5000) + monkeypatch.setattr(config, "GOOGLE_MAPS_MICROS_PER_REVIEW", 2000) + # 1 place (5000) + 10 worst-case reviews (20000) = 25000 required. + session = _gate_session(_OWNER, balance_micros=20_000) + + with pytest.raises(InsufficientCreditsError): + await gate_capability( + _FakePlatformInput(estimated_units=1, estimated_review_units=10), + BillingUnit.GOOGLE_MAPS_PLACE, + _ctx(session), + ) + + +async def test_platform_gate_passes_when_affordable( + monkeypatch, _enable_platform_billing +): + monkeypatch.setattr(config, "GOOGLE_SEARCH_MICROS_PER_SERP", 5500) + session = _gate_session(_OWNER, balance_micros=1_000_000) + + await gate_capability( + _FakePlatformInput(estimated_units=2), + BillingUnit.GOOGLE_SEARCH_SERP, + _ctx(session), + ) + + +async def test_platform_gate_disabled_is_noop(monkeypatch): + monkeypatch.setattr(config, "PLATFORM_SCRAPE_BILLING_ENABLED", False) + session = _gate_session(_OWNER, balance_micros=0) + + await gate_capability( + _FakePlatformInput(estimated_units=1000), + BillingUnit.REDDIT_ITEM, + _ctx(session), + ) + + session.execute.assert_not_called() diff --git a/surfsense_backend/tests/unit/services/test_web_crawl_credit_service.py b/surfsense_backend/tests/unit/services/test_web_crawl_credit_service.py index 270b9a6b7..7c7cd0e40 100644 --- a/surfsense_backend/tests/unit/services/test_web_crawl_credit_service.py +++ b/surfsense_backend/tests/unit/services/test_web_crawl_credit_service.py @@ -175,48 +175,7 @@ class TestChargeCredits: # =================================================================== -# E) Chat-scrape fold helper -# =================================================================== - - -class TestChatScrapeFold: - """The scrape_webpage tools fold one success into the live turn accumulator.""" - - def _import_helper(self): - from app.agents.chat.multi_agent_chat.main_agent.tools.scrape_webpage import ( - _bill_successful_scrape, - ) - - return _bill_successful_scrape - - def test_folds_into_active_turn_when_enabled(self): - from app.services.token_tracking_service import start_turn - - acc = start_turn() - self._import_helper()() - assert acc.total_cost_micros == 1000 - assert len(acc.calls) == 1 - assert acc.calls[0].call_kind == "web_crawl" - - def test_noop_when_billing_disabled(self, monkeypatch): - from app.services.token_tracking_service import start_turn - - monkeypatch.setattr(config, "WEB_CRAWL_CREDIT_BILLING_ENABLED", False) - acc = start_turn() - self._import_helper()() - assert acc.calls == [] - assert acc.total_cost_micros == 0 - - def test_noop_when_no_active_turn(self): - import contextvars - - # Run in a fresh context so the turn ContextVar resolves to its default - # (None) regardless of other tests — must not raise. - contextvars.Context().run(self._import_helper()) - - -# =================================================================== -# F) Captcha per-attempt unit (Phase 3d) +# E) Captcha per-attempt unit (Phase 3d) # =================================================================== @@ -284,41 +243,3 @@ class TestCheckBalance: session.execute.assert_not_called() -class TestCaptchaChatFold: - """The scrape tools fold captcha attempts into the live turn accumulator.""" - - def _import_helper(self): - from app.agents.chat.multi_agent_chat.main_agent.tools.scrape_webpage import ( - _bill_captcha_attempts, - ) - - return _bill_captcha_attempts - - def _outcome(self, attempts): - o = MagicMock() - o.captcha_attempts = attempts - return o - - def test_folds_attempts_when_enabled(self, _enable_captcha_billing): - from app.services.token_tracking_service import start_turn - - acc = start_turn() - self._import_helper()(self._outcome(2)) - assert acc.total_cost_micros == 6000 - assert len(acc.calls) == 1 - assert acc.calls[0].call_kind == "web_crawl_captcha" - - def test_noop_when_zero_attempts(self, _enable_captcha_billing): - from app.services.token_tracking_service import start_turn - - acc = start_turn() - self._import_helper()(self._outcome(0)) - assert acc.calls == [] - - def test_noop_when_billing_disabled(self, monkeypatch): - from app.services.token_tracking_service import start_turn - - monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_BILLING_ENABLED", False) - acc = start_turn() - self._import_helper()(self._outcome(3)) - assert acc.calls == []