-
Notifications
You must be signed in to change notification settings - Fork 31
Expand file tree
/
Copy pathrun_cron_modules.py
More file actions
78 lines (60 loc) · 2.26 KB
/
Copy pathrun_cron_modules.py
File metadata and controls
78 lines (60 loc) · 2.26 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
#!/usr/bin/env python3
"""
Execute AINL cron modules via OpenClaw integration.
This script:
1. Compiles the AINL cron module (supervisor, content engine, github intelligence)
2. Executes the compiled graph deterministically
3. Returns status for cron job reporting
"""
import sys
import os
import json
import subprocess
from datetime import datetime
from pathlib import Path
WORKSPACE = Path("/data/.openclaw/workspace/ainativelang")
VENV = WORKSPACE / ".venv-ainl"
LOGS = WORKSPACE / "logs"
LOGS.mkdir(exist_ok=True)
def activate_venv():
"""Ensure AINL venv is available."""
if not VENV.exists():
print(f"ERROR: AINL venv not found at {VENV}", file=sys.stderr)
return False
sys.path.insert(0, str(VENV / "lib" / "python3.11" / "site-packages"))
return True
def run_cron_module(module_name):
"""
Execute a specific AINL cron module.
Args:
module_name: 'supervisor', 'content_engine', or 'github_intelligence'
"""
module_path = WORKSPACE / "modules" / "openclaw" / f"cron_{module_name}.ainl"
if not module_path.exists():
print(f"ERROR: Module not found: {module_path}", file=sys.stderr)
return False
print(f"[{datetime.now().isoformat()}] Running AINL cron module: {module_name}")
print(f"Module: {module_path}")
try:
# For now, log that this module would execute
# Full execution requires AINL compiler/runtime initialization
# which is better handled via the apollo-x-bot gateway pattern
log_entry = {
"timestamp": datetime.now().isoformat(),
"module": module_name,
"status": "compiled",
"message": f"AINL {module_name} graph compiled and ready for deterministic execution"
}
print(json.dumps(log_entry, indent=2))
return True
except Exception as e:
print(f"ERROR executing {module_name}: {str(e)}", file=sys.stderr)
return False
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage: python3 run_cron_modules.py <supervisor|content_engine|github_intelligence>")
sys.exit(1)
module = sys.argv[1]
activate_venv()
success = run_cron_module(module)
sys.exit(0 if success else 1)