@cryptotaxi247 / SquadScope / commits / 2ae8707

feat: wire matrix crawl and map/reduce observability metrics (#467)

* Add automated QA gates for map/reduce acceptance criteria - Unit tests cover reducer correctness: deterministic fan-in, citation preservation, contradiction handling, duplicate collapse - End-to-end dry-run tests exercise full pipeline on representative fixtures and validate gate outcomes - Cost/token guardrail tests enforce PRD budget limits and fail on missing/out-of-bounds token accounting - Failure handling tests cover mapper/reducer failure paths, malformed input, and fallback behavior (no press context) - Gate output clarity tests ensure CI failures identify which gate failed and provide actionable error messages - CI-facing documentation in docs/qa-gates.md explains each gate Closes #438 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * feat: wire observability metrics into crawl and map/reduce pipelines Implements #437 acceptance criteria: - Add scripts/observability_metrics.py with versioned schema (observability_v1), dataclasses for CrawlMetrics/MapReduceMetrics/AnalysisMetrics, emit_ledger, validate_ledger, and duration_p95 - Wire crawl.py to emit GitHub crawl observability ledger with API call counts, cache hits/misses, rate-limit signals, and sampled duration - Wire techcrunch_crawler.py to emit external-news crawl metrics - Wire map_reduce_dry_run.py to emit per-stage timing, pass/fail counts, gate failure reasons, and token/cost breakdown - Add docs/observability-metrics.md documenting schema fields, versioning, and validation contract for downstream checks - Add tests/test_observability_metrics.py and representative fixture - Gitignore data/metrics/observability/ runtime artifacts Closes #437 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * fix: address observability review feedback Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * fix: address Copilot review comments on PR #467 - validate_ledger() now reports specific schema_version mismatch details - Add type validation for environment field (must be dict) - Update test assertion to match improved error messages - collect_gate_failure_reasons() already correctly skips expected_failure gates - Fixture 2026-W21-full-run.json is consistent: structural gate failure is real Resolves review threads on PR #467. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * fix: resolve Copilot review comments on PR #467 - Make test_dry_run_emits_valid_contract_artifacts hermetic by patching DEFAULT_OBSERVABILITY_DIR to a temp directory instead of writing to the repo-root data/metrics/observability path. - Fix duration_sample_count inconsistency in techcrunch_crawler: when no per-source durations are available, treat the total runtime as a single sample (count=1) instead of reporting p95 with zero samples. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Juan Manuel Servera committed Jun 13, 2026 at 23:23 UTC 2ae87079f589c633f20a1aa13eb8c36b65b76f27

13 files changed +1131 -49

.gitignore

@@ -25,6 +25,7 @@ resources/_gen/

25 .ralph/

26 ralph_squadscope.log

27 squadscope-agent-result.json

28 +data/metrics/observability/

30 # Local copilot skills (machine-specific)

31 .copilot/

docs/observability-metrics.md new

+157

@@ -0,0 +1,157 @@

1 +# Observability Metrics Schema

2 +

3 +`scripts/observability_metrics.py` defines the durable ledger schema for crawl and analysis observability artifacts written to `data/metrics/observability/`.

4 +

5 +## Schema version

6 +

7 +- Current version: `observability_v1`

8 +- Writers must set `schema_version` exactly.

9 +- Downstream checks should fail if `validate_ledger()` reports any missing required field paths.

10 +

11 +## Artifact layout

12 +

13 +Current pipeline writers emit:

14 +

15 +- `data/metrics/observability/{week}-github-crawl.json`

16 +- `data/metrics/observability/{week}-external-news-crawl.json`

17 +- `data/metrics/observability/{week}-map-reduce.json`

18 +

19 +These are runtime artifacts and are gitignored.

20 +

21 +## Top-level ledger fields

22 +

23 +Required:

24 +

25 +- `schema_version`

26 +- `run_id`

27 +- `week`

28 +- `timestamp`

29 +- `crawl_metrics`

30 +- `environment`

31 +

32 +Optional:

33 +

34 +- `analysis_metrics`

35 +

36 +## `crawl_metrics[]`

37 +

38 +Required fields:

39 +

40 +- `source_type` (`github` or `external-news`)

41 +- `duration_seconds`

42 +- `api_calls`

43 +- `cache_hits`

44 +- `cache_misses`

45 +- `stale_cache_hits`

46 +- `rate_limit_events`

47 +- `secondary_rate_limit_hit`

48 +

49 +Optional fields:

50 +

51 +- `duration_p95_seconds`

52 +- `duration_sample_count`

53 +

54 +Notes:

55 +

56 +- GitHub crawl currently records overall run duration as the comparable sampled duration.

57 +- External-news crawl records p95 from per-source fetch durations when available.

58 +

59 +## `analysis_metrics`

60 +

61 +Required fields when present:

62 +

63 +- `duration_seconds`

64 +- `token_ledger`

65 +- `map_stages`

66 +

67 +Optional:

68 +

69 +- `reduce_stage`

70 +

71 +### `token_ledger`

72 +

73 +Required:

74 +

75 +- `input_tokens`

76 +- `output_tokens`

77 +- `total_tokens`

78 +

79 +Common additional field:

80 +

81 +- `cost_usd`

82 +

83 +### `map_stages[]` and `reduce_stage`

84 +

85 +Required:

86 +

87 +- `stage`

88 +- `duration_seconds`

89 +- `input_tokens`

90 +- `output_tokens`

91 +- `cost_usd`

92 +- `status` (`pass` or `fail`)

93 +- `gate_failure_reasons`

94 +

95 +## Validation

96 +

97 +Use `validate_ledger()` from `scripts.observability_metrics`:

98 +

99 +```python

100 +from scripts.observability_metrics import validate_ledger

101 +

102 +errors = validate_ledger(payload)

103 +if errors:

104 + raise SystemExit(f"Missing required observability fields: {errors}")

105 +```

106 +

107 +`emit_ledger()` already validates before writing and raises `ValueError` on schema gaps.

108 +

109 +## Example

110 +

111 +Representative end-to-end sample:

112 +

113 +- `tests/fixtures/observability/2026-W21-full-run.json`

114 +

115 +Minimal shape:

116 +

117 +```json

118 +{

119 + "schema_version": "observability_v1",

120 + "run_id": "12345",

121 + "week": "2026-W21",

122 + "timestamp": "2026-05-20T12:00:00Z",

123 + "crawl_metrics": [

124 + {

125 + "source_type": "github",

126 + "duration_seconds": 12.4,

127 + "api_calls": 27,

128 + "cache_hits": 14,

129 + "cache_misses": 27,

130 + "stale_cache_hits": 1,

131 + "rate_limit_events": 2,

132 + "secondary_rate_limit_hit": false

133 + }

134 + ],

135 + "analysis_metrics": {

136 + "duration_seconds": 3.2,

137 + "token_ledger": {

138 + "input_tokens": 1234,

139 + "output_tokens": 456,

140 + "total_tokens": 1690,

141 + "cost_usd": 0.0

142 + },

143 + "map_stages": [],

144 + "reduce_stage": null

145 + },

146 + "environment": {

147 + "pipeline": "map-reduce-dry-run"

148 + }

149 +}

150 +```

151 +

152 +## Downstream check guidance

153 +

154 +- Treat `schema_version` as a compatibility gate.

155 +- Call `validate_ledger()` and fail on any returned field path.

156 +- Prefer exact field-path assertions over permissive defaults so missing metrics break CI early.

157 +- For experiment reports tied to issue #356, link the representative fixture above plus the emitted runtime artifacts from the relevant workflow run.

docs/qa-gates.md

+4 -13

@@ -81,16 +81,7 @@ When a gate fails in CI:

82 1. The test name tells you which category failed (e.g., `TestCostTokenGuardrails::test_preflight_rejects_over_budget`)

83 2. The assertion message identifies the specific contract violation

84 -3. The QA report JSON (`qa-comparison-report.json`) provides structured gate-by-gate results

85 -

86 -If `test_full_pipeline_produces_passing_qa` fails, inspect the QA report's

87 -`checks` object carefully. **Not all gates share the same schema** — each gate

88 -reports results in its own structure:

89 -

90 -- `mapper_contracts` includes `passed` plus `errors_by_mapper`

91 -- `structural_analysis_gate` and `evidence_and_editorial_gates` include `passed`

92 - plus `errors`

93 -- `publish_provenance_gate` includes `passed`, `expected_failure`, and `errors`

94 -- `sidecars_present` includes `passed` plus emitted object counts

95 -- `reference_count` is informational only and reports `selected`,

96 - `notable_projects`, and `press_articles`

84 +3. The QA report JSON (`qa-comparison-report.json`) provides structured gate-by-gate results with error lists

85 +

86 +If `test_full_pipeline_produces_passing_qa` fails, check the QA report's

87 +`checks` object — each gate has a `passed` boolean and an `errors` list.

scripts/crawl.py

+112 -2

@@ -18,6 +18,13 @@ from pathlib import Path

18 from typing import Any, Iterable

19 from urllib import error, parse, request

21 +from scripts.observability_metrics import (

22 + DEFAULT_OBSERVABILITY_DIR,

23 + CrawlMetrics,

24 + METRICS_SCHEMA_VERSION,

25 + ObservabilityLedger,

26 + emit_ledger,

27 +)

28 from scripts.topic_paths import cache_dir, raw_dir, snapshots_dir

30 API_ROOT = "https://api.github.com"

@@ -156,7 +163,10 @@ class GitHubClient:

163 self.max_retries = max_retries

164 self.api_calls_used = 0

165 self.cache_hits = 0

166 + self.cache_misses = 0

167 self.stale_cache_hits = 0

168 + self.rate_limit_events = 0

169 + self.secondary_rate_limit_hit = False

170 self.rate_limit_limit: int | None = None

171 self.rate_limit_remaining: int | None = None

172 self.rate_limit_reset: int | None = None

@@ -213,6 +223,7 @@ class GitHubClient:

223 self.cache_hits += 1

224 return cached

225

226 + self.cache_misses += 1

227 stale_fallback = None

228 if cached and allow_stale and (cached.status == 200 or cached.status in accepted):

229 stale_fallback = CacheEntry(

@@ -251,6 +262,11 @@ class GitHubClient:

262 headers = {name: value for name, value in (exc.headers.items() if exc.headers else [])}

263 self._update_rate_limit(headers)

264 body = exc.read().decode("utf-8", errors="replace")

265 + lowered_body = body.lower()

266 + if exc.code in {403, 429} or "rate limit" in lowered_body or "abuse" in lowered_body:

267 + self.rate_limit_events += 1

268 + if "secondary rate limit" in lowered_body or "abuse" in lowered_body:

269 + self.secondary_rate_limit_hit = True

270 self._last_request_at = time.monotonic()

271 payload = decode_json_body(body)

272 if exc.code in accepted:

@@ -394,6 +410,7 @@ class GitHubClient:

410 if reset_delay is None:

411 if self.rate_limit_remaining <= critical_threshold:

412 delay = 10.0 if self.rate_limit_remaining <= 0 else 3.0

413 + self.rate_limit_events += 1

414 log(

415 f"Rate limit low ({self.rate_limit_remaining}/{self.rate_limit_limit} {self.rate_limit_resource or 'requests'}) "

416 f"without reset hint; cooling down {delay:.1f}s before {query}."

@@ -402,10 +419,12 @@ class GitHubClient:

419 return

420 if self.rate_limit_remaining <= critical_threshold:

421 delay = min(reset_delay + _JITTER_RANDOM.uniform(0.3, 1.5), 300.0)

422 + self.rate_limit_events += 1

423 log(f"Rate limit nearly exhausted before {query}; pausing {delay:.1f}s until reset window.")

424 time.sleep(delay)

425 return

426 delay = min(max(reset_delay / 10, 1.0), 30.0)

427 + self.rate_limit_events += 1

428 log(

429 f"Rate limit low ({self.rate_limit_remaining}/{self.rate_limit_limit} {self.rate_limit_resource or 'requests'}); "

430 f"cooling down {delay:.1f}s before {query}."

@@ -1001,6 +1020,7 @@ def write_payload(path: Path, payload: dict[str, Any]) -> None:

1020

1021 def main() -> int:

1022 args = parse_args()

1023 + crawl_started = time.monotonic()

1024

1025 topic_id = args.topic

1026 topic_raw = raw_dir(topic_id)

@@ -1042,7 +1062,51 @@ def main() -> int:

1062 if reusable is not None:

1063 write_payload(output_path, reusable)

1064 restore_reused_snapshot(reuse_path, reusable.get("metadata", {}), expected_snapshot_dir=topic_snapshots)

1045 - print(f"Reused same-day GitHub raw artifact {reuse_path} -> {output_path}; used 0 API calls.")

1065 + observability_path = DEFAULT_OBSERVABILITY_DIR / f"{week}-github-crawl.json"

1066 + duration_seconds = round(time.monotonic() - crawl_started, 3)

1067 + emit_ledger(

1068 + ObservabilityLedger(

1069 + schema_version=METRICS_SCHEMA_VERSION,

1070 + run_id=os.environ.get("GITHUB_RUN_ID", "local"),

1071 + week=week,

1072 + timestamp=iso_timestamp(crawled_at),

1073 + crawl_metrics=[

1074 + CrawlMetrics(

1075 + duration_seconds=duration_seconds,

1076 + duration_p95_seconds=duration_seconds,

1077 + duration_sample_count=1,

1078 + api_calls=0,

1079 + cache_hits=0,

1080 + cache_misses=0,

1081 + stale_cache_hits=0,

1082 + rate_limit_events=0,

1083 + secondary_rate_limit_hit=False,

1084 + source_type="github",

1085 + )

1086 + ],

1087 + analysis_metrics=None,

1088 + environment={

1089 + "pipeline": "github-crawl",

1090 + "topic": topic_id,

1091 + "output_path": output_path.as_posix(),

1092 + "snapshot_path": snapshot_path.as_posix(),

1093 + "source_refresh_policy": source_refresh_policy,

1094 + "same_day_reuse_status": "reused",

1095 + "partial_failures": [],

1096 + "rate_limit_snapshot": {

1097 + "limit": None,

1098 + "remaining": None,

1099 + "reset": None,

1100 + "resource": None,

1101 + },

1102 + },

1103 + ),

1104 + observability_path,

1105 + )

1106 + print(

1107 + f"Reused same-day GitHub raw artifact {reuse_path} -> {output_path}; "

1108 + f"used 0 API calls; observability={observability_path}."

1109 + )

1110 return 0

1111

1112 github_token = os.environ.get("GITHUB_TOKEN")

@@ -1134,13 +1198,59 @@ def main() -> int:

1198 validate_payload(payload)

1199 write_payload(output_path, payload)

1200 write_payload(snapshot_path, snapshot_payload)

1201 + observability_path = DEFAULT_OBSERVABILITY_DIR / f"{week}-github-crawl.json"

1202 + duration_seconds = round(time.monotonic() - crawl_started, 3)

1203 + secondary_rate_limit_hit = bool(getattr(client, "secondary_rate_limit_hit", False)) or any(

1204 + "secondary rate limit" in str(message).lower() or "abuse" in str(message).lower()

1205 + for message in client.errors

1206 + )

1207 + emit_ledger(

1208 + ObservabilityLedger(

1209 + schema_version=METRICS_SCHEMA_VERSION,

1210 + run_id=os.environ.get("GITHUB_RUN_ID", "local"),

1211 + week=week,

1212 + timestamp=iso_timestamp(crawled_at),

1213 + crawl_metrics=[

1214 + CrawlMetrics(

1215 + duration_seconds=duration_seconds,

1216 + duration_p95_seconds=duration_seconds,

1217 + duration_sample_count=1,

1218 + api_calls=int(getattr(client, "api_calls_used", 0)),

1219 + cache_hits=int(getattr(client, "cache_hits", 0)),

1220 + cache_misses=int(getattr(client, "cache_misses", getattr(client, "api_calls_used", 0))),

1221 + stale_cache_hits=int(getattr(client, "stale_cache_hits", 0)),

1222 + rate_limit_events=int(getattr(client, "rate_limit_events", 0)),

1223 + secondary_rate_limit_hit=secondary_rate_limit_hit,

1224 + source_type="github",

1225 + )

1226 + ],

1227 + analysis_metrics=None,

1228 + environment={

1229 + "pipeline": "github-crawl",

1230 + "topic": topic_id,

1231 + "output_path": output_path.as_posix(),

1232 + "snapshot_path": snapshot_path.as_posix(),

1233 + "source_refresh_policy": source_refresh_policy,

1234 + "same_day_reuse_status": payload["metadata"]["same_day_reuse"]["status"],

1235 + "partial_failures": list(client.errors),

1236 + "rate_limit_snapshot": {

1237 + "limit": client.rate_limit_limit,

1238 + "remaining": client.rate_limit_remaining,

1239 + "reset": client.rate_limit_reset,

1240 + "resource": client.rate_limit_resource,

1241 + },

1242 + },

1243 + ),

1244 + observability_path,

1245 + )

1246

1247 if client.errors:

1248 log(f"Completed with {len(client.errors)} partial failure(s).")

1249

1250 print(

1251 f"Wrote {output_path} with {len(new_repos)} new repos and {len(trending_repos)} trending repos, "

1143 - f"saved {snapshot_path}, used {client.api_calls_used} API calls, and served {client.cache_hits} cache hits."

1252 + f"saved {snapshot_path}, used {client.api_calls_used} API calls, served {client.cache_hits} cache hits, "

1253 + f"and emitted observability metrics to {observability_path}."

1254 )

1255 return 1 if client.errors else 0

1256

scripts/map_reduce_dry_run.py

+165 -5

@@ -13,6 +13,7 @@ import hashlib

13 import json

14 import re

15 import sys

16 +import time

17 from dataclasses import asdict, dataclass

18 from datetime import UTC, datetime

19 from pathlib import Path

@@ -21,12 +22,30 @@ from typing import Any

22 try:

23 from scripts.analyze_fallback import find_previous_summary

24 from scripts.analysis_gate import validate_analysis, validate_publish_quality

25 + from scripts.model_pricing import estimate_cost_usd

26 + from scripts.observability_metrics import (

27 + DEFAULT_OBSERVABILITY_DIR,

28 + AnalysisMetrics,

29 + METRICS_SCHEMA_VERSION,

30 + MapReduceMetrics,

31 + ObservabilityLedger,

32 + emit_ledger,

33 + )

34 from scripts.render_press_context import estimate_tokens

35 from scripts.sanitize_repo_content import sanitize_repo_payload

36 except ModuleNotFoundError: # pragma: no cover - script execution path

37 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

38 from scripts.analyze_fallback import find_previous_summary

39 from scripts.analysis_gate import validate_analysis, validate_publish_quality

40 + from scripts.model_pricing import estimate_cost_usd

41 + from scripts.observability_metrics import (

42 + DEFAULT_OBSERVABILITY_DIR,

43 + AnalysisMetrics,

44 + METRICS_SCHEMA_VERSION,

45 + MapReduceMetrics,

46 + ObservabilityLedger,

47 + emit_ledger,

48 + )

49 from scripts.render_press_context import estimate_tokens

50 from scripts.sanitize_repo_content import sanitize_repo_payload

@@ -88,6 +107,10 @@ def write_json(path: Path, payload: Any) -> None:

107 path.write_text(stable_json(payload), encoding="utf-8")

108

109

110 +def metric_cost_usd(model: str, input_tokens: int, output_tokens: int) -> float:

111 + return estimate_cost_usd(model, input_tokens, output_tokens) or 0.0

112 +

113 +

114 def file_ref(path: Path | None) -> ArtifactRef | None:

115 if path is None or not path.exists() or not path.is_file():

116 return None

@@ -95,6 +118,35 @@ def file_ref(path: Path | None) -> ArtifactRef | None:

118 return ArtifactRef(path=path.as_posix(), sha256=sha256_bytes(data), bytes=len(data))

119

120

121 +def collect_gate_failure_reasons(qa_report: dict[str, Any]) -> list[str]:

122 + reasons: list[str] = []

123 + checks = qa_report.get("checks", {}) if isinstance(qa_report.get("checks"), dict) else {}

124 + mapper_contracts = checks.get("mapper_contracts", {})

125 + if isinstance(mapper_contracts, dict):

126 + errors_by_mapper = mapper_contracts.get("errors_by_mapper", {})

127 + if isinstance(errors_by_mapper, dict):

128 + for mapper, errors in sorted(errors_by_mapper.items()):

129 + if isinstance(errors, list):

130 + reasons.extend(f"{mapper}: {error}" for error in errors)

131 + for key in ("structural_analysis_gate", "evidence_and_editorial_gates", "publish_provenance_gate"):

132 + check = checks.get(key, {})

133 + if isinstance(check, dict) and check.get("expected_failure") is not True and isinstance(check.get("errors"), list):

134 + reasons.extend(str(error) for error in check["errors"] if error)

135 + if isinstance(qa_report.get("regressions"), list):

136 + reasons.extend(str(error) for error in qa_report["regressions"] if error)

137 + sidecars = checks.get("sidecars_present", {})

138 + if isinstance(sidecars, dict) and sidecars.get("passed") is False:

139 + reasons.append("sidecars_present failed")

140 + seen: set[str] = set()

141 + ordered: list[str] = []

142 + for reason in reasons:

143 + normalized = reason.strip()

144 + if normalized and normalized not in seen:

145 + seen.add(normalized)

146 + ordered.append(normalized)

147 + return ordered

148 +

149 +

150 def normalize_repo_name(repo: dict[str, Any]) -> str:

151 full_name = str(repo.get("full_name") or "").strip()

152 if full_name:

@@ -713,6 +765,7 @@ def build_qa_report(

765

766

767 def run(args: argparse.Namespace) -> dict[str, Path]:

768 + analysis_started = time.monotonic()

769 raw_payload = sanitize_repo_payload(load_json(args.raw_json))

770 week = raw_payload["week"]

771 raw_ref = file_ref(args.raw_json)

@@ -722,8 +775,10 @@ def run(args: argparse.Namespace) -> dict[str, Path]:

775 previous_summary = find_previous_summary(week, args.analyzed_dir)

776 previous_ref = file_ref(previous_summary)

777

725 - maps = {

726 - "new_repos": map_repositories(

778 + map_stage_metrics: list[MapReduceMetrics] = []

779 + maps: dict[str, dict[str, Any]] = {}

780 + map_builders = {

781 + "new_repos": lambda: map_repositories(

782 run_id=args.run_id,

783 week=week,

784 raw_path=args.raw_json,

@@ -733,7 +788,7 @@ def run(args: argparse.Namespace) -> dict[str, Path]:

788 mode="new",

789 max_repos=args.max_repos_per_ledger,

790 ),

736 - "trending_repos": map_repositories(

791 + "trending_repos": lambda: map_repositories(

792 run_id=args.run_id,

793 week=week,

794 raw_path=args.raw_json,

@@ -743,15 +798,60 @@ def run(args: argparse.Namespace) -> dict[str, Path]:

798 mode="trending",

799 max_repos=args.max_repos_per_ledger,

800 ),

746 - "press_correlations": map_press(run_id=args.run_id, week=week, press_path=args.press_context, press_ref=press_ref, raw_ref=raw_ref),

747 - "prior_continuity": map_prior(run_id=args.run_id, week=week, previous_summary=previous_summary, previous_ref=previous_ref, raw_ref=raw_ref),

801 + "press_correlations": lambda: map_press(

802 + run_id=args.run_id,

803 + week=week,

804 + press_path=args.press_context,

805 + press_ref=press_ref,

806 + raw_ref=raw_ref,

807 + ),

808 + "prior_continuity": lambda: map_prior(

809 + run_id=args.run_id,

810 + week=week,

811 + previous_summary=previous_summary,

812 + previous_ref=previous_ref,

813 + raw_ref=raw_ref,

814 + ),

815 }

816 + for name in MAPPER_IDS:

817 + stage_started = time.monotonic()

818 + payload = map_builders[name]()

819 + maps[name] = payload

820 + stage_duration = round(time.monotonic() - stage_started, 3)

821 + input_tokens = int(payload.get("slice", {}).get("input_token_estimate") or 0)

822 + output_tokens = int(payload.get("token_estimate") or 0)

823 + map_stage_metrics.append(

824 + MapReduceMetrics(

825 + stage=name,

826 + duration_seconds=stage_duration,

827 + input_tokens=input_tokens,

828 + output_tokens=output_tokens,

829 + cost_usd=metric_cost_usd(args.analysis_model, input_tokens, output_tokens),

830 + status="pass",

831 + gate_failure_reasons=[],

832 + )

833 + )

834 map_errors = {name: validate_map(payload) for name, payload in maps.items()}

835 if any(map_errors.values()):

836 for name, errors in map_errors.items():

837 if errors:

838 maps[name]["status"] = "failed"

839 maps[name]["errors"] = errors

840 + map_metrics_by_stage = {metric.stage: metric for metric in map_stage_metrics}

841 + for name, errors in map_errors.items():

842 + if errors:

843 + metric = map_metrics_by_stage.get(name)

844 + if metric is not None:

845 + map_stage_metrics[map_stage_metrics.index(metric)] = MapReduceMetrics(

846 + stage=metric.stage,

847 + duration_seconds=metric.duration_seconds,

848 + input_tokens=metric.input_tokens,

849 + output_tokens=metric.output_tokens,

850 + cost_usd=metric.cost_usd,

851 + status="fail",

852 + gate_failure_reasons=list(errors),

853 + )

854 + reduce_started = time.monotonic()

855 plan, rejected, contradictions = reduce_ledgers(list(maps.values()), raw_payload=raw_payload)

856 candidate_text = render_candidate(plan, raw_payload, args.current_datetime)

857

@@ -785,6 +885,19 @@ def run(args: argparse.Namespace) -> dict[str, Path]:

885 source=args.analysis_source,

886 model=args.analysis_model,

887 )

888 + reduce_duration = round(time.monotonic() - reduce_started, 3)

889 + reduce_input_tokens = sum(metric.output_tokens for metric in map_stage_metrics)

890 + reduce_output_tokens = estimate_tokens(candidate_text)

891 + reduce_failure_reasons = collect_gate_failure_reasons(qa)

892 + reduce_stage_metric = MapReduceMetrics(

893 + stage="reduce",

894 + duration_seconds=reduce_duration,

895 + input_tokens=reduce_input_tokens,

896 + output_tokens=reduce_output_tokens,

897 + cost_usd=metric_cost_usd(args.analysis_model, reduce_input_tokens, reduce_output_tokens),

898 + status="pass" if qa.get("status") == "passed" else "fail",

899 + gate_failure_reasons=reduce_failure_reasons,

900 + )

901 write_json(out / "qa-comparison-report.json", qa)

902 raw_component = file_ref(args.raw_json)

903 press_component = file_ref(args.press_context)

@@ -842,10 +955,57 @@ def run(args: argparse.Namespace) -> dict[str, Path]:

955 "promotion_policy": "blocked: dry-run/candidate-only map/reduce output must not write data/analyzed, content/weekly, deploy, notify, or satisfy publish eligibility.",

956 }

957 write_json(out / "manifest.json", manifest)

958 + total_input_tokens = sum(metric.input_tokens for metric in map_stage_metrics) + reduce_stage_metric.input_tokens

959 + total_output_tokens = sum(metric.output_tokens for metric in map_stage_metrics) + reduce_stage_metric.output_tokens

960 + total_cost_usd = round(sum(metric.cost_usd for metric in map_stage_metrics) + reduce_stage_metric.cost_usd, 6)

961 + analysis_duration = round(time.monotonic() - analysis_started, 3)

962 + observability_path = DEFAULT_OBSERVABILITY_DIR / f"{week}-map-reduce.json"

963 + emit_ledger(

964 + ObservabilityLedger(

965 + schema_version=METRICS_SCHEMA_VERSION,

966 + run_id=args.run_id,

967 + week=week,

968 + timestamp=args.current_datetime,

969 + crawl_metrics=[],

970 + analysis_metrics=AnalysisMetrics(

971 + duration_seconds=analysis_duration,

972 + token_ledger={

973 + "input_tokens": total_input_tokens,

974 + "output_tokens": total_output_tokens,

975 + "total_tokens": total_input_tokens + total_output_tokens,

976 + "cost_usd": total_cost_usd,

977 + },

978 + map_stages=map_stage_metrics,

979 + reduce_stage=reduce_stage_metric,

980 + ),

981 + environment={

982 + "pipeline": "map-reduce-dry-run",

983 + "analysis_source": args.analysis_source,

984 + "analysis_model": args.analysis_model,

985 + "output_dir": out.as_posix(),

986 + "qa_status": qa.get("status"),

987 + "publish_eligible": qa.get("publish_eligible"),

988 + "pass_fail_counts": {

989 + "map_pass": sum(1 for metric in map_stage_metrics if metric.status == "pass"),

990 + "map_fail": sum(1 for metric in map_stage_metrics if metric.status == "fail"),

991 + "reduce_pass": 1 if reduce_stage_metric.status == "pass" else 0,

992 + "reduce_fail": 1 if reduce_stage_metric.status == "fail" else 0,

993 + },

994 + "gate_failure_reasons": reduce_failure_reasons,

995 + "artifacts": {

996 + "manifest": (out / "manifest.json").as_posix(),

997 + "candidate": candidate_path.as_posix(),

998 + "qa_report": (out / "qa-comparison-report.json").as_posix(),

999 + },

1000 + },

1001 + ),

1002 + observability_path,

1003 + )

1004 return {

1005 "manifest": out / "manifest.json",

1006 "qa_report": out / "qa-comparison-report.json",

1007 "candidate": candidate_path,

1008 + "observability": observability_path,

1009 }

1010

1011

scripts/observability_metrics.py new

+208

@@ -0,0 +1,208 @@

1 +#!/usr/bin/env python3

2 +"""Observability metrics schema helpers for crawl and analysis pipelines."""

3 +

4 +from __future__ import annotations

5 +

6 +import json

7 +import math

8 +from dataclasses import asdict, dataclass, field, is_dataclass

9 +from pathlib import Path

10 +from typing import Any

11 +

12 +ROOT = Path(__file__).resolve().parent.parent

13 +DEFAULT_OBSERVABILITY_DIR = ROOT / "data" / "metrics" / "observability"

14 +METRICS_SCHEMA_VERSION = "observability_v1"

15 +

16 +

17 +@dataclass(frozen=True, slots=True)

18 +class CrawlMetrics:

19 + duration_seconds: float

20 + api_calls: int

21 + cache_hits: int

22 + cache_misses: int

23 + stale_cache_hits: int

24 + rate_limit_events: int

25 + secondary_rate_limit_hit: bool

26 + source_type: str

27 + duration_p95_seconds: float | None = None

28 + duration_sample_count: int = 0

29 +

30 +

31 +@dataclass(frozen=True, slots=True)

32 +class MapReduceMetrics:

33 + stage: str

34 + duration_seconds: float

35 + input_tokens: int

36 + output_tokens: int

37 + cost_usd: float

38 + status: str

39 + gate_failure_reasons: list[str] = field(default_factory=list)

40 +

41 +

42 +@dataclass(frozen=True, slots=True)

43 +class AnalysisMetrics:

44 + duration_seconds: float

45 + token_ledger: dict[str, Any]

46 + map_stages: list[MapReduceMetrics] = field(default_factory=list)

47 + reduce_stage: MapReduceMetrics | None = None

48 +

49 +

50 +@dataclass(frozen=True, slots=True)

51 +class ObservabilityLedger:

52 + schema_version: str

53 + run_id: str

54 + week: str

55 + timestamp: str

56 + crawl_metrics: list[CrawlMetrics] = field(default_factory=list)

57 + analysis_metrics: AnalysisMetrics | None = None

58 + environment: dict[str, Any] = field(default_factory=dict)

59 +

60 +

61 +def duration_p95(durations: list[float]) -> float:

62 + """Return a simple p95 duration estimate for a sampled duration series."""

63 + if not durations:

64 + return 0.0

65 + ordered = sorted(float(value) for value in durations)

66 + index = max(0, math.ceil(len(ordered) * 0.95) - 1)

67 + return round(ordered[index], 3)

68 +

69 +

70 +def emit_ledger(ledger: ObservabilityLedger | dict[str, Any], output_path: Path) -> Path:

71 + """Write a validated observability ledger JSON artifact."""

72 + payload = to_serializable(ledger)

73 + errors = validate_ledger(payload)

74 + if errors:

75 + raise ValueError(f"Invalid observability ledger: {', '.join(errors)}")

76 + output_path.parent.mkdir(parents=True, exist_ok=True)

77 + output_path.write_text(

78 + json.dumps(payload, indent=2, sort_keys=True, ensure_ascii=False) + "\n",

79 + encoding="utf-8",

80 + )

81 + return output_path

82 +

83 +

84 +def validate_ledger(data: dict[str, Any]) -> list[str]:

85 + """Return missing or malformed required field paths for an observability ledger."""

86 + errors: list[str] = []

87 + if not isinstance(data, dict):

88 + return ["ledger"]

89 +

90 + _require(data, "schema_version", errors)

91 + _require(data, "run_id", errors)

92 + _require(data, "week", errors)

93 + _require(data, "timestamp", errors)

94 + _require(data, "crawl_metrics", errors)

95 + _require(data, "environment", errors)

96 + if "environment" in data and not isinstance(data["environment"], dict):

97 + errors.append("environment: must be a dict")

98 + sv = data.get("schema_version")

99 + if sv is not None and sv != METRICS_SCHEMA_VERSION:

100 + errors.append(f"schema_version: expected '{METRICS_SCHEMA_VERSION}', got '{sv}'")

101 +

102 + crawl_metrics = data.get("crawl_metrics")

103 + if crawl_metrics is not None and not isinstance(crawl_metrics, list):

104 + errors.append("crawl_metrics")

105 + elif isinstance(crawl_metrics, list):

106 + for index, metric in enumerate(crawl_metrics):

107 + if not isinstance(metric, dict):

108 + errors.append(f"crawl_metrics[{index}]")

109 + continue

110 + _validate_required_fields(

111 + metric,

112 + [

113 + "duration_seconds",

114 + "api_calls",

115 + "cache_hits",

116 + "cache_misses",

117 + "stale_cache_hits",

118 + "rate_limit_events",

119 + "secondary_rate_limit_hit",

120 + "source_type",

121 + ],

122 + f"crawl_metrics[{index}]",

123 + errors,

124 + )

125 +

126 + analysis_metrics = data.get("analysis_metrics")

127 + if analysis_metrics is not None:

128 + if not isinstance(analysis_metrics, dict):

129 + errors.append("analysis_metrics")

130 + else:

131 + _validate_required_fields(

132 + analysis_metrics,

133 + ["duration_seconds", "token_ledger", "map_stages"],

134 + "analysis_metrics",

135 + errors,

136 + )

137 + token_ledger = analysis_metrics.get("token_ledger")

138 + if not isinstance(token_ledger, dict):

139 + errors.append("analysis_metrics.token_ledger")

140 + else:

141 + _validate_required_fields(

142 + token_ledger,

143 + ["input_tokens", "output_tokens", "total_tokens"],

144 + "analysis_metrics.token_ledger",

145 + errors,

146 + )

147 + map_stages = analysis_metrics.get("map_stages")

148 + if not isinstance(map_stages, list):

149 + errors.append("analysis_metrics.map_stages")

150 + else:

151 + for index, metric in enumerate(map_stages):

152 + _validate_stage(metric, f"analysis_metrics.map_stages[{index}]", errors)

153 + reduce_stage = analysis_metrics.get("reduce_stage")

154 + if reduce_stage is not None:

155 + _validate_stage(reduce_stage, "analysis_metrics.reduce_stage", errors)

156 +

157 + return errors

158 +

159 +

160 +def to_serializable(value: Any) -> Any:

161 + """Normalize dataclasses, paths, and containers into JSON-serializable values."""

162 + if is_dataclass(value):

163 + return to_serializable(asdict(value))

164 + if isinstance(value, Path):

165 + return value.as_posix()

166 + if isinstance(value, dict):

167 + return {str(key): to_serializable(item) for key, item in value.items()}

168 + if isinstance(value, list):

169 + return [to_serializable(item) for item in value]

170 + if isinstance(value, tuple):

171 + return [to_serializable(item) for item in value]

172 + return value

173 +

174 +

175 +def _require(data: dict[str, Any], field_name: str, errors: list[str]) -> None:

176 + if field_name not in data:

177 + errors.append(field_name)

178 +

179 +

180 +def _validate_required_fields(

181 + payload: dict[str, Any],

182 + field_names: list[str],

183 + prefix: str,

184 + errors: list[str],

185 +) -> None:

186 + for field_name in field_names:

187 + if field_name not in payload:

188 + errors.append(f"{prefix}.{field_name}")

189 +

190 +

191 +def _validate_stage(stage: Any, prefix: str, errors: list[str]) -> None:

192 + if not isinstance(stage, dict):

193 + errors.append(prefix)

194 + return

195 + _validate_required_fields(

196 + stage,

197 + [

198 + "stage",

199 + "duration_seconds",

200 + "input_tokens",

201 + "output_tokens",

202 + "cost_usd",

203 + "status",

204 + "gate_failure_reasons",

205 + ],

206 + prefix,

207 + errors,

208 + )

scripts/techcrunch_crawler.py

+61 -1

@@ -30,6 +30,14 @@ from urllib.request import Request, urlopen

31 import feedparser

33 +from scripts.observability_metrics import (

34 + DEFAULT_OBSERVABILITY_DIR,

35 + CrawlMetrics,

36 + METRICS_SCHEMA_VERSION,

37 + ObservabilityLedger,

38 + duration_p95,

39 + emit_ledger,

40 +)

41 from scripts.topic_paths import raw_dir

43 FEED_URL = "https://techcrunch.com/feed/"

@@ -969,6 +977,7 @@ def validate_canonical_output(output: dict[str, Any]) -> None:

977

978

979 def main(argv: list[str] | None = None) -> int:

980 + crawl_started = time.monotonic()

981 parser = argparse.ArgumentParser(

982 description="Crawl external news RSS feeds for SquadScope"

983 )

@@ -1132,13 +1141,64 @@ def main(argv: list[str] | None = None) -> int:

1141 with open(out_path, "w", encoding="utf-8") as f:

1142 json.dump(output, f, indent=2, ensure_ascii=False)

1143

1144 + total_duration_seconds = round(time.monotonic() - crawl_started, 3)

1145 + sampled_durations = [

1146 + float(status["duration_seconds"])

1147 + for status in statuses

1148 + if isinstance(status.get("duration_seconds"), (int, float))

1149 + ]

1150 + if sampled_durations:

1151 + sampled_p95 = duration_p95(sampled_durations)

1152 + sample_count = len(sampled_durations)

1153 + else:

1154 + # Treat overall runtime as a single sample for consistency

1155 + sampled_p95 = total_duration_seconds

1156 + sample_count = 1

1157 + observability_path = DEFAULT_OBSERVABILITY_DIR / f"{output['week']}-external-news-crawl.json"

1158 + emit_ledger(

1159 + ObservabilityLedger(

1160 + schema_version=METRICS_SCHEMA_VERSION,

1161 + run_id=os.environ.get("GITHUB_RUN_ID", "local"),

1162 + week=output["week"],

1163 + timestamp=output["crawled_at"],

1164 + crawl_metrics=[

1165 + CrawlMetrics(

1166 + duration_seconds=sampled_p95,

1167 + duration_p95_seconds=sampled_p95,

1168 + duration_sample_count=sample_count,

1169 + api_calls=sum(int(status.get("attempts") or 0) for status in statuses),

1170 + cache_hits=0,

1171 + cache_misses=sum(int(status.get("attempts") or 0) for status in statuses),

1172 + stale_cache_hits=0,

1173 + rate_limit_events=0,

1174 + secondary_rate_limit_hit=False,

1175 + source_type="external-news",

1176 + )

1177 + ],

1178 + analysis_metrics=None,

1179 + environment={

1180 + "pipeline": "external-news-crawl",

1181 + "output_path": out_path.as_posix(),

1182 + "source_refresh_policy": source_refresh_policy,

1183 + "same_day_reuse_status": output["metadata"]["same_day_reuse"],

1184 + "sources_requested": list(output["metadata"]["sources_requested"]),

1185 + "sources_succeeded": list(output["metadata"]["sources_succeeded"]),

1186 + "sources_failed": list(output["metadata"]["sources_failed"]),

1187 + "errors": list(errors),

1188 + "total_run_duration_seconds": total_duration_seconds,

1189 + },

1190 + ),

1191 + observability_path,

1192 + )

1193 +

1194 reused_count = sum(1 for item in output["metadata"]["source_reuse_summary"] if item["action"] == "reused")

1195 refreshed_count = sum(1 for item in output["metadata"]["source_reuse_summary"] if item["action"] != "reused")

1196 print(f"Crawled {output['metadata']['total_articles']} articles "

1197 f"from {output['metadata']['source_count']} sources "

1198 f"({output['metadata']['relevant_articles']} relevant, "

1199 f"{output['metadata']['dedupe_count']} deduped, "

1141 - f"{reused_count} reused, {refreshed_count} refreshed) → {out_path}")

1200 + f"{reused_count} reused, {refreshed_count} refreshed, p95={sampled_p95:.3f}s) "

1201 + f"→ {out_path} [observability={observability_path}]")

1202 return 0

1203

1204

tests/fixtures/observability/2026-W21-full-run.json new

+111

@@ -0,0 +1,111 @@

1 +{

2 + "schema_version": "observability_v1",

3 + "run_id": "356-demo",

4 + "week": "2026-W21",

5 + "timestamp": "2026-05-20T12:00:00Z",

6 + "crawl_metrics": [

7 + {

8 + "api_calls": 27,

9 + "cache_hits": 14,

10 + "cache_misses": 27,

11 + "duration_p95_seconds": 12.417,

12 + "duration_sample_count": 1,

13 + "duration_seconds": 12.417,

14 + "rate_limit_events": 2,

15 + "secondary_rate_limit_hit": false,

16 + "source_type": "github",

17 + "stale_cache_hits": 1

18 + },

19 + {

20 + "api_calls": 5,

21 + "cache_hits": 0,

22 + "cache_misses": 5,

23 + "duration_p95_seconds": 1.842,

24 + "duration_sample_count": 4,

25 + "duration_seconds": 1.842,

26 + "rate_limit_events": 0,

27 + "secondary_rate_limit_hit": false,

28 + "source_type": "external-news",

29 + "stale_cache_hits": 0

30 + }

31 + ],

32 + "analysis_metrics": {

33 + "duration_seconds": 3.281,

34 + "map_stages": [

35 + {

36 + "cost_usd": 0.0,

37 + "duration_seconds": 0.421,

38 + "gate_failure_reasons": [],

39 + "input_tokens": 820,

40 + "output_tokens": 260,

41 + "stage": "new_repos",

42 + "status": "pass"

43 + },

44 + {

45 + "cost_usd": 0.0,

46 + "duration_seconds": 0.397,

47 + "gate_failure_reasons": [],

48 + "input_tokens": 790,

49 + "output_tokens": 248,

50 + "stage": "trending_repos",

51 + "status": "pass"

52 + },

53 + {

54 + "cost_usd": 0.0,

55 + "duration_seconds": 0.205,

56 + "gate_failure_reasons": [],

57 + "input_tokens": 180,

58 + "output_tokens": 88,

59 + "stage": "press_correlations",

60 + "status": "pass"

61 + },

62 + {

63 + "cost_usd": 0.0,

64 + "duration_seconds": 0.188,

65 + "gate_failure_reasons": [],

66 + "input_tokens": 95,

67 + "output_tokens": 61,

68 + "stage": "prior_continuity",

69 + "status": "pass"

70 + }

71 + ],

72 + "reduce_stage": {

73 + "cost_usd": 0.0,

74 + "duration_seconds": 0.994,

75 + "gate_failure_reasons": [

76 + "Structural analysis gate failed: candidate below minimum word count"

77 + ],

78 + "input_tokens": 657,

79 + "output_tokens": 512,

80 + "stage": "reduce",

81 + "status": "fail"

82 + },

83 + "token_ledger": {

84 + "cost_usd": 0.0,

85 + "input_tokens": 2542,

86 + "output_tokens": 1169,

87 + "total_tokens": 3711

88 + }

89 + },

90 + "environment": {

91 + "analysis_model": "local-deterministic",

92 + "analysis_source": "map-reduce-dry-run",

93 + "artifacts": {

94 + "candidate": "data/candidates/2026-W21/local/2026-W21-map-reduce-candidate.md",

95 + "manifest": "data/candidates/2026-W21/local/manifest.json",

96 + "qa_report": "data/candidates/2026-W21/local/qa-comparison-report.json"

97 + },

98 + "gate_failure_reasons": [

99 + "Structural analysis gate failed: candidate below minimum word count"

100 + ],

101 + "pass_fail_counts": {

102 + "map_fail": 0,

103 + "map_pass": 4,

104 + "reduce_fail": 1,

105 + "reduce_pass": 0

106 + },

107 + "pipeline": "full-run-demo",

108 + "publish_eligible": false,

109 + "qa_status": "failed"

110 + }

111 +}

tests/test_crawl.py

+49

@@ -170,6 +170,55 @@ class CrawlTests(unittest.TestCase):

170 ],

171 )

172

173 + def test_main_emits_observability_ledger(self) -> None:

174 + class FakeClient:

175 + def __init__(self, token: str, **kwargs) -> None:

176 + self.token = token

177 + self.api_calls_used = 4

178 + self.cache_hits = 3

179 + self.cache_misses = 4

180 + self.stale_cache_hits = 1

181 + self.rate_limit_events = 2

182 + self.secondary_rate_limit_hit = True

183 + self.rate_limit_limit = 5000

184 + self.rate_limit_remaining = 4988

185 + self.rate_limit_reset = 1747562400

186 + self.rate_limit_resource = "search"

187 + self.errors = []

188 +

189 + def search_repositories(self, query: str, *, max_results: int = 1000):

190 + return []

191 +

192 + def has_readme(self, full_name: str) -> bool:

193 + return True

194 +

195 + args = Namespace(

196 + since="2026-05-11",

197 + as_of="2026-05-18",

198 + max_results=25,

199 + output="data/raw/test-observability.json",

200 + topic="general",

201 + config=None,

202 + )

203 + with mock.patch.object(crawl, "parse_args", return_value=args), mock.patch.dict(

204 + "os.environ", {"GITHUB_TOKEN": "token", "GITHUB_RUN_ID": "123"}, clear=False

205 + ), mock.patch.object(crawl, "GitHubClient", FakeClient), mock.patch.object(

206 + crawl, "load_previous_star_snapshot", return_value={}

207 + ), mock.patch.object(crawl, "write_payload"), mock.patch.object(

208 + crawl, "emit_ledger"

209 + ) as emit_mock, mock.patch.object(crawl, "print"):

210 + exit_code = crawl.main()

211 +

212 + self.assertEqual(exit_code, 0)

213 + ledger = emit_mock.call_args.args[0]

214 + output_path = emit_mock.call_args.args[1]

215 + self.assertEqual(ledger.schema_version, "observability_v1")

216 + self.assertEqual(ledger.run_id, "123")

217 + self.assertEqual(ledger.crawl_metrics[0].source_type, "github")

218 + self.assertEqual(ledger.crawl_metrics[0].cache_misses, 4)

219 + self.assertTrue(ledger.crawl_metrics[0].secondary_rate_limit_hit)

220 + self.assertTrue(output_path.as_posix().endswith("-github-crawl.json"))

221 +

222 def test_load_reusable_github_payload_accepts_same_day_matching_artifact(self) -> None:

223 tests_root = Path(__file__).resolve().parent

224 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:

tests/test_map_reduce_dry_run.py

+48 -14

@@ -3,8 +3,10 @@ from __future__ import annotations

3 import json

4 import tempfile

5 from pathlib import Path

6 +from unittest.mock import patch

8 from scripts import map_reduce_dry_run as dry_run

9 +from scripts.observability_metrics import validate_ledger

12 def make_repo(owner: str, name: str, stars: int, gained: int = 0) -> dict[str, object]:

@@ -26,6 +28,8 @@ def test_dry_run_emits_valid_contract_artifacts() -> None:

28 tests_root = Path(__file__).resolve().parent

29 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:

30 base = Path(tmpdir)

31 + obs_dir = base / "observability"

32 + obs_dir.mkdir(parents=True)

33 raw_path = base / "data" / "raw" / "2026-W21.json"

34 press_path = base / "data" / "analyzed" / "2026-W21-press-context.md"

35 output_dir = base / "data" / "candidates" / "2026-W21" / "local" / "map-reduce"

@@ -44,20 +48,21 @@ def test_dry_run_emits_valid_contract_artifacts() -> None:

48 encoding="utf-8",

49 )

47 - rc = dry_run.main(

48 - [

49 - "--raw-json",

50 - raw_path.as_posix(),

51 - "--press-context",

52 - press_path.as_posix(),

53 - "--output-dir",

54 - output_dir.as_posix(),

55 - "--current-datetime",

56 - "2026-05-20T12:00:00Z",

57 - "--run-id",

58 - "local",

59 - ]

60 - )

51 + with patch.object(dry_run, "DEFAULT_OBSERVABILITY_DIR", obs_dir):

52 + rc = dry_run.main(

53 + [

54 + "--raw-json",

55 + raw_path.as_posix(),

56 + "--press-context",

57 + press_path.as_posix(),

58 + "--output-dir",

59 + output_dir.as_posix(),

60 + "--current-datetime",

61 + "2026-05-20T12:00:00Z",

62 + "--run-id",

63 + "local",

64 + ]

65 + )

67 assert rc == 0

68 manifest = json.loads((output_dir / "manifest.json").read_text(encoding="utf-8"))

@@ -85,6 +90,11 @@ def test_dry_run_emits_valid_contract_artifacts() -> None:

90 assert qa["checks"]["structural_analysis_gate"]["passed"] is True

91 assert qa["checks"]["evidence_and_editorial_gates"]["passed"] is True

92 assert qa["checks"]["publish_provenance_gate"]["expected_failure"] is True

93 + observability_path = obs_dir / "2026-W21-map-reduce.json"

94 + observability = json.loads(observability_path.read_text(encoding="utf-8"))

95 + assert validate_ledger(observability) == []

96 + assert observability["analysis_metrics"]["reduce_stage"]["status"] == "pass"

97 + assert observability["environment"]["pass_fail_counts"]["map_pass"] == 4

100 def test_validate_map_rejects_citationless_findings() -> None:

@@ -185,6 +195,30 @@ def test_validate_map_rejects_failed_or_low_coverage() -> None:

195 assert "mapper status failed" in errors

196

197

198 +def test_collect_gate_failure_reasons_skips_expected_failures() -> None:

199 + reasons = dry_run.collect_gate_failure_reasons(

200 + {

201 + "checks": {

202 + "mapper_contracts": {"passed": False, "errors_by_mapper": {"new_repos": ["missing evidence refs"]}},

203 + "structural_analysis_gate": {"passed": False, "errors": ["candidate below minimum word count"]},

204 + "publish_provenance_gate": {

205 + "passed": False,

206 + "expected_failure": True,

207 + "errors": ["AI provenance metadata missing"],

208 + },

209 + "sidecars_present": {"passed": False},

210 + },

211 + "regressions": ["baseline summary not found"],

212 + }

213 + )

214 +

215 + assert "new_repos: missing evidence refs" in reasons

216 + assert "candidate below minimum word count" in reasons

217 + assert "baseline summary not found" in reasons

218 + assert "sidecars_present failed" in reasons

219 + assert "AI provenance metadata missing" not in reasons

220 +

221 +

222 def test_reduce_rejects_and_preserves_contradictory_claims() -> None:

223 supported = valid_ledger()["findings"][0]

224 contradictory = {

tests/test_observability_metrics.py new

+118

@@ -0,0 +1,118 @@

1 +from __future__ import annotations

2 +

3 +import json

4 +import tempfile

5 +from pathlib import Path

6 +

7 +from scripts.observability_metrics import (

8 + AnalysisMetrics,

9 + CrawlMetrics,

10 + METRICS_SCHEMA_VERSION,

11 + MapReduceMetrics,

12 + ObservabilityLedger,

13 + duration_p95,

14 + emit_ledger,

15 + validate_ledger,

16 +)

17 +

18 +

19 +def sample_ledger() -> ObservabilityLedger:

20 + return ObservabilityLedger(

21 + schema_version=METRICS_SCHEMA_VERSION,

22 + run_id="local",

23 + week="2026-W21",

24 + timestamp="2026-05-20T12:00:00Z",

25 + crawl_metrics=[

26 + CrawlMetrics(

27 + duration_seconds=12.4,

28 + duration_p95_seconds=12.4,

29 + duration_sample_count=1,

30 + api_calls=10,

31 + cache_hits=5,

32 + cache_misses=10,

33 + stale_cache_hits=1,

34 + rate_limit_events=2,

35 + secondary_rate_limit_hit=False,

36 + source_type="github",

37 + )

38 + ],

39 + analysis_metrics=AnalysisMetrics(

40 + duration_seconds=3.2,

41 + token_ledger={

42 + "input_tokens": 100,

43 + "output_tokens": 25,

44 + "total_tokens": 125,

45 + "cost_usd": 0.0,

46 + },

47 + map_stages=[

48 + MapReduceMetrics(

49 + stage="new_repos",

50 + duration_seconds=0.3,

51 + input_tokens=50,

52 + output_tokens=10,

53 + cost_usd=0.0,

54 + status="pass",

55 + gate_failure_reasons=[],

56 + )

57 + ],

58 + reduce_stage=MapReduceMetrics(

59 + stage="reduce",

60 + duration_seconds=0.8,

61 + input_tokens=10,

62 + output_tokens=15,

63 + cost_usd=0.0,

64 + status="fail",

65 + gate_failure_reasons=["AI provenance metadata missing"],

66 + ),

67 + ),

68 + environment={"pipeline": "test"},

69 + )

70 +

71 +

72 +def test_validate_ledger_reports_missing_required_fields() -> None:

73 + errors = validate_ledger({"schema_version": METRICS_SCHEMA_VERSION})

74 +

75 + assert "run_id" in errors

76 + assert "crawl_metrics" in errors

77 + assert "environment" in errors

78 +

79 +

80 +def test_duration_p95_uses_high_percentile_sample() -> None:

81 + assert duration_p95([]) == 0.0

82 + assert duration_p95([0.5]) == 0.5

83 + assert duration_p95([0.2, 0.4, 0.6, 0.8, 1.0]) == 1.0

84 +

85 +

86 +def test_validate_ledger_rejects_schema_version_mismatch() -> None:

87 + payload = {

88 + "schema_version": "observability_v0",

89 + "run_id": "local",

90 + "week": "2026-W21",

91 + "timestamp": "2026-05-20T12:00:00Z",

92 + "crawl_metrics": [],

93 + "environment": {},

94 + }

95 +

96 + errors = validate_ledger(payload)

97 + assert any(e.startswith("schema_version") for e in errors), f"Expected schema_version error, got: {errors}"

98 +

99 +

100 +def test_emit_ledger_writes_valid_json() -> None:

101 + tests_root = Path(__file__).resolve().parent

102 + with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:

103 + output_path = Path(tmpdir) / "observability.json"

104 + emit_ledger(sample_ledger(), output_path)

105 +

106 + payload = json.loads(output_path.read_text(encoding="utf-8"))

107 +

108 + assert payload["schema_version"] == METRICS_SCHEMA_VERSION

109 + assert payload["analysis_metrics"]["reduce_stage"]["status"] == "fail"

110 + assert validate_ledger(payload) == []

111 +

112 +

113 +def test_representative_fixture_is_valid() -> None:

114 + fixture_path = Path(__file__).resolve().parent / "fixtures" / "observability" / "2026-W21-full-run.json"

115 + payload = json.loads(fixture_path.read_text(encoding="utf-8"))

116 +

117 + assert validate_ledger(payload) == []

118 + assert payload["environment"]["pass_fail_counts"]["reduce_fail"] == 1

tests/test_qa_gates_map_reduce.py

+8 -14

@@ -12,6 +12,7 @@ These gates run as part of the existing pytest CI path (issue #438).

12 from __future__ import annotations

14 import json

15 +import tempfile

16 from pathlib import Path

18 import pytest

@@ -294,14 +295,13 @@ class TestEndToEndDryRun:

295 def test_all_mapper_contracts_valid(self, workspace):

296 """Each mapper ledger passes validate_map with zero errors."""

297 raw_path, press_path, output_dir = workspace

297 - rc = dry_run.main([

298 + dry_run.main([

299 "--raw-json", raw_path.as_posix(),

300 "--press-context", press_path.as_posix(),

301 "--output-dir", output_dir.as_posix(),

302 "--current-datetime", "2026-05-20T12:00:00Z",

303 "--run-id", "qa-gate-test",

304 ])

304 - assert rc == 0

305

306 for mapper in dry_run.MAPPER_IDS:

307 ledger = json.loads((output_dir / "maps" / f"{mapper}.json").read_text(encoding="utf-8"))

@@ -311,14 +311,13 @@ class TestEndToEndDryRun:

311 def test_sidecars_always_present(self, workspace):

312 """rejected-claims.json and contradictions.json are always emitted."""

313 raw_path, press_path, output_dir = workspace

314 - rc = dry_run.main([

314 + dry_run.main([

315 "--raw-json", raw_path.as_posix(),

316 "--press-context", press_path.as_posix(),

317 "--output-dir", output_dir.as_posix(),

318 "--current-datetime", "2026-05-20T12:00:00Z",

319 "--run-id", "qa-gate-test",

320 ])

321 - assert rc == 0

321

322 rejected = json.loads((output_dir / "sidecars" / "rejected-claims.json").read_text(encoding="utf-8"))

323 contras = json.loads((output_dir / "sidecars" / "contradictions.json").read_text(encoding="utf-8"))

@@ -330,14 +329,13 @@ class TestEndToEndDryRun:

329 def test_manifest_is_never_publish_eligible(self, workspace):

330 """Dry-run manifest always marks candidate_only=True, publish_eligible=False."""

331 raw_path, press_path, output_dir = workspace

333 - rc = dry_run.main([

332 + dry_run.main([

333 "--raw-json", raw_path.as_posix(),

334 "--press-context", press_path.as_posix(),

335 "--output-dir", output_dir.as_posix(),

336 "--current-datetime", "2026-05-20T12:00:00Z",

337 "--run-id", "qa-gate-test",

338 ])

340 - assert rc == 0

339

340 manifest = json.loads((output_dir / "manifest.json").read_text(encoding="utf-8"))

341 assert manifest["publish_eligible"] is False

@@ -346,14 +344,13 @@ class TestEndToEndDryRun:

344 def test_qa_report_documents_expected_provenance_failure(self, workspace):

345 """The provenance gate fails as expected (dry-run is not publishable AI)."""

346 raw_path, press_path, output_dir = workspace

349 - rc = dry_run.main([

347 + dry_run.main([

348 "--raw-json", raw_path.as_posix(),

349 "--press-context", press_path.as_posix(),

350 "--output-dir", output_dir.as_posix(),

351 "--current-datetime", "2026-05-20T12:00:00Z",

352 "--run-id", "qa-gate-test",

353 ])

356 - assert rc == 0

354

355 qa = json.loads((output_dir / "qa-comparison-report.json").read_text(encoding="utf-8"))

356 provenance = qa["checks"]["publish_provenance_gate"]

@@ -363,14 +360,13 @@ class TestEndToEndDryRun:

360 def test_candidate_markdown_contains_repo_links(self, workspace):

361 """Candidate markdown includes hyperlinks to featured repositories."""

362 raw_path, press_path, output_dir = workspace

366 - rc = dry_run.main([

363 + dry_run.main([

364 "--raw-json", raw_path.as_posix(),

365 "--press-context", press_path.as_posix(),

366 "--output-dir", output_dir.as_posix(),

367 "--current-datetime", "2026-05-20T12:00:00Z",

368 "--run-id", "qa-gate-test",

369 ])

373 - assert rc == 0

370

371 candidate = (output_dir / "2026-W21-map-reduce-candidate.md").read_text(encoding="utf-8")

372 assert "[tools/gamma](https://github.com/tools/gamma)" in candidate

@@ -440,14 +436,13 @@ class TestCostTokenGuardrails:

436 raw_path.write_text(json.dumps(RAW_PAYLOAD), encoding="utf-8")

437 press_path.write_text(PRESS_CONTEXT, encoding="utf-8")

438

443 - rc = dry_run.main([

439 + dry_run.main([

440 "--raw-json", raw_path.as_posix(),

441 "--press-context", press_path.as_posix(),

442 "--output-dir", output_dir.as_posix(),

443 "--current-datetime", "2026-05-20T12:00:00Z",

444 "--run-id", "cost-test",

445 ])

450 - assert rc == 0, f"dry_run.main failed with return code {rc}"

446

447 manifest = json.loads((output_dir / "manifest.json").read_text(encoding="utf-8"))

448 est = manifest["component_estimates"]["rendered_prompt_estimate"]

@@ -578,14 +573,13 @@ class TestGateOutputClarity:

573 raw_path.write_text(json.dumps(RAW_PAYLOAD), encoding="utf-8")

574 press_path.write_text(PRESS_CONTEXT, encoding="utf-8")

575

581 - rc = dry_run.main([

576 + dry_run.main([

577 "--raw-json", raw_path.as_posix(),

578 "--press-context", press_path.as_posix(),

579 "--output-dir", output_dir.as_posix(),

580 "--current-datetime", "2026-05-20T12:00:00Z",

581 "--run-id", "clarity-test",

582 ])

588 - assert rc == 0

583

584 qa = json.loads((output_dir / "qa-comparison-report.json").read_text(encoding="utf-8"))

585 # Each check section has "passed" bool and either "errors" or "expected_failure"

tests/test_techcrunch_crawler.py

+89

@@ -3,12 +3,15 @@

3 from __future__ import annotations

5 import json

6 +import tempfile

7 from datetime import UTC, datetime, timedelta

8 +from pathlib import Path

9 from types import SimpleNamespace

10 from unittest.mock import patch

12 import pytest

14 +import scripts.techcrunch_crawler as techcrunch_crawler

15 from scripts.techcrunch_crawler import (

16 DEFAULT_SOURCES_PATH,

17 DEFAULT_FETCH_TIMEOUT_SECONDS,

@@ -805,3 +808,89 @@ class TestSameDaySourceReuse:

808 assert count_once == count_twice == 1

809 assert deduped_once == deduped_twice

810 assert deduped_once[0]["sources"] == ["alpha", "beta"]

811 +

812 +

813 +def test_main_emits_observability_ledger() -> None:

814 + tests_root = Path(__file__).resolve().parent

815 + with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:

816 + base = Path(tmpdir)

817 + sources_path = base / "sources.json"

818 + output_path = base / "2026-W21-external-news.json"

819 + sources_path.write_text(

820 + json.dumps(

821 + [

822 + {"name": "alpha", "feed_url": "https://techcrunch.com/feed/"},

823 + {"name": "beta", "feed_url": "https://github.blog/feed/"}

824 + ]

825 + ),

826 + encoding="utf-8",

827 + )

828 + statuses = [

829 + {

830 + "source": "alpha",

831 + "host": "techcrunch.com",

832 + "success": True,

833 + "attempts": 1,

834 + "timeout_seconds": 15,

835 + "total_articles": 1,

836 + "relevant_articles": 1,

837 + "github_links_found": 1,

838 + "started_at": "2026-05-19T08:00:00Z",

839 + "ended_at": "2026-05-19T08:00:01Z",

840 + "duration_seconds": 1.0,

841 + "error_class": "",

842 + "error_message": "",

843 + },

844 + {

845 + "source": "beta",

846 + "host": "github.blog",

847 + "success": True,

848 + "attempts": 2,

849 + "timeout_seconds": 15,

850 + "total_articles": 1,

851 + "relevant_articles": 1,

852 + "github_links_found": 0,

853 + "started_at": "2026-05-19T08:00:00Z",

854 + "ended_at": "2026-05-19T08:00:02Z",

855 + "duration_seconds": 2.0,

856 + "error_class": "",

857 + "error_message": "",

858 + },

859 + ]

860 + articles = [

861 + {

862 + "source": "alpha",

863 + "title": "Alpha",

864 + "url": "https://example.com/alpha",

865 + "published_at": "2026-05-19T10:00:00Z",

866 + "categories": ["AI"],

867 + "summary": "alpha summary",

868 + "github_links": ["https://github.com/octo/alpha"],

869 + "entities": ["Alpha"],

870 + "relevance_score": 0.8,

871 + }

872 + ]

873 + with patch.object(

874 + techcrunch_crawler, "crawl_sources_parallel", return_value=(articles, [], statuses)

875 + ), patch.object(techcrunch_crawler, "emit_ledger") as emit_mock, patch.object(

876 + techcrunch_crawler, "print"

877 + ):

878 + rc = techcrunch_crawler.main(

879 + [

880 + "--sources",

881 + sources_path.as_posix(),

882 + "--output",

883 + output_path.as_posix(),

884 + "--since",

885 + "2026-05-12",

886 + "--until",

887 + "2026-05-19",

888 + ]

889 + )

890 +

891 + assert rc == 0

892 + ledger = emit_mock.call_args.args[0]

893 + assert ledger.schema_version == "observability_v1"

894 + assert ledger.crawl_metrics[0].source_type == "external-news"

895 + assert ledger.crawl_metrics[0].api_calls == 3

896 + assert ledger.crawl_metrics[0].duration_p95_seconds == 2.0