Add protocol sigmoid/inflation pricing

Adds dynamic pricing to assembler protocols:
- sigmoid: Number of discounted uses (linear 0→1 scaling)
- inflation: Compound rate for exponential cost after sigmoid uses

Author: daveey

packages/mettagrid/cpp/include/mettagrid/objects/protocol.hpp

@@ -4,6 +4,7 @@
 #include <pybind11/pybind11.h>
 #include <pybind11/stl.h>
 
+#include <cmath>
 #include <memory>
 #include <unordered_map>
 
@@ -16,13 +17,41 @@ class Protocol {
   std::unordered_map<InventoryItem, InventoryQuantity> input_resources;
   std::unordered_map<InventoryItem, InventoryQuantity> output_resources;
   unsigned short cooldown;
+  unsigned int sigmoid;  // Centroid for inflation curve
+  float inflation;       // Compound rate for cost adjustment
+
+  // Track activation count for inflation calculation
+  mutable unsigned int activation_count;
 
   Protocol(unsigned short min_agents = 0,
            const std::vector<ObservationType>& vibes = {},
            const std::unordered_map<InventoryItem, InventoryQuantity>& inputs = {},
            const std::unordered_map<InventoryItem, InventoryQuantity>& outputs = {},
-           unsigned short cooldown = 0)
-      : min_agents(min_agents), vibes(vibes), input_resources(inputs), output_resources(outputs), cooldown(cooldown) {}
+           unsigned short cooldown = 0,
+           unsigned int sigmoid = 0,
+           float inflation = 0.0f)
+      : min_agents(min_agents),
+        vibes(vibes),
+        input_resources(inputs),
+        output_resources(outputs),
+        cooldown(cooldown),
+        sigmoid(sigmoid),
+        inflation(inflation),
+        activation_count(0) {}
+
+  // Calculate cost multiplier based on activation count
+  // - Linear phase (0 to sigmoid uses): scales from 0 (free) to 1 (full price)
+  // - Exponential phase (after sigmoid uses): multiplier = (1 + inflation) ^ (activation_count - sigmoid)
+  float get_cost_multiplier() const {
+    // Linear phase: free to full price for first 'sigmoid' uses
+    if (sigmoid > 0 && activation_count < sigmoid) {
+      return static_cast<float>(activation_count) / static_cast<float>(sigmoid);
+    }
+    // Exponential phase: inflating costs after sigmoid uses
+    if (inflation == 0.0f) return 1.0f;
+    int exponent = static_cast<int>(activation_count) - static_cast<int>(sigmoid);
+    return std::pow(1.0f + inflation, static_cast<float>(exponent));
+  }
 };
 
 inline void bind_protocol(py::module& m) {
@@ -32,7 +61,10 @@ inline void bind_protocol(py::module& m) {
       .def_readwrite("vibes", &Protocol::vibes)
       .def_readwrite("input_resources", &Protocol::input_resources)
       .def_readwrite("output_resources", &Protocol::output_resources)
-      .def_readwrite("cooldown", &Protocol::cooldown);
+      .def_readwrite("cooldown", &Protocol::cooldown)
+      .def_readwrite("sigmoid", &Protocol::sigmoid)
+      .def_readwrite("inflation", &Protocol::inflation)
+      .def_readwrite("activation_count", &Protocol::activation_count);
 }
 
 #endif  // PACKAGES_METTAGRID_CPP_INCLUDE_METTAGRID_OBJECTS_PROTOCOL_HPP_

packages/mettagrid/python/src/mettagrid/config/mettagrid_c_config.py

@@ -286,6 +286,8 @@ def convert_to_cpp_game_config(mettagrid_config: dict | GameConfig):
                 cpp_protocol.input_resources = input_res
                 cpp_protocol.output_resources = output_res
                 cpp_protocol.cooldown = protocol_config.cooldown
+                cpp_protocol.sigmoid = protocol_config.sigmoid
+                cpp_protocol.inflation = protocol_config.inflation
                 protocols.append(cpp_protocol)
 
             # Convert tag names to IDs
@@ -522,6 +524,8 @@ def process_action_config(action_name: str, action_config):
                 output_res[key] = val
             cpp_protocol.output_resources = output_res
             cpp_protocol.cooldown = protocol_config.cooldown
+            cpp_protocol.sigmoid = protocol_config.sigmoid
+            cpp_protocol.inflation = protocol_config.inflation
             clipper_protocols.append(cpp_protocol)
         clipper_config = CppClipperConfig()
         clipper_config.unclipping_protocols = clipper_protocols

packages/mettagrid/python/src/mettagrid/config/mettagrid_config.py

@@ -384,6 +384,16 @@ class ProtocolConfig(Config):
     input_resources: dict[str, int] = Field(default_factory=dict)
     output_resources: dict[str, int] = Field(default_factory=dict)
     cooldown: int = Field(ge=0, default=0)
+    sigmoid: int = Field(
+        default=0,
+        ge=0,
+        description="Number of discounted uses. Cost scales linearly from 0 (free) to 1 (full price) over these uses.",
+    )
+    inflation: float = Field(
+        default=0.0,
+        ge=0.0,
+        description="Compound rate for exponential cost. Cost = base * (1+inflation)^(n - sigmoid).",
+    )
 
 
 class AssemblerConfig(GridObjectConfig):

packages/mettagrid/python/src/mettagrid/mettagrid_c.pyi

@@ -131,6 +131,9 @@ class Protocol:
     input_resources: dict[int, int]
     output_resources: dict[int, int]
     cooldown: int
+    sigmoid: int
+    inflation: float
+    activation_count: int
 
 class InventoryConfig:
     def __init__(

packages/mettagrid/tests/test_swap_frozen.py

@@ -7,6 +7,8 @@
     AgentConfig,
     AgentRewards,
     AttackActionConfig,
+    AttackOutcome,
+    ChangeVibeActionConfig,
     GameConfig,
     InventoryConfig,
     MettaGridConfig,
@@ -17,11 +19,20 @@
     WallConfig,
 )
 from mettagrid.simulator import Simulation
-from mettagrid.test_support.actions import attack, get_agent_position, move
+from mettagrid.test_support.actions import get_agent_position, move
 from mettagrid.test_support.map_builders import ObjectNameMapBuilder
 from mettagrid.test_support.orientation import Orientation
 
 
+def get_agent_frozen_status(sim: Simulation, agent_id: int) -> bool:
+    """Check if an agent is frozen."""
+    grid_objects = sim.grid_objects()
+    for obj in grid_objects.values():
+        if obj.get("agent_id") == agent_id:
+            return obj.get("is_frozen", False)
+    return False
+
+
 @pytest.fixture
 def base_config():
     """Base configuration for swap tests."""
@@ -40,7 +51,12 @@ def base_config():
                     "west",
                 ]
             ),
-            attack=AttackActionConfig(enabled=True, consumed_resources={"laser": 1}, defense_resources={"armor": 1}),
+            change_vibe=ChangeVibeActionConfig(number_of_vibes=10),
+            attack=AttackActionConfig(
+                enabled=False,  # Attack triggers via move, not standalone action
+                vibes=["charger"],  # Attack triggers when agent has charger vibe
+                success=AttackOutcome(freeze=10),
+            ),
         ),
         objects={
             "wall": WallConfig(),
@@ -135,26 +151,35 @@ def test_swap_with_frozen_agent(make_sim, adjacent_agents_map):
     assert pos_agent0_before == (1, 1), f"Agent 0 should be at (1, 1), got {pos_agent0_before}"
     assert pos_agent1_before == (1, 2), f"Agent 1 should be at (1, 2), got {pos_agent1_before}"
 
-    # Agent 0 attacks Agent 1 to freeze them
-    attack_result = attack(sim, target_arg=0, agent_idx=0)
-    print(f"Attack result: {attack_result}")
-    assert attack_result["success"], f"Attack should succeed: {attack_result}"
+    # Verify neither agent is frozen initially
+    assert not get_agent_frozen_status(sim, 0), "Agent 0 should not start frozen"
+    assert not get_agent_frozen_status(sim, 1), "Agent 1 should not start frozen"
 
-    # Verify agent 1 is frozen
-    grid_objects = sim.grid_objects()
-    agent1_obj = None
-    for _obj_id, obj in grid_objects.items():
-        if obj["type_name"] == "agent":
-            pos = (obj["r"], obj["c"])
-            if pos == pos_agent1_before:
-                agent1_obj = obj
-                break
-
-    assert agent1_obj is not None, "Should find agent 1"
-    assert agent1_obj.get("is_frozen", False), f"Agent 1 should be frozen: {agent1_obj}"
-    print(f"Agent 1 frozen status: {agent1_obj.get('is_frozen')}")
-
-    # Now agent 0 tries to move east onto frozen agent 1
+    # Agent 0 changes vibe to charger (to enable attack on move)
+    sim.agent(0).set_action("change_vibe_charger")
+    sim.agent(1).set_action("noop")
+    sim.step()
+
+    # Agent 0 moves east into Agent 1 - should trigger attack due to weapon vibe
+    sim.agent(0).set_action("move_east")
+    sim.agent(1).set_action("noop")
+    sim.step()
+
+    # Verify agent 1 is frozen from the attack
+    assert get_agent_frozen_status(sim, 1), "Agent 1 should be frozen after attack"
+    print("Agent 1 frozen status: True")
+
+    # Verify positions didn't change on first move (attack happened but no swap yet)
+    pos_agent0_mid = get_agent_position(sim, 0)
+    pos_agent1_mid = get_agent_position(sim, 1)
+    print(f"After attack - Agent 0: {pos_agent0_mid}, Agent 1: {pos_agent1_mid}")
+
+    # Agent 0 changes vibe back to default so next move doesn't trigger attack
+    sim.agent(0).set_action("change_vibe_default")
+    sim.agent(1).set_action("noop")
+    sim.step()
+
+    # Now agent 0 tries to move east onto frozen agent 1 - should swap
     move_result = move(sim, Orientation.EAST, agent_idx=0)
     print(f"Move result: {move_result}")