排查一次故障，我先得替 AI 翻完日志、监控和发布记录。

"""Executable synthetic incident queries; no production monitoring connection."""
from dataclasses import asdict, dataclass
from datetime import datetime, timedelta


def utc(value: str) -> datetime:
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None or parsed.utcoffset() != timedelta(0):
        raise ValueError("use an explicit UTC timestamp")
    return parsed


@dataclass(frozen=True)
class Query:
    service: str
    environment: str
    start: str
    end: str

    def __post_init__(self):
        if not self.service.strip() or not self.environment.strip():
            raise ValueError("service and environment are required")
        span = utc(self.end) - utc(self.start)
        if not timedelta(0) < span <= timedelta(hours=1):
            raise ValueError("window must be positive and at most one hour")


class FixtureSource:
    def __init__(self, rows):
        self.rows = rows
        self.queries = []

    def query(self, request: Query) -> list[dict]:
        self.queries.append(request)
        return [dict(row) for row in self.rows
                if row["service"] == request.service
                and row["environment"] == request.environment
                and utc(request.start) <= utc(row["at"]) < utc(request.end)]


class TimeoutSource:
    def __init__(self):
        self.queries = []

    def query(self, request: Query) -> list[dict]:
        self.queries.append(request)
        raise TimeoutError("synthetic source timeout")


class Collector:
    def __init__(self, sources):
        self.sources = dict(sources)

    def collect(self, request: Query) -> dict:
        evidence = []
        for name, source in self.sources.items():
            item = {"source": name, "query": asdict(request)}
            try:
                rows = source.query(request)
            except TimeoutError:
                item.update(state="unavailable", error="timeout")
            except Exception:
                # Do not leak credentials/URLs from upstream exception messages.
                item.update(state="unavailable", error="source_error")
            else:
                item.update(state="received", records=rows)
            evidence.append(item)
        return {"evidence": evidence}  # No inferred cause or repair action.


def demo_collector() -> Collector:
    return Collector({
        "errors": FixtureSource([
            {"service": "checkout", "environment": "production", "at": "2026-09-01T10:03:00Z", "event": "DEMO-E1", "code": "upstream_timeout"},
            {"service": "checkout", "environment": "staging", "at": "2026-09-01T10:04:00Z", "event": "DEMO-E2", "code": "test_failure"},
            {"service": "billing", "environment": "production", "at": "2026-09-01T10:04:00Z", "event": "DEMO-E3", "code": "other_service"},
        ]),
        "deployments": FixtureSource([
            {"service": "checkout", "environment": "production", "at": "2026-09-01T09:30:00Z", "event": "DEMO-D1", "release": "demo-v2"},
        ]),
        "latency": TimeoutSource(),
    })


if __name__ == "__main__":
    import json
    query = Query("checkout", "production", "2026-09-01T10:00:00Z", "2026-09-01T10:15:00Z")
    print(json.dumps(demo_collector().collect(query), indent=2))

"""The owner supplies read-only sources and the business authorization check."""
from collections.abc import Callable
from easyremote import ComputeNode, Context
from collector import Collector, Query


def create_node(collector: Collector, authorize: Callable[[str, str, str], None]) -> ComputeNode:
    node = ComputeNode(namespace="incident")

    @node.register
    def collect_evidence(ctx: Context, service: str, environment: str, start: str, end: str) -> dict:
        """Collect read-only evidence for one authorized service and UTC window."""
        authorize(ctx.caller, service, environment)  # Must raise on denial.
        return collector.collect(Query(service, environment, start, end))

    return node