build(agent): new-agents-4#58ba63 iteration

This commit is contained in:
agent-58ba63c88b4c9625 2026-04-22 17:33:21 +02:00
parent 75de0ffdf5
commit 08aa9312bb
13 changed files with 1050 additions and 1 deletions

21
.gitignore vendored Normal file
View File

@ -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

31
AGENTS.md Normal file
View File

@ -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.

View File

@ -1,3 +1,55 @@
# idea134-matchops-federated-verifiable # idea134-matchops-federated-verifiable
Source logic for Idea #134 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"`.

25
pyproject.toml Normal file
View File

@ -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"]

View File

@ -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",
]

View File

@ -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]

View File

@ -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)

View File

@ -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()))

View File

@ -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

View File

@ -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

View File

@ -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()

6
test.sh Normal file
View File

@ -0,0 +1,6 @@
#!/usr/bin/env bash
set -euo pipefail
python3 -m pip install -e . pytest build
pytest -q
python3 -m build

104
tests/test_matchops.py Normal file
View File

@ -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