@cryptotaxi247 / SquadScope-1 / commits / 8efce6c

feat: GitHub crawl shard experiment with wall-clock/API/rate-limit guardrails (#456)

Merging crawl shard experiment. Will be validated via a separate action before replacing the main crawl. Key metrics to judge: performance improvement and quality parity (map/reduce must not degrade analysis quality). The main value is unlocking more sources without context window pressure.

Juan Manuel Servera committed Jun 13, 2026 at 17:50 UTC 8efce6c9a588b5babdee1e82d72b97713c3fd744
3 files changed +1585
docs/experiments/shard-435-plan.md new
+71
@@ -0,0 +1,71 @@
1 +# Shard 435 Crawl Experiment Plan
2 +
3 +## Goal
4 +
5 +Run a local-only, no-publish experiment that compares the monolithic GitHub crawl with a sharded variant while enforcing:
6 +
7 +- at least 25% wall-clock improvement
8 +- no more than 10% GitHub API growth
9 +- no secondary-rate-limit regression
10 +- byte-stable downstream payloads after deterministic fan-in
11 +
12 +## Shard Boundaries
13 +
14 +1. **Shard 1 — new-search**
15 + - runs the new-repository search query group
16 + - in config mode, runs primary queries first and only falls back to secondary queries when needed
17 +
18 +2. **Shard 2 — trending-search**
19 + - runs the trending-repository search query group
20 + - skipped only when config mode intentionally has no trending bucket and must preserve existing semantics
21 +
22 +3. **Shard 3+ — validate-N**
23 + - consume deduplicated candidate chunks from a shared queue
24 + - perform README/detail validation and final filtering
25 + - requeue unfinished work when a shard hits its wall-clock limit so other shards can absorb the remainder
26 +
27 +## Deterministic Fan-in
28 +
29 +- keep new/trending buckets separate to preserve the canonical raw-artifact shape
30 +- dedupe within each bucket by `full_name`
31 +- sort final outputs like `crawl.py`
32 + - new repos by stars descending
33 + - trending repos by `stars_gained`, then stars descending
34 +- rebuild signals and star snapshots only after fan-in
35 +- compare canonical payload bytes after stripping volatile timestamp metadata
36 +
37 +## Guardrails
38 +
39 +- **Wall-clock budget:** each shard gets a configurable deadline (default 120s)
40 +- **API budget:** shard mode shares a global counter capped at `baseline_calls × 1.1`
41 +- **Shared backoff:** retry backoff windows are propagated across shard clients
42 +- **Rollback condition:** any secondary rate limit aborts remaining shard work and records a guardrail event
43 +
44 +## Assumptions
45 +
46 +- threading is sufficient because the workload is API I/O bound
47 +- cache reads are safe to share across shard clients
48 +- cross-bucket duplicates may still appear in both `new_repos` and `trending_repos` because the baseline crawl already allows that
49 +- at least three representative real runs are still required after implementation
50 +
51 +## Output Artifacts
52 +
53 +All experiment artifacts stay under `data/experiments/shard-435/EXPERIMENT_ID/`:
54 +
55 +- `baseline-raw.json`
56 +- `baseline-stars.json`
57 +- `shard-raw.json`
58 +- `shard-stars.json`
59 +- `report.json`
60 +
61 +## Suggested Command
62 +
63 +```bash
64 +python -m scripts.crawl_shard_experiment \
65 + --since 2026-06-06 \
66 + --as-of 2026-06-13 \
67 + --shards 3 \
68 + --wall-clock-budget 300 \
69 + --api-budget-multiplier 1.1 \
70 + --output-dir data/experiments/shard-435
71 +```
scripts/crawl_shard_experiment.py new
+1145
@@ -0,0 +1,1145 @@
1 +#!/usr/bin/env python3
2 +"""Run a local no-publish GitHub crawl sharding experiment for issue #435."""
3 +
4 +from __future__ import annotations
5 +
6 +import argparse
7 +import json
8 +import math
9 +import os
10 +import threading
11 +import time
12 +from collections import Counter
13 +from concurrent.futures import ThreadPoolExecutor, as_completed
14 +from dataclasses import dataclass, field
15 +from datetime import UTC, datetime
16 +from pathlib import Path
17 +from queue import Empty, Queue
18 +from typing import Any
19 +
20 +from scripts.crawl import (
21 + GitHubClient,
22 + RAW_ROOT,
23 + build_signals,
24 + build_star_snapshot,
25 + collect_repositories,
26 + github_artifact_checksum,
27 + github_crawl_config_checksum,
28 + github_schema_checksum,
29 + iso_timestamp,
30 + load_previous_star_snapshot,
31 + load_topic_queries,
32 + significance_skip_reason,
33 + to_repo_record,
34 + utc_now,
35 + validate_payload,
36 + week_slug,
37 + write_payload,
38 +)
39 +from scripts.topic_paths import cache_dir, raw_dir, snapshots_dir
40 +
41 +EXPERIMENT_ROOT = Path("data/experiments/shard-435")
42 +DEFAULT_WALL_CLOCK_BUDGET = 120
43 +DEFAULT_API_BUDGET_MULTIPLIER = 1.1
44 +
45 +
46 +class ExperimentAbort(RuntimeError):
47 + """Abort the shard experiment immediately."""
48 +
49 +
50 +class ShardBudgetExceeded(RuntimeError):
51 + """Stop a shard gracefully once its wall-clock budget is exhausted."""
52 +
53 +
54 +@dataclass(slots=True)
55 +class GuardrailEvent:
56 + kind: str
57 + shard: str
58 + message: str
59 + at: str
60 + details: dict[str, Any] = field(default_factory=dict)
61 +
62 + def to_dict(self) -> dict[str, Any]:
63 + payload = {"kind": self.kind, "shard": self.shard, "message": self.message, "at": self.at}
64 + if self.details:
65 + payload["details"] = self.details
66 + return payload
67 +
68 +
69 +@dataclass(slots=True)
70 +class SearchPlan:
71 + shard_name: str
72 + repo_group: str
73 + primary_queries: list[str]
74 + secondary_queries: list[str] = field(default_factory=list)
75 + min_repos_per_week: int = 0
76 +
77 +
78 +@dataclass(slots=True)
79 +class ValidationItem:
80 + repo_group: str
81 + sequence: int
82 + repo: dict[str, Any]
83 +
84 +
85 +@dataclass(slots=True)
86 +class ValidatedRecord:
87 + repo_group: str
88 + sequence: int
89 + record: dict[str, Any]
90 +
91 +
92 +@dataclass(slots=True)
93 +class CrawlContext:
94 + args: argparse.Namespace
95 + topic_id: str | None
96 + topic_raw: Path
97 + topic_snapshots: Path
98 + topic_cache: Path
99 + crawled_at: datetime
100 + run_started_at: datetime
101 + since: datetime
102 + window_end: datetime
103 + week: str
104 + max_results: int
105 + config_checksum: str
106 + current_code_sha: str
107 + source_refresh_policy: str
108 + baseline_output_path: Path
109 + baseline_snapshot_path: Path
110 + shard_output_path: Path
111 + shard_snapshot_path: Path
112 +
113 +
114 +@dataclass(slots=True)
115 +class RunResult:
116 + name: str
117 + payload: dict[str, Any]
118 + snapshot_payload: dict[str, Any]
119 + api_calls: int
120 + cache_hits: int
121 + stale_cache_hits: int
122 + rate_limit_events: int
123 + secondary_rate_limit_events: int
124 + partial_failures: list[str]
125 + wall_clock_s: float
126 + shards_used: int
127 + completed: bool
128 + guardrail_events: list[dict[str, Any]]
129 +
130 +
131 +@dataclass(slots=True)
132 +class ShardResult:
133 + shard_id: str
134 + repos_found: list[dict[str, Any]]
135 + api_calls: int
136 + errors: list[str]
137 + wall_clock_s: float
138 + rate_limit_detected: bool = False
139 + rate_limit_events: int = 0
140 +
141 +
142 +@dataclass(slots=True)
143 +class ExperimentReport:
144 + speedup_pct: float | None
145 + api_growth_pct: float | None
146 + rate_limit_regression: bool
147 + output_stable: bool
148 + partial_data: bool
149 + baseline_complete: bool
150 + shard_complete: bool
151 + verdict: str
152 +
153 + @classmethod
154 + def from_comparison(cls, comparison: dict[str, Any]) -> "ExperimentReport":
155 + speedup_pct = comparison.get("speedup_pct")
156 + api_growth_pct = comparison.get("api_growth_pct")
157 + rate_limit_regression = bool(
158 + comparison.get("rate_limit_regression", comparison.get("secondary_rate_limit_regression", False))
159 + )
160 + output_stable = bool(comparison.get("output_stable", False))
161 + partial_data = bool(comparison.get("partial_data", False))
162 + baseline_complete = bool(comparison.get("baseline_complete", True))
163 + shard_complete = bool(comparison.get("shard_complete", True))
164 + if partial_data or not baseline_complete or not shard_complete:
165 + verdict = "inconclusive"
166 + elif (
167 + speedup_pct is not None
168 + and api_growth_pct is not None
169 + and output_stable
170 + and not rate_limit_regression
171 + and float(speedup_pct) >= 25.0
172 + and float(api_growth_pct) <= 10.0
173 + ):
174 + verdict = "pass"
175 + else:
176 + verdict = "fail"
177 + return cls(
178 + speedup_pct=speedup_pct,
179 + api_growth_pct=api_growth_pct,
180 + rate_limit_regression=rate_limit_regression,
181 + output_stable=output_stable,
182 + partial_data=partial_data,
183 + baseline_complete=baseline_complete,
184 + shard_complete=shard_complete,
185 + verdict=verdict,
186 + )
187 +
188 +
189 +class SharedQuotaTracker:
190 + def __init__(self, cap: int) -> None:
191 + self.cap = max(0, int(cap))
192 + self.count = 0
193 + self._lock = threading.Lock()
194 +
195 + def increment(self) -> bool:
196 + with self._lock:
197 + if self.count >= self.cap:
198 + return False
199 + self.count += 1
200 + return True
201 +
202 + def reset(self) -> None:
203 + with self._lock:
204 + self.count = 0
205 +
206 +
207 +class WallClockBudget:
208 + def __init__(self, budget_s: float) -> None:
209 + self.budget_s = float(budget_s)
210 + self.started_at = time.monotonic()
211 +
212 + def elapsed_s(self) -> float:
213 + return time.monotonic() - self.started_at
214 +
215 + def is_exceeded(self) -> bool:
216 + return self.elapsed_s() > self.budget_s
217 +
218 +
219 +class SharedQuotaCoordinator:
220 + def __init__(self, api_hard_cap: int | None) -> None:
221 + self.api_hard_cap = api_hard_cap
222 + self.total_api_calls = 0
223 + self.global_backoff_until = 0.0
224 + self.secondary_rate_limit_hit = False
225 + self.abort_reason: str | None = None
226 + self._guardrail_events: list[GuardrailEvent] = []
227 + self._lock = threading.Lock()
228 +
229 + def before_request(self, shard_name: str, query: str, deadline: float | None) -> None:
230 + self._raise_if_aborted()
231 + self._raise_if_budget_exceeded(shard_name, deadline)
232 + while True:
233 + with self._lock:
234 + resume_at = self.global_backoff_until
235 + now = time.monotonic()
236 + if resume_at <= now:
237 + break
238 + time.sleep(min(resume_at - now, 1.0))
239 + self._raise_if_aborted()
240 + self._raise_if_budget_exceeded(shard_name, deadline)
241 + # Atomically reserve one API call slot to prevent concurrent over-cap
242 + with self._lock:
243 + if self.api_hard_cap is not None and self.total_api_calls >= self.api_hard_cap:
244 + if self.abort_reason is None:
245 + self.abort_reason = (
246 + f"API budget cap reached before {shard_name} requested {query} "
247 + f"({self.total_api_calls}/{self.api_hard_cap})."
248 + )
249 + self._guardrail_events.append(
250 + GuardrailEvent(
251 + kind="api_budget_cap",
252 + shard=shard_name,
253 + message=self.abort_reason,
254 + at=iso_timestamp(utc_now()),
255 + details={"api_calls_used": self.total_api_calls, "api_hard_cap": self.api_hard_cap},
256 + )
257 + )
258 + raise ExperimentAbort(self.abort_reason)
259 + # Reserve capacity for this request before releasing the lock
260 + self.total_api_calls += 1
261 +
262 + def register_api_calls(self, delta: int, shard_name: str) -> None:
263 + # Subtract the 1 call already reserved in before_request()
264 + additional = delta - 1
265 + if additional <= 0:
266 + return
267 + with self._lock:
268 + self.total_api_calls += additional
269 + if self.api_hard_cap is not None and self.total_api_calls > self.api_hard_cap and self.abort_reason is None:
270 + self.abort_reason = (
271 + f"API budget cap exceeded by {shard_name} "
272 + f"({self.total_api_calls}/{self.api_hard_cap})."
273 + )
274 + self._guardrail_events.append(
275 + GuardrailEvent(
276 + kind="api_budget_cap",
277 + shard=shard_name,
278 + message=self.abort_reason,
279 + at=iso_timestamp(utc_now()),
280 + details={"api_calls_used": self.total_api_calls, "api_hard_cap": self.api_hard_cap},
281 + )
282 + )
283 +
284 + def register_backoff(self, shard_name: str, delay: float, reason: str, *, secondary: bool = False) -> None:
285 + delay = max(delay, 1.0)
286 + with self._lock:
287 + self.global_backoff_until = max(self.global_backoff_until, time.monotonic() + delay)
288 + self._guardrail_events.append(
289 + GuardrailEvent(
290 + kind="secondary_rate_limit" if secondary else "rate_limit_backoff",
291 + shard=shard_name,
292 + message=reason,
293 + at=iso_timestamp(utc_now()),
294 + details={"delay_seconds": round(delay, 2)},
295 + )
296 + )
297 + if secondary:
298 + self.secondary_rate_limit_hit = True
299 + self.abort_reason = reason
300 +
301 + def record_budget_exceeded(self, shard_name: str, budget_seconds: int) -> None:
302 + with self._lock:
303 + self._guardrail_events.append(
304 + GuardrailEvent(
305 + kind="wall_clock_budget",
306 + shard=shard_name,
307 + message=f"{shard_name} exhausted its {budget_seconds}s wall-clock budget.",
308 + at=iso_timestamp(utc_now()),
309 + details={"budget_seconds": budget_seconds},
310 + )
311 + )
312 +
313 + def record_redistribution(self, shard_name: str, repo_count: int) -> None:
314 + if repo_count <= 0:
315 + return
316 + with self._lock:
317 + self._guardrail_events.append(
318 + GuardrailEvent(
319 + kind="work_redistributed",
320 + shard=shard_name,
321 + message=f"Redistributed {repo_count} remaining repositories from {shard_name}.",
322 + at=iso_timestamp(utc_now()),
323 + details={"repo_count": repo_count},
324 + )
325 + )
326 +
327 + def guardrail_events(self) -> list[dict[str, Any]]:
328 + with self._lock:
329 + return [event.to_dict() for event in self._guardrail_events]
330 +
331 + def _raise_if_aborted(self) -> None:
332 + with self._lock:
333 + if self.abort_reason is not None:
334 + raise ExperimentAbort(self.abort_reason)
335 +
336 + def _raise_if_budget_exceeded(self, shard_name: str, deadline: float | None) -> None:
337 + if deadline is not None and time.monotonic() >= deadline:
338 + raise ShardBudgetExceeded(f"{shard_name} exceeded its wall-clock budget.")
339 +
340 +
341 +class InstrumentedGitHubClient(GitHubClient):
342 + def __init__(
343 + self,
344 + token: str,
345 + *,
346 + cache_dir: Path,
347 + shard_name: str,
348 + coordinator: SharedQuotaCoordinator | None = None,
349 + deadline: float | None = None,
350 + timeout: int = 30,
351 + max_retries: int = 6,
352 + ) -> None:
353 + self._api_calls_used = 0
354 + self.shard_name = shard_name
355 + self.coordinator = coordinator
356 + self.deadline = deadline
357 + self.rate_limit_events = 0
358 + self.secondary_rate_limit_events = 0
359 + super().__init__(token, cache_dir=cache_dir, timeout=timeout, max_retries=max_retries)
360 +
361 + @property
362 + def api_calls_used(self) -> int:
363 + return self._api_calls_used
364 +
365 + @api_calls_used.setter
366 + def api_calls_used(self, value: int) -> None:
367 + prior = getattr(self, "_api_calls_used", 0)
368 + self._api_calls_used = value
369 + delta = value - prior
370 + if self.coordinator is not None and delta > 0:
371 + self.coordinator.register_api_calls(delta, self.shard_name)
372 +
373 + def _pause_for_rate_limit(self, query: str) -> None:
374 + if self.coordinator is not None:
375 + self.coordinator.before_request(self.shard_name, query, self.deadline)
376 + elif self.deadline is not None and time.monotonic() >= self.deadline:
377 + raise ShardBudgetExceeded(f"{self.shard_name} exceeded its wall-clock budget.")
378 + super()._pause_for_rate_limit(query)
379 +
380 + def _respect_min_interval(self, url: str) -> None:
381 + if self.deadline is not None and time.monotonic() >= self.deadline:
382 + raise ShardBudgetExceeded(f"{self.shard_name} exceeded its wall-clock budget.")
383 + super()._respect_min_interval(url)
384 +
385 + def _sleep_before_retry(
386 + self,
387 + attempt: int,
388 + headers: dict[str, str] | None,
389 + body: str,
390 + query: str,
391 + retry_limit: int,
392 + max_delay_seconds: float,
393 + ) -> None:
394 + lowered = body.lower()
395 + retry_after = None
396 + if headers and headers.get("Retry-After"):
397 + try:
398 + retry_after = max(float(headers["Retry-After"]), 1.0)
399 + except ValueError:
400 + retry_after = None
401 + reset_delay = self._reset_delay(headers)
402 + delay = retry_after or reset_delay or min(2**attempt, max_delay_seconds)
403 + if "secondary rate limit" in lowered:
404 + self.rate_limit_events += 1
405 + self.secondary_rate_limit_events += 1
406 + reason = f"{self.shard_name} hit a secondary rate limit while requesting {query}."
407 + if self.coordinator is not None:
408 + self.coordinator.register_backoff(self.shard_name, max(delay, 8.0), reason, secondary=True)
409 + raise ExperimentAbort(reason)
410 + if headers and (
411 + headers.get("Retry-After") is not None
412 + or headers.get("X-RateLimit-Remaining") == "0"
413 + or (self.rate_limit_remaining is not None and self.rate_limit_remaining <= 0)
414 + ):
415 + self.rate_limit_events += 1
416 + if self.coordinator is not None:
417 + self.coordinator.register_backoff(
418 + self.shard_name,
419 + max(delay, 1.0),
420 + f"{self.shard_name} backing off before retrying {query}.",
421 + )
422 + super()._sleep_before_retry(attempt, headers, body, query, retry_limit, max_delay_seconds)
423 +
424 +
425 +def parse_args() -> argparse.Namespace:
426 + parser = argparse.ArgumentParser(description=__doc__)
427 + parser.add_argument("--since", required=True, help="UTC crawl window start date (YYYY-MM-DD).")
428 + parser.add_argument("--as-of", required=True, help="UTC crawl window end date (YYYY-MM-DD).")
429 + parser.add_argument("--max-results", type=int, default=250, help="Maximum repositories per query.")
430 + parser.add_argument("--topic", default=None, help="Optional topic id.")
431 + parser.add_argument("--config", default=None, help="Optional crawl topic config file.")
432 + parser.add_argument("--shards", type=int, default=3, help="Total shards including search shards.")
433 + parser.add_argument("--wall-clock-budget", type=int, default=DEFAULT_WALL_CLOCK_BUDGET)
434 + parser.add_argument("--api-budget-multiplier", type=float, default=DEFAULT_API_BUDGET_MULTIPLIER)
435 + parser.add_argument("--output-dir", default=str(EXPERIMENT_ROOT))
436 + parser.add_argument("--experiment-id", default=None)
437 + return parser.parse_args()
438 +
439 +
440 +def build_context(args: argparse.Namespace, experiment_dir: Path) -> CrawlContext:
441 + crawled_at = utc_now()
442 + since = datetime.strptime(args.since, "%Y-%m-%d").replace(tzinfo=UTC)
443 + window_end = datetime.strptime(args.as_of, "%Y-%m-%d").replace(tzinfo=UTC)
444 + topic_id = args.topic
445 + max_results = max(1, min(int(args.max_results), 1000))
446 + config_args = argparse.Namespace(
447 + since=args.since,
448 + as_of=args.as_of,
449 + max_results=max_results,
450 + output=str(experiment_dir / "baseline-raw.json"),
451 + topic=topic_id,
452 + config=args.config,
453 + )
454 + return CrawlContext(
455 + args=args,
456 + topic_id=topic_id,
457 + topic_raw=raw_dir(topic_id),
458 + topic_snapshots=snapshots_dir(topic_id),
459 + topic_cache=cache_dir(topic_id),
460 + crawled_at=crawled_at,
461 + run_started_at=crawled_at,
462 + since=since,
463 + window_end=window_end,
464 + week=week_slug(window_end),
465 + max_results=max_results,
466 + config_checksum=github_crawl_config_checksum(config_args, since, window_end, max_results),
467 + current_code_sha=os.environ.get("CRAWLER_CODE_SHA", ""),
468 + source_refresh_policy="force-refresh",
469 + baseline_output_path=experiment_dir / "baseline-raw.json",
470 + baseline_snapshot_path=experiment_dir / "baseline-stars.json",
471 + shard_output_path=experiment_dir / "shard-raw.json",
472 + shard_snapshot_path=experiment_dir / "shard-stars.json",
473 + )
474 +
475 +
476 +def next_experiment_id(output_dir: Path) -> str:
477 + index = 0
478 + for candidate in output_dir.glob("shard-435-run-*"):
479 + suffix = candidate.name.removeprefix("shard-435-run-")
480 + if suffix.isdigit():
481 + index = max(index, int(suffix))
482 + return f"shard-435-run-{index + 1:03d}"
483 +
484 +
485 +def build_search_plans(context: CrawlContext) -> list[SearchPlan]:
486 + if context.args.config:
487 + queries = load_topic_queries(
488 + context.args.config,
489 + {"last_week": context.since.date().isoformat(), "today": context.window_end.date().isoformat()},
490 + )
491 + return [
492 + SearchPlan(
493 + shard_name="new-search",
494 + repo_group="new",
495 + primary_queries=list(queries["primary"]),
496 + secondary_queries=list(queries["secondary"]),
497 + min_repos_per_week=int(queries["min_repos_per_week"]),
498 + )
499 + ]
500 + return [
501 + SearchPlan(
502 + shard_name="new-search",
503 + repo_group="new",
504 + primary_queries=[f"created:{context.since.date().isoformat()}..{context.window_end.date().isoformat()} stars:>50"],
505 + ),
506 + SearchPlan(
507 + shard_name="trending-search",
508 + repo_group="trending",
509 + primary_queries=[f"pushed:{context.since.date().isoformat()}..{context.window_end.date().isoformat()} stars:>50"],
510 + ),
511 + ]
512 +
513 +
514 +def run_search_plan(
515 + plan: SearchPlan,
516 + client: InstrumentedGitHubClient,
517 + *,
518 + max_results: int,
519 + wall_clock_budget: int,
520 + coordinator: SharedQuotaCoordinator | None,
521 +) -> tuple[list[dict[str, Any]], dict[str, Any]]:
522 + candidates: list[dict[str, Any]] = []
523 + errors: list[str] = []
524 + completed = True
525 + try:
526 + for query in plan.primary_queries:
527 + candidates.extend(client.search_repositories(query, max_results=max_results))
528 + if plan.secondary_queries and len(candidates) < plan.min_repos_per_week:
529 + for query in plan.secondary_queries:
530 + candidates.extend(client.search_repositories(query, max_results=max_results))
531 + except ShardBudgetExceeded as exc:
532 + completed = False
533 + errors.append(str(exc))
534 + if coordinator is not None:
535 + coordinator.record_budget_exceeded(plan.shard_name, wall_clock_budget)
536 + except ExperimentAbort:
537 + raise
538 + return candidates, {
539 + "completed": completed,
540 + "errors": errors + list(client.errors),
541 + "api_calls": client.api_calls_used,
542 + "cache_hits": client.cache_hits,
543 + "stale_cache_hits": client.stale_cache_hits,
544 + "rate_limit_events": client.rate_limit_events,
545 + "secondary_rate_limit_events": client.secondary_rate_limit_events,
546 + }
547 +
548 +
549 +def chunk_validation_items(items: list[ValidationItem], worker_count: int) -> list[list[ValidationItem]]:
550 + if not items:
551 + return []
552 + chunk_size = max(1, math.ceil(len(items) / max(worker_count * 2, 1)))
553 + return [items[index : index + chunk_size] for index in range(0, len(items), chunk_size)]
554 +
555 +
556 +def prepare_validation_items(
557 + new_candidates: list[dict[str, Any]],
558 + trending_candidates: list[dict[str, Any]],
559 + worker_count: int,
560 +) -> tuple[Queue[list[ValidationItem]], dict[str, int]]:
561 + duplicate_counts = {"new": 0, "trending": 0}
562 + grouped_unique_items: list[ValidationItem] = []
563 + for repo_group, candidates in (("new", new_candidates), ("trending", trending_candidates)):
564 + seen: set[str] = set()
565 + for sequence, repo in enumerate(candidates):
566 + full_name = repo.get("full_name")
567 + if not full_name:
568 + grouped_unique_items.append(ValidationItem(repo_group=repo_group, sequence=sequence, repo=repo))
569 + continue
570 + if full_name in seen:
571 + duplicate_counts[repo_group] += 1
572 + continue
573 + seen.add(full_name)
574 + grouped_unique_items.append(ValidationItem(repo_group=repo_group, sequence=sequence, repo=repo))
575 + queue: Queue[list[ValidationItem]] = Queue()
576 + for chunk in chunk_validation_items(grouped_unique_items, worker_count):
577 + queue.put(chunk)
578 + return queue, duplicate_counts
579 +
580 +
581 +def trending_stars_gained(
582 + repo: dict[str, Any],
583 + *,
584 + previous_stars: dict[str, int] | None,
585 + trending_cutoff: datetime | None,
586 +) -> int | None:
587 + if previous_stars is None:
588 + return None
589 + previous_value = previous_stars.get(str(repo.get("full_name") or ""))
590 + current_stars = int(repo.get("stargazers_count") or 0)
591 + if previous_value is not None:
592 + return max(current_stars - previous_value, 0)
593 + if previous_stars and trending_cutoff is not None and isinstance(repo.get("created_at"), str):
594 + created = datetime.fromisoformat(str(repo["created_at"]).replace("Z", "+00:00"))
595 + if created >= trending_cutoff:
596 + return current_stars
597 + return None
598 +
599 +
600 +def validate_item(
601 + client: InstrumentedGitHubClient,
602 + item: ValidationItem,
603 + *,
604 + previous_stars: dict[str, int] | None,
605 + trending_cutoff: datetime | None,
606 +) -> tuple[ValidatedRecord | None, str | None]:
607 + full_name = item.repo.get("full_name")
608 + if not full_name:
609 + return None, "missing_full_name"
610 + skip_reason = significance_skip_reason(item.repo)
611 + if skip_reason:
612 + return None, skip_reason
613 + try:
614 + if not client.has_readme(full_name):
615 + return None, "missing_readme"
616 + except RuntimeError as exc:
617 + client.record_error(f"README lookup failed for {full_name}: {exc}")
618 + return None, "readme_lookup_failed"
619 + stars_gained = None
620 + if item.repo_group == "trending":
621 + stars_gained = trending_stars_gained(
622 + item.repo,
623 + previous_stars=previous_stars,
624 + trending_cutoff=trending_cutoff,
625 + )
626 + return ValidatedRecord(item.repo_group, item.sequence, to_repo_record(item.repo, stars_gained=stars_gained)), None
627 +
628 +
629 +def sort_validated_records(
630 + records: list[ValidatedRecord],
631 + *,
632 + previous_stars: dict[str, int] | None,
633 +) -> list[dict[str, Any]]:
634 + if records and records[0].repo_group == "trending" and previous_stars:
635 + ordered = sorted(
636 + records,
637 + key=lambda item: (
638 + -int(item.record.get("stars_gained", -1) if item.record.get("stars_gained") is not None else -1),
639 + -int(item.record.get("stars") or 0),
640 + item.sequence,
641 + ),
642 + )
643 + else:
644 + ordered = sorted(records, key=lambda item: (-int(item.record.get("stars") or 0), item.sequence))
645 + return [item.record for item in ordered]
646 +
647 +
648 +def validation_worker(
649 + worker_index: int,
650 + queue: Queue[list[ValidationItem]],
651 + client: InstrumentedGitHubClient,
652 + *,
653 + previous_stars: dict[str, int] | None,
654 + trending_cutoff: datetime | None,
655 + wall_clock_budget: int,
656 + coordinator: SharedQuotaCoordinator,
657 +) -> dict[str, Any]:
658 + shard_name = f"validate-{worker_index}"
659 + results = {"new": [], "trending": []}
660 + filters: dict[str, Counter[str]] = {"new": Counter(), "trending": Counter()}
661 + completed = True
662 + while True:
663 + try:
664 + chunk = queue.get_nowait()
665 + except Empty:
666 + break
667 + current_index = 0
668 + try:
669 + for current_index, item in enumerate(chunk):
670 + validated, skip_reason = validate_item(
671 + client,
672 + item,
673 + previous_stars=previous_stars,
674 + trending_cutoff=trending_cutoff,
675 + )
676 + if validated is not None:
677 + results[item.repo_group].append(validated)
678 + elif skip_reason is not None:
679 + filters[item.repo_group][skip_reason] += 1
680 + except ShardBudgetExceeded:
681 + completed = False
682 + coordinator.record_budget_exceeded(shard_name, wall_clock_budget)
683 + remaining_items = chunk[current_index:]
684 + if remaining_items:
685 + queue.put(remaining_items)
686 + coordinator.record_redistribution(shard_name, len(remaining_items))
687 + break
688 + except ExperimentAbort:
689 + raise
690 + return {
691 + "completed": completed,
692 + "results": results,
693 + "filters": {name: dict(counter) for name, counter in filters.items()},
694 + "errors": list(client.errors),
695 + "api_calls": client.api_calls_used,
696 + "cache_hits": client.cache_hits,
697 + "stale_cache_hits": client.stale_cache_hits,
698 + "rate_limit_events": client.rate_limit_events,
699 + "secondary_rate_limit_events": client.secondary_rate_limit_events,
700 + }
701 +
702 +
703 +def build_payload(
704 + *,
705 + context: CrawlContext,
706 + output_path: Path,
707 + snapshot_path: Path,
708 + new_repos: list[dict[str, Any]],
709 + trending_repos: list[dict[str, Any]],
710 + new_candidates: list[dict[str, Any]],
711 + trending_candidates: list[dict[str, Any]],
712 + api_calls: int,
713 + cache_hits: int,
714 + stale_cache_hits: int,
715 + rate_limit_limit: int | None,
716 + rate_limit_remaining: int | None,
717 + rate_limit_reset: int | None,
718 + rate_limit_resource: str | None,
719 + partial_failures: list[str],
720 + filter_summary: dict[str, dict[str, int]],
721 + run_mode: str,
722 +) -> tuple[dict[str, Any], dict[str, Any]]:
723 + star_snapshot = build_star_snapshot(new_candidates, trending_candidates)
724 + snapshot_payload = {
725 + "week": context.week,
726 + "captured_at": iso_timestamp(context.crawled_at),
727 + "repository_count": len(star_snapshot),
728 + "stars": star_snapshot,
729 + }
730 + payload = {
731 + "week": context.week,
732 + "crawled_at": iso_timestamp(context.crawled_at),
733 + "new_repos": new_repos,
734 + "trending_repos": trending_repos,
735 + "signals": build_signals(new_repos, trending_repos),
736 + "metadata": {
737 + "api_calls_used": api_calls,
738 + "cache_hits": cache_hits,
739 + "stale_cache_hits": stale_cache_hits,
740 + "rate_limit_limit": rate_limit_limit,
741 + "rate_limit_remaining": rate_limit_remaining,
742 + "rate_limit_reset": rate_limit_reset,
743 + "rate_limit_resource": rate_limit_resource,
744 + "partial_failures": partial_failures,
745 + "run_id": f"local-{run_mode}",
746 + "crawl_window": {
747 + "since": context.since.date().isoformat(),
748 + "until": context.window_end.date().isoformat(),
749 + },
750 + "crawl_config_checksum": context.config_checksum,
751 + "schema_checksum": github_schema_checksum(),
752 + "same_day_reuse": {"status": "not_reused", "source": "github", "source_id": "github-search"},
753 + "filter_summary": filter_summary,
754 + "snapshot_path": snapshot_path.as_posix(),
755 + "source_refresh_policy": context.source_refresh_policy,
756 + "crawler_code_sha": context.current_code_sha,
757 + "experiment_mode": run_mode,
758 + "output_path": output_path.as_posix(),
759 + },
760 + }
761 + payload["metadata"]["artifact_checksum"] = github_artifact_checksum(payload)
762 + validate_payload(payload)
763 + return payload, snapshot_payload
764 +
765 +
766 +def run_baseline(context: CrawlContext, token: str) -> RunResult:
767 + started_at = time.monotonic()
768 + search_plans = build_search_plans(context)
769 + client = InstrumentedGitHubClient(token, cache_dir=context.topic_cache, shard_name="baseline")
770 + previous_stars = load_previous_star_snapshot(
771 + context.topic_snapshots,
772 + context.week,
773 + context.baseline_output_path.parent,
774 + context.topic_raw,
775 + RAW_ROOT,
776 + )
777 + new_candidates: list[dict[str, Any]] = []
778 + trending_candidates: list[dict[str, Any]] = []
779 + if context.args.config:
780 + new_candidates, _ = run_search_plan(
781 + search_plans[0],
782 + client,
783 + max_results=context.max_results,
784 + wall_clock_budget=context.args.wall_clock_budget,
785 + coordinator=None,
786 + )
787 + else:
788 + for plan in search_plans:
789 + candidates, _ = run_search_plan(
790 + plan,
791 + client,
792 + max_results=context.max_results,
793 + wall_clock_budget=context.args.wall_clock_budget,
794 + coordinator=None,
795 + )
796 + if plan.repo_group == "new":
797 + new_candidates = candidates
798 + else:
799 + trending_candidates = candidates
800 + new_repos, new_filters = collect_repositories(client, new_candidates)
801 + trending_repos, trending_filters = collect_repositories(
802 + client,
803 + trending_candidates,
804 + previous_stars=previous_stars,
805 + trending_cutoff=context.since,
806 + )
807 + payload, snapshot_payload = build_payload(
808 + context=context,
809 + output_path=context.baseline_output_path,
810 + snapshot_path=context.baseline_snapshot_path,
811 + new_repos=new_repos,
812 + trending_repos=trending_repos,
813 + new_candidates=new_candidates,
814 + trending_candidates=trending_candidates,
815 + api_calls=client.api_calls_used,
816 + cache_hits=client.cache_hits,
817 + stale_cache_hits=client.stale_cache_hits,
818 + rate_limit_limit=client.rate_limit_limit,
819 + rate_limit_remaining=client.rate_limit_remaining,
820 + rate_limit_reset=client.rate_limit_reset,
821 + rate_limit_resource=client.rate_limit_resource,
822 + partial_failures=list(client.errors),
823 + filter_summary={"new_repos": new_filters, "trending_repos": trending_filters},
824 + run_mode="baseline",
825 + )
826 + write_payload(context.baseline_output_path, payload)
827 + write_payload(context.baseline_snapshot_path, snapshot_payload)
828 + return RunResult(
829 + name="baseline",
830 + payload=payload,
831 + snapshot_payload=snapshot_payload,
832 + api_calls=client.api_calls_used,
833 + cache_hits=client.cache_hits,
834 + stale_cache_hits=client.stale_cache_hits,
835 + rate_limit_events=client.rate_limit_events,
836 + secondary_rate_limit_events=client.secondary_rate_limit_events,
837 + partial_failures=list(client.errors),
838 + wall_clock_s=round(time.monotonic() - started_at, 3),
839 + shards_used=1,
840 + completed=not bool(client.errors),
841 + guardrail_events=[],
842 + )
843 +
844 +
845 +def run_sharded(context: CrawlContext, token: str, baseline_api_calls: int) -> RunResult:
846 + started_at = time.monotonic()
847 + search_plans = build_search_plans(context)
848 + validation_workers = max(1, int(context.args.shards) - len(search_plans))
849 + api_hard_cap = max(1, math.floor(baseline_api_calls * float(context.args.api_budget_multiplier)))
850 + coordinator = SharedQuotaCoordinator(api_hard_cap)
851 + previous_stars = load_previous_star_snapshot(
852 + context.topic_snapshots,
853 + context.week,
854 + context.shard_output_path.parent,
855 + context.topic_raw,
856 + RAW_ROOT,
857 + )
858 + aggregated_errors: list[str] = []
859 + aggregated_api_calls = 0
860 + aggregated_cache_hits = 0
861 + aggregated_stale_cache_hits = 0
862 + aggregated_rate_limit_events = 0
863 + aggregated_secondary_rate_limit_events = 0
864 + search_results: dict[str, list[dict[str, Any]]] = {"new": [], "trending": []}
865 +
866 + try:
867 + with ThreadPoolExecutor(max_workers=max(len(search_plans), 1)) as pool:
868 + futures = {
869 + pool.submit(
870 + run_search_plan,
871 + plan,
872 + InstrumentedGitHubClient(
873 + token,
874 + cache_dir=context.topic_cache,
875 + shard_name=plan.shard_name,
876 + coordinator=coordinator,
877 + deadline=time.monotonic() + int(context.args.wall_clock_budget),
878 + ),
879 + max_results=context.max_results,
880 + wall_clock_budget=int(context.args.wall_clock_budget),
881 + coordinator=coordinator,
882 + ): plan
883 + for plan in search_plans
884 + }
885 + for future in as_completed(futures):
886 + plan = futures[future]
887 + candidates, metrics = future.result()
888 + search_results[plan.repo_group] = candidates
889 + aggregated_errors.extend(metrics["errors"])
890 + aggregated_api_calls += metrics["api_calls"]
891 + aggregated_cache_hits += metrics["cache_hits"]
892 + aggregated_stale_cache_hits += metrics["stale_cache_hits"]
893 + aggregated_rate_limit_events += metrics["rate_limit_events"]
894 + aggregated_secondary_rate_limit_events += metrics["secondary_rate_limit_events"]
895 + except ExperimentAbort as exc:
896 + aggregated_errors.append(str(exc))
897 +
898 + validation_queue, duplicate_counts = prepare_validation_items(
899 + search_results["new"], search_results["trending"], validation_workers
900 + )
901 + validated_records: dict[str, list[ValidatedRecord]] = {"new": [], "trending": []}
902 + filter_summary: dict[str, Counter[str]] = {
903 + "new_repos": Counter({"duplicate": duplicate_counts["new"]}),
904 + "trending_repos": Counter({"duplicate": duplicate_counts["trending"]}),
905 + }
906 + if coordinator.abort_reason is None:
907 + try:
908 + with ThreadPoolExecutor(max_workers=validation_workers) as pool:
909 + futures = {
910 + pool.submit(
911 + validation_worker,
912 + index + 1,
913 + validation_queue,
914 + InstrumentedGitHubClient(
915 + token,
916 + cache_dir=context.topic_cache,
917 + shard_name=f"validate-{index + 1}",
918 + coordinator=coordinator,
919 + deadline=time.monotonic() + int(context.args.wall_clock_budget),
920 + ),
921 + previous_stars=previous_stars,
922 + trending_cutoff=context.since,
923 + wall_clock_budget=int(context.args.wall_clock_budget),
924 + coordinator=coordinator,
925 + ): index + 1
926 + for index in range(validation_workers)
927 + }
928 + for future in as_completed(futures):
929 + result = future.result()
930 + validated_records["new"].extend(result["results"]["new"])
931 + validated_records["trending"].extend(result["results"]["trending"])
932 + filter_summary["new_repos"].update(result["filters"]["new"])
933 + filter_summary["trending_repos"].update(result["filters"]["trending"])
934 + aggregated_errors.extend(result["errors"])
935 + aggregated_api_calls += result["api_calls"]
936 + aggregated_cache_hits += result["cache_hits"]
937 + aggregated_stale_cache_hits += result["stale_cache_hits"]
938 + aggregated_rate_limit_events += result["rate_limit_events"]
939 + aggregated_secondary_rate_limit_events += result["secondary_rate_limit_events"]
940 + except ExperimentAbort as exc:
941 + aggregated_errors.append(str(exc))
942 +
943 + new_repos = sort_validated_records(validated_records["new"], previous_stars=None)
944 + trending_repos = sort_validated_records(validated_records["trending"], previous_stars=previous_stars)
945 + payload, snapshot_payload = build_payload(
946 + context=context,
947 + output_path=context.shard_output_path,
948 + snapshot_path=context.shard_snapshot_path,
949 + new_repos=new_repos,
950 + trending_repos=trending_repos,
951 + new_candidates=search_results["new"],
952 + trending_candidates=search_results["trending"],
953 + api_calls=aggregated_api_calls,
954 + cache_hits=aggregated_cache_hits,
955 + stale_cache_hits=aggregated_stale_cache_hits,
956 + rate_limit_limit=None,
957 + rate_limit_remaining=None,
958 + rate_limit_reset=None,
959 + rate_limit_resource="mixed",
960 + partial_failures=aggregated_errors,
961 + filter_summary={name: dict(counter) for name, counter in filter_summary.items()},
962 + run_mode="shard",
963 + )
964 + write_payload(context.shard_output_path, payload)
965 + write_payload(context.shard_snapshot_path, snapshot_payload)
966 + return RunResult(
967 + name="shard",
968 + payload=payload,
969 + snapshot_payload=snapshot_payload,
970 + api_calls=aggregated_api_calls,
971 + cache_hits=aggregated_cache_hits,
972 + stale_cache_hits=aggregated_stale_cache_hits,
973 + rate_limit_events=aggregated_rate_limit_events,
974 + secondary_rate_limit_events=aggregated_secondary_rate_limit_events,
975 + partial_failures=aggregated_errors,
976 + wall_clock_s=round(time.monotonic() - started_at, 3),
977 + shards_used=len(search_plans) + validation_workers,
978 + completed=not aggregated_errors and validation_queue.empty() and not coordinator.secondary_rate_limit_hit,
979 + guardrail_events=coordinator.guardrail_events(),
980 + )
981 +
982 +
983 +def canonicalize_payload(payload: dict[str, Any]) -> bytes:
984 + canonical = {
985 + "week": payload.get("week"),
986 + "new_repos": payload.get("new_repos", []),
987 + "trending_repos": payload.get("trending_repos", []),
988 + "signals": payload.get("signals", {}),
989 + }
990 + return json.dumps(canonical, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
991 +
992 +
993 +def canonicalize_snapshot(snapshot_payload: dict[str, Any]) -> bytes:
994 + canonical = {
995 + "week": snapshot_payload.get("week"),
996 + "repository_count": snapshot_payload.get("repository_count"),
997 + "stars": snapshot_payload.get("stars", {}),
998 + }
999 + return json.dumps(canonical, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
1000 +
1001 +
1002 +def _normalize_for_comparison(value: Any) -> Any:
1003 + if isinstance(value, dict):
1004 + return {
1005 + key: _normalize_for_comparison(item)
1006 + for key, item in sorted(value.items())
1007 + if key not in {"crawled_at", "captured_at", "report_generated_at"}
1008 + }
1009 + if isinstance(value, list):
1010 + return [_normalize_for_comparison(item) for item in value]
1011 + return value
1012 +
1013 +
1014 +def deterministic_merge(shard_results: list[ShardResult]) -> dict[str, Any]:
1015 + merged: dict[str, dict[str, Any]] = {}
1016 + for result in shard_results:
1017 + for repo in result.repos_found:
1018 + full_name = str(repo.get("full_name") or "")
1019 + if full_name and full_name not in merged:
1020 + merged[full_name] = repo
1021 + repos = sorted(
1022 + merged.values(),
1023 + key=lambda item: (
1024 + -(int(item.get("stars_gained", -1) or -1)),
1025 + -(int(item.get("stars") or 0)),
1026 + str(item.get("full_name") or ""),
1027 + ),
1028 + )
1029 + return {"repos": repos}
1030 +
1031 +
1032 +def compare_results(baseline: dict[str, Any], shard: dict[str, Any]) -> dict[str, Any]:
1033 + baseline_wall = float(baseline.get("wall_clock_s", baseline.get("elapsed_s", 0.0)) or 0.0001)
1034 + shard_wall = float(shard.get("wall_clock_s", shard.get("elapsed_s", 0.0)) or 0.0)
1035 + baseline_api = int(baseline.get("api_calls", baseline.get("api_calls_used", 0)) or 0)
1036 + shard_api = int(shard.get("api_calls", shard.get("api_calls_used", 0)) or 0)
1037 + baseline_output = _normalize_for_comparison(baseline.get("canonical_output", baseline.get("output", {})))
1038 + shard_output = _normalize_for_comparison(shard.get("canonical_output", shard.get("output", {})))
1039 + return {
1040 + "speedup_pct": round(((baseline_wall - shard_wall) / baseline_wall) * 100, 2),
1041 + "api_growth_pct": round(((shard_api - baseline_api) / max(baseline_api, 1)) * 100, 2),
1042 + "output_stable": baseline_output == shard_output,
1043 + "rate_limit_regression": int(shard.get("rate_limit_events", 0) or 0)
1044 + > int(baseline.get("rate_limit_events", 0) or 0),
1045 + "partial_data": bool(shard.get("partial_data", False) or baseline.get("partial_data", False)),
1046 + "baseline_complete": not bool(baseline.get("partial_data", False)),
1047 + "shard_complete": not bool(shard.get("partial_data", False)),
1048 + }
1049 +
1050 +
1051 +def build_report(experiment_id: str, baseline: RunResult, shard: RunResult) -> dict[str, Any]:
1052 + baseline_wall = baseline.wall_clock_s or 0.0001
1053 + speedup_pct = round(((baseline_wall - shard.wall_clock_s) / baseline_wall) * 100, 2)
1054 + api_growth_pct = round(((shard.api_calls - baseline.api_calls) / max(baseline.api_calls, 1)) * 100, 2)
1055 + output_stable = canonicalize_payload(baseline.payload) == canonicalize_payload(shard.payload) and canonicalize_snapshot(
1056 + baseline.snapshot_payload
1057 + ) == canonicalize_snapshot(shard.snapshot_payload)
1058 + baseline_incomplete = bool(baseline.partial_failures) or not baseline.completed
1059 + shard_incomplete = bool(shard.partial_failures) or not shard.completed
1060 + report_card = ExperimentReport.from_comparison(
1061 + {
1062 + "speedup_pct": speedup_pct,
1063 + "api_growth_pct": api_growth_pct,
1064 + "rate_limit_regression": shard.secondary_rate_limit_events > baseline.secondary_rate_limit_events,
1065 + "output_stable": output_stable,
1066 + "partial_data": baseline_incomplete or shard_incomplete,
1067 + "baseline_complete": not baseline_incomplete,
1068 + "shard_complete": not shard_incomplete,
1069 + }
1070 + )
1071 + verdict = report_card.verdict
1072 + if report_card.partial_data and any(event["kind"] == "secondary_rate_limit" for event in shard.guardrail_events):
1073 + verdict = "fail"
1074 + return {
1075 + "experiment_id": experiment_id,
1076 + "baseline": {
1077 + "wall_clock_s": baseline.wall_clock_s,
1078 + "api_calls": baseline.api_calls,
1079 + "rate_limit_events": baseline.rate_limit_events,
1080 + "repos_new": len(baseline.payload.get("new_repos", [])),
1081 + "repos_trending": len(baseline.payload.get("trending_repos", [])),
1082 + },
1083 + "shard": {
1084 + "wall_clock_s": shard.wall_clock_s,
1085 + "api_calls": shard.api_calls,
1086 + "rate_limit_events": shard.rate_limit_events,
1087 + "shards_used": shard.shards_used,
1088 + "repos_new": len(shard.payload.get("new_repos", [])),
1089 + "repos_trending": len(shard.payload.get("trending_repos", [])),
1090 + },
1091 + "comparison": {
1092 + "speedup_pct": speedup_pct,
1093 + "api_growth_pct": api_growth_pct,
1094 + "output_stable": output_stable,
1095 + "secondary_rate_limit_regression": shard.secondary_rate_limit_events > baseline.secondary_rate_limit_events,
1096 + },
1097 + "verdict": verdict,
1098 + "guardrail_events": shard.guardrail_events,
1099 + }
1100 +
1101 +
1102 +def main() -> int:
1103 + args = parse_args()
1104 + if args.shards < 3:
1105 + print("--shards must be >= 3.", file=os.sys.stderr)
1106 + return 1
1107 + if args.wall_clock_budget <= 0:
1108 + print("--wall-clock-budget must be positive.", file=os.sys.stderr)
1109 + return 1
1110 + if args.api_budget_multiplier < 1.0:
1111 + print("--api-budget-multiplier must be >= 1.0.", file=os.sys.stderr)
1112 + return 1
1113 + token = os.environ.get("GITHUB_TOKEN")
1114 + if not token:
1115 + print("GITHUB_TOKEN is required", file=os.sys.stderr)
1116 + return 1
1117 + output_dir = Path(args.output_dir)
1118 + output_dir.mkdir(parents=True, exist_ok=True)
1119 + experiment_id = args.experiment_id or next_experiment_id(output_dir)
1120 + experiment_dir = output_dir / experiment_id
1121 + experiment_dir.mkdir(parents=True, exist_ok=True)
1122 + context = build_context(args, experiment_dir)
1123 + baseline = run_baseline(context, token)
1124 + shard = run_sharded(context, token, baseline.api_calls)
1125 + report = build_report(experiment_id, baseline, shard)
1126 + report_path = experiment_dir / "report.json"
1127 + write_payload(report_path, report)
1128 + print(
1129 + json.dumps(
1130 + {
1131 + "experiment_id": experiment_id,
1132 + "verdict": report["verdict"],
1133 + "speedup_pct": report["comparison"]["speedup_pct"],
1134 + "api_growth_pct": report["comparison"]["api_growth_pct"],
1135 + "output_stable": report["comparison"]["output_stable"],
1136 + "report_path": report_path.as_posix(),
1137 + },
1138 + ensure_ascii=False,
1139 + )
1140 + )
1141 + return 0 if report["verdict"] != "fail" else 1
1142 +
1143 +
1144 +if __name__ == "__main__":
1145 + raise SystemExit(main())
tests/test_crawl_shard_experiment.py new
+369
@@ -0,0 +1,369 @@
1 +from __future__ import annotations
2 +
3 +import dataclasses
4 +import importlib.util
5 +import inspect
6 +import json
7 +from concurrent.futures import ThreadPoolExecutor
8 +from typing import Any
9 +
10 +import pytest
11 +
12 +if importlib.util.find_spec("scripts.crawl_shard_experiment") is None:
13 + pytest.skip("scripts.crawl_shard_experiment is not implemented yet", allow_module_level=True)
14 +
15 +from scripts import crawl_shard_experiment as experiment
16 +
17 +
18 +def _build_instance(factory: Any, **values: Any) -> Any:
19 + signature = inspect.signature(factory)
20 + kwargs = {}
21 + for name, parameter in signature.parameters.items():
22 + if name in values:
23 + kwargs[name] = values[name]
24 + elif parameter.default is inspect._empty and parameter.kind in (
25 + inspect.Parameter.POSITIONAL_OR_KEYWORD,
26 + inspect.Parameter.KEYWORD_ONLY,
27 + ):
28 + raise AssertionError(f"missing required parameter {name!r} for {factory}")
29 + return factory(**kwargs)
30 +
31 +
32 +def _resolve_attr(target: Any, *names: str) -> Any:
33 + if isinstance(target, dict):
34 + for name in names:
35 + if name in target:
36 + return target[name]
37 + for name in names:
38 + if hasattr(target, name):
39 + return getattr(target, name)
40 + raise AssertionError(f"could not resolve any of {names!r} on {target!r}")
41 +
42 +
43 +def _resolve_method(target: Any, *names: str):
44 + for name in names:
45 + candidate = getattr(target, name, None)
46 + if callable(candidate):
47 + return candidate
48 + raise AssertionError(f"could not resolve any of {names!r} on {target!r}")
49 +
50 +
51 +def _make_tracker(cap: int):
52 + return _build_instance(
53 + experiment.SharedQuotaTracker,
54 + cap=cap,
55 + hard_cap=cap,
56 + limit=cap,
57 + quota_cap=cap,
58 + max_calls=cap,
59 + )
60 +
61 +
62 +def _tracker_cap(tracker: Any) -> int:
63 + return int(_resolve_attr(tracker, "cap", "hard_cap", "limit", "quota_cap", "max_calls"))
64 +
65 +
66 +def _tracker_count(tracker: Any) -> int:
67 + return int(_resolve_attr(tracker, "count", "used", "api_calls_used", "value", "calls_used"))
68 +
69 +
70 +def _consume_quota(tracker: Any) -> bool:
71 + method = _resolve_method(tracker, "increment", "try_acquire", "acquire", "consume", "record_call")
72 + try:
73 + result = method()
74 + except Exception as exc: # pragma: no cover
75 + message = str(exc).lower()
76 + if "quota" in message or "cap" in message or "limit" in message:
77 + return False
78 + raise
79 + if isinstance(result, bool):
80 + return result
81 + if isinstance(result, int):
82 + return result <= _tracker_cap(tracker)
83 + return _tracker_count(tracker) <= _tracker_cap(tracker)
84 +
85 +
86 +def _reset_tracker(tracker: Any) -> None:
87 + _resolve_method(tracker, "reset", "clear")()
88 +
89 +
90 +def _make_budget(seconds: float):
91 + return _build_instance(
92 + experiment.WallClockBudget,
93 + budget_s=seconds,
94 + budget_seconds=seconds,
95 + seconds=seconds,
96 + limit_s=seconds,
97 + )
98 +
99 +
100 +def _budget_exceeded(budget: Any) -> bool:
101 + status = _resolve_method(budget, "is_exceeded", "expired", "timed_out", "should_stop", "exhausted")
102 + return bool(status())
103 +
104 +
105 +def _elapsed_seconds(budget: Any) -> float:
106 + for name in ("elapsed_s", "elapsed_seconds", "elapsed"):
107 + if hasattr(budget, name):
108 + value = getattr(budget, name)
109 + return float(value() if callable(value) else value)
110 + raise AssertionError(f"could not resolve elapsed time on {budget!r}")
111 +
112 +
113 +def _make_shard_result(*, repos_found: list[dict[str, Any]], api_calls: int, errors: list[str], wall_clock_s: float):
114 + return _build_instance(
115 + experiment.ShardResult,
116 + shard_id="github:new-repos:q1",
117 + repos_found=repos_found,
118 + repos=repos_found,
119 + api_calls=api_calls,
120 + api_calls_used=api_calls,
121 + errors=errors,
122 + wall_clock_s=wall_clock_s,
123 + duration_s=wall_clock_s,
124 + rate_limit_detected=False,
125 + rate_limit_events=0,
126 + )
127 +
128 +
129 +def _extract_repos(payload: Any) -> list[dict[str, Any]]:
130 + if isinstance(payload, list):
131 + return payload
132 + for name in ("repos", "repos_found", "items", "results", "merged_repos"):
133 + if isinstance(payload, dict) and name in payload:
134 + return payload[name]
135 + if hasattr(payload, name):
136 + return getattr(payload, name)
137 + raise AssertionError(f"could not extract merged repos from {payload!r}")
138 +
139 +
140 +def _metric(result: Any, *names: str) -> Any:
141 + return _resolve_attr(result, *names)
142 +
143 +
144 +def _percent(value: Any) -> float:
145 + numeric = float(value)
146 + return numeric * 100.0 if abs(numeric) <= 1.0 else numeric
147 +
148 +
149 +def _canonical_json(payload: Any) -> str:
150 + return json.dumps(payload, separators=(",", ":"), ensure_ascii=False)
151 +
152 +
153 +def _make_report(comparison: dict[str, Any]):
154 + report_cls = experiment.ExperimentReport
155 + for factory_name in ("from_comparison", "from_metrics", "evaluate", "build"):
156 + factory = getattr(report_cls, factory_name, None)
157 + if callable(factory):
158 + signature = inspect.signature(factory)
159 + if "comparison" in signature.parameters:
160 + return factory(comparison=comparison)
161 + if any(parameter.kind == inspect.Parameter.VAR_KEYWORD for parameter in signature.parameters.values()):
162 + return factory(**comparison)
163 + allowed = {name: value for name, value in comparison.items() if name in signature.parameters}
164 + return factory(**allowed)
165 + signature = inspect.signature(report_cls)
166 + if "comparison" in signature.parameters:
167 + return report_cls(comparison=comparison)
168 + if any(parameter.kind == inspect.Parameter.VAR_KEYWORD for parameter in signature.parameters.values()):
169 + return report_cls(**comparison)
170 + allowed = {name: value for name, value in comparison.items() if name in signature.parameters}
171 + return report_cls(**allowed)
172 +
173 +
174 +def _verdict(report: Any) -> str:
175 + return str(_resolve_attr(report, "verdict", "status")).lower()
176 +
177 +
178 +@pytest.fixture
179 +def repo_alpha() -> dict[str, Any]:
180 + return {
181 + "full_name": "octo/alpha",
182 + "name": "alpha",
183 + "owner": "octo",
184 + "stars": 150,
185 + "stars_gained": 42,
186 + "language": "Python",
187 + "topics": ["ai", "testing"],
188 + "url": "https://github.com/octo/alpha",
189 + }
190 +
191 +
192 +@pytest.fixture
193 +def repo_beta() -> dict[str, Any]:
194 + return {
195 + "full_name": "octo/beta",
196 + "name": "beta",
197 + "owner": "octo",
198 + "stars": 120,
199 + "stars_gained": 15,
200 + "language": "Go",
201 + "topics": ["infra"],
202 + "url": "https://github.com/octo/beta",
203 + }
204 +
205 +
206 +@pytest.fixture
207 +def repo_gamma() -> dict[str, Any]:
208 + return {
209 + "full_name": "tools/gamma",
210 + "name": "gamma",
211 + "owner": "tools",
212 + "stars": 220,
213 + "stars_gained": 5,
214 + "language": "Rust",
215 + "topics": ["developer-tools"],
216 + "url": "https://github.com/tools/gamma",
217 + }
218 +
219 +
220 +class TestSharedQuotaTracker:
221 + def test_thread_safe_increment_and_cap_enforcement(self) -> None:
222 + tracker = _make_tracker(cap=10)
223 + with ThreadPoolExecutor(max_workers=8) as pool:
224 + successes = list(pool.map(lambda _: _consume_quota(tracker), range(25)))
225 + assert sum(bool(result) for result in successes) == 10
226 + assert _tracker_count(tracker) == 10
227 + assert _consume_quota(tracker) is False
228 +
229 + def test_reset_clears_accumulated_usage(self) -> None:
230 + tracker = _make_tracker(cap=3)
231 + assert _consume_quota(tracker) is True
232 + assert _consume_quota(tracker) is True
233 + assert _tracker_count(tracker) == 2
234 + _reset_tracker(tracker)
235 + assert _tracker_count(tracker) == 0
236 + assert _consume_quota(tracker) is True
237 +
238 +
239 +class TestWallClockBudget:
240 + def test_budget_not_exceeded_before_deadline(self, monkeypatch: pytest.MonkeyPatch) -> None:
241 + clock = {"now": 100.0}
242 + monkeypatch.setattr(experiment.time, "monotonic", lambda: clock["now"])
243 + budget = _make_budget(5.0)
244 + clock["now"] = 104.9
245 + assert _elapsed_seconds(budget) == pytest.approx(4.9)
246 + assert _budget_exceeded(budget) is False
247 +
248 + def test_budget_exceeded_triggers_graceful_stop(self, monkeypatch: pytest.MonkeyPatch) -> None:
249 + clock = {"now": 250.0}
250 + monkeypatch.setattr(experiment.time, "monotonic", lambda: clock["now"])
251 + budget = _make_budget(5.0)
252 + clock["now"] = 255.1
253 + assert _elapsed_seconds(budget) == pytest.approx(5.1)
254 + assert _budget_exceeded(budget) is True
255 +
256 +
257 +def test_shard_result_dataclass_contract(repo_alpha: dict[str, Any]) -> None:
258 + assert dataclasses.is_dataclass(experiment.ShardResult)
259 + field_names = {field.name for field in dataclasses.fields(experiment.ShardResult)}
260 + assert {"repos_found", "api_calls", "errors", "wall_clock_s"}.issubset(field_names)
261 + result = _make_shard_result(repos_found=[repo_alpha], api_calls=4, errors=[], wall_clock_s=1.75)
262 + assert _resolve_attr(result, "repos_found", "repos") == [repo_alpha]
263 + assert _resolve_attr(result, "api_calls", "api_calls_used") == 4
264 + assert result.errors == []
265 + assert _resolve_attr(result, "wall_clock_s", "duration_s") == pytest.approx(1.75)
266 +
267 +
268 +class TestDeterministicMerge:
269 + def test_deduplicates_by_full_name_and_preserves_star_gain(
270 + self, repo_alpha: dict[str, Any], repo_beta: dict[str, Any], repo_gamma: dict[str, Any]
271 + ) -> None:
272 + first = _make_shard_result(repos_found=[repo_alpha, repo_beta], api_calls=3, errors=[], wall_clock_s=1.1)
273 + duplicate_alpha = {**repo_alpha, "stars": 999, "stars_gained": 42}
274 + second = _make_shard_result(repos_found=[duplicate_alpha, repo_gamma], api_calls=2, errors=[], wall_clock_s=1.2)
275 + merged = experiment.deterministic_merge([first, second])
276 + merged_repos = _extract_repos(merged)
277 + assert [repo["full_name"] for repo in merged_repos].count("octo/alpha") == 1
278 + assert {repo["full_name"] for repo in merged_repos} == {"octo/alpha", "octo/beta", "tools/gamma"}
279 + alpha = next(repo for repo in merged_repos if repo["full_name"] == "octo/alpha")
280 + assert alpha["stars_gained"] == 42
281 +
282 + def test_same_inputs_produce_byte_identical_output_regardless_of_shard_order(
283 + self, repo_alpha: dict[str, Any], repo_beta: dict[str, Any], repo_gamma: dict[str, Any]
284 + ) -> None:
285 + shard_a = _make_shard_result(repos_found=[repo_alpha, repo_gamma], api_calls=3, errors=[], wall_clock_s=1.0)
286 + shard_b = _make_shard_result(repos_found=[repo_beta], api_calls=2, errors=[], wall_clock_s=1.0)
287 + merged_ab = experiment.deterministic_merge([shard_a, shard_b])
288 + merged_ba = experiment.deterministic_merge([shard_b, shard_a])
289 + assert _canonical_json(merged_ab) == _canonical_json(merged_ba)
290 +
291 +
292 +class TestCompareResults:
293 + def test_calculates_speedup_and_api_growth(self, repo_alpha: dict[str, Any], repo_beta: dict[str, Any]) -> None:
294 + canonical = {"repos": [repo_alpha, repo_beta], "metadata": {"note": "stable"}}
295 + baseline = {"wall_clock_s": 100.0, "api_calls": 100, "output": canonical, "canonical_output": canonical}
296 + shard = {"wall_clock_s": 70.0, "api_calls": 108, "output": canonical, "canonical_output": canonical}
297 + comparison = experiment.compare_results(baseline, shard)
298 + assert _percent(_metric(comparison, "speedup_pct", "speedup_percent", "speedup")) == pytest.approx(30.0)
299 + assert _percent(_metric(comparison, "api_growth_pct", "api_growth_percent", "api_growth")) == pytest.approx(8.0)
300 +
301 + def test_output_stability_ignores_timestamp_fields(
302 + self, repo_alpha: dict[str, Any], repo_beta: dict[str, Any]
303 + ) -> None:
304 + baseline = {
305 + "wall_clock_s": 100.0,
306 + "api_calls": 100,
307 + "output": {
308 + "week": "2026-W24",
309 + "crawled_at": "2026-06-13T12:00:00Z",
310 + "report_generated_at": "2026-06-13T12:00:01Z",
311 + "repos": [repo_alpha, repo_beta],
312 + },
313 + }
314 + shard = {
315 + "wall_clock_s": 75.0,
316 + "api_calls": 105,
317 + "output": {
318 + "week": "2026-W24",
319 + "crawled_at": "2026-06-13T12:10:00Z",
320 + "report_generated_at": "2026-06-13T12:10:01Z",
321 + "repos": [repo_alpha, repo_beta],
322 + },
323 + }
324 + comparison = experiment.compare_results(baseline, shard)
325 + assert bool(_metric(comparison, "output_stable", "stable_output", "is_output_stable")) is True
326 +
327 +
328 +class TestExperimentReport:
329 + def test_pass_verdict_when_all_guardrails_are_met(self) -> None:
330 + report = _make_report(
331 + {
332 + "speedup_pct": 30.0,
333 + "api_growth_pct": 8.0,
334 + "rate_limit_regression": False,
335 + "output_stable": True,
336 + "partial_data": False,
337 + "baseline_complete": True,
338 + "shard_complete": True,
339 + }
340 + )
341 + assert _verdict(report) == "pass"
342 +
343 + def test_fail_verdict_when_any_required_criterion_fails(self) -> None:
344 + report = _make_report(
345 + {
346 + "speedup_pct": 20.0,
347 + "api_growth_pct": 8.0,
348 + "rate_limit_regression": False,
349 + "output_stable": True,
350 + "partial_data": False,
351 + "baseline_complete": True,
352 + "shard_complete": True,
353 + }
354 + )
355 + assert _verdict(report) == "fail"
356 +
357 + def test_inconclusive_verdict_for_partial_data(self) -> None:
358 + report = _make_report(
359 + {
360 + "speedup_pct": None,
361 + "api_growth_pct": None,
362 + "rate_limit_regression": False,
363 + "output_stable": False,
364 + "partial_data": True,
365 + "baseline_complete": True,
366 + "shard_complete": False,
367 + }
368 + )
369 + assert _verdict(report) == "inconclusive"