From 6c3628447df0b9508e121b08089dfe0122589c5e Mon Sep 17 00:00:00 2001
From: Matheus Santana <130666046+mathsant-js@users.noreply.github.com>
Date: Tue, 1 Sep 2026 17:39:55 -0300
Subject: [PATCH 1/6] feat(simulation): add deterministic simulation clock

---
 backend/app/simulation/__init__.py     |  6 +-
 backend/app/simulation/clock.py        | 76 ++++++++++++++++++++
 backend/tests/test_simulation_clock.py | 97 ++++++++++++++++++++++++++
 3 files changed, 178 insertions(+), 1 deletion(-)
 create mode 100644 backend/app/simulation/clock.py
 create mode 100644 backend/tests/test_simulation_clock.py

diff --git a/backend/app/simulation/__init__.py b/backend/app/simulation/__init__.py
index 9d2eb79..ab7afc9 100644
--- a/backend/app/simulation/__init__.py
+++ b/backend/app/simulation/__init__.py
@@ -1 +1,5 @@
-"""Replaceable simulation data providers."""
+"""Replaceable, infrastructure-independent simulation primitives."""
+
+from app.simulation.clock import SimulationClock, SimulationClockState
+
+__all__ = ["SimulationClock", "SimulationClockState"]
diff --git a/backend/app/simulation/clock.py b/backend/app/simulation/clock.py
new file mode 100644
index 0000000..325dcfc
--- /dev/null
+++ b/backend/app/simulation/clock.py
@@ -0,0 +1,76 @@
+"""Deterministic clock primitives for the equipment simulator."""
+
+from dataclasses import dataclass, field
+from datetime import UTC, datetime, timedelta
+from enum import StrEnum
+
+DEFAULT_SIMULATION_SPEED = 60
+DEFAULT_REAL_TICK_INTERVAL = timedelta(seconds=1)
+
+
+class SimulationClockState(StrEnum):
+    """Lifecycle state of a simulation clock."""
+
+    STOPPED = "STOPPED"
+    RUNNING = "RUNNING"
+
+
+@dataclass(slots=True)
+class SimulationClock:
+    """A deterministic clock advanced explicitly by simulation ticks.
+
+    ``simulation_speed`` is the number of simulated seconds represented by one
+    real second and is intended to receive ``SystemConfiguration.simulation_speed``.
+    The defaults therefore implement the initial 1 real second = 1 simulated
+    minute configuration without coupling the clock to persistence or a run loop.
+    """
+
+    initial_instant: datetime
+    simulation_speed: int = DEFAULT_SIMULATION_SPEED
+    real_tick_interval: timedelta = DEFAULT_REAL_TICK_INTERVAL
+    _current_instant: datetime = field(init=False, repr=False)
+    state: SimulationClockState = field(init=False, default=SimulationClockState.STOPPED)
+
+    def __post_init__(self) -> None:
+        if self.initial_instant.tzinfo is None or self.initial_instant.utcoffset() is None:
+            raise ValueError("initial_instant must be timezone-aware")
+        if self.simulation_speed <= 0:
+            raise ValueError("simulation_speed must be positive")
+        if self.real_tick_interval <= timedelta(0):
+            raise ValueError("real_tick_interval must be positive")
+
+        self.initial_instant = self.initial_instant.astimezone(UTC)
+        self._current_instant = self.initial_instant
+
+    @property
+    def current_instant(self) -> datetime:
+        """Return the current simulated instant as a read-only UTC value."""
+
+        return self._current_instant
+
+    @property
+    def tick_duration(self) -> timedelta:
+        """Return the simulated time represented by one execution tick."""
+
+        return self.real_tick_interval * self.simulation_speed
+
+    def start(self) -> None:
+        """Put the clock in the running state."""
+
+        self.state = SimulationClockState.RUNNING
+
+    def stop(self) -> None:
+        """Put the clock in the stopped state."""
+
+        self.state = SimulationClockState.STOPPED
+
+    def advance(self, ticks: int = 1) -> datetime:
+        """Advance a running clock by an explicit number of ticks."""
+
+        if self.state is not SimulationClockState.RUNNING:
+            raise RuntimeError("simulation clock must be running to advance")
+        if isinstance(ticks, bool) or not isinstance(ticks, int) or ticks <= 0:
+            raise ValueError("ticks must be a positive integer")
+
+        self._current_instant += self.tick_duration * ticks
+        return self._current_instant
diff --git a/backend/tests/test_simulation_clock.py b/backend/tests/test_simulation_clock.py
new file mode 100644
index 0000000..43fa36b
--- /dev/null
+++ b/backend/tests/test_simulation_clock.py
@@ -0,0 +1,97 @@
+from datetime import UTC, datetime, timedelta, timezone
+
+import pytest
+
+from app.simulation.clock import SimulationClock, SimulationClockState
+
+FIXED_INSTANT = datetime(2026, 9, 1, 12, 0, tzinfo=UTC)
+
+
+def test_initial_state_uses_default_one_minute_tick() -> None:
+    clock = SimulationClock(initial_instant=FIXED_INSTANT)
+
+    assert clock.current_instant == FIXED_INSTANT
+    assert clock.tick_duration == timedelta(minutes=1)
+    assert clock.state is SimulationClockState.STOPPED
+
+
+def test_start_and_stop_control_clock_state() -> None:
+    clock = SimulationClock(initial_instant=FIXED_INSTANT)
+
+    clock.start()
+    assert clock.state is SimulationClockState.RUNNING
+
+    clock.stop()
+    assert clock.state is SimulationClockState.STOPPED
+
+
+def test_advance_one_or_multiple_ticks_without_waiting() -> None:
+    clock = SimulationClock(initial_instant=FIXED_INSTANT)
+    clock.start()
+
+    assert clock.advance() == FIXED_INSTANT + timedelta(minutes=1)
+    assert clock.advance(ticks=3) == FIXED_INSTANT + timedelta(minutes=4)
+
+
+def test_simulation_speed_accelerates_tick_duration() -> None:
+    clock = SimulationClock(
+        initial_instant=FIXED_INSTANT,
+        simulation_speed=120,
+        real_tick_interval=timedelta(seconds=2),
+    )
+    clock.start()
+
+    assert clock.tick_duration == timedelta(minutes=4)
+    assert clock.advance() == FIXED_INSTANT + timedelta(minutes=4)
+
+
+def test_initial_instant_is_normalized_to_utc() -> None:
+    source = datetime(2026, 9, 1, 9, 0, tzinfo=timezone(timedelta(hours=-3)))
+    clock = SimulationClock(initial_instant=source)
+
+    assert clock.current_instant == FIXED_INSTANT
+    assert clock.current_instant.tzinfo is UTC
+    clock.start()
+    assert clock.advance().tzinfo is UTC
+
+
+def test_fixed_initial_instant_produces_deterministic_results() -> None:
+    first = SimulationClock(initial_instant=FIXED_INSTANT, simulation_speed=30)
+    second = SimulationClock(initial_instant=FIXED_INSTANT, simulation_speed=30)
+    first.start()
+    second.start()
+
+    assert first.advance(ticks=10) == second.advance(ticks=10)
+
+
+@pytest.mark.parametrize("simulation_speed", [0, -1])
+def test_rejects_non_positive_simulation_speed(simulation_speed: int) -> None:
+    with pytest.raises(ValueError, match="simulation_speed must be positive"):
+        SimulationClock(initial_instant=FIXED_INSTANT, simulation_speed=simulation_speed)
+
+
+@pytest.mark.parametrize("interval", [timedelta(0), timedelta(microseconds=-1)])
+def test_rejects_non_positive_real_tick_interval(interval: timedelta) -> None:
+    with pytest.raises(ValueError, match="real_tick_interval must be positive"):
+        SimulationClock(initial_instant=FIXED_INSTANT, real_tick_interval=interval)
+
+
+def test_rejects_naive_initial_instant() -> None:
+    with pytest.raises(ValueError, match="initial_instant must be timezone-aware"):
+        SimulationClock(initial_instant=datetime(2026, 9, 1, 12, 0))
+
+
+def test_stopped_clock_cannot_advance() -> None:
+    clock = SimulationClock(initial_instant=FIXED_INSTANT)
+
+    with pytest.raises(RuntimeError, match="must be running"):
+        clock.advance()
+
+
+@pytest.mark.parametrize("ticks", [0, -1, True])
+def test_rejects_invalid_tick_count(ticks: int) -> None:
+    clock = SimulationClock(initial_instant=FIXED_INSTANT)
+    clock.start()
+
+    with pytest.raises(ValueError, match="ticks must be a positive integer"):
+        clock.advance(ticks=ticks)

From f23479ae4207c54cf4cef31d8e3502f74aa30c28 Mon Sep 17 00:00:00 2001
From: Matheus Santana <130666046+mathsant-js@users.noreply.github.com>
Date: Wed, 16 Sep 2026 21:09:06 -0300
Subject: [PATCH 2/6] feat(simulation): add deterministic solar generation per
 station

- Add a reproducible UTC solar curve and a configurable station solar peak.
- Include the Alembic migration and tests for simulation, persistence, and API.
---
 backend/app/models/infrastructure.py          |  8 ++-
 backend/app/schemas/station.py                |  8 ++-
 backend/app/simulation/__init__.py            |  8 ++-
 backend/app/simulation/energy_data.py         | 39 ++++++++++++
 .../20260916_0009_station_peak_solar.py       | 36 +++++++++++
 backend/tests/test_infrastructure.py          | 18 +++++-
 backend/tests/test_simulation_energy_data.py  | 59 +++++++++++++++++++
 7 files changed, 170 insertions(+), 6 deletions(-)
 create mode 100644 backend/app/simulation/energy_data.py
 create mode 100644 backend/migrations/versions/20260916_0009_station_peak_solar.py
 create mode 100644 backend/tests/test_simulation_energy_data.py

diff --git a/backend/app/models/infrastructure.py b/backend/app/models/infrastructure.py
index e2788d8..cfb71fc 100644
--- a/backend/app/models/infrastructure.py
+++ b/backend/app/models/infrastructure.py
@@ -16,12 +16,18 @@ class ChargerStatus(StrEnum):
 
 class ChargingStation(TimestampMixin, Base):
     __tablename__ = "charging_stations"
-    __table_args__ = (CheckConstraint("grid_limit_kw > 0", name="ck_stations_grid_limit_positive"),)
+    __table_args__ = (
+        CheckConstraint("grid_limit_kw > 0", name="ck_stations_grid_limit_positive"),
+        CheckConstraint(
+            "station_peak_solar_kw >= 0", name="ck_stations_peak_solar_nonnegative"
+        ),
+    )
 
     id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
     name: Mapped[str] = mapped_column(String(120), nullable=False)
     description: Mapped[str | None] = mapped_column(Text, nullable=True)
     grid_limit_kw: Mapped[float] = mapped_column(Float, nullable=False)
+    station_peak_solar_kw: Mapped[float] = mapped_column(Float, default=0.0, nullable=False)
     is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
 
     chargers: Mapped[list["Charger"]] = relationship(back_populates="station")
diff --git a/backend/app/schemas/station.py b/backend/app/schemas/station.py
index 07f5771..d069a77 100644
--- a/backend/app/schemas/station.py
+++ b/backend/app/schemas/station.py
@@ -1,16 +1,18 @@
 from typing import Annotated
 
-from pydantic import BaseModel, StringConstraints, field_validator
+from pydantic import BaseModel, Field, StringConstraints, field_validator
 
 from app.schemas.common import Name, ORMResponse, PositivePower
 
 Description = Annotated[str, StringConstraints(strip_whitespace=True, max_length=2000)]
+NonNegativePower = Annotated[float, Field(ge=0, allow_inf_nan=False)]
 
 
 class StationCreate(BaseModel):
     name: Name
     description: Description | None = None
     grid_limit_kw: PositivePower
+    station_peak_solar_kw: NonNegativePower = 0.0
     is_active: bool = True
 
 
@@ -18,9 +20,10 @@ class StationUpdate(BaseModel):
     name: Name | None = None
     description: Description | None = None
     grid_limit_kw: PositivePower | None = None
+    station_peak_solar_kw: NonNegativePower | None = None
     is_active: bool | None = None
 
-    @field_validator("name", "grid_limit_kw", "is_active")
+    @field_validator("name", "grid_limit_kw", "station_peak_solar_kw", "is_active")
     @classmethod
     def reject_null(cls, value: object) -> object:
         if value is None:
@@ -32,4 +35,5 @@ class StationResponse(ORMResponse):
     name: str
     description: str | None
     grid_limit_kw: float
+    station_peak_solar_kw: float
     is_active: bool
diff --git a/backend/app/simulation/__init__.py b/backend/app/simulation/__init__.py
index ab7afc9..b6469c9 100644
--- a/backend/app/simulation/__init__.py
+++ b/backend/app/simulation/__init__.py
@@ -1,5 +1,11 @@
 """Replaceable, infrastructure-independent simulation primitives."""
 
 from app.simulation.clock import SimulationClock, SimulationClockState
+from app.simulation.energy_data import EnergyDataProvider, SimulationEnergyDataProvider
 
-__all__ = ["SimulationClock", "SimulationClockState"]
+__all__ = [
+    "EnergyDataProvider",
+    "SimulationClock",
+    "SimulationClockState",
+    "SimulationEnergyDataProvider",
+]
diff --git a/backend/app/simulation/energy_data.py b/backend/app/simulation/energy_data.py
new file mode 100644
index 0000000..8528c6d
--- /dev/null
+++ b/backend/app/simulation/energy_data.py
@@ -0,0 +1,39 @@
+"""Deterministic energy data supplied to the simulator."""
+
+import math
+from datetime import UTC, datetime
+from typing import Protocol
+
+
+class EnergyDataProvider(Protocol):
+    """Source of available solar power for a station at a simulated instant."""
+
+    def solar_available_kw(self, timestamp: datetime, station_peak_solar_kw: float) -> float:
+        """Return available solar power in kW."""
+
+
+class SimulationEnergyDataProvider:
+    """Daily solar curve evaluated against UTC clock time.
+
+    Generation runs from 06:00 to 18:00 UTC. A half sine wave reaches the
+    configured station peak at 12:00 UTC. No state or randomness is involved.
+    """
+
+    def solar_available_kw(self, timestamp: datetime, station_peak_solar_kw: float) -> float:
+        if timestamp.tzinfo is None or timestamp.utcoffset() is None:
+            raise ValueError("timestamp must be timezone-aware")
+        if not math.isfinite(station_peak_solar_kw) or station_peak_solar_kw < 0:
+            raise ValueError("station_peak_solar_kw must be finite and nonnegative")
+
+        utc_time = timestamp.astimezone(UTC)
+        hour = (
+            utc_time.hour
+            + utc_time.minute / 60
+            + utc_time.second / 3600
+            + utc_time.microsecond / 3_600_000_000
+        )
+        if hour <= 6 or hour >= 18 or station_peak_solar_kw == 0:
+            return 0.0
+
+        power = station_peak_solar_kw * math.sin(math.pi * (hour - 6) / 12)
+        return min(station_peak_solar_kw, max(0.0, power))
diff --git a/backend/migrations/versions/20260916_0009_station_peak_solar.py b/backend/migrations/versions/20260916_0009_station_peak_solar.py
new file mode 100644
index 0000000..38521c0
--- /dev/null
+++ b/backend/migrations/versions/20260916_0009_station_peak_solar.py
@@ -0,0 +1,36 @@
+"""Add station peak solar power.
+
+Revision ID: 20260916_0009
+Revises: 20260901_0008
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "20260916_0009"
+down_revision: str | None = "20260901_0008"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("charging_stations") as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "station_peak_solar_kw",
+                sa.Float(),
+                nullable=False,
+                server_default="0",
+            )
+        )
+        batch_op.create_check_constraint(
+            "ck_stations_peak_solar_nonnegative", "station_peak_solar_kw >= 0"
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("charging_stations") as batch_op:
+        batch_op.drop_constraint("ck_stations_peak_solar_nonnegative", type_="check")
+        batch_op.drop_column("station_peak_solar_kw")
diff --git a/backend/tests/test_infrastructure.py b/backend/tests/test_infrastructure.py
index 9b7ba37..d3058b7 100644
--- a/backend/tests/test_infrastructure.py
+++ b/backend/tests/test_infrastructure.py
@@ -13,6 +13,7 @@ async def test_station_and_charger_crud(client: AsyncClient) -> None:
     assert station_response.status_code == 201
     station = station_response.json()
     assert station["is_active"] is True
+    assert station["station_peak_solar_kw"] == 0.0
 
     charger_response = await client.post(
         "/api/v1/chargers",
@@ -29,7 +30,7 @@ async def test_station_and_charger_crud(client: AsyncClient) -> None:
 
     station_update = await client.patch(
         f"/api/v1/stations/{station['id']}",
-        json={"grid_limit_kw": 55, "description": None},
+        json={"grid_limit_kw": 55, "description": None, "station_peak_solar_kw": 18.5},
     )
     charger_update = await client.patch(
         f"/api/v1/chargers/{charger['id']}",
@@ -38,13 +39,15 @@ async def test_station_and_charger_crud(client: AsyncClient) -> None:
     assert station_update.status_code == 200
     assert station_update.json()["description"] is None
     assert station_update.json()["grid_limit_kw"] == 55
+    assert station_update.json()["station_peak_solar_kw"] == 18.5
     assert charger_update.status_code == 200
     assert charger_update.json()["status"] == "UNAVAILABLE"
     assert charger_update.json()["is_active"] is False
 
     assert (await client.get("/api/v1/stations")).json() == [station_update.json()]
     assert (await client.get("/api/v1/chargers")).json() == [charger_update.json()]
-    assert (await client.get(f"/api/v1/stations/{station['id']}")).status_code == 200
+    persisted_station = (await client.get(f"/api/v1/stations/{station['id']}")).json()
+    assert persisted_station["station_peak_solar_kw"] == 18.5
     assert (await client.get(f"/api/v1/chargers/{charger['id']}")).status_code == 200
 
 
@@ -69,6 +72,17 @@ async def test_infrastructure_validates_references_enums_and_power(client: Async
     assert (
         await client.post("/api/v1/stations", json={"name": "Invalid", "grid_limit_kw": 0})
     ).status_code == 422
+    for peak in (-1, "NaN", "Infinity"):
+        invalid_solar = await client.post(
+            "/api/v1/stations",
+            json={"name": "Invalid solar", "grid_limit_kw": 60, "station_peak_solar_kw": peak},
+        )
+        assert invalid_solar.status_code == 422
+    assert (
+        await client.patch(
+            f"/api/v1/stations/{station_id}", json={"station_peak_solar_kw": None}
+        )
+    ).status_code == 422
 
     charger = await client.post(
         "/api/v1/chargers",
diff --git a/backend/tests/test_simulation_energy_data.py b/backend/tests/test_simulation_energy_data.py
new file mode 100644
index 0000000..0b6b33f
--- /dev/null
+++ b/backend/tests/test_simulation_energy_data.py
@@ -0,0 +1,59 @@
+"""Solar source remains deterministic and bounded across the simulated day."""
+
+import math
+from datetime import UTC, datetime, timedelta, timezone
+
+import pytest
+
+from app.simulation.energy_data import EnergyDataProvider, SimulationEnergyDataProvider
+
+
+@pytest.mark.parametrize(
+    ("hour", "expected"),
+    [
+        (0, 0.0),  # midnight
+        (5, 0.0),  # before sunrise
+        (6, 0.0),  # sunrise
+        (9, 5 * math.sqrt(2)),  # morning
+        (12, 10.0),  # noon
+        (15, 5 * math.sqrt(2)),  # afternoon
+        (18, 0.0),  # sunset
+        (21, 0.0),  # night
+    ],
+)
+def test_solar_curve(hour: int, expected: float) -> None:
+    provider: EnergyDataProvider = SimulationEnergyDataProvider()
+    instant = datetime(2026, 9, 16, hour, tzinfo=UTC)
+    assert provider.solar_available_kw(instant, 10.0) == pytest.approx(expected)
+
+
+def test_zero_peak_and_reproducibility() -> None:
+    provider = SimulationEnergyDataProvider()
+    instant = datetime(2026, 9, 16, 11, 37, tzinfo=UTC)
+    assert provider.solar_available_kw(instant, 0.0) == 0.0
+    assert provider.solar_available_kw(instant, 10.0) == provider.solar_available_kw(
+        instant, 10.0
+    )
+    assert provider.solar_available_kw(instant, 10.0) == provider.solar_available_kw(
+        instant.astimezone(timezone(timedelta(hours=-3))), 10.0
+    )
+
+
+def test_solar_power_is_finite_and_bounded() -> None:
+    provider = SimulationEnergyDataProvider()
+    for minute in range(24 * 60):
+        instant = datetime(2026, 9, 16, tzinfo=UTC) + timedelta(minutes=minute)
+        power = provider.solar_available_kw(instant, 32.5)
+        assert math.isfinite(power)
+        assert 0 <= power <= 32.5
+
+
+@pytest.mark.parametrize("peak", [-1.0, math.inf, -math.inf, math.nan])
+def test_invalid_peak_is_rejected(peak: float) -> None:
+    with pytest.raises(ValueError, match="station_peak_solar_kw"):
+        SimulationEnergyDataProvider().solar_available_kw(datetime(2026, 9, 16, tzinfo=UTC), peak)
+
+
+def test_naive_timestamp_is_rejected() -> None:
+    with pytest.raises(ValueError, match="timezone-aware"):
+        SimulationEnergyDataProvider().solar_available_kw(datetime(2026, 9, 16, 12), 10.0)

From 164dcd226a7cd887aef0f2108246d869a066581b Mon Sep 17 00:00:00 2001
From: Matheus Santana <130666046+mathsant-js@users.noreply.github.com>
Date: Wed, 16 Sep 2026 21:13:51 -0300
Subject: [PATCH 3/6] feat(energy): add interval energy calculation and reading
 data

---
 backend/app/services/energy_readings.py     | 101 +++++++++++
 backend/tests/test_energy_reading_domain.py | 186 ++++++++++++++++++++
 2 files changed, 287 insertions(+)
 create mode 100644 backend/app/services/energy_readings.py
 create mode 100644 backend/tests/test_energy_reading_domain.py

diff --git a/backend/app/services/energy_readings.py b/backend/app/services/energy_readings.py
new file mode 100644
index 0000000..3ccf2ab
--- /dev/null
+++ b/backend/app/services/energy_readings.py
@@ -0,0 +1,101 @@
+"""Pure interval energy calculations and data for an EnergyReading."""
+
+import math
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+from uuid import UUID
+
+# Float power and energy match the existing SQLAlchemy columns. Allow only
+# rounding error when comparing independently supplied power and energy parts.
+RELATIVE_TOLERANCE = 1e-9
+ABSOLUTE_TOLERANCE = 1e-12
+
+
+@dataclass(frozen=True, slots=True)
+class EnergyReadingData:
+    session_id: UUID
+    timestamp: datetime
+    requested_power_kw: float
+    allocated_power_kw: float
+    solar_power_kw: float
+    grid_power_kw: float
+    interval_energy_kwh: float
+    solar_energy_kwh: float
+    grid_energy_kwh: float
+
+
+def _validate_power(name: str, power_kw: float) -> None:
+    if not math.isfinite(power_kw) or power_kw < 0:
+        raise ValueError(f"{name} must be finite and nonnegative")
+
+
+def calculate_energy_kwh(power_kw: float, interval: timedelta) -> float:
+    """Convert a nonnegative power over an explicit duration into kWh."""
+
+    _validate_power("power_kw", power_kw)
+    if interval < timedelta(0):
+        raise ValueError("interval must be nonnegative")
+    energy_kwh = power_kw * (interval.total_seconds() / 3600)
+    if not math.isfinite(energy_kwh):
+        raise ValueError("interval energy must be finite")
+    return energy_kwh
+
+
+def build_energy_reading_data(
+    *,
+    session_id: UUID,
+    timestamp: datetime,
+    requested_power_kw: float,
+    allocated_power_kw: float,
+    solar_power_kw: float,
+    grid_power_kw: float,
+    interval: timedelta,
+) -> EnergyReadingData:
+    """Build reading fields from power values already chosen by the allocator.
+
+    Power and energy conservation use 1e-9 relative and 1e-12 absolute
+    tolerance for floating point rounding. Zero allocation requires zero
+    solar and grid power exactly.
+    """
+
+    for name, power in (
+        ("requested_power_kw", requested_power_kw),
+        ("allocated_power_kw", allocated_power_kw),
+        ("solar_power_kw", solar_power_kw),
+        ("grid_power_kw", grid_power_kw),
+    ):
+        _validate_power(name, power)
+    if allocated_power_kw > requested_power_kw:
+        raise ValueError("allocated_power_kw must not exceed requested_power_kw")
+    if allocated_power_kw == 0 and (solar_power_kw != 0 or grid_power_kw != 0):
+        raise ValueError("zero allocation requires zero solar and grid power")
+    if not math.isclose(
+        solar_power_kw + grid_power_kw,
+        allocated_power_kw,
+        rel_tol=RELATIVE_TOLERANCE,
+        abs_tol=ABSOLUTE_TOLERANCE,
+    ):
+        raise ValueError("solar and grid power must sum to allocated_power_kw")
+
+    interval_energy_kwh = calculate_energy_kwh(allocated_power_kw, interval)
+    solar_energy_kwh = calculate_energy_kwh(solar_power_kw, interval)
+    grid_energy_kwh = calculate_energy_kwh(grid_power_kw, interval)
+    if not math.isclose(
+        solar_energy_kwh + grid_energy_kwh,
+        interval_energy_kwh,
+        rel_tol=RELATIVE_TOLERANCE,
+        abs_tol=ABSOLUTE_TOLERANCE,
+    ):
+        raise ValueError("solar and grid energy must sum to interval_energy_kwh")
+
+    return EnergyReadingData(
+        session_id=session_id,
+        timestamp=timestamp,
+        requested_power_kw=requested_power_kw,
+        allocated_power_kw=allocated_power_kw,
+        solar_power_kw=solar_power_kw,
+        grid_power_kw=grid_power_kw,
+        interval_energy_kwh=interval_energy_kwh,
+        solar_energy_kwh=solar_energy_kwh,
+        grid_energy_kwh=grid_energy_kwh,
+    )
diff --git a/backend/tests/test_energy_reading_domain.py b/backend/tests/test_energy_reading_domain.py
new file mode 100644
index 0000000..9a2ff02
--- /dev/null
+++ b/backend/tests/test_energy_reading_domain.py
@@ -0,0 +1,186 @@
+"""Interval energy is deterministic and conserved across solar and grid."""
+
+import math
+from dataclasses import asdict
+from datetime import UTC, datetime, timedelta
+from uuid import UUID
+
+import pytest
+
+from app.services.energy_readings import (
+    ABSOLUTE_TOLERANCE,
+    RELATIVE_TOLERANCE,
+    build_energy_reading_data,
+    calculate_energy_kwh,
+)
+
+SESSION_ID = UUID("00000000-0000-0000-0000-000000000001")
+INSTANT = datetime(2026, 9, 16, 12, tzinfo=UTC)
+
+
+@pytest.mark.parametrize(
+    ("power_kw", "interval", "expected_kwh"),
+    [
+        (15.0, timedelta(minutes=5), 1.25),
+        (0.0, timedelta(minutes=5), 0.0),
+        (15.0, timedelta(0), 0.0),
+    ],
+)
+def test_calculate_energy_kwh(
+    power_kw: float, interval: timedelta, expected_kwh: float
+) -> None:
+    assert calculate_energy_kwh(power_kw, interval) == pytest.approx(expected_kwh)
+
+
+@pytest.mark.parametrize(
+    ("allocated", "solar", "grid", "total_kwh", "solar_kwh", "grid_kwh"),
+    [
+        (15.0, 15.0, 0.0, 1.25, 1.25, 0.0),
+        (15.0, 0.0, 15.0, 1.25, 0.0, 1.25),
+        (15.0, 6.0, 9.0, 1.25, 0.5, 0.75),
+        (0.0, 0.0, 0.0, 0.0, 0.0, 0.0),
+    ],
+)
+def test_build_energy_reading_data(
+    allocated: float,
+    solar: float,
+    grid: float,
+    total_kwh: float,
+    solar_kwh: float,
+    grid_kwh: float,
+) -> None:
+    reading = build_energy_reading_data(
+        session_id=SESSION_ID,
+        timestamp=INSTANT,
+        requested_power_kw=15.0,
+        allocated_power_kw=allocated,
+        solar_power_kw=solar,
+        grid_power_kw=grid,
+        interval=timedelta(minutes=5),
+    )
+
+    assert reading.interval_energy_kwh == pytest.approx(total_kwh)
+    assert reading.solar_energy_kwh == pytest.approx(solar_kwh)
+    assert reading.grid_energy_kwh == pytest.approx(grid_kwh)
+    assert reading.interval_energy_kwh == pytest.approx(
+        reading.solar_energy_kwh + reading.grid_energy_kwh
+    )
+    assert asdict(reading) == {
+        "session_id": SESSION_ID,
+        "timestamp": INSTANT,
+        "requested_power_kw": 15.0,
+        "allocated_power_kw": allocated,
+        "solar_power_kw": solar,
+        "grid_power_kw": grid,
+        "interval_energy_kwh": reading.interval_energy_kwh,
+        "solar_energy_kwh": reading.solar_energy_kwh,
+        "grid_energy_kwh": reading.grid_energy_kwh,
+    }
+
+
+def test_zero_interval_builds_zero_energy_with_nonzero_power() -> None:
+    reading = build_energy_reading_data(
+        session_id=SESSION_ID,
+        timestamp=INSTANT,
+        requested_power_kw=15,
+        allocated_power_kw=15,
+        solar_power_kw=6,
+        grid_power_kw=9,
+        interval=timedelta(0),
+    )
+    assert (reading.interval_energy_kwh, reading.solar_energy_kwh, reading.grid_energy_kwh) == (
+        0,
+        0,
+        0,
+    )
+
+
+@pytest.mark.parametrize("power_kw", [-1.0, -math.inf, math.inf, math.nan])
+def test_calculation_rejects_invalid_power(power_kw: float) -> None:
+    with pytest.raises(ValueError, match="power_kw must be finite and nonnegative"):
+        calculate_energy_kwh(power_kw, timedelta(minutes=5))
+
+
+def test_calculation_rejects_negative_interval() -> None:
+    with pytest.raises(ValueError, match="interval must be nonnegative"):
+        calculate_energy_kwh(15, timedelta(microseconds=-1))
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("requested_power_kw", -1.0),
+        ("allocated_power_kw", -1.0),
+        ("solar_power_kw", -1.0),
+        ("grid_power_kw", -1.0),
+        ("solar_power_kw", math.nan),
+    ],
+)
+def test_reading_rejects_invalid_power(field: str, value: float) -> None:
+    powers = dict(requested_power_kw=15.0, allocated_power_kw=15.0,
+                  solar_power_kw=6.0, grid_power_kw=9.0)
+    powers[field] = value
+    with pytest.raises(ValueError, match=f"{field} must be finite and nonnegative"):
+        build_energy_reading_data(
+            session_id=SESSION_ID, timestamp=INSTANT, interval=timedelta(minutes=5), **powers
+        )
+
+
+@pytest.mark.parametrize(
+    ("allocated", "solar", "grid"),
+    [(15.0, 6.0, 8.0), (0.0, 1e-13, 0.0)],
+)
+def test_reading_rejects_inconsistent_power(
+    allocated: float, solar: float, grid: float
+) -> None:
+    with pytest.raises(ValueError, match="power"):
+        build_energy_reading_data(
+            session_id=SESSION_ID,
+            timestamp=INSTANT,
+            requested_power_kw=15,
+            allocated_power_kw=allocated,
+            solar_power_kw=solar,
+            grid_power_kw=grid,
+            interval=timedelta(minutes=5),
+        )
+
+
+def test_reading_rejects_negative_interval_and_excess_allocation() -> None:
+    with pytest.raises(ValueError, match="interval must be nonnegative"):
+        build_energy_reading_data(
+            session_id=SESSION_ID,
+            timestamp=INSTANT,
+            requested_power_kw=15,
+            allocated_power_kw=15,
+            solar_power_kw=6,
+            grid_power_kw=9,
+            interval=timedelta(seconds=-1),
+        )
+    with pytest.raises(ValueError, match="must not exceed"):
+        build_energy_reading_data(
+            session_id=SESSION_ID,
+            timestamp=INSTANT,
+            requested_power_kw=14,
+            allocated_power_kw=15,
+            solar_power_kw=6,
+            grid_power_kw=9,
+            interval=timedelta(minutes=5),
+        )
+
+
+def test_float_rounding_is_accepted_and_energy_is_conserved_within_tolerance() -> None:
+    reading = build_energy_reading_data(
+        session_id=SESSION_ID,
+        timestamp=INSTANT,
+        requested_power_kw=0.3,
+        allocated_power_kw=0.3,
+        solar_power_kw=0.1,
+        grid_power_kw=0.2,
+        interval=timedelta(minutes=5),
+    )
+    assert math.isclose(
+        reading.interval_energy_kwh,
+        reading.solar_energy_kwh + reading.grid_energy_kwh,
+        rel_tol=RELATIVE_TOLERANCE,
+        abs_tol=ABSOLUTE_TOLERANCE,
+    )

From ea4a24f70b4bab0c31e11537095731a46b9cf2a2 Mon Sep 17 00:00:00 2001
From: Matheus Santana <130666046+mathsant-js@users.noreply.github.com>
Date: Wed, 16 Sep 2026 21:31:24 -0300
Subject: [PATCH 4/6] feat(simulation): execute and persist deterministic ticks

- Process CHARGING sessions by station using an injected power resolver
- Persist solar and energy readings in a single transaction
- Update session accumulators and enforce energy limits
- Prevent duplicate readings for the same station or session and timestamp
- Add tests for rollback, idempotency, and logging
---
 README.md                                     |  17 +-
 backend/app/models/energy.py                  |  14 +-
 backend/app/simulation/tick.py                | 201 ++++++++++++++
 .../versions/20260916_0010_tick_uniqueness.py |  32 +++
 backend/tests/test_simulation_tick.py         | 259 ++++++++++++++++++
 5 files changed, 518 insertions(+), 5 deletions(-)
 create mode 100644 backend/app/simulation/tick.py
 create mode 100644 backend/migrations/versions/20260916_0010_tick_uniqueness.py
 create mode 100644 backend/tests/test_simulation_tick.py

diff --git a/README.md b/README.md
index fc38d2d..6ac5201 100644
--- a/README.md
+++ b/README.md
@@ -75,7 +75,16 @@ evidências documentados. O backend entrega Users/Auth, Vehicles, Stations, Char
 e Sessions sob `/api/v1`, com JWT, autorização por papel e propriedade, persistência
 via Alembic e regras de início/encerramento de sessão na camada de serviço.
 
-O próximo incremento funcional é a Fase 3. Relógio/ticks do simulador, gestão e
-alocação energética, dashboards, ESG e treinamento/inferência de ML ainda não estão
-implementados. Estruturas preparatórias de dados de fases futuras existentes no
-backend não devem ser confundidas com esses fluxos completos.
+A Fase 3 já contém relógio determinístico, provedor solar e o serviço
+`app.simulation.tick.execute_tick`. O serviço recebe uma resolução de potência
+injetável por estação, valida os limites físicos e persiste as leituras e os
+acumuladores em uma transação. Não há loop automático nem rota pública para
+acioná-lo. A implementação de alocação e rateio pertence à Fase 4.
+
+Cada estação processada recebe uma `SolarReading` única por timestamp simulado;
+cada sessão recebe uma `EnergyReading` única por timestamp. Uma reexecução no
+mesmo instante ignora estações já concluídas. Constraints no banco impedem
+duplicatas mesmo em escrita concorrente, e uma falha desfaz todo o tick. O
+relógio avança apenas depois do commit. O chamador deve passar uma sessão de
+banco sem transação ativa e um relógio iniciado; em caso de erro pode repetir
+o mesmo tick após corrigir a causa.
diff --git a/backend/app/models/energy.py b/backend/app/models/energy.py
index a39ddd7..cda8800 100644
--- a/backend/app/models/energy.py
+++ b/backend/app/models/energy.py
@@ -3,7 +3,17 @@
 from decimal import Decimal
 from enum import StrEnum
 
-from sqlalchemy import CheckConstraint, DateTime, Enum, Float, ForeignKey, Index, Numeric, text
+from sqlalchemy import (
+    CheckConstraint,
+    DateTime,
+    Enum,
+    Float,
+    ForeignKey,
+    Index,
+    Numeric,
+    UniqueConstraint,
+    text,
+)
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from app.db.base import Base
@@ -87,6 +97,7 @@ class ChargingSession(TimestampMixin, Base):
 class EnergyReading(Base):
     __tablename__ = "energy_readings"
     __table_args__ = (
+        UniqueConstraint("session_id", "timestamp", name="uq_energy_readings_session_tick"),
         CheckConstraint("requested_power_kw >= 0", name="ck_energy_readings_requested_nonnegative"),
         CheckConstraint("allocated_power_kw >= 0", name="ck_energy_readings_allocated_nonnegative"),
         CheckConstraint(
@@ -119,6 +130,7 @@ class EnergyReading(Base):
 class SolarReading(Base):
     __tablename__ = "solar_readings"
     __table_args__ = (
+        UniqueConstraint("station_id", "timestamp", name="uq_solar_readings_station_tick"),
         CheckConstraint("available_power_kw >= 0", name="ck_solar_readings_power_nonnegative"),
     )
 
diff --git a/backend/app/simulation/tick.py b/backend/app/simulation/tick.py
new file mode 100644
index 0000000..64094b8
--- /dev/null
+++ b/backend/app/simulation/tick.py
@@ -0,0 +1,201 @@
+"""Persist one deterministic simulator tick using an injected power resolver.
+
+The resolver owns allocation and solar distribution (Phase 4). This service
+only checks its safety invariants and records the resulting interval energy.
+"""
+
+import logging
+import math
+from collections import defaultdict
+from collections.abc import Mapping, Sequence
+from dataclasses import asdict, dataclass
+from datetime import datetime
+from typing import Protocol
+from uuid import UUID
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.models.energy import ChargingSession, ChargingSessionStatus, EnergyReading, SolarReading
+from app.models.infrastructure import Charger, ChargingStation
+from app.models.vehicle import Vehicle
+from app.services.energy_readings import ABSOLUTE_TOLERANCE, build_energy_reading_data
+from app.simulation.clock import SimulationClock, SimulationClockState
+from app.simulation.energy_data import EnergyDataProvider
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True, slots=True)
+class SessionPowerRequest:
+    session_id: UUID
+    requested_power_kw: float
+    charger_max_power_kw: float
+    vehicle_max_charge_power_kw: float
+
+
+@dataclass(frozen=True, slots=True)
+class PowerBreakdown:
+    allocated_power_kw: float
+    solar_power_kw: float
+    grid_power_kw: float
+
+
+class PowerResolver(Protocol):
+    """Phase 4 supplies one breakdown for each charging session at a station."""
+
+    def resolve(
+        self,
+        *,
+        station_id: UUID,
+        grid_limit_kw: float,
+        solar_available_kw: float,
+        sessions: Sequence[SessionPowerRequest],
+    ) -> Mapping[UUID, PowerBreakdown]: ...
+
+
+@dataclass(frozen=True, slots=True)
+class TickResult:
+    timestamp: datetime
+    stations_processed: int
+    energy_readings_created: int
+    stations_already_processed: int
+
+
+def execute_tick(
+    db: Session,
+    *,
+    clock: SimulationClock,
+    solar_provider: EnergyDataProvider,
+    power_resolver: PowerResolver,
+) -> TickResult:
+    """Commit all readings and accumulators atomically, then advance the clock.
+
+    A station with a SolarReading at the same timestamp is already complete:
+    the previous tick committed atomically. The database unique constraints
+    also reject concurrent or external duplicate writes.
+    """
+
+    if clock.state is not SimulationClockState.RUNNING:
+        raise RuntimeError("simulation clock must be running")
+    timestamp = clock.current_instant
+    processed = 0
+    created = 0
+    skipped = 0
+    try:
+        with db.begin():
+            rows = db.execute(
+                select(ChargingSession, Charger, Vehicle, ChargingStation)
+                .join(Charger, ChargingSession.charger_id == Charger.id)
+                .join(Vehicle, ChargingSession.vehicle_id == Vehicle.id)
+                .join(ChargingStation, Charger.station_id == ChargingStation.id)
+                .where(ChargingSession.status == ChargingSessionStatus.CHARGING)
+                .order_by(ChargingStation.id, ChargingSession.id)
+            ).all()
+            stations: dict[UUID, list[tuple[ChargingSession, Charger, Vehicle]]] = defaultdict(list)
+            for charging_session, charger, vehicle, station in rows:
+                stations[station.id].append((charging_session, charger, vehicle))
+
+            for station_id, members in stations.items():
+                # Serialize same-station ticks on PostgreSQL before checking idempotency.
+                station = db.scalar(
+                    select(ChargingStation)
+                    .where(ChargingStation.id == station_id)
+                    .with_for_update()
+                )
+                if station is None:
+                    raise RuntimeError("charging station disappeared during tick")
+                existing = db.scalar(
+                    select(SolarReading.id).where(
+                        SolarReading.station_id == station_id,
+                        SolarReading.timestamp == timestamp,
+                    )
+                )
+                if existing is not None:
+                    skipped += 1
+                    continue
+
+                solar_available = solar_provider.solar_available_kw(
+                    timestamp, station.station_peak_solar_kw
+                )
+                if (
+                    not math.isfinite(solar_available)
+                    or not 0 <= solar_available <= station.station_peak_solar_kw
+                ):
+                    raise ValueError("solar provider returned power outside station capacity")
+                requests = [
+                    SessionPowerRequest(
+                        session_id=session.id,
+                        requested_power_kw=session.requested_power_kw,
+                        charger_max_power_kw=charger.max_power_kw,
+                        vehicle_max_charge_power_kw=vehicle.max_charge_power_kw,
+                    )
+                    for session, charger, vehicle in members
+                ]
+                powers = power_resolver.resolve(
+                    station_id=station_id,
+                    grid_limit_kw=station.grid_limit_kw,
+                    solar_available_kw=solar_available,
+                    sessions=requests,
+                )
+                if set(powers) != {request.session_id for request in requests}:
+                    raise ValueError(
+                        "power resolver must return every charging session exactly once"
+                    )
+                grid_total = sum(power.grid_power_kw for power in powers.values())
+                solar_total = sum(power.solar_power_kw for power in powers.values())
+                if (
+                    not math.isfinite(grid_total)
+                    or grid_total > station.grid_limit_kw + ABSOLUTE_TOLERANCE
+                ):
+                    raise ValueError("grid power exceeds station limit")
+                if (
+                    not math.isfinite(solar_total)
+                    or solar_total > solar_available + ABSOLUTE_TOLERANCE
+                ):
+                    raise ValueError("solar power exceeds available generation")
+
+                db.add(
+                    SolarReading(
+                        station_id=station_id,
+                        timestamp=timestamp,
+                        available_power_kw=solar_available,
+                    )
+                )
+                for (session, _, _), request in zip(members, requests, strict=True):
+                    power = powers[session.id]
+                    if power.allocated_power_kw > min(
+                        request.charger_max_power_kw, request.vehicle_max_charge_power_kw
+                    ):
+                        raise ValueError("allocation exceeds charger or vehicle limit")
+                    reading = build_energy_reading_data(
+                        session_id=session.id,
+                        timestamp=timestamp,
+                        requested_power_kw=request.requested_power_kw,
+                        allocated_power_kw=power.allocated_power_kw,
+                        solar_power_kw=power.solar_power_kw,
+                        grid_power_kw=power.grid_power_kw,
+                        interval=clock.tick_duration,
+                    )
+                    db.add(EnergyReading(**asdict(reading)))
+                    session.allocated_power_kw = reading.allocated_power_kw
+                    session.energy_consumed_kwh += reading.interval_energy_kwh
+                    session.solar_energy_kwh += reading.solar_energy_kwh
+                    session.grid_energy_kwh += reading.grid_energy_kwh
+                    created += 1
+                processed += 1
+        clock.advance()
+    except Exception:
+        logger.exception("simulation_tick_failed", extra={"tick_timestamp": timestamp.isoformat()})
+        raise
+
+    logger.info(
+        "simulation_tick_completed",
+        extra={
+            "tick_timestamp": timestamp.isoformat(),
+            "stations_processed": processed,
+            "energy_readings_created": created,
+            "stations_already_processed": skipped,
+        },
+    )
+    return TickResult(timestamp, processed, created, skipped)
diff --git a/backend/migrations/versions/20260916_0010_tick_uniqueness.py b/backend/migrations/versions/20260916_0010_tick_uniqueness.py
new file mode 100644
index 0000000..bf45958
--- /dev/null
+++ b/backend/migrations/versions/20260916_0010_tick_uniqueness.py
@@ -0,0 +1,32 @@
+"""Prevent duplicate readings for one simulated instant.
+
+Revision ID: 20260916_0010
+Revises: 20260916_0009
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+
+revision: str = "20260916_0010"
+down_revision: str | None = "20260916_0009"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("energy_readings") as batch_op:
+        batch_op.create_unique_constraint(
+            "uq_energy_readings_session_tick", ["session_id", "timestamp"]
+        )
+    with op.batch_alter_table("solar_readings") as batch_op:
+        batch_op.create_unique_constraint(
+            "uq_solar_readings_station_tick", ["station_id", "timestamp"]
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("solar_readings") as batch_op:
+        batch_op.drop_constraint("uq_solar_readings_station_tick", type_="unique")
+    with op.batch_alter_table("energy_readings") as batch_op:
+        batch_op.drop_constraint("uq_energy_readings_session_tick", type_="unique")
diff --git a/backend/tests/test_simulation_tick.py b/backend/tests/test_simulation_tick.py
new file mode 100644
index 0000000..f6bcd15
--- /dev/null
+++ b/backend/tests/test_simulation_tick.py
@@ -0,0 +1,259 @@
+"""One tick persists only supplied safe power, atomically and once."""
+
+from collections.abc import Sequence
+from datetime import UTC, datetime
+from decimal import Decimal
+from uuid import UUID, uuid4
+
+import pytest
+from sqlalchemy import func, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.models.energy import ChargingSession, ChargingSessionStatus, EnergyReading, SolarReading
+from app.models.infrastructure import Charger, ChargingStation
+from app.models.user import User
+from app.models.vehicle import Vehicle
+from app.simulation.clock import SimulationClock
+from app.simulation.energy_data import SimulationEnergyDataProvider
+from app.simulation.tick import PowerBreakdown, SessionPowerRequest, execute_tick
+
+INSTANT = datetime(2026, 9, 16, 12, tzinfo=UTC)
+
+
+class FixedResolver:
+    def __init__(self, powers: dict[UUID, PowerBreakdown]) -> None:
+        self.powers = powers
+        self.calls: list[UUID] = []
+
+    def resolve(
+        self,
+        *,
+        station_id: UUID,
+        grid_limit_kw: float,
+        solar_available_kw: float,
+        sessions: Sequence[SessionPowerRequest],
+    ) -> dict[UUID, PowerBreakdown]:
+        self.calls.append(station_id)
+        return {request.session_id: self.powers[request.session_id] for request in sessions}
+
+
+def clock() -> SimulationClock:
+    value = SimulationClock(initial_instant=INSTANT)
+    value.start()
+    return value
+
+
+def add_session(
+    db: Session,
+    *,
+    station: ChargingStation | None = None,
+    status: ChargingSessionStatus = ChargingSessionStatus.CHARGING,
+) -> tuple[ChargingStation, ChargingSession]:
+    if station is None:
+        station = ChargingStation(name="Station", grid_limit_kw=20, station_peak_solar_kw=10)
+        db.add(station)
+        db.flush()
+    user = User(name="Driver", email=f"{uuid4()}@example.com", password_hash="hash")
+    db.add(user)
+    db.flush()
+    vehicle = Vehicle(
+        user_id=user.id,
+        name="EV",
+        brand="Brand",
+        model="Model",
+        license_plate=str(uuid4())[:8],
+        max_charge_power_kw=11,
+    )
+    charger = Charger(station_id=station.id, name="Charger", code=str(uuid4()), max_power_kw=11)
+    db.add_all([vehicle, charger])
+    db.flush()
+    session = ChargingSession(
+        user_id=user.id,
+        vehicle_id=vehicle.id,
+        charger_id=charger.id,
+        status=status,
+        requested_power_kw=11,
+        allocated_power_kw=0,
+        energy_consumed_kwh=0,
+        solar_energy_kwh=0,
+        grid_energy_kwh=0,
+        tariff_per_kwh=Decimal("1"),
+        total_cost=Decimal("0"),
+    )
+    db.add(session)
+    db.commit()
+    return station, session
+
+
+def counts(db: Session) -> tuple[int, int]:
+    return (
+        db.scalar(select(func.count()).select_from(EnergyReading)) or 0,
+        db.scalar(select(func.count()).select_from(SolarReading)) or 0,
+    )
+
+
+def test_empty_and_noncharging_sessions_produce_nothing(db_session: Session) -> None:
+    resolver = FixedResolver({})
+    first = execute_tick(
+        db_session,
+        clock=clock(),
+        solar_provider=SimulationEnergyDataProvider(),
+        power_resolver=resolver,
+    )
+    assert first.energy_readings_created == 0
+    assert counts(db_session) == (0, 0)
+    db_session.rollback()
+    for status in (
+        ChargingSessionStatus.CREATED,
+        ChargingSessionStatus.PAUSED,
+        ChargingSessionStatus.COMPLETED,
+        ChargingSessionStatus.CANCELLED,
+    ):
+        _, session = add_session(db_session, status=status)
+        assert session.energy_consumed_kwh == 0
+    result = execute_tick(
+        db_session,
+        clock=clock(),
+        solar_provider=SimulationEnergyDataProvider(),
+        power_resolver=resolver,
+    )
+    assert result.stations_processed == 0
+    assert counts(db_session) == (0, 0)
+    assert not resolver.calls
+
+
+def test_multiple_sessions_stations_and_accumulators(
+    db_session: Session, caplog: pytest.LogCaptureFixture
+) -> None:
+    station_a, first = add_session(db_session)
+    _, second = add_session(db_session, station=station_a)
+    _, paused = add_session(
+        db_session, station=station_a, status=ChargingSessionStatus.PAUSED
+    )
+    station_b, third = add_session(db_session)
+    resolver = FixedResolver(
+        {
+            first.id: PowerBreakdown(9, 4, 5),
+            second.id: PowerBreakdown(7, 3, 4),
+            third.id: PowerBreakdown(6, 2, 4),
+        }
+    )
+    tick_clock = clock()
+    with caplog.at_level("INFO", logger="app.simulation.tick"):
+        result = execute_tick(
+            db_session,
+            clock=tick_clock,
+            solar_provider=SimulationEnergyDataProvider(),
+            power_resolver=resolver,
+        )
+    assert (result.stations_processed, result.energy_readings_created) == (2, 3)
+    assert set(resolver.calls) == {station_a.id, station_b.id}
+    assert counts(db_session) == (3, 2)
+    db_session.refresh(paused)
+    assert paused.energy_consumed_kwh == 0
+    assert paused.allocated_power_kw == 0
+    success = next(
+        record for record in caplog.records if record.message == "simulation_tick_completed"
+    )
+    assert success.energy_readings_created == 3
+    assert success.tick_timestamp == INSTANT.isoformat()
+    readings = db_session.scalars(select(EnergyReading)).all()
+    solar = db_session.scalars(select(SolarReading)).all()
+    assert {reading.timestamp.replace(tzinfo=UTC) for reading in [*readings, *solar]} == {INSTANT}
+    assert all(reading.available_power_kw == 10 for reading in solar)
+    for session in (first, second, third):
+        db_session.refresh(session)
+        own = next(reading for reading in readings if reading.session_id == session.id)
+        assert session.allocated_power_kw == own.allocated_power_kw
+        assert session.energy_consumed_kwh == pytest.approx(own.interval_energy_kwh)
+        assert session.solar_energy_kwh == pytest.approx(own.solar_energy_kwh)
+        assert session.grid_energy_kwh == pytest.approx(own.grid_energy_kwh)
+        assert own.interval_energy_kwh == pytest.approx(own.solar_energy_kwh + own.grid_energy_kwh)
+    db_session.rollback()
+    execute_tick(
+        db_session,
+        clock=tick_clock,
+        solar_provider=SimulationEnergyDataProvider(),
+        power_resolver=resolver,
+    )
+    for session in (first, second, third):
+        db_session.refresh(session)
+        own_readings = db_session.scalars(
+            select(EnergyReading).where(EnergyReading.session_id == session.id)
+        ).all()
+        assert len(own_readings) == 2
+        assert session.energy_consumed_kwh == pytest.approx(
+            sum(reading.interval_energy_kwh for reading in own_readings)
+        )
+        assert session.solar_energy_kwh == pytest.approx(
+            sum(reading.solar_energy_kwh for reading in own_readings)
+        )
+        assert session.grid_energy_kwh == pytest.approx(
+            sum(reading.grid_energy_kwh for reading in own_readings)
+        )
+
+
+def test_repeat_same_instant_is_idempotent_and_constraints_reject_duplicates(
+    db_session: Session,
+) -> None:
+    station, session = add_session(db_session)
+    resolver = FixedResolver({session.id: PowerBreakdown(9, 4, 5)})
+    provider = SimulationEnergyDataProvider()
+    execute_tick(db_session, clock=clock(), solar_provider=provider, power_resolver=resolver)
+    db_session.rollback()
+    repeat = execute_tick(
+        db_session, clock=clock(), solar_provider=provider, power_resolver=resolver
+    )
+    assert repeat.stations_already_processed == 1
+    assert counts(db_session) == (1, 1)
+    db_session.refresh(session)
+    assert session.energy_consumed_kwh == pytest.approx(9 / 60)
+    db_session.rollback()
+    db_session.add(SolarReading(station_id=station.id, timestamp=INSTANT, available_power_kw=10))
+    with pytest.raises(IntegrityError):
+        db_session.commit()
+    db_session.rollback()
+    db_session.add(
+        EnergyReading(
+            session_id=session.id,
+            timestamp=INSTANT,
+            requested_power_kw=11,
+            allocated_power_kw=9,
+            solar_power_kw=4,
+            grid_power_kw=5,
+            interval_energy_kwh=9 / 60,
+            solar_energy_kwh=4 / 60,
+            grid_energy_kwh=5 / 60,
+        )
+    )
+    with pytest.raises(IntegrityError):
+        db_session.commit()
+
+
+def test_failure_rolls_back_all_stations_and_logs(
+    db_session: Session, caplog: pytest.LogCaptureFixture
+) -> None:
+    station_a, first = add_session(db_session)
+    station_b, second = add_session(db_session)
+    ordered = sorted(((station_a.id, first.id), (station_b.id, second.id)))
+    resolver = FixedResolver(
+        {
+            ordered[0][1]: PowerBreakdown(9, 4, 5),
+            ordered[1][1]: PowerBreakdown(11, 0, 21),
+        }
+    )
+    tick_clock = clock()
+    with pytest.raises(ValueError), caplog.at_level("INFO", logger="app.simulation.tick"):
+        execute_tick(
+            db_session,
+            clock=tick_clock,
+            solar_provider=SimulationEnergyDataProvider(),
+            power_resolver=resolver,
+        )
+    assert set(resolver.calls) == {station_a.id, station_b.id}
+    assert counts(db_session) == (0, 0)
+    assert tick_clock.current_instant == INSTANT
+    db_session.refresh(first)
+    assert first.energy_consumed_kwh == 0
+    assert any(record.message == "simulation_tick_failed" for record in caplog.records)

From 46229408acaaa4aafa6eac8d451dfb59e09b83bd Mon Sep 17 00:00:00 2001
From: Matheus Santana <130666046+mathsant-js@users.noreply.github.com>
Date: Wed, 16 Sep 2026 21:47:38 -0300
Subject: [PATCH 5/6] feat(simulation): add admin controls and manual tick API

- Add ADMIN-only status, start, stop, reset and manual tick endpoints.
- Cover authentication, lifecycle, readings and OpenAPI with API tests.
- Document Phase 3 behavior and the pending PostgreSQL validation.
---
 README.md                            |  15 +-
 backend/app/api/router.py            |   2 +
 backend/app/api/routes/simulation.py |  89 ++++++++++
 backend/app/schemas/energy.py        |   7 +
 backend/app/schemas/simulation.py    |  21 +++
 backend/app/simulation/control.py    | 112 ++++++++++++
 backend/tests/test_simulation_api.py | 250 +++++++++++++++++++++++++++
 docs/PHASE_3.md                      |  31 ++++
 8 files changed, 522 insertions(+), 5 deletions(-)
 create mode 100644 backend/app/api/routes/simulation.py
 create mode 100644 backend/app/schemas/simulation.py
 create mode 100644 backend/app/simulation/control.py
 create mode 100644 backend/tests/test_simulation_api.py
 create mode 100644 docs/PHASE_3.md

diff --git a/README.md b/README.md
index 6ac5201..f4aa078 100644
--- a/README.md
+++ b/README.md
@@ -67,19 +67,24 @@ O alvo executa testes, lint e verificação de tipos no backend e no frontend, a
 - [Arquitetura](docs/ARCHITECTURE.md)
 - [Fase 1 — Fundação (concluída)](docs/PHASE_1.md)
 - [Fase 2 — Domínio (concluída)](docs/PHASE_2.md)
+- [Fase 3 — Simulação e leituras (validação pendente)](docs/PHASE_3.md)
 
 ## Estado atual
 
 As Fases 1 (fundação) e 2 (domínio) estão concluídas, com critérios de saída e
-evidências documentados. O backend entrega Users/Auth, Vehicles, Stations, Chargers
-e Sessions sob `/api/v1`, com JWT, autorização por papel e propriedade, persistência
+evidências documentados. A Fase 3 tem implementação e testes concluídos, mas
+aguarda validação online das migrations em PostgreSQL. O backend entrega
+Users/Auth, Vehicles, Stations, Chargers e Sessions sob `/api/v1`, com JWT,
+autorização por papel e propriedade, persistência
 via Alembic e regras de início/encerramento de sessão na camada de serviço.
 
-A Fase 3 já contém relógio determinístico, provedor solar e o serviço
+A Fase 3 contém relógio determinístico, provedor solar e o serviço
 `app.simulation.tick.execute_tick`. O serviço recebe uma resolução de potência
 injetável por estação, valida os limites físicos e persiste as leituras e os
-acumuladores em uma transação. Não há loop automático nem rota pública para
-acioná-lo. A implementação de alocação e rateio pertence à Fase 4.
+acumuladores em uma transação. O controle ADMIN em `/api/v1/simulation` expõe
+status, start, stop, reset e um tick manual (`POST /ticks`), sem loop automático.
+Até a Fase 4, o resolvedor HTTP aloca zero kW. Alocação, limite de rede na
+distribuição e prioridade solar pertencem à Fase 4.
 
 Cada estação processada recebe uma `SolarReading` única por timestamp simulado;
 cada sessão recebe uma `EnergyReading` única por timestamp. Uma reexecução no
diff --git a/backend/app/api/router.py b/backend/app/api/router.py
index f12682f..446beea 100644
--- a/backend/app/api/router.py
+++ b/backend/app/api/router.py
@@ -8,6 +8,7 @@
 from app.api.routes.health import router as health_router
 from app.api.routes.predictions import configuration_router, predictions_router
 from app.api.routes.sessions import router as sessions_router
+from app.api.routes.simulation import router as simulation_router
 from app.api.routes.solar import router as solar_router
 from app.api.routes.stations import router as stations_router
 from app.api.routes.users import router as users_router
@@ -21,6 +22,7 @@
 api_router.include_router(stations_router)
 api_router.include_router(chargers_router)
 api_router.include_router(sessions_router)
+api_router.include_router(simulation_router)
 api_router.include_router(energy_router)
 api_router.include_router(solar_router)
 api_router.include_router(billing_router)
diff --git a/backend/app/api/routes/simulation.py b/backend/app/api/routes/simulation.py
new file mode 100644
index 0000000..40151b5
--- /dev/null
+++ b/backend/app/api/routes/simulation.py
@@ -0,0 +1,89 @@
+"""Administrative control of the explicitly advanced Phase 3 simulator."""
+
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, status
+
+from app.api.dependencies import AdminUser
+from app.api.routes.common import (
+    CONFLICT_RESPONSE,
+    FORBIDDEN_RESPONSE,
+    UNAUTHORIZED_RESPONSE,
+    DbSession,
+)
+from app.schemas.simulation import SimulationStatusResponse, SimulationTickResponse
+from app.simulation.control import SimulationController, get_simulation_controller
+
+router = APIRouter(prefix="/simulation", tags=["simulation"])
+Controller = Annotated[SimulationController, Depends(get_simulation_controller)]
+AUTH_RESPONSES = UNAUTHORIZED_RESPONSE | FORBIDDEN_RESPONSE
+
+
+def _status(control: SimulationController) -> SimulationStatusResponse:
+    with control.lock:
+        return SimulationStatusResponse(
+            state=control.clock.state,
+            current_instant=control.clock.current_instant,
+            tick_duration_seconds=control.clock.tick_duration.total_seconds(),
+            simulation_speed=control.clock.simulation_speed,
+            last_tick=control.last_tick,
+        )
+
+
+@router.get("/status", response_model=SimulationStatusResponse, responses=AUTH_RESPONSES)
+async def simulation_status(_: AdminUser, control: Controller) -> SimulationStatusResponse:
+    """Return this API process's clock state and most recent successful tick."""
+    return _status(control)
+
+
+@router.post("/start", response_model=SimulationStatusResponse, responses=AUTH_RESPONSES)
+async def start_simulation(
+    db: DbSession, _: AdminUser, control: Controller
+) -> SimulationStatusResponse:
+    """Idempotently start manual control; load the configured simulation speed."""
+    control.start(db)
+    return _status(control)
+
+
+@router.post("/stop", response_model=SimulationStatusResponse, responses=AUTH_RESPONSES)
+async def stop_simulation(_: AdminUser, control: Controller) -> SimulationStatusResponse:
+    """Idempotently stop manual ticks; no worker or timer is started."""
+    control.stop()
+    return _status(control)
+
+
+@router.post(
+    "/reset",
+    response_model=SimulationStatusResponse,
+    responses=AUTH_RESPONSES | CONFLICT_RESPONSE,
+)
+async def reset_simulation(
+    db: DbSession, _: AdminUser, control: Controller
+) -> SimulationStatusResponse:
+    """While stopped, clear clock metadata and move beyond saved readings without deleting data."""
+    if not control.reset(db):
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT, detail="Stop simulation before reset"
+        )
+    return _status(control)
+
+
+@router.post(
+    "/ticks",
+    response_model=SimulationTickResponse,
+    responses=AUTH_RESPONSES | CONFLICT_RESPONSE,
+)
+async def run_simulation_tick(
+    db: DbSession, _: AdminUser, control: Controller
+) -> SimulationTickResponse:
+    """Run exactly one transactional tick while RUNNING. Phase 3 allocates zero power."""
+    result = control.tick(db)
+    if result is None:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Simulation is stopped")
+    return SimulationTickResponse(
+        timestamp=result.timestamp,
+        stations_processed=result.stations_processed,
+        energy_readings_created=result.energy_readings_created,
+        stations_already_processed=result.stations_already_processed,
+        current_instant=control.clock.current_instant,
+    )
diff --git a/backend/app/schemas/energy.py b/backend/app/schemas/energy.py
index bdc60b8..42f2451 100644
--- a/backend/app/schemas/energy.py
+++ b/backend/app/schemas/energy.py
@@ -50,6 +50,13 @@ class ReadingResponse(BaseModel):
     id: UUID
     timestamp: datetime
 
+    @field_validator("timestamp")
+    @classmethod
+    def normalize_timestamp_utc(cls, value: datetime) -> datetime:
+        if value.tzinfo is None:
+            return value.replace(tzinfo=UTC)
+        return value.astimezone(UTC)
+
 
 class EnergyReadingResponse(ReadingResponse):
     session_id: UUID
diff --git a/backend/app/schemas/simulation.py b/backend/app/schemas/simulation.py
new file mode 100644
index 0000000..4f750c1
--- /dev/null
+++ b/backend/app/schemas/simulation.py
@@ -0,0 +1,21 @@
+from datetime import datetime
+
+from pydantic import BaseModel
+
+from app.simulation.clock import SimulationClockState
+
+
+class SimulationStatusResponse(BaseModel):
+    state: SimulationClockState
+    current_instant: datetime
+    tick_duration_seconds: float
+    simulation_speed: int
+    last_tick: datetime | None
+
+
+class SimulationTickResponse(BaseModel):
+    timestamp: datetime
+    stations_processed: int
+    energy_readings_created: int
+    stations_already_processed: int
+    current_instant: datetime
diff --git a/backend/app/simulation/control.py b/backend/app/simulation/control.py
new file mode 100644
index 0000000..281d4cb
--- /dev/null
+++ b/backend/app/simulation/control.py
@@ -0,0 +1,112 @@
+"""Single-process, manually driven simulator control for the Phase 3 MVP."""
+
+import logging
+from collections.abc import Sequence
+from datetime import UTC, datetime
+from threading import RLock
+from uuid import UUID
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from app.models.energy import EnergyReading, SolarReading
+from app.models.prediction import SystemConfiguration
+from app.simulation.clock import SimulationClock, SimulationClockState
+from app.simulation.energy_data import SimulationEnergyDataProvider
+from app.simulation.tick import PowerBreakdown, SessionPowerRequest, TickResult, execute_tick
+
+logger = logging.getLogger(__name__)
+
+
+class ZeroPowerResolver:
+    """Safe Phase 3 placeholder until Phase 4 supplies power allocation."""
+
+    def resolve(
+        self,
+        *,
+        station_id: UUID,
+        grid_limit_kw: float,
+        solar_available_kw: float,
+        sessions: Sequence[SessionPowerRequest],
+    ) -> dict[UUID, PowerBreakdown]:
+        return {item.session_id: PowerBreakdown(0, 0, 0) for item in sessions}
+
+
+class SimulationController:
+    def __init__(self, clock: SimulationClock | None = None) -> None:
+        self.clock = clock or SimulationClock(initial_instant=datetime.now(UTC))
+        self.last_tick: datetime | None = None
+        self.lock = RLock()
+
+    def start(self, db: Session) -> None:
+        with self.lock:
+            if self.clock.state is SimulationClockState.RUNNING:
+                return
+            configuration = db.scalar(select(SystemConfiguration).limit(1))
+            if configuration is not None:
+                self.clock.simulation_speed = configuration.simulation_speed
+            db.rollback()
+            self.clock.start()
+            logger.info("simulation_started")
+
+    def stop(self) -> None:
+        with self.lock:
+            if self.clock.state is SimulationClockState.STOPPED:
+                return
+            self.clock.stop()
+            logger.info("simulation_stopped")
+
+    def reset(self, db: Session) -> bool:
+        """Reset only while stopped, preserving readings and moving past their timestamps."""
+        with self.lock:
+            if self.clock.state is SimulationClockState.RUNNING:
+                return False
+            latest_energy = db.scalar(select(func.max(EnergyReading.timestamp)))
+            latest_solar = db.scalar(select(func.max(SolarReading.timestamp)))
+            db.rollback()
+            latest = max(
+                (
+                    value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)
+                    for value in (latest_energy, latest_solar)
+                    if value is not None
+                ),
+                default=None,
+            )
+            next_instant = datetime.now(UTC)
+            if latest is not None:
+                next_instant = max(next_instant, latest + self.clock.tick_duration)
+            self.clock = SimulationClock(
+                initial_instant=next_instant,
+                simulation_speed=self.clock.simulation_speed,
+                real_tick_interval=self.clock.real_tick_interval,
+            )
+            self.last_tick = None
+            logger.info("simulation_reset", extra={"simulated_at": next_instant.isoformat()})
+            return True
+
+    def tick(self, db: Session) -> TickResult | None:
+        with self.lock:
+            if self.clock.state is SimulationClockState.STOPPED:
+                return None
+            # Authentication performed a SELECT on this request's session.
+            # execute_tick owns its transaction and must enter with a clean session.
+            db.rollback()
+            try:
+                result = execute_tick(
+                    db,
+                    clock=self.clock,
+                    solar_provider=SimulationEnergyDataProvider(),
+                    power_resolver=ZeroPowerResolver(),
+                )
+            except Exception:
+                logger.exception("simulation_control_tick_failed")
+                raise
+            self.last_tick = result.timestamp
+            return result
+
+
+controller = SimulationController()
+
+
+async def get_simulation_controller() -> SimulationController:
+    return controller
diff --git a/backend/tests/test_simulation_api.py b/backend/tests/test_simulation_api.py
new file mode 100644
index 0000000..21cc85f
--- /dev/null
+++ b/backend/tests/test_simulation_api.py
@@ -0,0 +1,250 @@
+"""Administrative simulation lifecycle and read API integration."""
+
+from datetime import UTC, datetime, timedelta
+from decimal import Decimal
+from uuid import UUID, uuid4
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from app.core.security import create_access_token
+from app.main import app
+from app.models.energy import ChargingSession, ChargingSessionStatus, EnergyReading, SolarReading
+from app.models.infrastructure import Charger, ChargingStation
+from app.models.prediction import SystemConfiguration
+from app.models.user import User, UserRole
+from app.models.vehicle import Vehicle
+from app.simulation.clock import SimulationClock
+from app.simulation.control import SimulationController, get_simulation_controller
+
+INSTANT = datetime(2026, 9, 16, 12, tzinfo=UTC)
+
+
+@pytest.fixture
+def control() -> SimulationController:
+    value = SimulationController(SimulationClock(initial_instant=INSTANT))
+
+    async def override() -> SimulationController:
+        return value
+
+    app.dependency_overrides[get_simulation_controller] = override
+    yield value
+    app.dependency_overrides.pop(get_simulation_controller, None)
+
+
+def seed_charging_session(db: Session) -> tuple[UUID, UUID]:
+    user = User(name="Driver", email="driver-simulation@example.com", password_hash="hash")
+    station = ChargingStation(name="Solar station", grid_limit_kw=20, station_peak_solar_kw=10)
+    db.add_all([user, station])
+    db.flush()
+    charger = Charger(station_id=station.id, name="C1", code="SIM-1", max_power_kw=11)
+    vehicle = Vehicle(
+        user_id=user.id,
+        name="EV",
+        brand="B",
+        model="M",
+        license_plate="SIM-EV",
+        max_charge_power_kw=11,
+    )
+    db.add_all([charger, vehicle])
+    db.flush()
+    charging = ChargingSession(
+        user_id=user.id,
+        vehicle_id=vehicle.id,
+        charger_id=charger.id,
+        status=ChargingSessionStatus.CHARGING,
+        requested_power_kw=11,
+        allocated_power_kw=0,
+        energy_consumed_kwh=0,
+        solar_energy_kwh=0,
+        grid_energy_kwh=0,
+        tariff_per_kwh=Decimal("1"),
+        total_cost=Decimal("0"),
+    )
+    db.add(charging)
+    db.commit()
+    return station.id, charging.id
+
+
+@pytest.mark.anyio
+async def test_simulation_authentication_and_authorization(
+    client: AsyncClient, db_session: Session, control: SimulationController
+) -> None:
+    user = User(
+        name="Regular",
+        email="regular-simulation@example.com",
+        password_hash="hash",
+        role=UserRole.USER,
+    )
+    db_session.add(user)
+    db_session.commit()
+    paths = [
+        ("get", "/status"),
+        ("post", "/start"),
+        ("post", "/stop"),
+        ("post", "/reset"),
+        ("post", "/ticks"),
+    ]
+    for method, suffix in paths:
+        path = f"/api/v1/simulation{suffix}"
+        for header in (None, "Bearer invalid"):
+            headers = {"Authorization": header} if header else {"Authorization": ""}
+            response = await client.request(method, path, headers=headers)
+            assert response.status_code == 401
+        response = await client.request(
+            method, path, headers={"Authorization": f"Bearer {create_access_token(user.id)}"}
+        )
+        assert response.status_code == 403
+    assert control.clock.current_instant == INSTANT
+
+
+@pytest.mark.anyio
+async def test_simulation_lifecycle_readings_filters_and_reset(
+    client: AsyncClient,
+    db_session: Session,
+    control: SimulationController,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    station_id, session_id = seed_charging_session(db_session)
+    db_session.add(
+        SystemConfiguration(
+            simulation_speed=120,
+            grid_emission_factor_kg_per_kwh=0.4,
+            high_demand_threshold=0.8,
+            medium_peak_threshold=0.5,
+            high_peak_threshold=0.8,
+        )
+    )
+    db_session.commit()
+    base = "/api/v1/simulation"
+    initial = (await client.get(f"{base}/status")).json()
+    assert initial == {
+        "state": "STOPPED",
+        "current_instant": INSTANT.isoformat().replace("+00:00", "Z"),
+        "tick_duration_seconds": 60,
+        "simulation_speed": 60,
+        "last_tick": None,
+    }
+    assert (await client.post(f"{base}/ticks")).status_code == 409
+    with caplog.at_level("INFO"):
+        started = await client.post(f"{base}/start")
+        repeated_start = await client.post(f"{base}/start")
+        assert started.status_code == repeated_start.status_code == 200
+        assert started.json() == repeated_start.json()
+        assert started.json()["state"] == "RUNNING"
+        assert started.json()["simulation_speed"] == 120
+        assert started.json()["tick_duration_seconds"] == 120
+        assert (await client.post(f"{base}/reset")).status_code == 409
+        tick = await client.post(f"{base}/ticks")
+        assert tick.status_code == 200
+        assert tick.json()["timestamp"] == initial["current_instant"]
+        assert tick.json()["stations_processed"] == 1
+        assert tick.json()["energy_readings_created"] == 1
+        current = INSTANT + timedelta(minutes=2)
+        assert datetime.fromisoformat(tick.json()["current_instant"]) == current
+        status = (await client.get(f"{base}/status")).json()
+        assert datetime.fromisoformat(status["current_instant"]) == current
+        assert datetime.fromisoformat(status["last_tick"]) == INSTANT
+        stopped = await client.post(f"{base}/stop")
+        repeated_stop = await client.post(f"{base}/stop")
+        assert stopped.status_code == repeated_stop.status_code == 200
+        assert stopped.json() == repeated_stop.json()
+        assert stopped.json()["state"] == "STOPPED"
+        reset = await client.post(f"{base}/reset")
+        assert reset.status_code == 200
+        assert reset.json()["last_tick"] is None
+        assert datetime.fromisoformat(reset.json()["current_instant"]) >= current
+    assert {record.message for record in caplog.records} >= {
+        "simulation_started",
+        "simulation_stopped",
+        "simulation_reset",
+        "simulation_tick_completed",
+    }
+    db_session.rollback()
+    assert db_session.scalar(select(func.count()).select_from(EnergyReading)) == 1
+    assert db_session.scalar(select(func.count()).select_from(SolarReading)) == 1
+    session = db_session.get(ChargingSession, session_id)
+    assert session is not None
+    assert session.energy_consumed_kwh == 0
+    for resource in ("energy", "solar"):
+        current_response = await client.get(
+            f"/api/v1/{resource}/current", params={"station_id": str(station_id)}
+        )
+        assert current_response.status_code == 200
+        assert datetime.fromisoformat(current_response.json()["timestamp"]) == INSTANT
+        history = await client.get(
+            f"/api/v1/{resource}/history",
+            params={
+                "station_id": str(station_id),
+                "from": INSTANT.isoformat(),
+                "to": INSTANT.isoformat(),
+            },
+        )
+        assert history.status_code == 200
+        assert len(history.json()) == 1
+        excluded = await client.get(
+            f"/api/v1/{resource}/history",
+            params={
+                "station_id": str(uuid4()),
+                "from": INSTANT.isoformat(),
+                "to": INSTANT.isoformat(),
+            },
+        )
+        assert excluded.json() == []
+        after = await client.get(
+            f"/api/v1/{resource}/history",
+            params={"from": (INSTANT + timedelta(seconds=1)).isoformat()},
+        )
+        assert after.json() == []
+    assert (await client.get("/api/v1/energy/current")).json()["allocated_power_kw"] == 0
+    assert (await client.get("/api/v1/solar/current")).json()["available_power_kw"] == 10
+
+
+@pytest.mark.anyio
+async def test_failed_tick_keeps_clock_and_hides_internal_error(
+    client: AsyncClient, control: SimulationController, monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    def fail_tick(*args: object, **kwargs: object) -> None:
+        raise RuntimeError("private simulator traceback marker")
+
+    monkeypatch.setattr("app.simulation.control.execute_tick", fail_tick)
+    assert (await client.post("/api/v1/simulation/start")).status_code == 200
+    transport = ASGITransport(app=app, raise_app_exceptions=False)
+    with caplog.at_level("ERROR"):
+        async with AsyncClient(
+            transport=transport, base_url="http://test", headers=dict(client.headers)
+        ) as safe_client:
+            response = await safe_client.post("/api/v1/simulation/ticks")
+    assert response.status_code == 500
+    assert response.json() == {"detail": "Internal server error"}
+    assert "private simulator" not in response.text
+    assert control.clock.current_instant == INSTANT
+    assert control.last_tick is None
+    assert any(record.message == "simulation_control_tick_failed" for record in caplog.records)
+
+
+@pytest.mark.anyio
+async def test_simulation_openapi_contract(
+    client: AsyncClient, control: SimulationController
+) -> None:
+    response = await client.get("/openapi.json")
+    assert response.status_code == 200
+    paths = response.json()["paths"]
+    for suffix, method, schema in (
+        ("status", "get", "SimulationStatusResponse"),
+        ("start", "post", "SimulationStatusResponse"),
+        ("stop", "post", "SimulationStatusResponse"),
+        ("reset", "post", "SimulationStatusResponse"),
+        ("ticks", "post", "SimulationTickResponse"),
+    ):
+        operation = paths[f"/api/v1/simulation/{suffix}"][method]
+        assert operation["security"] == [{"HTTPBearer": []}]
+        assert operation["responses"]["200"]["content"]["application/json"]["schema"] == {
+            "$ref": f"#/components/schemas/{schema}"
+        }
+        assert {"401", "403"} <= operation["responses"].keys()
+        if suffix in ("reset", "ticks"):
+            assert "409" in operation["responses"]
diff --git a/docs/PHASE_3.md b/docs/PHASE_3.md
new file mode 100644
index 0000000..34a2e53
--- /dev/null
+++ b/docs/PHASE_3.md
@@ -0,0 +1,31 @@
+# Fase 3 — Simulação e leituras
+
+## Escopo entregue
+
+O backend possui relógio simulado determinístico, curva solar por estação, serviço de tick transacional e APIs de leitura de energia e solar. O controle administrativo fica em `/api/v1/simulation`: `GET /status`, `POST /start`, `POST /stop`, `POST /reset` e `POST /ticks` (um tick manual). Todos exigem JWT de ADMIN; token ausente ou inválido retorna 401 e USER retorna 403. Os contratos e erros esperados aparecem no OpenAPI.
+
+`status` informa estado, instante UTC atual, duração simulada do tick em segundos, `simulation_speed` e timestamp do último tick concluído. `start` e `stop` são idempotentes e retornam 200 com o mesmo estado quando repetidos. `start` carrega `simulation_speed` da `SystemConfiguration`, quando presente. Sem configuração, o padrão é 60 segundos simulados por segundo real; o tick manual representa um segundo real.
+
+`POST /ticks` exige estado RUNNING, executa uma vez `execute_tick` e retorna 409 se o simulador estiver parado. O serviço de tick abre uma transação para as leituras e acumuladores, confirma antes de avançar o relógio e registra sucesso ou falha. Sem sessões CHARGING, o relógio ainda avança. A resolução de potência provisória da Fase 3 aloca **zero kW** para cada sessão ativa. Isso gera leituras de energia com zero consumo e mantém os limites físicos, enquanto a curva solar gera `SolarReading` para cada estação com sessão ativa. O resolvedor é injetável; os testes do serviço exercitam leituras com potência não nula por um resolvedor de teste.
+
+`reset` exige STOPPED e retorna 409 durante RUNNING. Ele limpa somente o estado do relógio e o marcador do último tick. O novo instante é o maior entre o horário UTC atual e o último timestamp persistido de energia ou solar mais uma duração de tick. Assim a próxima execução não reutiliza timestamps. **Nenhuma `EnergyReading` ou `SolarReading` é apagada**, nem são zerados acumuladores de sessões. Leituras continuam disponíveis nos endpoints `/energy/current`, `/energy/history`, `/solar/current` e `/solar/history`, inclusive com filtros `station_id`, `from` e `to` nos históricos.
+
+## Arquitetura e limites
+
+O controle usa um relógio e lock por processo web. `start` habilita apenas ticks manuais: não cria thread, timer ou loop automático. O serviço transacional permanece em `app.simulation.tick`, isolado do transporte HTTP. Logs registram start, stop, reset, tick concluído e falhas; erros inesperados recebem a resposta genérica 500 do aplicativo.
+
+Este controle em memória é adequado para a demonstração com **um processo web**. Um reinício perde estado RUNNING, relógio e último tick, mas mantém leituras no banco. Vários workers teriam relógios independentes e não são suportados nesta fase. A idempotência de leituras por estação e timestamp é reforçada por constraints existentes.
+
+## Migrations e validação
+
+A Fase 3 usa as migrations existentes de `station_peak_solar_kw` e unicidade de tick (`20260916_0009` e `20260916_0010`). O controle administrativo não altera o schema, portanto não há migration nova. Os testes cobrem relógio, curva solar, transação, rollback, idempotência, autenticação, autorização, ciclo administrativo, leitura e OpenAPI.
+
+Na validação desta entrega, `alembic heads` encontrou somente `20260916_0010 (head)` e `alembic upgrade head --sql` gerou SQL offline para PostgreSQL. O `alembic check` e o upgrade online não puderam ser concluídos: não havia PostgreSQL em `localhost:5432` e o daemon Docker recusou acesso ao socket. Uma tentativa de upgrade sobre SQLite parou na migration `0006`, que usa alteração de constraint não suportada pelo dialect SQLite. Não houve alteração de schema nesta entrega. A validação online das migrations existentes permanece como limitação operacional.
+
+Ruff e mypy passaram; pytest concluiu 144 testes backend. Vitest concluiu 2 testes frontend; ESLint, typecheck TypeScript e build Vite passaram.
+
+## Próxima fase
+
+**Alocação de potência, limite de rede na distribuição e prioridade solar permanecem para a Fase 4.** O serviço de tick já valida os limites de uma resolução injetada; o resolvedor de produção ainda precisa implementar essas regras.
+
+Alertas e analytics derivados por tick previstos no fluxo completo do SPEC ainda dependem das fases posteriores desses módulos.

From fd945562417d2733ebc9b809c2076fb339e592dc Mon Sep 17 00:00:00 2001
From: Matheus Santana <130666046+mathsant-js@users.noreply.github.com>
Date: Wed, 16 Sep 2026 21:58:42 -0300
Subject: [PATCH 6/6] fix: align enum check constraints with Alembic schema
 comparison

- Declare enum check constraints explicitly in the SQLAlchemy models so Alembic recognizes the constraints already created by migrations.
- Update Phase 3 documentation with the PostgreSQL validation results.
---
 README.md                            | 8 ++++----
 backend/app/models/energy.py         | 6 +++++-
 backend/app/models/infrastructure.py | 9 +++++++--
 backend/app/models/user.py           | 5 +++--
 docs/PHASE_3.md                      | 2 +-
 5 files changed, 20 insertions(+), 10 deletions(-)

diff --git a/README.md b/README.md
index f4aa078..31fce72 100644
--- a/README.md
+++ b/README.md
@@ -67,13 +67,13 @@ O alvo executa testes, lint e verificação de tipos no backend e no frontend, a
 - [Arquitetura](docs/ARCHITECTURE.md)
 - [Fase 1 — Fundação (concluída)](docs/PHASE_1.md)
 - [Fase 2 — Domínio (concluída)](docs/PHASE_2.md)
-- [Fase 3 — Simulação e leituras (validação pendente)](docs/PHASE_3.md)
+- [Fase 3 — Simulação e leituras (concluída)](docs/PHASE_3.md)
 
 ## Estado atual
 
-As Fases 1 (fundação) e 2 (domínio) estão concluídas, com critérios de saída e
-evidências documentados. A Fase 3 tem implementação e testes concluídos, mas
-aguarda validação online das migrations em PostgreSQL. O backend entrega
+As Fases 1 (fundação), 2 (domínio) e 3 (simulação e leituras) estão concluídas,
+com critérios de saída e evidências documentados. As migrations da Fase 3 foram
+validadas online em PostgreSQL. O backend entrega
 Users/Auth, Vehicles, Stations, Chargers e Sessions sob `/api/v1`, com JWT,
 autorização por papel e propriedade, persistência
 via Alembic e regras de início/encerramento de sessão na camada de serviço.
diff --git a/backend/app/models/energy.py b/backend/app/models/energy.py
index cda8800..d904d78 100644
--- a/backend/app/models/energy.py
+++ b/backend/app/models/energy.py
@@ -41,6 +41,10 @@ class ChargingSession(TimestampMixin, Base):
         CheckConstraint("grid_energy_kwh >= 0", name="ck_sessions_grid_energy_nonnegative"),
         CheckConstraint("tariff_per_kwh >= 0", name="ck_sessions_tariff_nonnegative"),
         CheckConstraint("total_cost >= 0", name="ck_sessions_total_cost_nonnegative"),
+        CheckConstraint(
+            "status IN ('CREATED', 'CHARGING', 'PAUSED', 'COMPLETED', 'CANCELLED')",
+            name="charging_session_status",
+        ),
         Index(
             "uq_charging_sessions_active_charger",
             "charger_id",
@@ -72,7 +76,7 @@ class ChargingSession(TimestampMixin, Base):
             ChargingSessionStatus,
             name="charging_session_status",
             native_enum=False,
-            create_constraint=True,
+            create_constraint=False,
             validate_strings=True,
         ),
         default=ChargingSessionStatus.CREATED,
diff --git a/backend/app/models/infrastructure.py b/backend/app/models/infrastructure.py
index cfb71fc..d3df808 100644
--- a/backend/app/models/infrastructure.py
+++ b/backend/app/models/infrastructure.py
@@ -35,7 +35,12 @@ class ChargingStation(TimestampMixin, Base):
 
 class Charger(TimestampMixin, Base):
     __tablename__ = "chargers"
-    __table_args__ = (CheckConstraint("max_power_kw > 0", name="ck_chargers_max_power_positive"),)
+    __table_args__ = (
+        CheckConstraint("max_power_kw > 0", name="ck_chargers_max_power_positive"),
+        CheckConstraint(
+            "status IN ('AVAILABLE', 'CHARGING', 'UNAVAILABLE')", name="charger_status"
+        ),
+    )
 
     id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
     station_id: Mapped[uuid.UUID] = mapped_column(
@@ -49,7 +54,7 @@ class Charger(TimestampMixin, Base):
             ChargerStatus,
             name="charger_status",
             native_enum=False,
-            create_constraint=True,
+            create_constraint=False,
             validate_strings=True,
         ),
         default=ChargerStatus.AVAILABLE,
diff --git a/backend/app/models/user.py b/backend/app/models/user.py
index 700dd82..ee8006c 100644
--- a/backend/app/models/user.py
+++ b/backend/app/models/user.py
@@ -1,7 +1,7 @@
 import uuid
 from enum import StrEnum
 
-from sqlalchemy import Boolean, Enum, String
+from sqlalchemy import Boolean, CheckConstraint, Enum, String
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from app.db.base import Base
@@ -15,6 +15,7 @@ class UserRole(StrEnum):
 
 class User(TimestampMixin, Base):
     __tablename__ = "users"
+    __table_args__ = (CheckConstraint("role IN ('ADMIN', 'USER')", name="user_role"),)
 
     id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
     name: Mapped[str] = mapped_column(String(120), nullable=False)
@@ -25,7 +26,7 @@ class User(TimestampMixin, Base):
             UserRole,
             name="user_role",
             native_enum=False,
-            create_constraint=True,
+            create_constraint=False,
             validate_strings=True,
         ),
         default=UserRole.USER,
diff --git a/docs/PHASE_3.md b/docs/PHASE_3.md
index 34a2e53..b635c7b 100644
--- a/docs/PHASE_3.md
+++ b/docs/PHASE_3.md
@@ -20,7 +20,7 @@ Este controle em memória é adequado para a demonstração com **um processo we
 
 A Fase 3 usa as migrations existentes de `station_peak_solar_kw` e unicidade de tick (`20260916_0009` e `20260916_0010`). O controle administrativo não altera o schema, portanto não há migration nova. Os testes cobrem relógio, curva solar, transação, rollback, idempotência, autenticação, autorização, ciclo administrativo, leitura e OpenAPI.
 
-Na validação desta entrega, `alembic heads` encontrou somente `20260916_0010 (head)` e `alembic upgrade head --sql` gerou SQL offline para PostgreSQL. O `alembic check` e o upgrade online não puderam ser concluídos: não havia PostgreSQL em `localhost:5432` e o daemon Docker recusou acesso ao socket. Uma tentativa de upgrade sobre SQLite parou na migration `0006`, que usa alteração de constraint não suportada pelo dialect SQLite. Não houve alteração de schema nesta entrega. A validação online das migrations existentes permanece como limitação operacional.
+Na validação desta entrega, `alembic heads` encontrou somente `20260916_0010 (head)` e `alembic upgrade head --sql` gerou SQL offline para PostgreSQL. Posteriormente, `alembic upgrade head` foi executado em um PostgreSQL 16 temporário e chegou à revisão `20260916_0010`; `alembic check` concluiu sem operações pendentes. As três constraints de enum das migrations `0006` e `0007` passaram a ser declaradas explicitamente nos modelos: o Alembic 1.19 não compara constraints vinculadas ao tipo `Enum`, embora elas estivessem presentes no banco. Não houve alteração do schema nem migration nova. O banco temporário foi removido após a validação.
 
 Ruff e mypy passaram; pytest concluiu 144 testes backend. Vitest concluiu 2 testes frontend; ESLint, typecheck TypeScript e build Vite passaram.