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
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117 | #!/usr/bin/env python3
"""
Mansu Musa weekly cadence — scheduled forensic revenue audit for the Chairman.
Watches: the weekly MANSU_MUSA audit (persona MANSU_MUSA, MUSA_TASKS["audit"]) —
the same run the /musa audit command triggers, on a timer instead of a tap.
Schedule: once weekly via a LaunchAgent (e.g. com.chairman.musa, Monday 09:00),
RunAtLoad false, no KeepAlive — one fire per week, exits when done.
Cost: $0 of its own. It triggers the SAME paid audit /musa audit runs; but every
failure mode here costs $0 — an empty Anthropic balance or a tripped budget
cutoff becomes a clean, throttled skip, never a launchd crash-loop and never a
partial paid call.
Delivery: run_persona() files an agent_reports row (type "audit") + task-ledger
rows to Supabase but does NOT touch Telegram; it returns the report text. This
cadence sends that text to 19Keys via tb.send_text so the scheduled audit reaches
the phone exactly like /musa audit does — no double-send (run_persona sends
nothing itself).
Import note: `import telegram_bridge` runs the bridge's top-level env loading and
api-metering wrap only; agent_loop / any API / any Telegram call sits behind the
module's __main__ guard, so importing it is side-effect-safe and cost-free.
"""
import os
import sys
import traceback
from datetime import datetime, timezone
from pathlib import Path
HOME = Path("/Users/19keys/sovereign-empire-sweep/chairman-agent")
LOG_FILE = HOME / "logs" / "musa-cadence.log"
sys.path.insert(0, str(HOME))
import telegram_bridge as tb # noqa: E402 (top-level = env load + metering wrap; no API/TG call)
import credit_guard # noqa: E402
JOB = "musa"
HEADER = "MANSU MUSA — weekly audit (scheduled)\n\n"
def _log(line):
ts = datetime.now(timezone.utc).isoformat()
try:
LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
with LOG_FILE.open("a") as f:
f.write(ts + " " + line + "\n")
except Exception:
pass
def _dry():
return os.environ.get("LOOPS_DRY") == "1"
def _alert(text):
"""Single notification path — honors LOOPS_DRY, never inlines Telegram."""
if _dry():
print("[DRY] " + JOB + ": " + text)
else:
tb.send_text(tb.USER_ID, text)
def _skip(reason, notice, detail=""):
"""Record a throttled skip, notify at most once per 24h, log, exit 0."""
credit_guard.record_skip(JOB)
if credit_guard.should_notify(JOB):
_alert(notice)
credit_guard.mark_notified(JOB)
print("skip: " + reason)
_log("skip: " + reason + ((" | " + detail) if detail else ""))
sys.exit(0)
def main():
# Respect the bridge's budget cutoff first — the same gate /musa honors.
if tb.cost_guard.is_cutoff_active():
_skip(
"budget cutoff active",
"Musa weekly audit skipped — budget cutoff active (80% monthly "
"cap reached). /unlock to resume paid APIs.",
)
try:
result = tb.run_persona(
"MANSU_MUSA", tb.MUSA_TASKS["audit"], report_type="audit")
except Exception as e:
if credit_guard.is_credit_error(e):
_skip(
"API credits exhausted",
"Musa weekly audit skipped — API credits exhausted (owner "
"action: add credits at console.anthropic.com).",
detail=str(e),
)
# Any other failure: stderr traceback + one log line, never crash-loop.
traceback.print_exc()
_log("fail: " + str(e))
sys.exit(0)
# run_persona filed the Supabase report; deliver the text to the phone here.
if _dry():
print("[DRY] " + JOB + ": " + HEADER + result)
else:
tb.send_text(tb.USER_ID, HEADER + result)
print("sent: weekly audit delivered")
_log("sent: weekly audit delivered")
sys.exit(0)
if __name__ == "__main__":
try:
main()
except SystemExit:
raise
except Exception as exc: # belt-and-suspenders: launchd must never crash-loop
traceback.print_exc()
_log("fatal: " + str(exc))
sys.exit(0)
|