| 1 | #!/usr/bin/env python3 |
| 2 | """Shared run-context schema for crawl matrix legs and fan-in validation. |
| 3 | |
| 4 | This module defines the canonical run-context schema that all crawl legs, |
| 5 | fan-in validators, and downstream consumers share. No matrix leg may |
| 6 | independently compute its own time window from the local wall clock. |
| 7 | |
| 8 | The run context is the single source of truth for: |
| 9 | - run_id: stable idempotency key |
| 10 | - week: ISO week identifier |
| 11 | - since/until: inclusive start and exclusive end of the collection window |
| 12 | - config checksums: detect drift between legs |
| 13 | - code_sha: pipeline version pinning |
| 14 | |
| 15 | References: |
| 16 | - Issue #333: Define crawl matrix readiness and fan-in validation path |
| 17 | - docs/matrix-crawl-fan-in-contracts.md: Full contract specification |
| 18 | """ |
| 19 | |
| 20 | from __future__ import annotations |
| 21 | |
| 22 | import hashlib |
| 23 | import json |
| 24 | import re |
| 25 | from dataclasses import asdict, dataclass |
| 26 | from datetime import UTC, datetime |
| 27 | from pathlib import Path |
| 28 | from typing import Any |
| 29 | |
| 30 | SCHEMA_VERSION = "run_context_v1" |
| 31 | |
| 32 | # ISO week pattern: YYYY-WNN |
| 33 | _WEEK_RE = re.compile(r"^\d{4}-W(?:0[1-9]|[1-4]\d|5[0-3])$") |
| 34 | |
| 35 | # ISO-8601 timestamp pattern (basic check) |
| 36 | _ISO_TS_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})$") |
| 37 | |
| 38 | |
| 39 | @dataclass(frozen=True, slots=True) |
| 40 | class RunContext: |
| 41 | """Immutable shared run context distributed to all crawl legs.""" |
| 42 | |
| 43 | schema_version: str |
| 44 | run_id: str |
| 45 | week: str |
| 46 | since: str |
| 47 | until: str |
| 48 | source_config_checksum: str |
| 49 | topic_config_checksum: str |
| 50 | code_sha: str |
| 51 | created_at: str |
| 52 | |
| 53 | def to_dict(self) -> dict[str, Any]: |
| 54 | return asdict(self) |
| 55 | |
| 56 | def to_json(self, **kwargs: Any) -> str: |
| 57 | return json.dumps(self.to_dict(), sort_keys=True, ensure_ascii=False, **kwargs) |
| 58 | |
| 59 | @classmethod |
| 60 | def from_dict(cls, data: dict[str, Any]) -> "RunContext": |
| 61 | return cls( |
| 62 | schema_version=data["schema_version"], |
| 63 | run_id=data["run_id"], |
| 64 | week=data["week"], |
| 65 | since=data["since"], |
| 66 | until=data["until"], |
| 67 | source_config_checksum=data["source_config_checksum"], |
| 68 | topic_config_checksum=data["topic_config_checksum"], |
| 69 | code_sha=data["code_sha"], |
| 70 | created_at=data["created_at"], |
| 71 | ) |
| 72 | |
| 73 | @classmethod |
| 74 | def from_json(cls, text: str) -> "RunContext": |
| 75 | return cls.from_dict(json.loads(text)) |
| 76 | |
| 77 | |
| 78 | class RunContextValidationError(Exception): |
| 79 | """Raised when run context validation fails.""" |
| 80 | |
| 81 | pass |
| 82 | |
| 83 | |
| 84 | def build_run_context( |
| 85 | *, |
| 86 | week: str, |
| 87 | since: datetime, |
| 88 | until: datetime, |
| 89 | source_config_checksum: str, |
| 90 | topic_config_checksum: str, |
| 91 | code_sha: str, |
| 92 | created_at: datetime | None = None, |
| 93 | run_id: str | None = None, |
| 94 | ) -> RunContext: |
| 95 | """Build an immutable run context for a crawl run. |
| 96 | |
| 97 | The run_id is derived deterministically from the week and checksums |
| 98 | unless explicitly provided. |
| 99 | """ |
| 100 | now = created_at or datetime.now(UTC) |
| 101 | since_str = since.strftime("%Y-%m-%dT%H:%M:%SZ") |
| 102 | until_str = until.strftime("%Y-%m-%dT%H:%M:%SZ") |
| 103 | created_str = now.strftime("%Y-%m-%dT%H:%M:%SZ") |
| 104 | |
| 105 | if run_id is None: |
| 106 | # Deterministic run_id from week + checksums |
| 107 | id_payload = f"{week}:{source_config_checksum}:{topic_config_checksum}:{code_sha}" |
| 108 | sha_prefix = hashlib.sha256(id_payload.encode()).hexdigest()[:12] |
| 109 | run_id = f"{week}-{sha_prefix}" |
| 110 | |
| 111 | return RunContext( |
| 112 | schema_version=SCHEMA_VERSION, |
| 113 | run_id=run_id, |
| 114 | week=week, |
| 115 | since=since_str, |
| 116 | until=until_str, |
| 117 | source_config_checksum=source_config_checksum, |
| 118 | topic_config_checksum=topic_config_checksum, |
| 119 | code_sha=code_sha, |
| 120 | created_at=created_str, |
| 121 | ) |
| 122 | |
| 123 | |
| 124 | def validate_run_context(ctx: RunContext | dict[str, Any]) -> list[str]: |
| 125 | """Validate a run context structure. Returns a list of error strings (empty = valid).""" |
| 126 | if isinstance(ctx, RunContext): |
| 127 | data = ctx.to_dict() |
| 128 | else: |
| 129 | data = ctx |
| 130 | |
| 131 | errors: list[str] = [] |
| 132 | |
| 133 | # Required fields |
| 134 | required_fields = [ |
| 135 | "schema_version", |
| 136 | "run_id", |
| 137 | "week", |
| 138 | "since", |
| 139 | "until", |
| 140 | "source_config_checksum", |
| 141 | "topic_config_checksum", |
| 142 | "code_sha", |
| 143 | "created_at", |
| 144 | ] |
| 145 | for f in required_fields: |
| 146 | if f not in data or not data[f]: |
| 147 | errors.append(f"missing or empty required field: {f}") |
| 148 | |
| 149 | if errors: |
| 150 | return errors |
| 151 | |
| 152 | # Schema version |
| 153 | if data["schema_version"] != SCHEMA_VERSION: |
| 154 | errors.append( |
| 155 | f"schema_version mismatch: expected '{SCHEMA_VERSION}', got '{data['schema_version']}'" |
| 156 | ) |
| 157 | |
| 158 | # Week format |
| 159 | if not _WEEK_RE.match(data["week"]): |
| 160 | errors.append(f"invalid week format: '{data['week']}' (expected YYYY-WNN)") |
| 161 | |
| 162 | # Timestamp formats |
| 163 | for ts_field in ("since", "until", "created_at"): |
| 164 | val = data.get(ts_field, "") |
| 165 | if val and not _ISO_TS_RE.match(val): |
| 166 | errors.append(f"invalid ISO-8601 timestamp in '{ts_field}': '{val}'") |
| 167 | |
| 168 | # Checksum format (should be hex strings) |
| 169 | for cksum_field in ("source_config_checksum", "topic_config_checksum", "code_sha"): |
| 170 | val = data.get(cksum_field, "") |
| 171 | if val and not re.match(r"^[a-f0-9]+$", val): |
| 172 | errors.append(f"invalid hex checksum in '{cksum_field}': '{val}'") |
| 173 | |
| 174 | return errors |
| 175 | |
| 176 | |
| 177 | def compute_source_config_checksum(config_path: Path) -> str: |
| 178 | """Compute SHA-256 checksum of the source configuration file.""" |
| 179 | content = config_path.read_bytes() |
| 180 | return hashlib.sha256(content).hexdigest() |
| 181 | |
| 182 | |
| 183 | def compute_topic_config_checksum(config_path: Path) -> str: |
| 184 | """Compute SHA-256 checksum of the topic configuration file.""" |
| 185 | content = config_path.read_bytes() |
| 186 | return hashlib.sha256(content).hexdigest() |
| 187 | |
| 188 | |
| 189 | def compute_code_sha(source_files: list[Path]) -> str: |
| 190 | """Compute combined SHA-256 of relevant pipeline source files.""" |
| 191 | h = hashlib.sha256() |
| 192 | for f in sorted(source_files): |
| 193 | if f.exists(): |
| 194 | h.update(f.read_bytes()) |
| 195 | return h.hexdigest() |
| 196 | |
| 197 | |
| 198 | def contexts_compatible(a: RunContext, b: RunContext) -> list[str]: |
| 199 | """Check if two run contexts are compatible for fan-in merge. |
| 200 | |
| 201 | Returns list of mismatch descriptions (empty = compatible). |
| 202 | """ |
| 203 | mismatches: list[str] = [] |
| 204 | |
| 205 | if a.schema_version != b.schema_version: |
| 206 | mismatches.append(f"schema_version: {a.schema_version} vs {b.schema_version}") |
| 207 | if a.week != b.week: |
| 208 | mismatches.append(f"week: {a.week} vs {b.week}") |
| 209 | if a.since != b.since: |
| 210 | mismatches.append(f"since: {a.since} vs {b.since}") |
| 211 | if a.until != b.until: |
| 212 | mismatches.append(f"until: {a.until} vs {b.until}") |
| 213 | if a.source_config_checksum != b.source_config_checksum: |
| 214 | mismatches.append( |
| 215 | f"source_config_checksum: {a.source_config_checksum} vs {b.source_config_checksum}" |
| 216 | ) |
| 217 | if a.topic_config_checksum != b.topic_config_checksum: |
| 218 | mismatches.append( |
| 219 | f"topic_config_checksum: {a.topic_config_checksum} vs {b.topic_config_checksum}" |
| 220 | ) |
| 221 | if a.code_sha != b.code_sha: |
| 222 | mismatches.append(f"code_sha: {a.code_sha} vs {b.code_sha}") |
| 223 | |
| 224 | return mismatches |