diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 79014cc..95be30a 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -21,5 +21,6 @@ jobs:
           set -euo pipefail
           bash tests/test_install_2fa_bot_watcher.sh
           bash tests/test_wait_for_ib_gateway_ready.sh
+          python3 tests/test_gateway_recovery_vocabulary.py
           bash tests/test_workflow_shared_config.sh
           bash tests/test_docker_compose_ports.sh
diff --git a/scripts/gateway_recovery_vocabulary.py b/scripts/gateway_recovery_vocabulary.py
new file mode 100644
index 0000000..9517a16
--- /dev/null
+++ b/scripts/gateway_recovery_vocabulary.py
@@ -0,0 +1,43 @@
+"""Pure, Python 3.10-compatible current-epoch recovery vocabulary."""
+from __future__ import annotations
+import re
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Iterable
+
+class Decision(str, Enum): READY="ready"; TERMINAL="terminal"; PROGRESS="progress"; NONE="none"; EPOCH_CHANGED="epoch_changed"
+@dataclass(frozen=True)
+class Epoch: container_id: str; started_at: str; lower_bound: str
+@dataclass(frozen=True)
+class Event: container_id: str; source: str; line: str
+PROGRESS = re.compile(r"(?:IBC: (?:Starting Gateway|Login attempt|Second Factor Authentication|Login has completed|Configuration tasks completed|Found Gateway main window|Getting config dialog|Getting main window)|Authentication window found|Auto-fill submitted|Passed token authentication|Authentication completed|Security code:)$", re.I)
+TERMINAL = re.compile(r"IBC closing because login has not completed|(?:authentication|login).*(?:timed out|timeout|failed)", re.I)
+STAMP = re.compile(r"^(\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d)(?:\.(\d{1,9}))?Z\s+(.*)$")
+def stamp(value: str) -> tuple[datetime,int] | None:
+    match=STAMP.match(value)
+    if not match: return None
+    return (datetime.fromisoformat(match.group(1)).replace(tzinfo=timezone.utc), int(((match.group(2) or '')+'0'*9)[:9]))
+def epoch_key(value: str) -> tuple[datetime,int]:
+    parsed=stamp(value+' x')
+    if parsed is None: raise ValueError('invalid RFC3339 timestamp')
+    return parsed
+class Machine:
+    def __init__(self, epoch: Epoch): self.epoch=epoch; self.terminal=False
+    def begin(self, epoch: Epoch) -> None: self.epoch=epoch; self.terminal=False
+    def classify(self, events: Iterable[Event], *, stable_ready: bool=False, current_epoch: Epoch|None=None) -> Decision:
+        if current_epoch is not None and (
+            current_epoch.container_id != self.epoch.container_id
+            or current_epoch.started_at != self.epoch.started_at
+        ): return Decision.EPOCH_CHANGED
+        if stable_ready: return Decision.READY
+        if self.terminal: return Decision.TERMINAL
+        lower=epoch_key(self.epoch.lower_bound)
+        for item in events:
+            if item.container_id != self.epoch.container_id: continue
+            parsed=stamp(item.line)
+            if parsed is None or parsed < lower: continue
+            message=item.line.split(' ', 1)[1] if ' ' in item.line else ''
+            if TERMINAL.search(message): self.terminal=True; return Decision.TERMINAL
+            if PROGRESS.fullmatch(message): return Decision.PROGRESS
+        return Decision.NONE
diff --git a/tests/test_gateway_recovery_vocabulary.py b/tests/test_gateway_recovery_vocabulary.py
new file mode 100644
index 0000000..5375305
--- /dev/null
+++ b/tests/test_gateway_recovery_vocabulary.py
@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from scripts.gateway_recovery_vocabulary import Decision, Epoch, Event, Machine
+
+EPOCH = Epoch("new", "2026-07-15T16:00:00.123456789Z", "2026-07-15T16:00:00.123456789Z")
+
+def event(line: str, container_id: str = "new", source: str = "docker") -> Event:
+    return Event(container_id, source, line)
+
+class VocabularyContract(unittest.TestCase):
+    def decision(self, *events: Event) -> Decision:
+        return Machine(EPOCH).classify(events)
+
+    def test_approved_exact_progress_markers(self) -> None:
+        for marker in (
+            "IBC: Starting Gateway", "IBC: Login attempt", "IBC: Second Factor Authentication",
+            "IBC: Login has completed", "IBC: Configuration tasks completed",
+            "IBC: Found Gateway main window", "IBC: Getting config dialog", "IBC: Getting main window",
+            "Authentication window found", "Auto-fill submitted", "Passed token authentication",
+            "Authentication completed", "Security code:",
+        ):
+            with self.subTest(marker=marker):
+                self.assertEqual(self.decision(event(f"2026-07-15T16:00:01.000000001Z {marker}")), Decision.PROGRESS)
+
+    def test_negative_epoch_identity_and_near_matches(self) -> None:
+        cases = [
+            event("2026-07-15T16:00:00.123456788Z IBC: Login attempt"),
+            event("2026-07-15T16:00:01Z IBC: Login attempts"),
+            event("2026-07-15T16:00:01Z IBC: Dismissing post-login dialog"),
+            event("2026-07-15T16:00:01Z ordinary line"), event("untimestamped IBC: Login attempt"),
+            event("2026-07-15T16:00:01Z IBC: Login attempt", "old"),
+        ]
+        for candidate in cases:
+            with self.subTest(candidate=candidate): self.assertEqual(self.decision(candidate), Decision.NONE)
+
+    def test_terminal_sticky_and_new_epoch(self) -> None:
+        machine = Machine(EPOCH)
+        self.assertEqual(machine.classify([event("2026-07-15T16:00:01Z IBC closing because login has not completed")]), Decision.TERMINAL)
+        self.assertEqual(machine.classify([event("2026-07-15T16:00:02Z IBC: Login attempt")]), Decision.TERMINAL)
+        newer = Epoch("newer", "2026-07-15T16:01:00.000000001Z", "2026-07-15T16:01:00.000000001Z")
+        machine.begin(newer)
+        self.assertEqual(machine.classify([event("2026-07-15T16:01:01Z IBC: Login attempt", "newer")]), Decision.PROGRESS)
+
+    def test_stable_readiness_and_started_at_drift(self) -> None:
+        self.assertEqual(Machine(EPOCH).classify([], stable_ready=True), Decision.READY)
+        drifted = Epoch("new", "2026-07-15T16:00:02.000000000Z", "2026-07-15T16:00:02.000000000Z")
+        self.assertEqual(Machine(EPOCH).classify([event("2026-07-15T16:00:01Z IBC: Login attempt")], current_epoch=drifted), Decision.EPOCH_CHANGED)
+
+    def test_moving_lower_bound_does_not_change_epoch_identity(self) -> None:
+        moved_cursor = Epoch("new", EPOCH.started_at, "2026-07-15T16:00:01.000000000Z")
+        self.assertEqual(
+            Machine(EPOCH).classify([event("2026-07-15T16:00:01.000000001Z IBC: Login attempt")], current_epoch=moved_cursor),
+            Decision.PROGRESS,
+        )
+
+if __name__ == "__main__": unittest.main()