From 08aa9312bb444f9ab8b143e026a76b3e70877f28 Mon Sep 17 00:00:00 2001 From: agent-58ba63c88b4c9625 Date: Wed, 22 Apr 2026 17:33:21 +0200 Subject: [PATCH] build(agent): new-agents-4#58ba63 iteration --- .gitignore | 21 ++ AGENTS.md | 31 +++ README.md | 54 +++- pyproject.toml | 25 ++ .../__init__.py | 58 ++++ .../adapters.py | 93 +++++++ .../dsl.py | 133 +++++++++ .../ir.py | 253 ++++++++++++++++++ .../ledger.py | 119 ++++++++ .../optimizer.py | 139 ++++++++++ .../utils.py | 15 ++ test.sh | 6 + tests/test_matchops.py | 104 +++++++ 13 files changed, 1050 insertions(+), 1 deletion(-) create mode 100644 .gitignore create mode 100644 AGENTS.md create mode 100644 pyproject.toml create mode 100644 src/idea134_matchops_federated_verifiable/__init__.py create mode 100644 src/idea134_matchops_federated_verifiable/adapters.py create mode 100644 src/idea134_matchops_federated_verifiable/dsl.py create mode 100644 src/idea134_matchops_federated_verifiable/ir.py create mode 100644 src/idea134_matchops_federated_verifiable/ledger.py create mode 100644 src/idea134_matchops_federated_verifiable/optimizer.py create mode 100644 src/idea134_matchops_federated_verifiable/utils.py create mode 100644 test.sh create mode 100644 tests/test_matchops.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..bd5590b --- /dev/null +++ b/.gitignore @@ -0,0 +1,21 @@ +node_modules/ +.npmrc +.env +.env.* +__tests__/ +coverage/ +.nyc_output/ +dist/ +build/ +.cache/ +*.log +.DS_Store +tmp/ +.tmp/ +__pycache__/ +*.pyc +.venv/ +venv/ +*.egg-info/ +.pytest_cache/ +READY_TO_PUBLISH diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..54ea394 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,31 @@ +# AGENTS.md + +## Architecture + +- `src/idea134_matchops_federated_verifiable/` contains the package code. +- `dsl.py` compiles the declarative MatchOps spec into canonical IR. +- `ir.py` defines the canonical data model, verifiable execution graph, and plan delta structures. +- `optimizer.py` implements the federated, privacy-preserving scheduling loop. +- `adapters.py` contains the Graph-of-Contracts registry and starter adapters. +- `ledger.py` provides the tamper-evident governance ledger with SQLite storage. +- `tests/` contains unit tests for the compiler, optimizer, adapters, and ledger. + +## Tech Stack + +- Python 3.10+ +- `pydantic` for strict schema validation and deterministic serialization +- `sqlite3` from the standard library for durable governance ledger storage +- `pytest` for tests + +## Rules + +- Keep serialization canonical; hashes depend on stable JSON ordering. +- Preserve the src layout and package name `idea134_matchops_federated_verifiable`. +- Prefer small, deterministic functions over ad hoc logic. +- Do not weaken ledger integrity checks or execution graph attestations. +- Add or update tests whenever behavior changes. + +## Testing + +- Run `bash test.sh` from the repository root. +- `test.sh` must run both the test suite and `python3 -m build` successfully. diff --git a/README.md b/README.md index 68cabf0..aba161f 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,55 @@ # idea134-matchops-federated-verifiable -Source logic for Idea #134 \ No newline at end of file +MatchOps is a federated session-orchestration toolkit for multiplayer games. It models regions, servers, session types, capacity budgets, latency budgets, fairness rules, and execution policies in a declarative DSL, then compiles that spec into canonical IR, a verifiable execution graph, and governance artifacts. + +This repository focuses on the control-plane primitives: + +- declarative DSL compilation +- canonical IR and plan deltas +- cryptographically hashed execution attestations +- a Graph-of-Contracts adapter registry +- a privacy-preserving federated scheduler +- a tamper-evident governance ledger backed by SQLite + +## What It Provides + +- `MatchOpsSpec`: the input DSL model. +- `compile_spec(...)`: converts a spec into canonical IR and a verifiable execution graph. +- `FederatedOptimizer`: allocates session demand from aggregated regional signals without exposing raw player data. +- `GoCRegistry`: registers versioned adapters for cloud APIs and game-server routing. +- `SQLiteGovernanceLedger`: stores append-only governance events with a hash chain. + +## Repository Layout + +```text +src/idea134_matchops_federated_verifiable/ +tests/ +AGENTS.md +test.sh +pyproject.toml +``` + +## Quick Start + +```bash +python3 -m pip install -e . +pytest -q +python3 -m build +``` + +Or run the full verification script: + +```bash +bash test.sh +``` + +## Design Notes + +- Canonical JSON serialization is used for all hashes so decisions are reproducible. +- The execution graph uses per-step attestations derived from the previous step and the current payload. +- The optimizer only consumes aggregated regional signals and session demand, never raw player identity data. +- The governance ledger is append-only and can be persisted to SQLite for durable audit trails. + +## Public Package Metadata + +The package is published as `idea134-matchops-federated-verifiable` and its README is wired into `pyproject.toml` via `readme = "README.md"`. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..9bbe9b2 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,25 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "idea134-matchops-federated-verifiable" +version = "0.1.0" +description = "Federated, verifiable session orchestration primitives for large-scale multiplayer games" +readme = "README.md" +requires-python = ">=3.10" +dependencies = [ + "pydantic>=2.7,<3", +] + +[project.urls] +Homepage = "https://example.invalid/matchops" + +[tool.setuptools] +package-dir = {"" = "src"} + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/src/idea134_matchops_federated_verifiable/__init__.py b/src/idea134_matchops_federated_verifiable/__init__.py new file mode 100644 index 0000000..e3955da --- /dev/null +++ b/src/idea134_matchops_federated_verifiable/__init__.py @@ -0,0 +1,58 @@ +from .adapters import CloudApiAdapter, GameServerRouterAdapter, GoCRegistry +from .dsl import compile_spec +from .ir import ( + AuditLog, + AuditLogEntry, + CanonicalIR, + CapacityBudget, + ConstraintEdge, + ConstraintGraph, + ConstraintNode, + ExecutionGraph, + ExecutionPolicies, + ExecutionStep, + LatencyBudget, + LocalRegion, + MatchOpsSpec, + ObjectiveEdge, + ObjectiveGraph, + ObjectiveNode, + PlanDelta, + Region, + Server, + SessionType, +) +from .ledger import SQLiteGovernanceLedger +from .optimizer import AggregatedSignal, FederatedOptimizer, SessionDemand, reconcile_plan_deltas + +__all__ = [ + "AggregatedSignal", + "AuditLog", + "AuditLogEntry", + "CanonicalIR", + "CapacityBudget", + "CloudApiAdapter", + "ConstraintEdge", + "ConstraintGraph", + "ConstraintNode", + "ExecutionGraph", + "ExecutionPolicies", + "ExecutionStep", + "FederatedOptimizer", + "GameServerRouterAdapter", + "GoCRegistry", + "LatencyBudget", + "LocalRegion", + "MatchOpsSpec", + "ObjectiveEdge", + "ObjectiveGraph", + "ObjectiveNode", + "PlanDelta", + "Region", + "Server", + "SessionDemand", + "SessionType", + "SQLiteGovernanceLedger", + "compile_spec", + "reconcile_plan_deltas", +] diff --git a/src/idea134_matchops_federated_verifiable/adapters.py b/src/idea134_matchops_federated_verifiable/adapters.py new file mode 100644 index 0000000..22e2980 --- /dev/null +++ b/src/idea134_matchops_federated_verifiable/adapters.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +from abc import ABC, abstractmethod +from collections.abc import Iterable + +from pydantic import BaseModel, ConfigDict, Field + +from .ir import CanonicalIR +from .utils import sha256_digest + + +class AdapterAction(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + action: str + target: str + payload: dict[str, object] = Field(default_factory=dict) + + +class AdapterTranslation(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + adapter_name: str + adapter_version: str + actions: list[AdapterAction] = Field(default_factory=list) + + @property + def translation_hash(self) -> str: + return sha256_digest(self.model_dump(mode="python")) + + +class ContractAdapter(ABC): + name: str + version: str + + @abstractmethod + def translate(self, ir: CanonicalIR) -> AdapterTranslation: + raise NotImplementedError + + +class CloudApiAdapter(ContractAdapter): + name = "cloud-api" + version = "1.0" + + def translate(self, ir: CanonicalIR) -> AdapterTranslation: + actions: list[AdapterAction] = [] + for region in ir.local_regions: + target = region.region.name + payload = { + "provider": region.region.provider, + "servers": [server.name for server in region.servers], + "capacity_budget": region.capacity_budget.model_dump(mode="python") if region.capacity_budget else None, + } + actions.append(AdapterAction(action="ensure_region_capacity", target=target, payload=payload)) + return AdapterTranslation(adapter_name=self.name, adapter_version=self.version, actions=actions) + + +class GameServerRouterAdapter(ContractAdapter): + name = "game-server-router" + version = "1.0" + + def translate(self, ir: CanonicalIR) -> AdapterTranslation: + actions = [] + for delta in ir.plan_deltas: + payload = {"change_type": delta.change_type, "payload": delta.payload, "version": delta.version} + actions.append(AdapterAction(action="route_delta", target=delta.region, payload=payload)) + return AdapterTranslation(adapter_name=self.name, adapter_version=self.version, actions=actions) + + +class GoCRegistry: + def __init__(self) -> None: + self._adapters: dict[tuple[str, str], ContractAdapter] = {} + + def register(self, adapter: ContractAdapter) -> None: + self._adapters[(adapter.name, adapter.version)] = adapter + + def get(self, name: str, version: str | None = None) -> ContractAdapter: + if version is not None: + return self._adapters[(name, version)] + + matches = [adapter for (adapter_name, _), adapter in self._adapters.items() if adapter_name == name] + if not matches: + raise KeyError(name) + if len(matches) > 1: + raise KeyError(f"multiple versions registered for {name}; specify version") + return matches[0] + + def available(self) -> list[str]: + return sorted(f"{name}@{version}" for name, version in self._adapters) + + def translate_all(self, ir: CanonicalIR, names: Iterable[str] | None = None) -> list[AdapterTranslation]: + selected = list(names) if names is not None else sorted({name for name, _ in self._adapters}) + return [self.get(name).translate(ir) for name in selected] diff --git a/src/idea134_matchops_federated_verifiable/dsl.py b/src/idea134_matchops_federated_verifiable/dsl.py new file mode 100644 index 0000000..2ebfc4c --- /dev/null +++ b/src/idea134_matchops_federated_verifiable/dsl.py @@ -0,0 +1,133 @@ +from __future__ import annotations + +from .ir import ( + AuditLog, + AuditLogEntry, + CanonicalIR, + CompiledPlan, + ConstraintEdge, + ConstraintGraph, + ConstraintNode, + ExecutionGraph, + ExecutionStep, + LocalRegion, + MatchOpsSpec, + ObjectiveEdge, + ObjectiveGraph, + ObjectiveNode, + PlanDelta, + group_servers_by_region, +) +from .utils import sha256_digest + + +def _build_execution_graph(spec_hash: str, ir_hash: str) -> ExecutionGraph: + steps: list[ExecutionStep] = [] + current_input = "genesis" + stages = [ + ("compile-spec", "compile", spec_hash), + ("normalize-ir", "validate", ir_hash), + ("seal-plan", "audit", sha256_digest({"spec_hash": spec_hash, "ir_hash": ir_hash})), + ] + for name, kind, output_hash in stages: + attestation = sha256_digest( + {"name": name, "input_hash": current_input, "output_hash": output_hash, "kind": kind} + ) + steps.append( + ExecutionStep( + name=name, + kind=kind, + input_hash=current_input, + output_hash=output_hash, + attestation=attestation, + ) + ) + current_input = output_hash + return ExecutionGraph(steps=steps) + + +def compile_spec(spec: MatchOpsSpec) -> CompiledPlan: + spec_hash = sha256_digest(spec.model_dump(mode="python")) + grouped_servers = group_servers_by_region(spec.servers) + + local_regions = [] + for region in sorted(spec.regions, key=lambda item: item.name): + local_regions.append( + LocalRegion( + region=region, + servers=grouped_servers.get(region.name, []), + capacity_budget=next((budget for budget in spec.capacity_budgets if budget.region == region.name), None), + latency_budget=None, + ) + ) + + objective_nodes = [ObjectiveNode(name=f"latency:{session.name}", kind="latency", weight=1.0) for session in spec.session_types] + objective_nodes.extend( + ObjectiveNode(name=f"fairness:{region.name}", kind="fairness", weight=region.fairness_weight) + for region in spec.regions + ) + objective_nodes.append(ObjectiveNode(name="capacity", kind="capacity", weight=1.0)) + objective_nodes.append(ObjectiveNode(name="routing", kind="routing", weight=1.0)) + + objective_edges = [ + ObjectiveEdge(source="routing", target=node.name, relation="influences") + for node in objective_nodes + if node.kind in {"latency", "fairness", "capacity"} + ] + + constraint_nodes = [ + ConstraintNode( + name=f"capacity:{budget.region}", + kind="capacity", + expression=f"players <= {budget.max_players} and sessions <= {budget.max_sessions}", + threshold=budget.max_players, + ) + for budget in spec.capacity_budgets + ] + constraint_nodes.extend( + ConstraintNode( + name=f"latency:{budget.session_type}", + kind="latency", + expression=f"p95_ms <= {budget.p95_ms}", + threshold=budget.p95_ms, + ) + for budget in spec.latency_budgets + ) + constraint_nodes.append( + ConstraintNode(name="policy:failover", kind="policy", expression=f"failover_mode == '{spec.policies.failover_mode}'") + ) + + constraint_edges = [ + ConstraintEdge(source="policy:failover", target=node.name, relation="gates") + for node in constraint_nodes + if node.kind in {"capacity", "latency"} + ] + + plan_deltas = [ + PlanDelta( + version=1, + region=region.name, + change_type="seed-region", + payload={"servers": [server.name for server in grouped_servers.get(region.name, [])]}, + ) + for region in sorted(spec.regions, key=lambda item: item.name) + ] + + audit_entry = AuditLogEntry( + sequence=0, + actor="compiler", + action="compile-spec", + object_hash=spec_hash, + details={"regions": len(spec.regions), "servers": len(spec.servers), "session_types": len(spec.session_types)}, + ) + + ir = CanonicalIR( + local_regions=local_regions, + objective_graph=ObjectiveGraph(nodes=objective_nodes, edges=objective_edges), + constraint_graph=ConstraintGraph(nodes=constraint_nodes, edges=constraint_edges), + plan_deltas=plan_deltas, + audit_log=AuditLog(entries=[audit_entry]), + ) + execution_graph = _build_execution_graph(spec_hash, ir.ir_hash) + return CompiledPlan(spec_hash=spec_hash, ir=ir, execution_graph=execution_graph) + diff --git a/src/idea134_matchops_federated_verifiable/ir.py b/src/idea134_matchops_federated_verifiable/ir.py new file mode 100644 index 0000000..d8fb112 --- /dev/null +++ b/src/idea134_matchops_federated_verifiable/ir.py @@ -0,0 +1,253 @@ +from __future__ import annotations + +from collections import defaultdict +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field, computed_field, model_validator + +from .utils import sha256_digest + + +class Region(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + name: str + provider: str + fairness_weight: float = Field(default=1.0, ge=0.0) + + +class Server(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + name: str + region: str + capacity: int = Field(gt=0) + labels: dict[str, str] = Field(default_factory=dict) + + +class SessionType(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + name: str + max_players: int = Field(gt=0) + min_players: int = Field(default=1, ge=1) + latency_budget_ms: int = Field(gt=0) + + +class CapacityBudget(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + region: str + max_sessions: int = Field(ge=0) + max_players: int = Field(ge=0) + + +class LatencyBudget(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + session_type: str + p95_ms: int = Field(gt=0) + + +class ExecutionPolicies(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + session_quota_per_region: dict[str, int] = Field(default_factory=dict) + hot_path_routing: bool = True + failover_mode: Literal["regional", "global", "deny"] = "regional" + + +class PlanDelta(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + version: int = Field(ge=0) + region: str + change_type: str + payload: dict[str, Any] = Field(default_factory=dict) + previous_hash: str | None = None + + @computed_field + @property + def delta_hash(self) -> str: + return sha256_digest(self.model_dump(mode="python", exclude={"delta_hash"})) + + +class AuditLogEntry(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + sequence: int = Field(ge=0) + actor: str + action: str + object_hash: str + prev_hash: str | None = None + details: dict[str, Any] = Field(default_factory=dict) + + @computed_field + @property + def entry_hash(self) -> str: + return sha256_digest(self.model_dump(mode="python", exclude={"entry_hash"})) + + +class AuditLog(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + entries: list[AuditLogEntry] = Field(default_factory=list) + + def head_hash(self) -> str | None: + return self.entries[-1].entry_hash if self.entries else None + + +class LocalRegion(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + region: Region + servers: list[Server] = Field(default_factory=list) + capacity_budget: CapacityBudget | None = None + latency_budget: LatencyBudget | None = None + + +class ObjectiveNode(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + name: str + kind: Literal["latency", "fairness", "capacity", "routing"] + weight: float = Field(gt=0) + + +class ObjectiveEdge(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + source: str + target: str + relation: str + + +class ObjectiveGraph(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + nodes: list[ObjectiveNode] = Field(default_factory=list) + edges: list[ObjectiveEdge] = Field(default_factory=list) + + +class ConstraintNode(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + name: str + kind: Literal["capacity", "latency", "fairness", "policy"] + expression: str + threshold: float | int | None = None + + +class ConstraintEdge(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + source: str + target: str + relation: str + + +class ConstraintGraph(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + nodes: list[ConstraintNode] = Field(default_factory=list) + edges: list[ConstraintEdge] = Field(default_factory=list) + + +class ExecutionStep(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + name: str + input_hash: str + output_hash: str + attestation: str + kind: Literal["compile", "validate", "optimize", "route", "audit"] + + def verify(self) -> bool: + expected = sha256_digest( + { + "name": self.name, + "input_hash": self.input_hash, + "output_hash": self.output_hash, + "kind": self.kind, + } + ) + return expected == self.attestation + + +class ExecutionGraph(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + steps: list[ExecutionStep] = Field(default_factory=list) + + @computed_field + @property + def root_hash(self) -> str: + return sha256_digest([step.model_dump(mode="python") for step in self.steps]) + + def verify(self) -> bool: + previous = "genesis" + for step in self.steps: + if step.input_hash != previous or not step.verify(): + return False + previous = step.output_hash + return True + + +class CanonicalIR(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + local_regions: list[LocalRegion] = Field(default_factory=list) + objective_graph: ObjectiveGraph = Field(default_factory=ObjectiveGraph) + constraint_graph: ConstraintGraph = Field(default_factory=ConstraintGraph) + plan_deltas: list[PlanDelta] = Field(default_factory=list) + audit_log: AuditLog = Field(default_factory=AuditLog) + + @computed_field + @property + def ir_hash(self) -> str: + return sha256_digest(self.model_dump(mode="python", exclude={"ir_hash"})) + + +class MatchOpsSpec(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + regions: list[Region] + servers: list[Server] + session_types: list[SessionType] + capacity_budgets: list[CapacityBudget] = Field(default_factory=list) + latency_budgets: list[LatencyBudget] = Field(default_factory=list) + policies: ExecutionPolicies = Field(default_factory=ExecutionPolicies) + + @model_validator(mode="after") + def validate_references(self) -> "MatchOpsSpec": + region_names = {region.name for region in self.regions} + server_regions = {server.region for server in self.servers} + missing_regions = server_regions - region_names + if missing_regions: + raise ValueError(f"servers reference unknown regions: {sorted(missing_regions)}") + + session_names = {session.name for session in self.session_types} + for budget in self.latency_budgets: + if budget.session_type not in session_names: + raise ValueError(f"latency budget references unknown session type: {budget.session_type}") + + for budget in self.capacity_budgets: + if budget.region not in region_names: + raise ValueError(f"capacity budget references unknown region: {budget.region}") + return self + + +class CompiledPlan(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + spec_hash: str + ir: CanonicalIR + execution_graph: ExecutionGraph + + +def group_servers_by_region(servers: list[Server]) -> dict[str, list[Server]]: + grouped: dict[str, list[Server]] = defaultdict(list) + for server in servers: + grouped[server.region].append(server) + return dict(sorted(grouped.items())) + diff --git a/src/idea134_matchops_federated_verifiable/ledger.py b/src/idea134_matchops_federated_verifiable/ledger.py new file mode 100644 index 0000000..d99528a --- /dev/null +++ b/src/idea134_matchops_federated_verifiable/ledger.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +import json +import sqlite3 +from dataclasses import dataclass +from pathlib import Path + +from pydantic import BaseModel, ConfigDict, Field + +from .ir import AuditLogEntry +from .utils import sha256_digest + + +class LedgerEvent(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + sequence: int = Field(ge=0) + actor: str + action: str + payload: dict[str, object] = Field(default_factory=dict) + prev_hash: str | None = None + + @property + def entry_hash(self) -> str: + return sha256_digest(self.model_dump(mode="python")) + + +@dataclass(slots=True) +class LedgerVerification: + ok: bool + entries: int + last_hash: str | None + + +class SQLiteGovernanceLedger: + def __init__(self, db_path: str | Path = ":memory:") -> None: + self.db_path = str(db_path) + self._ensure_schema() + + def _connect(self) -> sqlite3.Connection: + conn = sqlite3.connect(self.db_path) + conn.row_factory = sqlite3.Row + return conn + + def _ensure_schema(self) -> None: + with self._connect() as conn: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS governance_ledger ( + sequence INTEGER PRIMARY KEY, + actor TEXT NOT NULL, + action TEXT NOT NULL, + payload_json TEXT NOT NULL, + prev_hash TEXT, + entry_hash TEXT NOT NULL UNIQUE + ) + """ + ) + conn.commit() + + def last_hash(self) -> str | None: + with self._connect() as conn: + row = conn.execute("SELECT entry_hash FROM governance_ledger ORDER BY sequence DESC LIMIT 1").fetchone() + return row["entry_hash"] if row else None + + def append(self, actor: str, action: str, payload: dict[str, object]) -> LedgerEvent: + with self._connect() as conn: + row = conn.execute("SELECT sequence, entry_hash FROM governance_ledger ORDER BY sequence DESC LIMIT 1").fetchone() + sequence = 0 if row is None else int(row["sequence"]) + 1 + prev_hash = None if row is None else row["entry_hash"] + event = LedgerEvent(sequence=sequence, actor=actor, action=action, payload=payload, prev_hash=prev_hash) + conn.execute( + "INSERT INTO governance_ledger(sequence, actor, action, payload_json, prev_hash, entry_hash) VALUES (?, ?, ?, ?, ?, ?)", + (event.sequence, event.actor, event.action, json.dumps(payload, sort_keys=True, separators=(",", ":")), event.prev_hash, event.entry_hash), + ) + conn.commit() + return event + + def list_entries(self) -> list[LedgerEvent]: + with self._connect() as conn: + rows = conn.execute("SELECT sequence, actor, action, payload_json, prev_hash, entry_hash FROM governance_ledger ORDER BY sequence ASC").fetchall() + return [ + LedgerEvent( + sequence=int(row["sequence"]), + actor=row["actor"], + action=row["action"], + payload=json.loads(row["payload_json"]), + prev_hash=row["prev_hash"], + ) + for row in rows + ] + + def verify(self) -> LedgerVerification: + entries = self.list_entries() + previous_hash = None + for entry in entries: + if entry.prev_hash != previous_hash: + return LedgerVerification(ok=False, entries=len(entries), last_hash=previous_hash) + if entry.entry_hash != sha256_digest(entry.model_dump(mode="python")): + return LedgerVerification(ok=False, entries=len(entries), last_hash=previous_hash) + previous_hash = entry.entry_hash + return LedgerVerification(ok=True, entries=len(entries), last_hash=previous_hash) + + def as_audit_entries(self) -> list[AuditLogEntry]: + entries = self.list_entries() + audit_entries = [] + for event in entries: + audit_entries.append( + AuditLogEntry( + sequence=event.sequence, + actor=event.actor, + action=event.action, + object_hash=sha256_digest(event.payload), + prev_hash=event.prev_hash, + details=event.payload, + ) + ) + return audit_entries + diff --git a/src/idea134_matchops_federated_verifiable/optimizer.py b/src/idea134_matchops_federated_verifiable/optimizer.py new file mode 100644 index 0000000..39560b5 --- /dev/null +++ b/src/idea134_matchops_federated_verifiable/optimizer.py @@ -0,0 +1,139 @@ +from __future__ import annotations + +from collections import defaultdict +from dataclasses import dataclass + +from pydantic import BaseModel, ConfigDict, Field + +from .ir import MatchOpsSpec, PlanDelta +from .utils import sha256_digest + + +class AggregatedSignal(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + region: str + queue_depth: int = Field(ge=0) + load_factor: float = Field(ge=0.0) + avg_latency_ms: float = Field(gt=0) + available_capacity: int = Field(ge=0) + version: int = Field(default=0, ge=0) + + +class SessionDemand(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + session_type: str + players: int = Field(gt=0) + max_latency_ms: int = Field(gt=0) + + +class SessionAllocation(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + session_type: str + region: str + players: int = Field(gt=0) + score: float + reason: str + + +class OptimizationResult(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + allocations: list[SessionAllocation] = Field(default_factory=list) + regional_load: dict[str, int] = Field(default_factory=dict) + delta: PlanDelta + round_hash: str + + +@dataclass(slots=True) +class _DualState: + values: dict[str, float] + + +class FederatedOptimizer: + def __init__(self, iterations: int = 4) -> None: + if iterations <= 0: + raise ValueError("iterations must be positive") + self.iterations = iterations + + def optimize(self, spec: MatchOpsSpec, signals: list[AggregatedSignal], demand: list[SessionDemand]) -> OptimizationResult: + if not signals: + raise ValueError("signals must not be empty") + + signal_map = {signal.region: signal for signal in signals} + regions = sorted(spec.regions, key=lambda region: region.name) + if any(region.name not in signal_map for region in regions): + missing = sorted(region.name for region in regions if region.name not in signal_map) + raise ValueError(f"missing aggregated signal for regions: {missing}") + + fairness_weights = {region.name: region.fairness_weight for region in regions} + dual = _DualState(values={region.name: 0.0 for region in regions}) + allocations: list[SessionAllocation] = [] + + for _ in range(self.iterations): + allocations = [] + region_counts = defaultdict(int) + for session in sorted(demand, key=lambda item: (item.session_type, item.players, item.max_latency_ms)): + ranked = [] + for region in regions: + signal = signal_map[region.name] + latency_penalty = signal.avg_latency_ms / session.max_latency_ms + load_penalty = signal.load_factor + (signal.queue_depth / (signal.available_capacity + 1)) + dual_penalty = dual.values[region.name] + fairness_bonus = 1.0 / fairness_weights[region.name] + score = latency_penalty + load_penalty + dual_penalty + fairness_bonus + ranked.append((score, region.name)) + + ranked.sort(key=lambda item: (item[0], item[1])) + chosen_score, chosen_region = ranked[0] + signal = signal_map[chosen_region] + if signal.available_capacity < session.players: + fallback = sorted( + regions, + key=lambda region: ( + signal_map[region.name].available_capacity < session.players, + signal_map[region.name].avg_latency_ms, + region.name, + ), + )[0] + chosen_region = fallback.name + signal = signal_map[chosen_region] + chosen_score = signal.avg_latency_ms / session.max_latency_ms + signal.load_factor + dual.values[chosen_region] + + region_counts[chosen_region] += session.players + allocations.append( + SessionAllocation( + session_type=session.session_type, + region=chosen_region, + players=session.players, + score=round(chosen_score, 6), + reason="latency-load-dual-fairness-minimization", + ) + ) + + total_players = sum(region_counts.values()) or 1 + quota = {region.name: total_players * fairness_weights[region.name] / sum(fairness_weights.values()) for region in regions} + for region in regions: + dual.values[region.name] += 0.1 * (region_counts[region.name] - quota[region.name]) + + regional_load = {region.name: sum(allocation.players for allocation in allocations if allocation.region == region.name) for region in regions} + delta = PlanDelta( + version=max((signal.version for signal in signals), default=0) + 1, + region="federated", + change_type="allocation", + payload={ + "regional_load": regional_load, + "allocations": [allocation.model_dump(mode="python") for allocation in allocations], + }, + ) + round_hash = sha256_digest({"allocations": [allocation.model_dump(mode="python") for allocation in allocations], "delta": delta.model_dump(mode="python")}) + return OptimizationResult(allocations=allocations, regional_load=regional_load, delta=delta, round_hash=round_hash) + + +def reconcile_plan_deltas(deltas: list[PlanDelta]) -> tuple[list[PlanDelta], str]: + unique = {delta.delta_hash: delta for delta in deltas} + ordered = sorted(unique.values(), key=lambda delta: (delta.version, delta.region, delta.change_type, delta.delta_hash)) + sync_hash = sha256_digest([delta.model_dump(mode="python") for delta in ordered]) + return ordered, sync_hash diff --git a/src/idea134_matchops_federated_verifiable/utils.py b/src/idea134_matchops_federated_verifiable/utils.py new file mode 100644 index 0000000..9c21581 --- /dev/null +++ b/src/idea134_matchops_federated_verifiable/utils.py @@ -0,0 +1,15 @@ +from __future__ import annotations + +import hashlib +import json +from typing import Any + + +def canonical_json(value: Any) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + + +def sha256_digest(value: Any) -> str: + payload = canonical_json(value).encode("utf-8") + return hashlib.sha256(payload).hexdigest() + diff --git a/test.sh b/test.sh new file mode 100644 index 0000000..77db31c --- /dev/null +++ b/test.sh @@ -0,0 +1,6 @@ +#!/usr/bin/env bash +set -euo pipefail + +python3 -m pip install -e . pytest build +pytest -q +python3 -m build diff --git a/tests/test_matchops.py b/tests/test_matchops.py new file mode 100644 index 0000000..7082e32 --- /dev/null +++ b/tests/test_matchops.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +import sqlite3 + +from idea134_matchops_federated_verifiable import ( + AggregatedSignal, + CloudApiAdapter, + FederatedOptimizer, + GameServerRouterAdapter, + GoCRegistry, + MatchOpsSpec, + Region, + Server, + SessionDemand, + SessionType, + SQLiteGovernanceLedger, + CapacityBudget, + LatencyBudget, + compile_spec, +) +from idea134_matchops_federated_verifiable.optimizer import reconcile_plan_deltas + + +def sample_spec() -> MatchOpsSpec: + return MatchOpsSpec( + regions=[ + Region(name="us-east", provider="aws", fairness_weight=1.0), + Region(name="eu-west", provider="gcp", fairness_weight=2.0), + ], + servers=[ + Server(name="se-1", region="us-east", capacity=200), + Server(name="se-2", region="eu-west", capacity=150), + ], + session_types=[SessionType(name="ranked", max_players=10, latency_budget_ms=80)], + capacity_budgets=[ + CapacityBudget(region="us-east", max_sessions=10, max_players=100), + CapacityBudget(region="eu-west", max_sessions=8, max_players=80), + ], + latency_budgets=[LatencyBudget(session_type="ranked", p95_ms=80)], + ) + + +def test_compile_spec_builds_verifiable_execution_graph(): + plan = compile_spec(sample_spec()) + + assert plan.execution_graph.verify() + assert plan.ir.ir_hash + assert len(plan.ir.local_regions) == 2 + assert plan.ir.constraint_graph.nodes + + +def test_registry_and_adapters_translate_ir(): + plan = compile_spec(sample_spec()) + registry = GoCRegistry() + registry.register(CloudApiAdapter()) + registry.register(GameServerRouterAdapter()) + + translations = registry.translate_all(plan.ir) + + assert {translation.adapter_name for translation in translations} == {"cloud-api", "game-server-router"} + assert all(translation.translation_hash for translation in translations) + assert any(action.action == "ensure_region_capacity" for action in translations[0].actions + translations[1].actions) + + +def test_optimizer_prefers_lower_latency_region_and_is_deterministic(): + spec = sample_spec() + optimizer = FederatedOptimizer(iterations=3) + signals = [ + AggregatedSignal(region="us-east", queue_depth=12, load_factor=0.8, avg_latency_ms=95, available_capacity=20, version=7), + AggregatedSignal(region="eu-west", queue_depth=5, load_factor=0.3, avg_latency_ms=40, available_capacity=20, version=8), + ] + demand = [SessionDemand(session_type="ranked", players=10, max_latency_ms=80)] + + result = optimizer.optimize(spec, signals, demand) + repeat = optimizer.optimize(spec, signals, demand) + + assert result == repeat + assert result.allocations[0].region == "eu-west" + assert result.delta.region == "federated" + + +def test_plan_delta_reconciliation_and_ledger_integrity(tmp_path): + ledger = SQLiteGovernanceLedger(tmp_path / "ledger.sqlite") + ledger.append("compiler", "compile-spec", {"spec": "matchops"}) + ledger.append("optimizer", "allocate-sessions", {"allocations": 2}) + + verification = ledger.verify() + assert verification.ok + assert verification.entries == 2 + + entries, sync_hash = reconcile_plan_deltas( + [ + compile_spec(sample_spec()).ir.plan_deltas[1], + compile_spec(sample_spec()).ir.plan_deltas[0], + ] + ) + assert len(entries) == 2 + assert sync_hash + + with sqlite3.connect(tmp_path / "ledger.sqlite") as conn: + conn.execute("UPDATE governance_ledger SET action = 'tampered' WHERE sequence = 0") + conn.commit() + + assert not ledger.verify().ok