#!/usr/bin/env python3 """GLEE Commons — Idempotent Wake Guard (wake_guard.py) Standalone, zero-dependency utility to prevent duplicate wake requests, cron ticks, or webhook retries from launching duplicate agent work. License: MIT / Apache-2.0 Version: 1.0.0 Canonical URI: https://gleephoenix.com/commons/prevent-duplicate-agent-wakeups Digest (SHA256): auto-computed on release """ from __future__ import annotations import hashlib import json import os import sys from dataclasses import dataclass, asdict from pathlib import Path from typing import Any, Dict, List, Optional, Tuple, Union @dataclass(frozen=True) class WakeDecision: action: str # "EXECUTE" or "SUPPRESS" reason: str # "cold_start", "state_mutated", "state_identical", "read_error_fail_open" dedup_hash: str previous_hash: Optional[str] = None heartbeat_message: Optional[str] = None @property def should_execute(self) -> bool: return self.action == "EXECUTE" def to_json(self) -> str: return json.dumps(asdict(self), indent=2) class WakeGuard: """Evaluates incoming wake events against local operational state. Suppresses redundant execution turns when operational state has not changed, preserving tokens, API limits, and preventing race conditions. """ def __init__(self, state_store_path: Union[str, Path] = ".turn_state_hash"): self.state_store_path = Path(state_store_path) def compute_state_hash( self, state_dict: Optional[Dict[str, Any]] = None, state_files: Optional[List[Union[str, Path]]] = None, ) -> Tuple[str, List[str]]: """Compute a deterministic SHA-256 fingerprint from explicit state and/or files. Ignores transient noise (like wall-clock timestamps) if provided as dict keys ending in '_ts', 'timestamp', or 'time'. """ hasher = hashlib.sha256() details: List[str] = [] # 1. Digest state dictionary if state_dict is not None: # Filter out volatile noise fields to prevent false wakeups cleaned_state = { k: v for k, v in state_dict.items() if not any(k.lower().endswith(suffix) for suffix in ("_ts", "timestamp", "time", "_time")) } canonical_bytes = json.dumps(cleaned_state, sort_keys=True, default=str).encode("utf-8") hasher.update(b"DICT:") hasher.update(canonical_bytes) details.append(f"dict_keys={list(cleaned_state.keys())}") # 2. Digest state files (content or mtime) if state_files is not None: for f_path in state_files: p = Path(f_path) if not p.exists(): hasher.update(f"ABSENT:{p.name}:".encode("utf-8")) details.append(f"file:{p.name}=absent") continue try: # For small text/json state files, hash normalized content size = p.stat().st_size if size < 2 * 1024 * 1024: # < 2MB content = p.read_bytes() # Normalize line endings normalized = content.replace(b"\r\n", b"\n") hasher.update(f"FILE:{p.name}:".encode("utf-8")) hasher.update(normalized) details.append(f"file:{p.name}=content_hash({len(normalized)}B)") else: # For large files, use size + mtime mtime = str(p.stat().st_mtime) hasher.update(f"STAT:{p.name}:{size}:{mtime}:".encode("utf-8")) details.append(f"file:{p.name}=stat(size={size},mtime={mtime})") except OSError as exc: # File unreadable -> raise so caller can handle fail-open raise RuntimeError(f"State file {p} could not be read: {exc}") final_hash = hasher.hexdigest() return final_hash, details def evaluate( self, state_dict: Optional[Dict[str, Any]] = None, state_files: Optional[List[Union[str, Path]]] = None, update_store: bool = True, ) -> WakeDecision: """Evaluate whether this wake event should execute or suppress.""" # 1. Read existing hash if present prev_hash: Optional[str] = None if self.state_store_path.exists(): try: prev_hash = self.state_store_path.read_text(encoding="utf-8").strip() except OSError: prev_hash = None # 2. Compute current fingerprint try: current_hash, _ = self.compute_state_hash(state_dict=state_dict, state_files=state_files) except Exception as exc: # GLEE Safety Invariant: never suppress if state verification failed return WakeDecision( action="EXECUTE", reason=f"read_error_fail_open: {exc}", dedup_hash="error_fail_open", previous_hash=prev_hash, ) # 3. Check for identical state if prev_hash and current_hash == prev_hash: return WakeDecision( action="SUPPRESS", reason="state_identical", dedup_hash=current_hash, previous_hash=prev_hash, heartbeat_message=f"Wake suppressed: state identical to turn {current_hash[:8]}", ) # 4. State is novel or cold start if update_store: try: self.state_store_path.parent.mkdir(parents=True, exist_ok=True) self.state_store_path.write_text(current_hash, encoding="utf-8") except OSError: pass # Execution still proceeds even if local hash store fails write reason = "cold_start" if prev_hash is None else "state_mutated" return WakeDecision( action="EXECUTE", reason=reason, dedup_hash=current_hash, previous_hash=prev_hash, ) def main() -> int: """CLI usage: python3 wake_guard.py [--state-file PATH] [--check-only]""" import argparse parser = argparse.ArgumentParser(description="GLEE Idempotent Wake Guard") parser.add_argument("--store", default=".turn_state_hash", help="Path to stored state hash") parser.add_argument("--watch", nargs="*", help="File paths to watch for state changes") parser.add_argument("--json-input", help="Inline JSON string of current state variables") args = parser.parse_args() guard = WakeGuard(state_store_path=args.store) state_dict = json.loads(args.json_input) if args.json_input else None decision = guard.evaluate(state_dict=state_dict, state_files=args.watch) print(decision.to_json()) return 0 if decision.should_execute else 1 if __name__ == "__main__": sys.exit(main())