Gateway Freeze-Thaw Recovery

Detect involuntary host-suspend gaps (laptop sleep, VM pause/migrate) and drive gateway recovery on resume

from praisonaiagents import Agent
from praisonai.bots import BotOS, Bot
from praisonaiagents.gateway import WallClockGapThawPolicy

agent = Agent(name="assistant", instructions="Be helpful.")

botos = BotOS(
    bots=[Bot("telegram", agent=agent)],
    thaw_policy=WallClockGapThawPolicy(),  # sensible defaults
)
botos.run()

The gateway detects when the host was frozen (laptop sleep, VM pause / snapshot / live-migrate) and drives recovery on thaw — restart channel sockets when idle, refresh presence, and coalesce missed cron fires.

Quick Start

1. Default — sensible thresholds

   from praisonaiagents import Agent
   from praisonai.bots import BotOS, Bot
   from praisonaiagents.gateway import WallClockGapThawPolicy
   
   agent = Agent(name="assistant", instructions="Help users")
   
   botos = BotOS(
       bots=[Bot("telegram", agent=agent)],
       thaw_policy=WallClockGapThawPolicy(),
   )
   botos.run()

   The gateway compares wall-clock and monotonic time on every run-loop tick. A wall-clock jump that runs far ahead of monotonic elapsed time is a freeze — recovery fires on the next tick after resume.

2. Tuned thresholds — slower loop, higher tolerance

   from praisonaiagents import Agent
   from praisonai.bots import BotOS, Bot
   from praisonaiagents.gateway import WallClockGapThawPolicy
   
   agent = Agent(name="assistant", instructions="Help users")
   
   thaw_policy = WallClockGapThawPolicy(
       tick_interval_s=30.0,
       gap_threshold_s=120.0,
   )
   
   botos = BotOS(
       bots=[Bot("telegram", agent=agent)],
       thaw_policy=thaw_policy,
   )
   botos.run()

   Match tick_interval_s to your run-loop cadence and set gap_threshold_s above your worst legitimate stall.

How It Works

sequenceDiagram
    participant Loop as RunLoop
    participant Policy as WallClockGapThawPolicy
    participant Decision as ThawDecision

    Loop->>Policy: observe(monotonic_now, wall_now)
    Note over Policy: First tick seeds baseline only
    Policy-->>Decision: suspended=False
    Note over Loop,Policy: ... host suspends ...
    Loop->>Policy: observe(monotonic_now, wall_now)
    Note over Policy: wall_delta − monotonic_delta > gap_threshold_s<br/>AND monotonic_delta < tick_interval_s
    Policy-->>Decision: suspended=True, gap_seconds≈divergence
    Decision-->>Loop: restart_transports=True, reconcile_schedule=True

On Linux CLOCK_MONOTONIC does not advance while the host is suspended, so wall-clock running far ahead of monotonic is a positive signature of a freeze the process could not otherwise observe.

Detection rule — a gap is reported only when both are true:

The second condition rejects the NTP / manual-clock-step false positive: if monotonic advanced a full tick or more, the process was alive and no recovery fires.

Choosing between Freeze-Thaw and Scale-to-Zero

Freeze-Thaw handles the involuntary suspend; Scale-to-Zero handles the intentional one.

Configuration Options

WallClockGapThawPolicy constructor

from praisonaiagents.gateway import WallClockGapThawPolicy

policy = WallClockGapThawPolicy(
    tick_interval_s=15.0,
    gap_threshold_s=60.0,
    enabled=True,
)

ThawDecision — result of every observe() call

@dataclass(frozen=True) — assigning to a field raises.

ThawPolicyProtocol — custom detectors

@runtime_checkable protocol with a single method. Implement your own detector and pass it wherever a ThawPolicyProtocol is accepted — isinstance(policy, ThawPolicyProtocol) succeeds without inheritance.

def observe(self, *, monotonic_now: float, wall_now: float) -> ThawDecision: ...

Imports

from praisonaiagents.gateway import (
    ThawDecision,
    ThawPolicyProtocol,
    WallClockGapThawPolicy,
)

Common Patterns

Disable in tests

from praisonaiagents.gateway import WallClockGapThawPolicy

policy = WallClockGapThawPolicy(enabled=False)
# observe() always returns ThawDecision(suspended=False), even for a 2400s wall jump

Use enabled=False to keep the wiring in place while turning detection off — no code removal.

Custom detector implementing ThawPolicyProtocol

from pathlib import Path
from praisonaiagents.gateway import ThawDecision, WallClockGapThawPolicy


class WakeCountThawPolicy:
    """Also consult /sys/power/wakeup_count on Linux for a second signal."""

    def __init__(self):
        self._base = WallClockGapThawPolicy()

    def observe(self, *, monotonic_now: float, wall_now: float) -> ThawDecision:
        decision = self._base.observe(
            monotonic_now=monotonic_now, wall_now=wall_now
        )
        if decision.suspended:
            wakeups = Path("/sys/power/wakeup_count").read_text().strip()
            print(f"Freeze confirmed by wakeup_count={wakeups}")
        return decision

Custom detectors must still ignore NTP steps — reuse the built-in guard or replicate it.

Log the observed freeze window

from praisonaiagents.gateway import WallClockGapThawPolicy

policy = WallClockGapThawPolicy()

decision = policy.observe(monotonic_now=mono, wall_now=wall)
if decision.suspended:
    print(f"Host was frozen for ~{decision.gap_seconds:.0f}s")

Read ThawDecision.gap_seconds in a hook to record how long the host was suspended.

Best Practices

Related