From 31f2c6451fb6e441e4e3be3e013a128c8bca63e0 Mon Sep 17 00:00:00 2001 From: typosaurus Date: Fri, 7 Aug 2026 19:16:25 +0000 Subject: [PATCH] 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 | 14 ++ src/typosaurus_sandbox/research/frontier.py | 226 ++++++++++++++++++++ 2 files changed, 240 insertions(+) create mode 100644 src/typosaurus_sandbox/research/frontier.py diff --git a/src/typosaurus_sandbox/research/__init__.py b/src/typosaurus_sandbox/research/__init__.py index 387c848..7d21db3 100644 --- a/src/typosaurus_sandbox/research/__init__.py +++ b/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", ] + + diff --git a/src/typosaurus_sandbox/research/frontier.py b/src/typosaurus_sandbox/research/frontier.py new file mode 100644 index 0000000..c8e74f4 --- /dev/null +++ b/src/typosaurus_sandbox/research/frontier.py @@ -0,0 +1,226 @@ +# retoor + +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, + ) +