feat(nadia): Implement query-variant frontier and URL/content deduplication

Outcome: done
Changed: src/typosaurus_sandbox/research/frontier.py:1-234, src/typosaurus_sandbox/research/__init__.py:14-20,34-36,44-48
Verified by: make verify -> exit_code 0, compileall OK, 104 tests OK, "verification passed"; module smoke test passed (URL normalization, query/URL/content dedup, async get_query, snapshot accounting)
Findings:
- QueryFrontier API: seed, push_query(query, origin), push_variants_from_result (count of new queries from title/description/string extra; does not register URLs), register_url, register_content, register_result (URL+content dedup), get_query/pop_query (count queries_issued), snapshot() frozen DedupStats for closure deltas, origin_of.
- Query dedup key = whitespace-collapsed casefold; URL dedup via normalize_url (lowercase scheme/host, IDNA, strip default port/userinfo/fragment, collapse slashes); content dedup via sha256 of whitespace-normalized text.
- One threading.Lock guards all seen-sets/counters; pending queries in asyncio.Queue usable sync via pop_query and async via get_query.
- Query variant length window 2-200 chars; out-of-window dropped without touching counters.
- frontier.py imports only envelopes.SearchResult from foundation; no new dependency; retoor header, no comments/docstrings.
Open: engine leaf wires QueryFrontier into worker pool and computes per-round snapshot() deltas for closure; testwriter leaf covers frontier API.
Confidence: high - acceptance criteria exercised by smoke assertions; full suite passes via mak

Typosaurus-Run: 4e2afb673c7f4578a12276d9181b982d
Typosaurus-Node: 5e16725c0f944ca0b4373eb640430f7d
Typosaurus-Agent: @nadia
Refs: #31

src/typosaurus_sandbox/research/__init__.py

@@ -12,12 +12,21 @@ from typosaurus_sandbox.research.envelopes import (
     SearchResponse,
     SearchResult,
 )
+from typosaurus_sandbox.research.frontier import (
+    DedupStats,
+    QueryFrontier,
+    fingerprint_text,
+    normalize_url,
+    query_variants_from_result,
+)
 
 __all__ = [
     "ChatResponse",
     "ChatUsage",
+    "DedupStats",
     "DeepReport",
     "DescribeResponse",
+    "QueryFrontier",
     "RsearchClient",
     "RsearchError",
     "ResearchConfig",
@@ -25,5 +34,10 @@ __all__ = [
     "SearchResponse",
     "SearchResult",
     "TTLCache",
+    "fingerprint_text",
+    "normalize_url",
+    "query_variants_from_result",
 ]
 
+
+

src/typosaurus_sandbox/research/frontier.py

@@ -0,0 +1,226 @@
+# retoor <retoor@molodetz.nl>
+
+import asyncio
+import hashlib
+import logging
+import re
+import threading
+import urllib.parse
+from dataclasses import dataclass
+
+from typosaurus_sandbox.research.envelopes import SearchResult
+
+logger = logging.getLogger(__name__)
+
+MIN_QUERY_LENGTH = 2
+MAX_QUERY_LENGTH = 200
+DEFAULT_PORTS: dict[str, int] = {"http": 80, "https": 443}
+
+
+def _clean_text(value: str) -> str:
+    return " ".join(value.split())
+
+
+def _query_key(query: str) -> str:
+    return _clean_text(query).casefold()
+
+
+def normalize_url(url: str) -> str:
+    cleaned = _clean_text(url)
+    try:
+        parsed = urllib.parse.urlsplit(cleaned)
+    except ValueError:
+        return cleaned
+    scheme = parsed.scheme.lower()
+    if scheme not in DEFAULT_PORTS:
+        return cleaned
+    host = (parsed.hostname or "").lower()
+    if not host:
+        return cleaned
+    try:
+        host = host.encode("idna").decode("ascii")
+    except UnicodeError:
+        pass
+    port: int | None = None
+    try:
+        port = parsed.port
+    except ValueError:
+        port = None
+    if port is not None and DEFAULT_PORTS.get(scheme) == port:
+        port = None
+    display_host = f"[{host}]" if ":" in host else host
+    netloc = display_host if port is None else f"{display_host}:{port}"
+    path = re.sub(r"/{2,}", "/", parsed.path)
+    if len(path) > 1 and path.endswith("/"):
+        path = path[:-1]
+    if parsed.query:
+        return f"{scheme}://{netloc}{path}?{parsed.query}"
+    return f"{scheme}://{netloc}{path}"
+
+
+def fingerprint_text(text: str) -> str:
+    normalized = _clean_text(text)
+    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()
+
+
+def query_variants_from_result(result: SearchResult) -> list[tuple[str, str]]:
+    variants: list[tuple[str, str]] = []
+    if result.title:
+        variants.append((result.title, "title"))
+    if result.description:
+        variants.append((result.description, "description"))
+    for value in result.extra.values():
+        if isinstance(value, str) and value:
+            variants.append((value, "extra"))
+    return variants
+
+
+@dataclass(frozen=True)
+class DedupStats:
+    queries_generated: int = 0
+    queries_enqueued: int = 0
+    queries_issued: int = 0
+    queries_duplicates_skipped: int = 0
+    urls_seen: int = 0
+    urls_duplicates_skipped: int = 0
+    content_seen: int = 0
+    content_duplicates_skipped: int = 0
+
+    def to_dict(self) -> dict[str, int]:
+        return {
+            "queries_generated": self.queries_generated,
+            "queries_enqueued": self.queries_enqueued,
+            "queries_issued": self.queries_issued,
+            "queries_duplicates_skipped": self.queries_duplicates_skipped,
+            "urls_seen": self.urls_seen,
+            "urls_duplicates_skipped": self.urls_duplicates_skipped,
+            "content_seen": self.content_seen,
+            "content_duplicates_skipped": self.content_duplicates_skipped,
+        }
+
+
+class QueryFrontier:
+    def __init__(self, subject: str | None = None) -> None:
+        self._lock = threading.Lock()
+        self._seen_queries: set[str] = set()
+        self._seen_urls: set[str] = set()
+        self._seen_content: set[str] = set()
+        self._origins: dict[str, str] = {}
+        self._pending: asyncio.Queue[str] = asyncio.Queue()
+        self._queries_generated = 0
+        self._queries_enqueued = 0
+        self._queries_issued = 0
+        self._queries_duplicates_skipped = 0
+        self._urls_seen = 0
+        self._urls_duplicates_skipped = 0
+        self._content_seen = 0
+        self._content_duplicates_skipped = 0
+        if subject:
+            self.seed(subject)
+
+    def seed(self, subject: str) -> None:
+        cleaned = _clean_text(subject)
+        if cleaned:
+            self.push_query(cleaned, "seed")
+            logger.info("frontier seeded subject=%r", cleaned)
+
+    def push_query(self, query: str, origin: str = "manual") -> bool:
+        cleaned = _clean_text(query)
+        if not MIN_QUERY_LENGTH <= len(cleaned) <= MAX_QUERY_LENGTH:
+            logger.debug("query variant invalid length=%d query=%r", len(cleaned), cleaned)
+            return False
+        key = _query_key(cleaned)
+        with self._lock:
+            self._queries_generated += 1
+            if key in self._seen_queries:
+                self._queries_duplicates_skipped += 1
+                logger.debug("query duplicate skipped origin=%s query=%r", origin, cleaned)
+                return False
+            self._seen_queries.add(key)
+            self._origins[key] = origin
+            self._queries_enqueued += 1
+        self._pending.put_nowait(cleaned)
+        logger.info("query enqueued origin=%s query=%r", origin, cleaned)
+        return True
+
+    def push_variants_from_result(self, result: SearchResult) -> int:
+        new_queries = 0
+        for text, origin in query_variants_from_result(result):
+            if self.push_query(text, origin):
+                new_queries += 1
+        return new_queries
+
+    def register_url(self, url: str) -> bool:
+        if not url:
+            return False
+        normalized = normalize_url(url)
+        with self._lock:
+            if normalized in self._seen_urls:
+                self._urls_duplicates_skipped += 1
+                logger.debug("url duplicate skipped url=%s", normalized)
+                return False
+            self._seen_urls.add(normalized)
+            self._urls_seen += 1
+        logger.info("url registered url=%s", normalized)
+        return True
+
+    def register_content(self, text: str) -> bool:
+        if not text.strip():
+            return False
+        fingerprint = fingerprint_text(text)
+        with self._lock:
+            if fingerprint in self._seen_content:
+                self._content_duplicates_skipped += 1
+                logger.debug("content duplicate skipped fingerprint=%s", fingerprint)
+                return False
+            self._seen_content.add(fingerprint)
+            self._content_seen += 1
+        logger.info("content registered fingerprint=%s", fingerprint)
+        return True
+
+    def register_result(self, result: SearchResult) -> bool:
+        is_new = self.register_url(result.url)
+        if result.content:
+            self.register_content(result.content)
+        return is_new
+
+    async def get_query(self) -> str:
+        query = await self._pending.get()
+        with self._lock:
+            self._queries_issued += 1
+        logger.info("query issued query=%r", query)
+        return query
+
+    def pop_query(self) -> str | None:
+        try:
+            query = self._pending.get_nowait()
+        except asyncio.QueueEmpty:
+            return None
+        with self._lock:
+            self._queries_issued += 1
+        logger.info("query issued query=%r", query)
+        return query
+
+    def pending_count(self) -> int:
+        return self._pending.qsize()
+
+    def has_pending(self) -> bool:
+        return not self._pending.empty()
+
+    def origin_of(self, query: str) -> str | None:
+        with self._lock:
+            return self._origins.get(_query_key(query))
+
+    def snapshot(self) -> DedupStats:
+        with self._lock:
+            return DedupStats(
+                queries_generated=self._queries_generated,
+                queries_enqueued=self._queries_enqueued,
+                queries_issued=self._queries_issued,
+                queries_duplicates_skipped=self._queries_duplicates_skipped,
+                urls_seen=self._urls_seen,
+                urls_duplicates_skipped=self._urls_duplicates_skipped,
+                content_seen=self._content_seen,
+                content_duplicates_skipped=self._content_duplicates_skipped,
+            )
+