Upload folder using huggingface_hub

common/games.py

@@ -8,34 +8,15 @@ from typing import Callable
 from constant_definitions.game_constants import (
     DEFAULT_ZERO_FLOAT,
     DEFAULT_ZERO_INT,
-    # Prisoner's Dilemma
-    PD_CC_PAYOFF,
-    PD_CD_PAYOFF,
-    PD_DC_PAYOFF,
-    PD_DD_PAYOFF,
-    # Stag Hunt
-    SH_SS_PAYOFF,
-    SH_SH_PAYOFF,
-    SH_HS_PAYOFF,
-    SH_HH_PAYOFF,
-    # Hawk-Dove
-    HD_HH_PAYOFF,
-    HD_HD_PAYOFF,
-    HD_DH_PAYOFF,
-    HD_DD_PAYOFF,
-    # Ultimatum
+    PD_CC_PAYOFF, PD_CD_PAYOFF, PD_DC_PAYOFF, PD_DD_PAYOFF,
+    SH_SS_PAYOFF, SH_SH_PAYOFF, SH_HS_PAYOFF, SH_HH_PAYOFF,
+    HD_HH_PAYOFF, HD_HD_PAYOFF, HD_DH_PAYOFF, HD_DD_PAYOFF,
     ULTIMATUM_POT,
-    # Trust
-    TRUST_MULTIPLIER,
-    TRUST_ENDOWMENT,
-    # Public Goods
-    PG_MULTIPLIER_NUMERATOR,
-    PG_MULTIPLIER_DENOMINATOR,
-    PG_ENDOWMENT,
-    PG_DEFAULT_NUM_PLAYERS,
-    # Round counts
-    DEFAULT_NUM_ROUNDS,
-    SINGLE_SHOT_ROUNDS,
+    TRUST_MULTIPLIER, TRUST_ENDOWMENT,
+    PG_MULTIPLIER_NUMERATOR, PG_MULTIPLIER_DENOMINATOR,
+    PG_ENDOWMENT, PG_DEFAULT_NUM_PLAYERS,
+    DEFAULT_NUM_ROUNDS, SINGLE_SHOT_ROUNDS, DEFAULT_TWO_PLAYERS,
+    OPPONENT_MODE_STRATEGY,
 )
 
 # ---------------------------------------------------------------------------
@@ -50,30 +31,39 @@ class GameConfig:
     name: str
     description: str
     actions: list[str]
-    game_type: str  # "matrix" | "ultimatum" | "trust" | "public_goods"
+    game_type: str
     default_rounds: int
-    payoff_fn: Callable[[str, str], tuple[float, float]]
+    payoff_fn: Callable
+    num_players: int = DEFAULT_TWO_PLAYERS
+    applied_variants: tuple[str, ...] = ()
+    base_game_key: str = ""
+    enforcement: str = ""
+    penalty_numerator: int = DEFAULT_ZERO_INT
+    penalty_denominator: int = SINGLE_SHOT_ROUNDS
+    allow_side_payments: bool = False
+    opponent_mode: str = OPPONENT_MODE_STRATEGY
+    opponent_actions: tuple[str, ...] | None = None
 
 
 # ---------------------------------------------------------------------------
 # Matrix-game payoff helpers
 # ---------------------------------------------------------------------------
 
-_PD_MATRIX: dict[tuple[str, str], tuple[float, float]] = {
+_PD_MATRIX = {
     ("cooperate", "cooperate"): (float(PD_CC_PAYOFF), float(PD_CC_PAYOFF)),
     ("cooperate", "defect"):    (float(PD_CD_PAYOFF), float(PD_DC_PAYOFF)),
     ("defect", "cooperate"):    (float(PD_DC_PAYOFF), float(PD_CD_PAYOFF)),
     ("defect", "defect"):       (float(PD_DD_PAYOFF), float(PD_DD_PAYOFF)),
 }
 
-_SH_MATRIX: dict[tuple[str, str], tuple[float, float]] = {
+_SH_MATRIX = {
     ("stag", "stag"): (float(SH_SS_PAYOFF), float(SH_SS_PAYOFF)),
     ("stag", "hare"): (float(SH_SH_PAYOFF), float(SH_HS_PAYOFF)),
     ("hare", "stag"): (float(SH_HS_PAYOFF), float(SH_SH_PAYOFF)),
     ("hare", "hare"): (float(SH_HH_PAYOFF), float(SH_HH_PAYOFF)),
 }
 
-_HD_MATRIX: dict[tuple[str, str], tuple[float, float]] = {
+_HD_MATRIX = {
     ("hawk", "hawk"): (float(HD_HH_PAYOFF), float(HD_HH_PAYOFF)),
     ("hawk", "dove"): (float(HD_HD_PAYOFF), float(HD_DH_PAYOFF)),
     ("dove", "hawk"): (float(HD_DH_PAYOFF), float(HD_HD_PAYOFF)),
@@ -81,9 +71,7 @@ _HD_MATRIX: dict[tuple[str, str], tuple[float, float]] = {
 }
 
 
-def _matrix_payoff_fn(
-    matrix: dict[tuple[str, str], tuple[float, float]],
-) -> Callable[[str, str], tuple[float, float]]:
+def _matrix_payoff_fn(matrix: dict) -> Callable:
     """Return a payoff function backed by a pre-built matrix dict."""
 
     def _payoff(player_action: str, opponent_action: str) -> tuple[float, float]:
@@ -284,6 +272,7 @@ def _load_extensions() -> None:
         "common.games_market.advanced", "common.games_coop.cooperative",
         "common.games_coop.dynamic", "common.games_coop.pd_variants",
         "common.games_coop.infinite", "common.games_coop.stochastic",
+        "common.meta.meta_games",
     ]:
         try:
             importlib.import_module(mod)

common/games_coop/pd_variants.py

@@ -1,13 +1,14 @@
 """Prisoner's Dilemma variants for KantBench."""
 from __future__ import annotations
 
+from dataclasses import replace
+
 from common.games import GAMES, GameConfig, _matrix_payoff_fn
+from common.variants import apply_exit
 from constant_definitions.game_constants import (
-    PD_CC_PAYOFF, PD_CD_PAYOFF, PD_DC_PAYOFF, PD_DD_PAYOFF,
     DEFAULT_NUM_ROUNDS, SINGLE_SHOT_ROUNDS,
 )
 from constant_definitions.var.pd_variant_constants import (
-    OPD_EXIT_PAYOFF,
     APD_A_TEMPTATION, APD_A_REWARD, APD_A_PUNISHMENT, APD_A_SUCKER,
     APD_B_TEMPTATION, APD_B_REWARD, APD_B_PUNISHMENT, APD_B_SUCKER,
     DONATION_BENEFIT, DONATION_COST,
@@ -18,22 +19,6 @@ from constant_definitions.var.pd_variant_constants import (
 _ZERO_F = float()
 
 
-# -- Optional PD (cooperate / defect / exit) --
-_OPD_EXIT_F = float(OPD_EXIT_PAYOFF)
-_OPD_BASE: dict[tuple[str, str], tuple[float, float]] = {
-    ("cooperate", "cooperate"): (float(PD_CC_PAYOFF), float(PD_CC_PAYOFF)),
-    ("cooperate", "defect"):    (float(PD_CD_PAYOFF), float(PD_DC_PAYOFF)),
-    ("defect", "cooperate"):    (float(PD_DC_PAYOFF), float(PD_CD_PAYOFF)),
-    ("defect", "defect"):       (float(PD_DD_PAYOFF), float(PD_DD_PAYOFF)),
-}
-
-
-def _optional_pd_payoff(pa: str, oa: str) -> tuple[float, float]:
-    if pa == "exit" or oa == "exit":
-        return (_OPD_EXIT_F, _OPD_EXIT_F)
-    return _OPD_BASE[(pa, oa)]
-
-
 # -- Asymmetric PD (alibi game: different payoffs per player) --
 _ASYM_PD: dict[tuple[str, str], tuple[float, float]] = {
     ("cooperate", "cooperate"): (float(APD_A_REWARD), float(APD_B_REWARD)),
@@ -74,20 +59,21 @@ _PW: dict[tuple[str, str], tuple[float, float]] = {
 
 
 # -- Register --
-PD_VARIANT_GAMES: dict[str, GameConfig] = {
-    "optional_pd": GameConfig(
-        name="Optional Prisoner's Dilemma",
-        description=(
-            "A Prisoner's Dilemma with a third action: exit. Exiting gives "
-            "a safe intermediate payoff regardless of the opponent's choice. "
-            "Tests whether outside options change cooperation dynamics and "
-            "models situations where players can walk away from interactions."
-        ),
-        actions=["cooperate", "defect", "exit"],
-        game_type="matrix",
-        default_rounds=DEFAULT_NUM_ROUNDS,
-        payoff_fn=_optional_pd_payoff,
+_PD_KEY = "prisoners_dilemma"
+_optional_pd_composed = apply_exit(GAMES[_PD_KEY], base_key=_PD_KEY)
+_optional_pd = replace(
+    _optional_pd_composed,
+    name="Optional Prisoner's Dilemma",
+    description=(
+        "A Prisoner's Dilemma with a third action: exit. Exiting gives "
+        "a safe intermediate payoff regardless of the opponent's choice. "
+        "Tests whether outside options change cooperation dynamics and "
+        "models situations where players can walk away from interactions."
     ),
+)
+
+PD_VARIANT_GAMES: dict[str, GameConfig] = {
+    "optional_pd": _optional_pd,
     "asymmetric_pd": GameConfig(
         name="Asymmetric Prisoner's Dilemma",
         description=(

common/games_info/communication.py

@@ -1,11 +1,12 @@
 """Communication and mediation games for KantBench."""
 from __future__ import annotations
 
+from dataclasses import replace
+
 from common.games import GAMES, GameConfig, _matrix_payoff_fn
+from common.variants import apply_cheap_talk, apply_binding_commitment
 from constant_definitions.game_constants import DEFAULT_NUM_ROUNDS, SINGLE_SHOT_ROUNDS
 from constant_definitions.var.communication_constants import (
-    CTPD_REWARD, CTPD_TEMPTATION, CTPD_PUNISHMENT, CTPD_SUCKER,
-    COMMIT_COST,
     CE_FOLLOW_FOLLOW, CE_FOLLOW_DEVIATE,
     CE_DEVIATE_FOLLOW, CE_DEVIATE_DEVIATE,
     FP_MATCH_PAYOFF, FP_MISMATCH_PAYOFF,
@@ -13,49 +14,35 @@ from constant_definitions.var.communication_constants import (
     MG_REJECT_ACCEPT, MG_REJECT_REJECT,
 )
 
-_ONE = int(bool(True))
 _ZERO_F = float()
 
-# -- Cheap Talk PD (message + action, messages are non-binding) --
-_CTPD_BASE: dict[tuple[str, str], tuple[float, float]] = {
-    ("cooperate", "cooperate"): (float(CTPD_REWARD), float(CTPD_REWARD)),
-    ("cooperate", "defect"):    (float(CTPD_SUCKER), float(CTPD_TEMPTATION)),
-    ("defect", "cooperate"):    (float(CTPD_TEMPTATION), float(CTPD_SUCKER)),
-    ("defect", "defect"):       (float(CTPD_PUNISHMENT), float(CTPD_PUNISHMENT)),
-}
-
+# -- Cheap Talk PD via composition --
+_PD_KEY = "prisoners_dilemma"
+_cheap_talk_pd_composed = apply_cheap_talk(GAMES[_PD_KEY], base_key=_PD_KEY)
+_cheap_talk_pd = replace(
+    _cheap_talk_pd_composed,
+    name="Cheap Talk Prisoner's Dilemma",
+    description=(
+        "A Prisoner's Dilemma where each player sends a non-binding "
+        "message before acting. Messages are cheap talk: costless and "
+        "unenforceable. Payoffs depend only on actual actions. Tests "
+        "whether non-binding communication improves cooperation."
+    ),
+    game_type="cheap_talk_pd",
+)
 
-def _cheap_talk_pd_payoff(pa: str, oa: str) -> tuple[float, float]:
-    """Message is cheap talk; payoff depends only on actual action."""
-    actual_p = pa.rsplit("_", _ONE)[_ONE]
-    actual_o = oa.rsplit("_", _ONE)[_ONE]
-    return _CTPD_BASE[(actual_p, actual_o)]
-
-
-_CTPD_ACTS = [
-    "msg_coop_cooperate", "msg_coop_defect",
-    "msg_def_cooperate", "msg_def_defect",
-]
-
-
-# -- Binding Commitment (costly commitment mechanism) --
-_CC = float(CTPD_REWARD)
-_CS = float(CTPD_SUCKER)
-_CT = float(CTPD_TEMPTATION)
-_CP = float(CTPD_PUNISHMENT)
-_COST = float(COMMIT_COST)
-
-_BIND_MATRIX: dict[tuple[str, str], tuple[float, float]] = {
-    ("commit_coop", "commit_coop"):  (_CC - _COST, _CC - _COST),
-    ("commit_coop", "free_coop"):    (_CC - _COST, _CC),
-    ("commit_coop", "free_defect"):  (_CS - _COST, _CT),
-    ("free_coop", "commit_coop"):    (_CC, _CC - _COST),
-    ("free_coop", "free_coop"):      (_CC, _CC),
-    ("free_coop", "free_defect"):    (_CS, _CT),
-    ("free_defect", "commit_coop"):  (_CT, _CS - _COST),
-    ("free_defect", "free_coop"):    (_CT, _CS),
-    ("free_defect", "free_defect"):  (_CP, _CP),
-}
+# -- Binding Commitment via composition --
+_binding_composed = apply_binding_commitment(GAMES[_PD_KEY], base_key=_PD_KEY)
+_binding_commitment = replace(
+    _binding_composed,
+    name="Binding Commitment Game",
+    description=(
+        "A Prisoner's Dilemma where players can pay a cost to make a "
+        "binding commitment to cooperate. The commitment is credible "
+        "but costly. Tests whether costly signaling through commitment "
+        "mechanisms changes equilibrium behavior."
+    ),
+)
 
 
 # -- Correlated Equilibrium (follow external mediator or deviate) --
@@ -90,32 +77,8 @@ _MED: dict[tuple[str, str], tuple[float, float]] = {
 
 # -- Register --
 COMMUNICATION_GAMES: dict[str, GameConfig] = {
-    "cheap_talk_pd": GameConfig(
-        name="Cheap Talk Prisoner's Dilemma",
-        description=(
-            "A Prisoner's Dilemma where each player sends a non-binding "
-            "message before acting. Messages are cheap talk: costless and "
-            "unenforceable. Payoffs depend only on actual actions. Tests "
-            "whether non-binding communication improves cooperation."
-        ),
-        actions=_CTPD_ACTS,
-        game_type="cheap_talk_pd",
-        default_rounds=DEFAULT_NUM_ROUNDS,
-        payoff_fn=_cheap_talk_pd_payoff,
-    ),
-    "binding_commitment": GameConfig(
-        name="Binding Commitment Game",
-        description=(
-            "A Prisoner's Dilemma where players can pay a cost to make a "
-            "binding commitment to cooperate. The commitment is credible "
-            "but costly. Tests whether costly signaling through commitment "
-            "mechanisms changes equilibrium behavior."
-        ),
-        actions=["commit_coop", "free_coop", "free_defect"],
-        game_type="matrix",
-        default_rounds=DEFAULT_NUM_ROUNDS,
-        payoff_fn=_matrix_payoff_fn(_BIND_MATRIX),
-    ),
+    "cheap_talk_pd": _cheap_talk_pd,
+    "binding_commitment": _binding_commitment,
     "correlated_equilibrium": GameConfig(
         name="Correlated Equilibrium Game",
         description=(

common/games_meta/coalition_config.py

@@ -2,10 +2,8 @@
 
 from __future__ import annotations
 
-from dataclasses import dataclass
-from typing import Callable
-
-from common.games_meta.nplayer_config import NPlayerGameConfig, NPLAYER_GAMES
+from common.games import GameConfig
+from common.games_meta.nplayer_config import NPLAYER_GAMES
 from constant_definitions.nplayer.coalition_constants import (
     COALITION_DEFAULT_ROUNDS, COALITION_DEFAULT_PENALTY_NUMERATOR,
     COALITION_DEFAULT_PENALTY_DENOMINATOR,
@@ -27,32 +25,18 @@ from constant_definitions.nplayer.coalition_constants import (
     COMMONS_LOW_DEPLETED, COMMONS_HIGH_DEPLETED,
 )
 
+CoalitionGameConfig = GameConfig
+
 _ONE = int(bool(True))
 _ZERO = int()
 _PEN_N = COALITION_DEFAULT_PENALTY_NUMERATOR
 _PEN_D = COALITION_DEFAULT_PENALTY_DENOMINATOR
 
 
-@dataclass(frozen=True)
-class CoalitionGameConfig:
-    """Immutable specification for a coalition-enabled N-player game."""
-
-    name: str
-    description: str
-    actions: list[str]
-    num_players: int
-    default_rounds: int
-    payoff_fn: Callable[[tuple[str, ...]], tuple[float, ...]]
-    enforcement: str
-    penalty_numerator: int
-    penalty_denominator: int
-    allow_side_payments: bool
+COALITION_GAMES: dict[str, GameConfig] = {}
 
 
-COALITION_GAMES: dict[str, CoalitionGameConfig] = {}
-
-
-def get_coalition_game(name: str) -> CoalitionGameConfig:
+def get_coalition_game(name: str) -> GameConfig:
     """Look up a coalition game by name. Raises KeyError if not found."""
     return COALITION_GAMES[name]
 
@@ -160,16 +144,17 @@ def _commons_governance_payoff(actions: tuple[str, ...]) -> tuple[float, ...]:
 # ---------------------------------------------------------------------------
 
 def _cfg(name: str, desc: str, actions: list[str], n: int,
-         fn: object, enf: str, side: bool = False) -> CoalitionGameConfig:
-    return CoalitionGameConfig(
-        name=name, description=desc, actions=actions, num_players=n,
-        default_rounds=COALITION_DEFAULT_ROUNDS, payoff_fn=fn,  # type: ignore[arg-type]
+         fn: object, enf: str, side: bool = False) -> GameConfig:
+    return GameConfig(
+        name=name, description=desc, actions=actions, game_type="coalition",
+        num_players=n, default_rounds=COALITION_DEFAULT_ROUNDS,
+        payoff_fn=fn,  # type: ignore[arg-type]
         enforcement=enf, penalty_numerator=_PEN_N, penalty_denominator=_PEN_D,
         allow_side_payments=side,
     )
 
 
-_BUILTIN_COALITION_GAMES: dict[str, CoalitionGameConfig] = {
+_BUILTIN_COALITION_GAMES: dict[str, GameConfig] = {
     "coalition_cartel": _cfg(
         "Cartel",
         "Players collude or compete. If enough collude the cartel holds. "
@@ -218,10 +203,6 @@ _BUILTIN_COALITION_GAMES: dict[str, CoalitionGameConfig] = {
 
 COALITION_GAMES.update(_BUILTIN_COALITION_GAMES)
 
-# Dual registration as plain NPlayerGameConfig
+# Register coalition games as N-player games too (same GameConfig instances)
 for _key, _c in _BUILTIN_COALITION_GAMES.items():
-    NPLAYER_GAMES[_key] = NPlayerGameConfig(
-        name=_c.name, description=_c.description, actions=_c.actions,
-        num_players=_c.num_players, default_rounds=_c.default_rounds,
-        payoff_fn=_c.payoff_fn,
-    )
+    NPLAYER_GAMES[_key] = _c

common/games_meta/game_tags.py

@@ -78,8 +78,8 @@ GAME_TAGS: dict[str, frozenset[str]] = {
     "gift_exchange": frozenset({NO_COMMUNICATION, COMPLETE_INFORMATION, SEQUENTIAL, SINGLE_SHOT, ASYMMETRIC_PAYOFF, SOCIAL_DILEMMA, LARGE_CHOICE}),
 
     # ── games_info/communication.py ──
-    "cheap_talk_pd": frozenset({CHEAP_TALK, COMPLETE_INFORMATION, SIMULTANEOUS, REPEATED, SYMMETRIC_PAYOFF, SOCIAL_DILEMMA, BINARY_CHOICE}),
-    "binding_commitment": frozenset({BINDING_COMMITMENT, COMPLETE_INFORMATION, SEQUENTIAL, SINGLE_SHOT, SYMMETRIC_PAYOFF, SOCIAL_DILEMMA, BINARY_CHOICE}),
+    "cheap_talk_pd": frozenset({CHEAP_TALK, COMPLETE_INFORMATION, SIMULTANEOUS, REPEATED, SYMMETRIC_PAYOFF, SOCIAL_DILEMMA, SMALL_CHOICE}),
+    "binding_commitment": frozenset({BINDING_COMMITMENT, COMPLETE_INFORMATION, SEQUENTIAL, SINGLE_SHOT, SYMMETRIC_PAYOFF, SOCIAL_DILEMMA, SMALL_CHOICE}),
     "correlated_equilibrium": frozenset({MEDIATED, COMPLETE_INFORMATION, SIMULTANEOUS, SINGLE_SHOT, SYMMETRIC_PAYOFF, SOCIAL_DILEMMA, BINARY_CHOICE}),
     "focal_point": frozenset({NO_COMMUNICATION, COMPLETE_INFORMATION, SIMULTANEOUS, SINGLE_SHOT, COORDINATION, SOCIAL_DILEMMA, SMALL_CHOICE}),
     "mediated_game": frozenset({MEDIATED, COMPLETE_INFORMATION, SIMULTANEOUS, SINGLE_SHOT, SYMMETRIC_PAYOFF, SOCIAL_DILEMMA, BINARY_CHOICE}),
@@ -175,6 +175,19 @@ GAME_TAGS: dict[str, frozenset[str]] = {
     "coalition_resource_trading": frozenset({CHEAP_TALK, COMPLETE_INFORMATION, SIMULTANEOUS, REPEATED, SYMMETRIC_PAYOFF, MARKET_COMPETITION, BINARY_CHOICE, MULTIPLAYER, COALITION_FORMATION, META_GOVERNANCE}),
     "coalition_rule_voting": frozenset({BINDING_COMMITMENT, COMPLETE_INFORMATION, SIMULTANEOUS, REPEATED, SYMMETRIC_PAYOFF, VOTING, BINARY_CHOICE, MULTIPLAYER, COALITION_FORMATION, BINDING_ENFORCEMENT, META_GOVERNANCE}),
     "coalition_commons": frozenset({CHEAP_TALK, COMPLETE_INFORMATION, SIMULTANEOUS, REPEATED, SYMMETRIC_PAYOFF, SOCIAL_DILEMMA, BINARY_CHOICE, MULTIPLAYER, COALITION_FORMATION, PENALTY_ENFORCEMENT, META_GOVERNANCE}),
+
+    # ── meta/meta_games.py ──
+    "rule_proposal_prisoners_dilemma": frozenset({BINDING_COMMITMENT, COMPLETE_INFORMATION, SIMULTANEOUS, REPEATED, SYMMETRIC_PAYOFF, SOCIAL_DILEMMA, LARGE_CHOICE, META_GOVERNANCE}),
+    "rule_proposal_stag_hunt": frozenset({BINDING_COMMITMENT, COMPLETE_INFORMATION, SIMULTANEOUS, REPEATED, COORDINATION, SOCIAL_DILEMMA, LARGE_CHOICE, META_GOVERNANCE}),
+    "rule_proposal_hawk_dove": frozenset({BINDING_COMMITMENT, COMPLETE_INFORMATION, SIMULTANEOUS, REPEATED, ANTI_COORDINATION, SOCIAL_DILEMMA, LARGE_CHOICE, META_GOVERNANCE}),
+    "rule_signal_prisoners_dilemma": frozenset({CHEAP_TALK, COMPLETE_INFORMATION, SIMULTANEOUS, REPEATED, SYMMETRIC_PAYOFF, SOCIAL_DILEMMA, LARGE_CHOICE, META_GOVERNANCE}),
+    "rule_signal_stag_hunt": frozenset({CHEAP_TALK, COMPLETE_INFORMATION, SIMULTANEOUS, REPEATED, COORDINATION, SOCIAL_DILEMMA, LARGE_CHOICE, META_GOVERNANCE}),
+    "rule_signal_hawk_dove": frozenset({CHEAP_TALK, COMPLETE_INFORMATION, SIMULTANEOUS, REPEATED, ANTI_COORDINATION, SOCIAL_DILEMMA, LARGE_CHOICE, META_GOVERNANCE}),
+
+    # ── meta/meta_games.py (gossip) ──
+    "gossip_prisoners_dilemma": frozenset({CHEAP_TALK, COMPLETE_INFORMATION, SIMULTANEOUS, REPEATED, SYMMETRIC_PAYOFF, SOCIAL_DILEMMA, LARGE_CHOICE, META_GOVERNANCE}),
+    "gossip_stag_hunt": frozenset({CHEAP_TALK, COMPLETE_INFORMATION, SIMULTANEOUS, REPEATED, COORDINATION, SOCIAL_DILEMMA, LARGE_CHOICE, META_GOVERNANCE}),
+    "gossip_hawk_dove": frozenset({CHEAP_TALK, COMPLETE_INFORMATION, SIMULTANEOUS, REPEATED, ANTI_COORDINATION, SOCIAL_DILEMMA, LARGE_CHOICE, META_GOVERNANCE}),
 }
 
 
@@ -183,6 +196,33 @@ GAME_TAGS: dict[str, frozenset[str]] = {
 # ---------------------------------------------------------------------------
 
 
+def derive_variant_tags(
+    base_tags: frozenset[str], variant_name: str,
+) -> frozenset[str]:
+    """Compute tags for a composed variant game from base game tags."""
+    tags = set(base_tags)
+    if variant_name == "cheap_talk":
+        tags.discard(NO_COMMUNICATION)
+        tags.add(CHEAP_TALK)
+        tags.discard(BINARY_CHOICE)
+        tags.add(SMALL_CHOICE)
+    elif variant_name == "binding_commitment":
+        tags.discard(NO_COMMUNICATION)
+        tags.add(BINDING_COMMITMENT)
+        tags.discard(BINARY_CHOICE)
+        tags.add(SMALL_CHOICE)
+    elif variant_name == "exit":
+        tags.discard(BINARY_CHOICE)
+        tags.add(SMALL_CHOICE)
+    elif variant_name == "gossip":
+        tags.discard(NO_COMMUNICATION)
+        tags.add(CHEAP_TALK)
+        tags.discard(BINARY_CHOICE)
+        tags.add(LARGE_CHOICE)
+        tags.add(META_GOVERNANCE)
+    return frozenset(tags)
+
+
 def get_games_by_tag(tag: str) -> list[str]:
     """Return all game keys that have the given tag."""
     return [key for key, tags in GAME_TAGS.items() if tag in tags]

common/games_meta/nplayer_config.py

@@ -2,29 +2,13 @@
 
 from __future__ import annotations
 
-from dataclasses import dataclass
-from typing import Callable
+from common.games import GameConfig
 
-from constant_definitions.nplayer.nplayer_constants import (
-    NPLAYER_DEFAULT_ROUNDS,
-)
+NPlayerGameConfig = GameConfig
 
+NPLAYER_GAMES: dict[str, GameConfig] = {}
 
-@dataclass(frozen=True)
-class NPlayerGameConfig:
-    """Immutable specification for an N-player game type."""
 
-    name: str
-    description: str
-    actions: list[str]
-    num_players: int
-    default_rounds: int
-    payoff_fn: Callable[[tuple[str, ...]], tuple[float, ...]]
-
-
-NPLAYER_GAMES: dict[str, NPlayerGameConfig] = {}
-
-
-def get_nplayer_game(name: str) -> NPlayerGameConfig:
+def get_nplayer_game(name: str) -> GameConfig:
     """Look up an N-player game by name. Raises KeyError if not found."""
     return NPLAYER_GAMES[name]

common/games_meta/nplayer_games.py

@@ -2,7 +2,8 @@
 
 from __future__ import annotations
 
-from common.games_meta.nplayer_config import NPlayerGameConfig, NPLAYER_GAMES
+from common.games import GameConfig
+from common.games_meta.nplayer_config import NPLAYER_GAMES
 from constant_definitions.nplayer.nplayer_constants import (
     NPLAYER_DEFAULT_ROUNDS,
     NPG_ENDOWMENT,
@@ -90,19 +91,20 @@ def _el_farol_payoff(actions: tuple[str, ...]) -> tuple[float, ...]:
 _THREE = _ONE + _ONE + _ONE
 _FIVE = _THREE + _ONE + _ONE
 
-_BUILTIN_NPLAYER_GAMES: dict[str, NPlayerGameConfig] = {
-    "nplayer_public_goods": NPlayerGameConfig(
+_BUILTIN_NPLAYER_GAMES: dict[str, GameConfig] = {
+    "nplayer_public_goods": GameConfig(
         name="N-Player Public Goods",
         description=(
             "Each player contributes from an endowment. The total pot is "
             "multiplied and split equally among all players."
         ),
         actions=_PG_ACTIONS,
+        game_type="public_goods",
         num_players=_FIVE,
         default_rounds=NPLAYER_DEFAULT_ROUNDS,
         payoff_fn=_public_goods_payoff,
     ),
-    "nplayer_volunteer_dilemma": NPlayerGameConfig(
+    "nplayer_volunteer_dilemma": GameConfig(
         name="N-Player Volunteer's Dilemma",
         description=(
             "Players choose to volunteer or abstain. If at least one "
@@ -110,17 +112,19 @@ _BUILTIN_NPLAYER_GAMES: dict[str, NPlayerGameConfig] = {
             "If nobody volunteers, everyone gets nothing."
         ),
         actions=["volunteer", "abstain"],
+        game_type="matrix",
         num_players=_FIVE,
         default_rounds=NPLAYER_DEFAULT_ROUNDS,
         payoff_fn=_volunteer_dilemma_payoff,
     ),
-    "nplayer_el_farol": NPlayerGameConfig(
+    "nplayer_el_farol": GameConfig(
         name="N-Player El Farol Bar",
         description=(
             "Players decide whether to attend a bar. The bar is fun when "
             "not crowded but unpleasant when too many people show up."
         ),
         actions=["attend", "stay_home"],
+        game_type="matrix",
         num_players=_FIVE,
         default_rounds=NPLAYER_DEFAULT_ROUNDS,
         payoff_fn=_el_farol_payoff,

common/meta/memory_store.py

@@ -0,0 +1,183 @@
+"""Persistent cross-episode memory backed by cognee knowledge graph.
+
+Records episode summaries, gossip ratings, and opponent statistics.
+Uses in-memory stats when cognee is not installed.
+"""
+from __future__ import annotations
+
+import asyncio
+import threading
+from typing import Any
+
+from constant_definitions.var.meta.reputation_constants import (
+    COGNEE_DATASET_NAME,
+    COGNEE_SEARCH_TYPE,
+    DEFAULT_REPUTATION_SCORE_NUMERATOR,
+    DEFAULT_REPUTATION_SCORE_DENOMINATOR,
+    REPUTATION_DECAY_NUMERATOR,
+    REPUTATION_DECAY_DENOMINATOR,
+    META_KEY_COOPERATION_RATE,
+    META_KEY_INTERACTION_COUNT,
+    META_KEY_GOSSIP_HISTORY,
+)
+
+_ZERO = int()
+_ONE = int(bool(True))
+_DEFAULT_SCORE = (
+    DEFAULT_REPUTATION_SCORE_NUMERATOR / DEFAULT_REPUTATION_SCORE_DENOMINATOR
+)
+_DECAY = REPUTATION_DECAY_NUMERATOR / REPUTATION_DECAY_DENOMINATOR
+
+try:
+    import cognee as _cognee  # type: ignore[import-untyped]
+    _HAS_COGNEE = True
+except ImportError:
+    _cognee = None
+    _HAS_COGNEE = False
+
+
+class AsyncBridge:
+    """Runs async coroutines from sync code via a dedicated thread."""
+
+    def __init__(self) -> None:
+        self._loop = asyncio.new_event_loop()
+        self._thread = threading.Thread(
+            target=self._loop.run_forever, daemon=True,
+        )
+        self._thread.start()
+
+    def run(self, coro: Any) -> Any:
+        """Submit *coro* to the background loop and block for the result."""
+        future = asyncio.run_coroutine_threadsafe(coro, self._loop)
+        return future.result()
+
+
+def _default_reputation() -> dict[str, Any]:
+    """Return a neutral default reputation dict."""
+    return {
+        "score": _DEFAULT_SCORE,
+        META_KEY_COOPERATION_RATE: _DEFAULT_SCORE,
+        META_KEY_INTERACTION_COUNT: _ZERO,
+        META_KEY_GOSSIP_HISTORY: [],
+    }
+
+
+def _format_episode_text(
+    agent_id: str,
+    opponent_id: str,
+    game: str,
+    history: list[Any],
+    cooperation_rate: float,
+    scores: tuple[float, float],
+) -> str:
+    """Format an episode summary for cognee ingestion."""
+    rounds = len(history)
+    p_score, o_score = scores
+    actions = "; ".join(
+        f"R{r.round_number}: {r.player_action} vs {r.opponent_action}"
+        for r in history
+    )
+    return (
+        f"Game Interaction Report\n"
+        f"Agent: {agent_id} | Opponent: {opponent_id} | Game: {game}\n"
+        f"Rounds: {rounds} | Agent Score: {p_score} | "
+        f"Opponent Score: {o_score}\n"
+        f"Cooperation Rate: {cooperation_rate}\n"
+        f"Actions: {actions}\n"
+    )
+
+
+def _parse_reputation(
+    results: Any, stats: dict[str, Any],
+) -> dict[str, Any]:
+    """Merge cognee search results with in-memory stats."""
+    rep = dict(stats) if stats else _default_reputation()
+    if results:
+        rep["cognee_context"] = str(results)
+    return rep
+
+
+class CogneeMemoryStore:
+    """Persistent memory backed by cognee knowledge graph."""
+
+    def __init__(self) -> None:
+        self._bridge = AsyncBridge() if _HAS_COGNEE else None
+        self._stats: dict[str, dict[str, Any]] = {}
+
+    def record_episode(
+        self,
+        agent_id: str,
+        opponent_id: str,
+        game: str,
+        history: list[Any],
+        cooperation_rate: float,
+        scores: tuple[float, float],
+    ) -> None:
+        """Format episode as text and add to cognee, then cognify."""
+        text = _format_episode_text(
+            agent_id, opponent_id, game, history,
+            cooperation_rate, scores,
+        )
+        if self._bridge is not None and _HAS_COGNEE:
+            try:
+                self._bridge.run(
+                    _cognee.add(text, dataset_name=COGNEE_DATASET_NAME),
+                )
+                self._bridge.run(_cognee.cognify())
+            except Exception:
+                pass
+        self._update_stats(opponent_id, cooperation_rate, scores)
+
+    def query_reputation(self, opponent_id: str) -> dict[str, Any]:
+        """Query cognee for opponent reputation. Uses stats if unavailable."""
+        stats = self._stats.get(opponent_id, _default_reputation())
+        if self._bridge is None or not _HAS_COGNEE:
+            return stats
+        try:
+            results = self._bridge.run(
+                _cognee.search(
+                    f"reputation and behavior of {opponent_id}",
+                    search_type=COGNEE_SEARCH_TYPE,
+                ),
+            )
+            return _parse_reputation(results, stats)
+        except Exception:
+            return stats
+
+    def record_gossip(
+        self, rater_id: str, target_id: str, rating: str,
+    ) -> None:
+        """Record a gossip rating in cognee."""
+        text = f"{rater_id} rated {target_id} as {rating}."
+        if self._bridge is not None and _HAS_COGNEE:
+            try:
+                self._bridge.run(
+                    _cognee.add(text, dataset_name=COGNEE_DATASET_NAME),
+                )
+            except Exception:
+                pass
+        target_stats = self._stats.setdefault(
+            target_id, _default_reputation(),
+        )
+        gossip_list = target_stats.setdefault(META_KEY_GOSSIP_HISTORY, [])
+        gossip_list.append({"rater": rater_id, "rating": rating})
+
+    def get_stats(self, opponent_id: str) -> dict[str, Any]:
+        """Fast in-memory stats (no LLM call)."""
+        return self._stats.get(opponent_id, _default_reputation())
+
+    def _update_stats(
+        self,
+        opponent_id: str,
+        coop_rate: float,
+        scores: tuple[float, float],
+    ) -> None:
+        """Update running statistics for an opponent."""
+        current = self._stats.get(opponent_id, _default_reputation())
+        count = current.get(META_KEY_INTERACTION_COUNT, _ZERO) + _ONE
+        old_coop = current.get(META_KEY_COOPERATION_RATE, _DEFAULT_SCORE)
+        blended = old_coop * _DECAY + coop_rate * (_ONE - _DECAY)
+        current["score"] = blended
+        current[META_KEY_COOPERATION_RATE] = blended
+        current[META_KEY_INTERACTION_COUNT] = count
+        self._stats[opponent_id] = current

common/meta/meta_games.py

@@ -0,0 +1,70 @@
+"""Pre-registered meta-gaming rule-proposal games for KantBench.
+
+Registers symmetric meta-games (rule_proposal, rule_signal) for the
+three core matrix games.  Constitutional and proposer-responder are
+composed per-episode via ``compose_game()`` because constitutional uses
+mutable closure state that must be fresh per episode.
+"""
+from __future__ import annotations
+
+from dataclasses import replace
+
+from common.games import GAMES
+from common.meta.variants_meta import apply_rule_proposal, apply_rule_signal
+from common.meta.variants_reputation import apply_gossip
+
+_BASE_KEYS = ("prisoners_dilemma", "stag_hunt", "hawk_dove")
+
+_FRIENDLY_NAMES = {
+    "prisoners_dilemma": "Prisoner's Dilemma",
+    "stag_hunt": "Stag Hunt",
+    "hawk_dove": "Hawk-Dove",
+}
+
+META_GAMES: dict = {}
+
+for _key in _BASE_KEYS:
+    _base = GAMES[_key]
+    _fname = _FRIENDLY_NAMES[_key]
+
+    _rp = apply_rule_proposal(_base, base_key=_key)
+    _rp = replace(
+        _rp,
+        name=f"Rule Proposal {_fname}",
+        description=(
+            f"{_fname} with simultaneous binding rule proposals. "
+            "Both players propose a rule and choose an action. "
+            "If proposals match the agreed rule modifies payoffs."
+        ),
+    )
+    META_GAMES[f"rule_proposal_{_key}"] = _rp
+
+    _rs = apply_rule_signal(_base, base_key=_key)
+    _rs = replace(
+        _rs,
+        name=f"Rule Signal {_fname}",
+        description=(
+            f"{_fname} with simultaneous non-binding rule signals. "
+            "Both players signal a preferred rule and choose an action. "
+            "Signals are visible but never enforced."
+        ),
+    )
+    META_GAMES[f"rule_signal_{_key}"] = _rs
+
+for _key in _BASE_KEYS:
+    _base = GAMES[_key]
+    _fname = _FRIENDLY_NAMES[_key]
+
+    _gp = apply_gossip(_base, base_key=_key)
+    _gp = replace(
+        _gp,
+        name=f"Gossip {_fname}",
+        description=(
+            f"{_fname} with reputation gossip. "
+            "Players rate opponents as trustworthy, untrustworthy, "
+            "or neutral alongside each action."
+        ),
+    )
+    META_GAMES[f"gossip_{_key}"] = _gp
+
+GAMES.update(META_GAMES)

common/meta/meta_rules.py

@@ -0,0 +1,134 @@
+"""Rule catalog and payoff transforms for the meta-gaming variant system.
+
+Each rule is a payoff transform: given base payoffs and actions, return
+modified payoffs. The ``apply_rule`` dispatcher looks up a rule by name
+and delegates to the corresponding transform function.
+"""
+from __future__ import annotations
+
+from typing import Callable
+
+from constant_definitions.var.meta.meta_rule_constants import (
+    RULE_NONE, RULE_EQUAL_SPLIT, RULE_COOP_BONUS,
+    RULE_DEFECT_PENALTY, RULE_MIN_GUARANTEE, RULE_BAN_DEFECT,
+    COOP_BONUS_NUMERATOR, COOP_BONUS_DENOMINATOR,
+    DEFECT_PENALTY_NUMERATOR, DEFECT_PENALTY_DENOMINATOR,
+    MIN_GUARANTEE_NUMERATOR, MIN_GUARANTEE_DENOMINATOR,
+    BAN_DEFECT_PENALTY_NUMERATOR, BAN_DEFECT_PENALTY_DENOMINATOR,
+    EQUAL_SPLIT_DENOMINATOR,
+    META_SEPARATOR, META_SPLIT_LIMIT,
+)
+
+RuleTransform = Callable[
+    [float, float, str, str], tuple[float, float]
+]
+
+_COOPERATIVE_ACTIONS = frozenset({"cooperate", "stag", "dove"})
+
+_COOP_BONUS = COOP_BONUS_NUMERATOR / COOP_BONUS_DENOMINATOR
+_DEFECT_PENALTY = DEFECT_PENALTY_NUMERATOR / DEFECT_PENALTY_DENOMINATOR
+_MIN_GUARANTEE = MIN_GUARANTEE_NUMERATOR / MIN_GUARANTEE_DENOMINATOR
+_BAN_PENALTY = BAN_DEFECT_PENALTY_NUMERATOR / BAN_DEFECT_PENALTY_DENOMINATOR
+
+
+def _is_cooperative(action: str) -> bool:
+    """Return True if *action* is a cooperative action."""
+    return action in _COOPERATIVE_ACTIONS
+
+
+def _rule_none(
+    base_p: float, base_o: float,
+    p_action: str, o_action: str,
+) -> tuple[float, float]:
+    return (base_p, base_o)
+
+
+def _rule_equal_split(
+    base_p: float, base_o: float,
+    p_action: str, o_action: str,
+) -> tuple[float, float]:
+    total = base_p + base_o
+    share = total / EQUAL_SPLIT_DENOMINATOR
+    return (share, share)
+
+
+def _rule_coop_bonus(
+    base_p: float, base_o: float,
+    p_action: str, o_action: str,
+) -> tuple[float, float]:
+    p_pay = base_p + (_COOP_BONUS if _is_cooperative(p_action) else float())
+    o_pay = base_o + (_COOP_BONUS if _is_cooperative(o_action) else float())
+    return (p_pay, o_pay)
+
+
+def _rule_defect_penalty(
+    base_p: float, base_o: float,
+    p_action: str, o_action: str,
+) -> tuple[float, float]:
+    p_pay = base_p - (
+        _DEFECT_PENALTY if not _is_cooperative(p_action) else float()
+    )
+    o_pay = base_o - (
+        _DEFECT_PENALTY if not _is_cooperative(o_action) else float()
+    )
+    return (p_pay, o_pay)
+
+
+def _rule_min_guarantee(
+    base_p: float, base_o: float,
+    p_action: str, o_action: str,
+) -> tuple[float, float]:
+    p_pay = max(base_p, _MIN_GUARANTEE)
+    o_pay = max(base_o, _MIN_GUARANTEE)
+    return (p_pay, o_pay)
+
+
+def _rule_ban_defect(
+    base_p: float, base_o: float,
+    p_action: str, o_action: str,
+) -> tuple[float, float]:
+    p_pay = base_p - (
+        _BAN_PENALTY if not _is_cooperative(p_action) else float()
+    )
+    o_pay = base_o - (
+        _BAN_PENALTY if not _is_cooperative(o_action) else float()
+    )
+    return (p_pay, o_pay)
+
+
+RULE_CATALOG: dict[str, RuleTransform] = {
+    RULE_NONE: _rule_none,
+    RULE_EQUAL_SPLIT: _rule_equal_split,
+    RULE_COOP_BONUS: _rule_coop_bonus,
+    RULE_DEFECT_PENALTY: _rule_defect_penalty,
+    RULE_MIN_GUARANTEE: _rule_min_guarantee,
+    RULE_BAN_DEFECT: _rule_ban_defect,
+}
+
+
+def apply_rule(
+    rule_name: str,
+    base_p: float, base_o: float,
+    p_action: str, o_action: str,
+) -> tuple[float, float]:
+    """Look up *rule_name* in the catalog and apply its transform."""
+    return RULE_CATALOG[rule_name](base_p, base_o, p_action, o_action)
+
+
+_ZERO = int()
+_ONE = int(bool(True))
+_TWO = _ONE + _ONE
+
+
+def parse_meta_action(
+    action: str,
+    split_limit: int = META_SPLIT_LIMIT,
+) -> tuple[str, str, str]:
+    """Parse an encoded meta-action into (prefix, rule, base_action).
+
+    The action format is ``prefix_rule_baseaction`` where *rule* is a
+    single token (no underscores).  Using ``split`` with the configured
+    split limit yields exactly three parts.
+    """
+    parts = action.split(META_SEPARATOR, split_limit)
+    return (parts[_ZERO], parts[_ONE], parts[_TWO])

common/meta/variants_meta.py

@@ -0,0 +1,193 @@
+"""Meta-gaming variant transforms for rule proposal, signaling, and negotiation.
+
+Four composable transforms following the ``apply_*`` pattern from
+``variants.py``.  Each expands the action space to encode both a rule
+proposal and a base-game action in a single string.
+"""
+from __future__ import annotations
+
+from dataclasses import replace
+from typing import Callable
+
+from common.games import GameConfig
+from common.meta.meta_rules import apply_rule, parse_meta_action
+
+from constant_definitions.var.meta.meta_rule_constants import (
+    VARIANT_RULE_PROPOSAL, VARIANT_RULE_SIGNAL,
+    VARIANT_CONSTITUTIONAL, VARIANT_PROPOSER_RESPONDER,
+    META_PROP_PREFIX, META_SIG_PREFIX, META_CONST_PREFIX,
+    META_RPROP_PREFIX, META_RACCEPT_PREFIX, META_RREJECT_PREFIX,
+    META_SEPARATOR,
+    DEFAULT_RULE_CATALOG, RULE_NONE,
+)
+
+_ONE = int(bool(True))
+_ZERO = int()
+
+
+def _build_prefixed_actions(
+    prefix: str,
+    rules: tuple[str, ...],
+    base_actions: list[str],
+) -> list[str]:
+    """Build action list: prefix_rule_baseaction for each combination."""
+    sep = META_SEPARATOR
+    return [
+        sep.join([prefix, rule, act])
+        for rule in rules
+        for act in base_actions
+    ]
+
+
+def apply_rule_proposal(
+    base: GameConfig,
+    rules: tuple[str, ...] = DEFAULT_RULE_CATALOG,
+    base_key: str = "",
+) -> GameConfig:
+    """Simultaneous, binding, per-round rule proposal.
+
+    Both players choose ``prop_<rule>_<action>``.  If both propose the
+    same rule the rule's payoff transform is applied; otherwise base
+    payoffs are used.
+    """
+    prefix = META_PROP_PREFIX
+    new_actions = _build_prefixed_actions(prefix, rules, base.actions)
+    original_payoff = base.payoff_fn
+
+    def _payoff(pa: str, oa: str) -> tuple[float, float]:
+        _, p_rule, p_act = parse_meta_action(pa)
+        _, o_rule, o_act = parse_meta_action(oa)
+        base_p, base_o = original_payoff(p_act, o_act)
+        if p_rule == o_rule:
+            return apply_rule(p_rule, base_p, base_o, p_act, o_act)
+        return (base_p, base_o)
+
+    return replace(
+        base,
+        actions=new_actions,
+        payoff_fn=_payoff,
+        applied_variants=base.applied_variants + (VARIANT_RULE_PROPOSAL,),
+        base_game_key=base_key or base.base_game_key,
+    )
+
+
+def apply_rule_signal(
+    base: GameConfig,
+    rules: tuple[str, ...] = DEFAULT_RULE_CATALOG,
+    base_key: str = "",
+) -> GameConfig:
+    """Simultaneous, non-binding, per-round rule signal.
+
+    Both players choose ``sig_<rule>_<action>``.  Proposals are visible
+    in history but never enforced -- payoffs always come from the base game.
+    """
+    prefix = META_SIG_PREFIX
+    new_actions = _build_prefixed_actions(prefix, rules, base.actions)
+    original_payoff = base.payoff_fn
+
+    def _payoff(pa: str, oa: str) -> tuple[float, float]:
+        _, _p_rule, p_act = parse_meta_action(pa)
+        _, _o_rule, o_act = parse_meta_action(oa)
+        return original_payoff(p_act, o_act)
+
+    return replace(
+        base,
+        actions=new_actions,
+        payoff_fn=_payoff,
+        applied_variants=base.applied_variants + (VARIANT_RULE_SIGNAL,),
+        base_game_key=base_key or base.base_game_key,
+    )
+
+
+def apply_constitutional(
+    base: GameConfig,
+    rules: tuple[str, ...] = DEFAULT_RULE_CATALOG,
+    base_key: str = "",
+) -> GameConfig:
+    """Multi-round negotiation with binding lock-in once agreed.
+
+    Both players choose ``const_<rule>_<action>``.  The first round
+    where both propose the same non-none rule locks that rule in for
+    ALL subsequent rounds.  Before agreement, base payoffs apply.
+
+    A fresh mutable closure is created per call so each episode via
+    ``compose_game()`` gets clean state.
+    """
+    prefix = META_CONST_PREFIX
+    new_actions = _build_prefixed_actions(prefix, rules, base.actions)
+    original_payoff = base.payoff_fn
+    adopted_rule: list[str] = []
+
+    def _payoff(pa: str, oa: str) -> tuple[float, float]:
+        _, p_rule, p_act = parse_meta_action(pa)
+        _, o_rule, o_act = parse_meta_action(oa)
+        base_p, base_o = original_payoff(p_act, o_act)
+
+        if adopted_rule:
+            return apply_rule(adopted_rule[_ZERO], base_p, base_o, p_act, o_act)
+
+        if p_rule == o_rule and p_rule != RULE_NONE:
+            adopted_rule.append(p_rule)
+            return apply_rule(p_rule, base_p, base_o, p_act, o_act)
+
+        return (base_p, base_o)
+
+    return replace(
+        base,
+        actions=new_actions,
+        payoff_fn=_payoff,
+        applied_variants=base.applied_variants + (VARIANT_CONSTITUTIONAL,),
+        base_game_key=base_key or base.base_game_key,
+    )
+
+
+def apply_proposer_responder(
+    base: GameConfig,
+    rules: tuple[str, ...] = DEFAULT_RULE_CATALOG,
+    base_key: str = "",
+) -> GameConfig:
+    """Asymmetric: player proposes a rule, opponent accepts or rejects.
+
+    Player actions: ``rprop_<rule>_<action>`` (propose + play).
+    Opponent actions: ``raccept_<action>`` or ``rreject_<action>``
+    (respond + play).
+
+    Accept -> rule applies to base payoffs.  Reject -> base payoffs.
+    """
+    sep = META_SEPARATOR
+    player_actions = _build_prefixed_actions(
+        META_RPROP_PREFIX, rules, base.actions,
+    )
+    opp_actions: list[str] = []
+    for act in base.actions:
+        opp_actions.append(sep.join([META_RACCEPT_PREFIX, act]))
+        opp_actions.append(sep.join([META_RREJECT_PREFIX, act]))
+
+    original_payoff = base.payoff_fn
+
+    def _payoff(pa: str, oa: str) -> tuple[float, float]:
+        _, p_rule, p_act = parse_meta_action(pa)
+        o_parts = oa.split(sep, _ONE)
+        o_prefix = o_parts[_ZERO]
+        o_act = o_parts[_ONE]
+        base_p, base_o = original_payoff(p_act, o_act)
+        if o_prefix == META_RACCEPT_PREFIX:
+            return apply_rule(p_rule, base_p, base_o, p_act, o_act)
+        return (base_p, base_o)
+
+    return replace(
+        base,
+        actions=player_actions,
+        payoff_fn=_payoff,
+        applied_variants=base.applied_variants + (VARIANT_PROPOSER_RESPONDER,),
+        base_game_key=base_key or base.base_game_key,
+        opponent_actions=tuple(opp_actions),
+    )
+
+
+_META_VARIANT_REGISTRY: dict[str, Callable[..., GameConfig]] = {
+    VARIANT_RULE_PROPOSAL: apply_rule_proposal,
+    VARIANT_RULE_SIGNAL: apply_rule_signal,
+    VARIANT_CONSTITUTIONAL: apply_constitutional,
+    VARIANT_PROPOSER_RESPONDER: apply_proposer_responder,
+}

common/meta/variants_reputation.py

@@ -0,0 +1,73 @@
+"""Reputation gossip variant transform for the composable variant system.
+
+Follows the ``apply_*`` pattern from ``variants.py`` and ``variants_meta.py``.
+Adds ``gossip_<rating>_<base_action>`` actions to any base game.
+Payoffs depend only on the base action, like cheap_talk.
+"""
+from __future__ import annotations
+
+from dataclasses import replace
+from typing import Callable
+
+from common.games import GameConfig
+
+from constant_definitions.var.meta.reputation_constants import (
+    VARIANT_GOSSIP,
+    DEFAULT_RATINGS,
+    GOSSIP_PREFIX,
+    GOSSIP_SEPARATOR,
+    GOSSIP_SPLIT_LIMIT,
+)
+
+_ONE = int(bool(True))
+_ZERO = int()
+_TWO = _ONE + _ONE
+
+
+def apply_gossip(
+    base: GameConfig,
+    ratings: tuple[str, ...] = DEFAULT_RATINGS,
+    base_key: str = "",
+) -> GameConfig:
+    """Add reputation gossip to a base game.
+
+    For base actions ``[A, B]`` and ratings ``[trustworthy, untrustworthy,
+    neutral]``, produces ``[gossip_trustworthy_A, gossip_trustworthy_B,
+    gossip_untrustworthy_A, ...]``.  Payoffs depend only on the actual
+    action (last segment), like cheap_talk.
+    """
+    sep = GOSSIP_SEPARATOR
+    prefix = GOSSIP_PREFIX
+    new_actions = [
+        sep.join([prefix, rating, act])
+        for rating in ratings
+        for act in base.actions
+    ]
+    original_payoff = base.payoff_fn
+
+    def _payoff(pa: str, oa: str) -> tuple[float, float]:
+        actual_p = pa.rsplit(sep, _ONE)[_ONE]
+        actual_o = oa.rsplit(sep, _ONE)[_ONE]
+        return original_payoff(actual_p, actual_o)
+
+    return replace(
+        base,
+        actions=new_actions,
+        payoff_fn=_payoff,
+        applied_variants=base.applied_variants + (VARIANT_GOSSIP,),
+        base_game_key=base_key or base.base_game_key,
+    )
+
+
+def parse_gossip_action(action: str) -> tuple[str, str, str]:
+    """Parse ``gossip_<rating>_<base_action>`` into components.
+
+    Returns ``(prefix, rating, base_action)``.
+    """
+    parts = action.split(GOSSIP_SEPARATOR, GOSSIP_SPLIT_LIMIT)
+    return (parts[_ZERO], parts[_ONE], parts[_TWO])
+
+
+_REPUTATION_VARIANT_REGISTRY: dict[str, Callable[..., GameConfig]] = {
+    VARIANT_GOSSIP: apply_gossip,
+}

common/variants.py

@@ -0,0 +1,279 @@
+"""Composable game variant transforms for KantBench.
+
+Each ``apply_*`` function takes a :class:`GameConfig` and returns a new
+:class:`GameConfig` with modified actions, payoff function, and metadata.
+Variants compose: ``apply_exit(apply_cheap_talk(base))`` works.
+"""
+
+from __future__ import annotations
+
+from dataclasses import replace
+from typing import Callable
+
+from common.games import GAMES, GameConfig
+from constant_definitions.game_constants import (
+    DEFAULT_TWO_PLAYERS,
+    OPPONENT_MODE_SELF,
+    OPPONENT_MODE_CROSS,
+)
+from constant_definitions.var.pd_variant_constants import (
+    OPD_EXIT_PAYOFF,
+    VARIANT_CHEAP_TALK,
+    VARIANT_EXIT,
+    VARIANT_BINDING_COMMITMENT,
+    VARIANT_NOISY_ACTIONS,
+    VARIANT_NOISY_PAYOFFS,
+    VARIANT_SELF_PLAY,
+    VARIANT_CROSS_MODEL,
+    CT_MSG_PREFIX,
+    CT_SEPARATOR,
+    BC_COMMIT_PREFIX,
+    BC_FREE_PREFIX,
+    EXIT_ACTION,
+    DEFAULT_TREMBLE_PROB_NUMERATOR,
+    DEFAULT_TREMBLE_PROB_DENOMINATOR,
+    DEFAULT_NOISE_SCALE_NUMERATOR,
+    DEFAULT_NOISE_SCALE_DENOMINATOR,
+)
+from constant_definitions.var.communication_constants import COMMIT_COST
+
+_ONE = int(bool(True))
+_ZERO = int()
+
+
+def apply_cheap_talk(
+    base: GameConfig,
+    base_key: str = "",
+) -> GameConfig:
+    """Add a non-binding message phase to a base game.
+
+    For base actions ``[A, B]`` produces ``[msg_A_A, msg_A_B, msg_B_A,
+    msg_B_B]``.  Payoffs depend only on the actual action (last segment).
+    """
+    sep = CT_SEPARATOR
+    prefix = CT_MSG_PREFIX
+    base_actions = base.actions
+    new_actions = [
+        sep.join([prefix, msg, act])
+        for msg in base_actions
+        for act in base_actions
+    ]
+
+    original_payoff = base.payoff_fn
+
+    def _payoff(pa: str, oa: str) -> tuple[float, float]:
+        actual_p = pa.rsplit(sep, _ONE)[_ONE]
+        actual_o = oa.rsplit(sep, _ONE)[_ONE]
+        return original_payoff(actual_p, actual_o)
+
+    return replace(
+        base,
+        actions=new_actions,
+        payoff_fn=_payoff,
+        applied_variants=base.applied_variants + (VARIANT_CHEAP_TALK,),
+        base_game_key=base_key or base.base_game_key,
+    )
+
+
+def apply_exit(
+    base: GameConfig,
+    base_key: str = "",
+    exit_payoff: int = OPD_EXIT_PAYOFF,
+) -> GameConfig:
+    """Add an exit option that gives both players a safe payoff.
+
+    Appends ``"exit"`` to the action list.  If either player exits both
+    receive *exit_payoff*; otherwise delegates to the base payoff function.
+    """
+    exit_f = float(exit_payoff)
+    exit_act = EXIT_ACTION
+    new_actions = list(base.actions) + [exit_act]
+    original_payoff = base.payoff_fn
+
+    def _payoff(pa: str, oa: str) -> tuple[float, float]:
+        if pa == exit_act or oa == exit_act:
+            return (exit_f, exit_f)
+        return original_payoff(pa, oa)
+
+    return replace(
+        base,
+        actions=new_actions,
+        payoff_fn=_payoff,
+        applied_variants=base.applied_variants + (VARIANT_EXIT,),
+        base_game_key=base_key or base.base_game_key,
+    )
+
+
+def apply_binding_commitment(
+    base: GameConfig,
+    base_key: str = "",
+    commit_cost: int = COMMIT_COST,
+) -> GameConfig:
+    """Add a costly binding commitment mechanism.
+
+    For base actions ``[A, B, ...]`` the first action *A* gets a
+    ``commit_A`` variant (player locked to *A*, pays *commit_cost*).
+    All actions get a ``free_X`` variant (no cost, free choice).
+    """
+    sep = CT_SEPARATOR
+    commit_pfx = BC_COMMIT_PREFIX
+    free_pfx = BC_FREE_PREFIX
+    cost_f = float(commit_cost)
+    base_actions = base.actions
+    commit_action = base_actions[_ZERO]
+
+    new_actions = [sep.join([commit_pfx, commit_action])]
+    for act in base_actions:
+        new_actions.append(sep.join([free_pfx, act]))
+
+    original_payoff = base.payoff_fn
+
+    def _parse(action: str) -> tuple[str, bool]:
+        """Return (actual_action, is_committed)."""
+        parts = action.split(sep, _ONE)
+        return parts[_ONE], parts[_ZERO] == commit_pfx
+
+    def _payoff(pa: str, oa: str) -> tuple[float, float]:
+        p_act, p_committed = _parse(pa)
+        o_act, o_committed = _parse(oa)
+        p_pay, o_pay = original_payoff(p_act, o_act)
+        if p_committed:
+            p_pay = p_pay - cost_f
+        if o_committed:
+            o_pay = o_pay - cost_f
+        return (p_pay, o_pay)
+
+    return replace(
+        base,
+        actions=new_actions,
+        payoff_fn=_payoff,
+        applied_variants=base.applied_variants + (VARIANT_BINDING_COMMITMENT,),
+        base_game_key=base_key or base.base_game_key,
+    )
+
+
+_DEFAULT_TREMBLE = DEFAULT_TREMBLE_PROB_NUMERATOR / DEFAULT_TREMBLE_PROB_DENOMINATOR
+_DEFAULT_NOISE = DEFAULT_NOISE_SCALE_NUMERATOR / DEFAULT_NOISE_SCALE_DENOMINATOR
+_NOISY_ONLY_TWO_PLAYER = "apply_noisy variant only supports two-player games"
+
+
+def apply_noisy_actions(
+    base: GameConfig,
+    base_key: str = "",
+    tremble_prob: float = _DEFAULT_TREMBLE,
+) -> GameConfig:
+    """With probability *tremble_prob* each player's action is replaced by a random one."""
+    if base.num_players != DEFAULT_TWO_PLAYERS:
+        raise ValueError(_NOISY_ONLY_TWO_PLAYER)
+    import random as _rng_mod
+    original_payoff = base.payoff_fn
+    actions = base.actions
+
+    def _payoff(pa: str, oa: str) -> tuple[float, float]:
+        actual_p = _rng_mod.choice(actions) if _rng_mod.random() < tremble_prob else pa
+        actual_o = _rng_mod.choice(actions) if _rng_mod.random() < tremble_prob else oa
+        return original_payoff(actual_p, actual_o)
+
+    return replace(
+        base,
+        payoff_fn=_payoff,
+        applied_variants=base.applied_variants + (VARIANT_NOISY_ACTIONS,),
+        base_game_key=base_key or base.base_game_key,
+    )
+
+
+def apply_noisy_payoffs(
+    base: GameConfig,
+    base_key: str = "",
+    noise_scale: float = _DEFAULT_NOISE,
+) -> GameConfig:
+    """Add Gaussian noise N(zero, noise_scale) to each payoff independently."""
+    if base.num_players != DEFAULT_TWO_PLAYERS:
+        raise ValueError(_NOISY_ONLY_TWO_PLAYER)
+    import random as _rng_mod
+    original_payoff = base.payoff_fn
+
+    def _payoff(pa: str, oa: str) -> tuple[float, float]:
+        p, o = original_payoff(pa, oa)
+        return (p + _rng_mod.gauss(float(_ZERO), noise_scale),
+                o + _rng_mod.gauss(float(_ZERO), noise_scale))
+
+    return replace(
+        base,
+        payoff_fn=_payoff,
+        applied_variants=base.applied_variants + (VARIANT_NOISY_PAYOFFS,),
+        base_game_key=base_key or base.base_game_key,
+    )
+
+
+_OPPONENT_ONLY_TWO_PLAYER = "opponent mode variants only support two-player games"
+
+
+def apply_self_play(
+    base: GameConfig,
+    base_key: str = "",
+) -> GameConfig:
+    """Mark a game for self-play: the model plays against itself."""
+    if base.num_players != DEFAULT_TWO_PLAYERS:
+        raise ValueError(_OPPONENT_ONLY_TWO_PLAYER)
+    return replace(
+        base,
+        opponent_mode=OPPONENT_MODE_SELF,
+        applied_variants=base.applied_variants + (VARIANT_SELF_PLAY,),
+        base_game_key=base_key or base.base_game_key,
+    )
+
+
+def apply_cross_model(
+    base: GameConfig,
+    base_key: str = "",
+) -> GameConfig:
+    """Mark a game for cross-model play: model vs a different model."""
+    if base.num_players != DEFAULT_TWO_PLAYERS:
+        raise ValueError(_OPPONENT_ONLY_TWO_PLAYER)
+    return replace(
+        base,
+        opponent_mode=OPPONENT_MODE_CROSS,
+        applied_variants=base.applied_variants + (VARIANT_CROSS_MODEL,),
+        base_game_key=base_key or base.base_game_key,
+    )
+
+
+_VARIANT_REGISTRY: dict[str, Callable[..., GameConfig]] = {
+    VARIANT_CHEAP_TALK: apply_cheap_talk,
+    VARIANT_EXIT: apply_exit,
+    VARIANT_BINDING_COMMITMENT: apply_binding_commitment,
+    VARIANT_NOISY_ACTIONS: apply_noisy_actions,
+    VARIANT_NOISY_PAYOFFS: apply_noisy_payoffs,
+    VARIANT_SELF_PLAY: apply_self_play,
+    VARIANT_CROSS_MODEL: apply_cross_model,
+}
+
+from common.meta.variants_meta import (  # noqa: E402
+    apply_rule_proposal, apply_rule_signal,
+    apply_constitutional, apply_proposer_responder,
+    _META_VARIANT_REGISTRY,
+)
+
+_VARIANT_REGISTRY.update(_META_VARIANT_REGISTRY)
+
+from common.meta.variants_reputation import (  # noqa: E402
+    apply_gossip,
+    _REPUTATION_VARIANT_REGISTRY,
+)
+
+_VARIANT_REGISTRY.update(_REPUTATION_VARIANT_REGISTRY)
+
+
+def compose_game(base_key: str, *variant_names: str) -> GameConfig:
+    """Build a game by applying named variants to a base game.
+
+    Example::
+
+        compose_game("stag_hunt", "cheap_talk", "exit")
+    """
+    game = GAMES[base_key]
+    for vname in variant_names:
+        apply_fn = _VARIANT_REGISTRY[vname]
+        game = apply_fn(game, base_key=base_key)
+    return game

constant_definitions/game_constants.py

@@ -8,6 +8,7 @@ MIN_STEP_COUNT = int()
 # Episode configuration
 DEFAULT_NUM_ROUNDS = 10
 SINGLE_SHOT_ROUNDS = 1
+DEFAULT_TWO_PLAYERS = 2
 
 # --- Prisoner's Dilemma payoffs ---
 PD_CC_PAYOFF = 3  # Both cooperate
@@ -71,6 +72,11 @@ SERVER_PORT = 8000
 # Max concurrent environments
 MAX_CONCURRENT_ENVS = 1
 
+# --- Opponent mode ---
+OPPONENT_MODE_STRATEGY = "strategy"
+OPPONENT_MODE_SELF = "self_play"
+OPPONENT_MODE_CROSS = "cross_model"
+
 # --- Evaluation module constants ---
 EVAL_ZERO = 0
 EVAL_ONE = 1
@@ -86,6 +92,10 @@ EVAL_ONE_FLOAT = 1.0
 EVAL_HALF = 0.5
 EVAL_NEGATIVE_ONE = -1
 
+# --- N-player / coalition evaluation constants ---
+NPLAYER_EVAL_DEFAULT_EPISODES = 3
+COALITION_EVAL_DEFAULT_EPISODES = 3
+
 # --- External benchmark constants ---
 EVAL_EIGHT = 8
 EVAL_TEN = 10

constant_definitions/train/agent_constants.py

@@ -33,3 +33,33 @@ SYSTEM_PROMPT = (
 
 # Sentinel returned when LLM output cannot be parsed
 PARSE_FAILURE_SENTINEL = "__PARSE_FAILURE__"
+
+# --- N-player prompt section headers ---
+NPLAYER_PROMPT_SECTION_PLAYERS = "PLAYERS"
+NPLAYER_PROMPT_SECTION_ALL_SCORES = "ALL SCORES"
+
+# --- Coalition prompt section headers ---
+COALITION_PROMPT_SECTION_PHASE = "PHASE"
+COALITION_PROMPT_SECTION_PROPOSALS = "PENDING PROPOSALS"
+COALITION_PROMPT_SECTION_COALITIONS = "ACTIVE COALITIONS"
+
+# --- Governance prompt section headers ---
+GOVERNANCE_PROMPT_SECTION_RULES = "GOVERNANCE RULES"
+GOVERNANCE_PROMPT_SECTION_PENDING = "PENDING GOVERNANCE"
+
+# N-player system prompt
+NPLAYER_SYSTEM_PROMPT = (
+    "You are playing an N-player game-theory game. Analyse the situation "
+    "and choose the best action. Respond with ONLY the action name, "
+    "nothing else."
+)
+
+# Coalition system prompt
+COALITION_SYSTEM_PROMPT = (
+    "You are playing a coalition formation game. You can form coalitions "
+    "with other players and propose governance changes. Respond with "
+    "valid JSON when negotiating, or ONLY the action name when acting."
+)
+
+# Maximum tokens for coalition JSON response
+COALITION_MAX_ACTION_TOKENS = 256

constant_definitions/var/meta/meta_rule_constants.py

@@ -0,0 +1,41 @@
+# Meta-gaming rule proposal variant constants
+
+# Variant names
+VARIANT_RULE_PROPOSAL = "rule_proposal"
+VARIANT_RULE_SIGNAL = "rule_signal"
+VARIANT_CONSTITUTIONAL = "constitutional"
+VARIANT_PROPOSER_RESPONDER = "proposer_responder"
+
+# Action prefixes (single tokens)
+META_PROP_PREFIX = "prop"
+META_SIG_PREFIX = "sig"
+META_CONST_PREFIX = "const"
+META_RPROP_PREFIX = "rprop"
+META_RACCEPT_PREFIX = "raccept"
+META_RREJECT_PREFIX = "rreject"
+META_SEPARATOR = "_"
+META_SPLIT_LIMIT = 2
+
+# Rule names (single tokens, no underscores)
+RULE_NONE = "none"
+RULE_EQUAL_SPLIT = "equalsplit"
+RULE_COOP_BONUS = "coopbonus"
+RULE_DEFECT_PENALTY = "defectpenalty"
+RULE_MIN_GUARANTEE = "minguarantee"
+RULE_BAN_DEFECT = "bandefect"
+
+DEFAULT_RULE_CATALOG = (
+    RULE_NONE, RULE_EQUAL_SPLIT, RULE_COOP_BONUS,
+    RULE_DEFECT_PENALTY, RULE_MIN_GUARANTEE, RULE_BAN_DEFECT,
+)
+
+# Payoff parameters (numerator/denominator to avoid inline literals)
+COOP_BONUS_NUMERATOR = 2
+COOP_BONUS_DENOMINATOR = 1
+DEFECT_PENALTY_NUMERATOR = 3
+DEFECT_PENALTY_DENOMINATOR = 1
+MIN_GUARANTEE_NUMERATOR = 1
+MIN_GUARANTEE_DENOMINATOR = 1
+BAN_DEFECT_PENALTY_NUMERATOR = 10
+BAN_DEFECT_PENALTY_DENOMINATOR = 1
+EQUAL_SPLIT_DENOMINATOR = 2

constant_definitions/var/meta/reputation_constants.py

@@ -0,0 +1,32 @@
+# Reputation and gossip variant constants
+
+# Variant names
+VARIANT_GOSSIP = "gossip"
+VARIANT_MEMORY = "memory"
+
+# Gossip ratings (single tokens, no underscores)
+RATING_TRUSTWORTHY = "trustworthy"
+RATING_UNTRUSTWORTHY = "untrustworthy"
+RATING_NEUTRAL = "neutral"
+DEFAULT_RATINGS = (RATING_TRUSTWORTHY, RATING_UNTRUSTWORTHY, RATING_NEUTRAL)
+
+# Action prefixes
+GOSSIP_PREFIX = "gossip"
+GOSSIP_SEPARATOR = "_"
+GOSSIP_SPLIT_LIMIT = 2
+
+# Reputation defaults (numerator / denominator)
+DEFAULT_REPUTATION_SCORE_NUMERATOR = 5
+DEFAULT_REPUTATION_SCORE_DENOMINATOR = 10
+REPUTATION_DECAY_NUMERATOR = 9
+REPUTATION_DECAY_DENOMINATOR = 10
+
+# Metadata keys
+META_KEY_REPUTATION = "opponent_reputation"
+META_KEY_GOSSIP_HISTORY = "gossip_history"
+META_KEY_INTERACTION_COUNT = "interaction_count"
+META_KEY_COOPERATION_RATE = "cooperation_rate"
+
+# Cognee dataset name
+COGNEE_DATASET_NAME = "kant_interactions"
+COGNEE_SEARCH_TYPE = "GRAPH_COMPLETION"

constant_definitions/var/pd_variant_constants.py

@@ -24,3 +24,25 @@ PW_DISARM_DISARM = 4
 PW_DISARM_ARM = -1
 PW_ARM_DISARM = 6
 PW_ARM_ARM = 0
+
+# Composable variant system -- name strings and prefixes
+VARIANT_CHEAP_TALK = "cheap_talk"
+VARIANT_EXIT = "exit"
+VARIANT_BINDING_COMMITMENT = "binding_commitment"
+VARIANT_NOISY_ACTIONS = "noisy_actions"
+VARIANT_NOISY_PAYOFFS = "noisy_payoffs"
+CT_MSG_PREFIX = "msg"
+CT_SEPARATOR = "_"
+BC_COMMIT_PREFIX = "commit"
+BC_FREE_PREFIX = "free"
+EXIT_ACTION = "exit"
+
+# Opponent mode variants
+VARIANT_SELF_PLAY = "self_play"
+VARIANT_CROSS_MODEL = "cross_model"
+
+# Bayesian variant parameters
+DEFAULT_TREMBLE_PROB_NUMERATOR = 1
+DEFAULT_TREMBLE_PROB_DENOMINATOR = 10
+DEFAULT_NOISE_SCALE_NUMERATOR = 1
+DEFAULT_NOISE_SCALE_DENOMINATOR = 2

env/environment.py

@@ -155,6 +155,8 @@ class KantEnvironment(Environment[GameObservation, GameAction, GameState]):
 
     def _opponent_actions(self) -> list[str]:
         assert self._game is not None
+        if self._game.opponent_actions is not None:
+            return list(self._game.opponent_actions)
         gt = self._game.game_type
         if gt == "ultimatum":
             return ["accept", "reject"]

env/reputation/__init__.py

@@ -0,0 +1,4 @@
+"""Reputation environment subpackage."""
+from env.reputation.reputation_env import ReputationEnvironment
+
+__all__ = ["ReputationEnvironment"]

env/reputation/reputation_env.py

@@ -0,0 +1,111 @@
+"""Environment wrapper adding cross-episode reputation via cognee.
+
+Injects opponent reputation into obs.metadata before each episode.
+Records episode outcomes and gossip ratings in CogneeMemoryStore after.
+"""
+from __future__ import annotations
+
+from typing import Any, Optional
+
+from env.environment import KantEnvironment
+from env.models import GameAction, GameObservation, GameState
+from common.meta.memory_store import CogneeMemoryStore
+from common.meta.variants_reputation import parse_gossip_action
+
+from constant_definitions.var.meta.reputation_constants import (
+    GOSSIP_PREFIX,
+    GOSSIP_SEPARATOR,
+    META_KEY_REPUTATION,
+    META_KEY_INTERACTION_COUNT,
+)
+
+_ZERO = int()
+_ONE = int(bool(True))
+_COOPERATIVE_ACTIONS = frozenset({"cooperate", "stag", "dove"})
+
+
+def _compute_coop_rate(history: list[Any]) -> float:
+    """Compute cooperation rate from round history."""
+    if not history:
+        return float(_ZERO)
+    coop_count = _ZERO
+    for rnd in history:
+        base_action = rnd.player_action
+        if GOSSIP_SEPARATOR in base_action:
+            base_action = base_action.rsplit(GOSSIP_SEPARATOR, _ONE)[_ONE]
+        if base_action in _COOPERATIVE_ACTIONS:
+            coop_count = coop_count + _ONE
+    return coop_count / len(history)
+
+
+class ReputationEnvironment:
+    """Environment wrapper that adds cross-episode reputation via cognee.
+
+    Injects opponent reputation into obs.metadata before each episode.
+    Records episode outcomes and gossip ratings in CogneeMemoryStore.
+    """
+
+    def __init__(
+        self,
+        memory_store: CogneeMemoryStore,
+        env: Optional[KantEnvironment] = None,
+    ) -> None:
+        self._env = env if env is not None else KantEnvironment()
+        self._store = memory_store
+        self._agent_id: str = ""
+        self._opponent_id: str = ""
+
+    def reset(
+        self,
+        *,
+        agent_id: str = "agent",
+        **kwargs: Any,
+    ) -> GameObservation:
+        """Reset environment and inject reputation into metadata."""
+        self._agent_id = agent_id
+        self._opponent_id = kwargs.get("strategy", "unknown")
+        obs = self._env.reset(**kwargs)
+        reputation = self._store.query_reputation(self._opponent_id)
+        updated_meta = dict(obs.metadata)
+        updated_meta[META_KEY_REPUTATION] = reputation
+        updated_meta[META_KEY_INTERACTION_COUNT] = reputation.get(
+            META_KEY_INTERACTION_COUNT, _ZERO,
+        )
+        obs = obs.model_copy(update={"metadata": updated_meta})
+        return obs
+
+    def step(
+        self,
+        action: GameAction,
+        **kwargs: Any,
+    ) -> GameObservation:
+        """Step environment, extracting gossip and recording episodes."""
+        gossip_marker = GOSSIP_PREFIX + GOSSIP_SEPARATOR
+        if action.action.startswith(gossip_marker):
+            _, rating, _ = parse_gossip_action(action.action)
+            self._store.record_gossip(
+                self._agent_id, self._opponent_id, rating,
+            )
+
+        obs = self._env.step(action, **kwargs)
+
+        if obs.done:
+            self._store.record_episode(
+                agent_id=self._agent_id,
+                opponent_id=self._opponent_id,
+                game=obs.game_name,
+                history=obs.history,
+                cooperation_rate=_compute_coop_rate(obs.history),
+                scores=(obs.player_score, obs.opponent_score),
+            )
+
+        reputation = self._store.get_stats(self._opponent_id)
+        updated_meta = dict(obs.metadata)
+        updated_meta[META_KEY_REPUTATION] = reputation
+        obs = obs.model_copy(update={"metadata": updated_meta})
+        return obs
+
+    @property
+    def state(self) -> GameState:
+        """Delegate to wrapped environment state."""
+        return self._env.state

server/KantBench_environment.py

@@ -1,8 +1,8 @@
 """KantBench environment adapter for the HF Space.
 
 Thin wrapper that delegates to the real KantEnvironment (90+ 2-player games,
-17 strategies) and NPlayerEnvironment (3 N-player games) instead of a
-standalone reimplementation.
+17 strategies, meta-games, composable variants) and NPlayerEnvironment
+(3 N-player games) instead of a standalone reimplementation.
 """
 
 from __future__ import annotations
@@ -22,12 +22,18 @@ from env.nplayer.models import NPlayerAction, NPlayerObservation
 import common.games_meta.nplayer_games  # noqa: F401
 from common.games_meta.nplayer_config import NPLAYER_GAMES
 
+from common.games import GAMES
+from common.variants import compose_game
+
 
 class KantbenchEnvironment(Environment):
     """Game theory environment exposing 90+ two-player and N-player games.
 
     Wraps the real KantEnvironment and NPlayerEnvironment, routing
     automatically based on the requested game name.
+
+    Supports a ``variant`` reset parameter for dynamic game composition
+    (e.g. ``variant="constitutional"`` or ``variant="cheap_talk"``).
     """
 
     SUPPORTS_CONCURRENT_SESSIONS: bool = True
@@ -39,6 +45,16 @@ class KantbenchEnvironment(Environment):
 
     def reset(self, **kwargs: Any) -> KantBenchObservation:
         game_name: str = kwargs.get("game", "prisoners_dilemma")
+        variant: Optional[str] = kwargs.pop("variant", None)
+
+        # Dynamic variant composition — compose game on-the-fly and
+        # register it so KantEnvironment can look it up via get_game().
+        # Constitutional variant creates fresh mutable closure per call.
+        if variant and game_name in GAMES:
+            composed = compose_game(game_name, variant)
+            composed_key = f"_composed_{variant}_{game_name}"
+            GAMES[composed_key] = composed
+            kwargs["game"] = composed_key
 
         if game_name in NPLAYER_GAMES:
             self._is_nplayer = True