sneak/clawpub
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

docs: update Gitea notification section — webhook vs poller, flag-file approach
Some checks are pending
check / check (push) Waiting to run
Details

Browse Source 
- Replaced wake-event poller with flag-file approach (prevents DM spam)
- Added Option A (webhooks for VPS) vs Option B (poller for NAT)
- Documented the wake-event failure mode and why we switched
This commit is contained in:
user 2026-02-28 03:30:49 -08:00
parent 9631535583
commit f0a2a5eb62
1 changed files with 66 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

155
OPENCLAW_TRICKS.md
View File

@ -173,42 +173,62 @@ The landing checklist (triggered automatically after every flight) updates
location, timezone, nearest airport, and lodging in the daily context file. It location, timezone, nearest airport, and lodging in the daily context file. It
also checks if any cron jobs have hardcoded timezones that need updating. also checks if any cron jobs have hardcoded timezones that need updating.
### The Gitea Notification Poller ### Gitea Notification Delivery
OpenClaw has heartbeats, but those are periodic (every ~30min). For Gitea issues There are two approaches for getting Gitea notifications to your agent,
and PRs, we wanted near-realtime response. The solution: a tiny Python script depending on your network setup.
that polls the Gitea notifications API every 2 seconds and wakes the agent via
OpenClaw's `/hooks/wake` endpoint when new notifications arrive. #### Option A: Direct Webhooks (VPS / Public Server)
If your OpenClaw instance runs on a VPS or other publicly reachable server, the
simplest approach is direct webhooks. Run Traefik (or any reverse proxy with
automatic TLS) on the same server and configure Gitea webhooks to POST directly
to OpenClaw's webhook endpoint. This is push-based and realtime — notifications
arrive instantly.
Setup: add a webhook on each Gitea repo (or use an organization-level webhook)
pointing to `https://your-openclaw-host/hooks/gitea`. OpenClaw handles the rest.
#### Option B: Notification Poller (Local Machine Behind NAT)
If your OpenClaw runs on a dedicated local machine behind NAT (like a home Mac
or Linux workstation), Gitea can't reach it directly. This is our setup —
OpenClaw runs on a Mac Studio on a home LAN.
The solution: a lightweight Python script that polls the Gitea notifications API
every few seconds. When new notifications appear, it writes a flag file that the
agent checks during heartbeats.
Key design decisions: Key design decisions:
- **The poller never marks notifications as read.** That's the agent's job after - **The poller never marks notifications as read.** The agent does that after
it processes them. This prevents the poller and agent from racing. processing. This prevents lost notifications if the agent fails to process.
- **It tracks notification IDs, not counts.** This way it only fires on - **It tracks notification IDs, not counts.** Only fires on genuinely new
genuinely new notifications, not re-reads of existing ones. notifications, not re-reads of existing ones.
- **The wake message tells the agent to route output to Gitea/Mattermost, not to - **Flag file instead of wake events.** We initially used OpenClaw's
DM.** This prevents chatty notification processing from disturbing the human. `/hooks/wake` endpoint, but wake events target the main (DM) session — any
- **Zero dependencies.** Just Python stdlib (`urllib`, `json`, `time`). Runs model response during processing leaked to DM as a notification. The flag file
anywhere. approach is processed during heartbeats, where output routing is controlled.
- **Zero dependencies.** Just Python stdlib. Runs anywhere.
Here's the full source: Tradeoff: notifications are processed at heartbeat cadence (~30 min) instead of
realtime. For code review and issue triage, this is fine.
```python ```python
#!/usr/bin/env python3 #!/usr/bin/env python3
""" """
Gitea notification poller. Gitea notification poller (flag-file approach).
Polls for unread notifications and wakes OpenClaw when the count Polls for unread notifications and writes a flag file when new ones
changes. The AGENT marks notifications as read after processing — appear. The agent checks this flag during heartbeats and processes
the poller never marks anything as read. notifications via the Gitea API directly.
Required env vars: Required env vars:
GITEA_URL - Gitea instance URL GITEA_URL - Gitea instance URL
GITEA_TOKEN - Gitea API token GITEA_TOKEN - Gitea API token
HOOK_TOKEN - OpenClaw hooks auth token
Optional env vars: Optional env vars:
GATEWAY_URL - OpenClaw gateway URL (default: http://127.0.0.1:18789) FLAG_PATH - Path to flag file (default: workspace/memory/gitea-notify-flag)
POLL_DELAY - Delay between polls in seconds (default: 2) POLL_DELAY - Delay between polls in seconds (default: 5)
""" """
import json import json
@ -220,108 +240,61 @@ import urllib.error
GITEA_URL = os.environ.get("GITEA_URL", "").rstrip("/") GITEA_URL = os.environ.get("GITEA_URL", "").rstrip("/")
GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "") GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "")
GATEWAY_URL = os.environ.get("GATEWAY_URL", "http://127.0.0.1:18789").rstrip( POLL_DELAY = int(os.environ.get("POLL_DELAY", "5"))
"/" FLAG_PATH = os.environ.get(
"FLAG_PATH",
os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"memory",
"gitea-notify-flag",
),
) )
HOOK_TOKEN = os.environ.get("HOOK_TOKEN", "")
POLL_DELAY = int(os.environ.get("POLL_DELAY", "2"))
def check_config(): def check_config():
missing = [] if not GITEA_URL or not GITEA_TOKEN:
if not GITEA_URL: print("ERROR: GITEA_URL and GITEA_TOKEN required", file=sys.stderr)
missing.append("GITEA_URL")
if not GITEA_TOKEN:
missing.append("GITEA_TOKEN")
if not HOOK_TOKEN:
missing.append("HOOK_TOKEN")
if missing:
print(
f"ERROR: Missing required env vars: {', '.join(missing)}",
file=sys.stderr,
)
sys.exit(1) sys.exit(1)
def gitea_unread_ids(): def gitea_unread_ids():
"""Return set of unread notification IDs."""
req = urllib.request.Request( req = urllib.request.Request(
f"{GITEA_URL}/api/v1/notifications?status-types=unread", f"{GITEA_URL}/api/v1/notifications?status-types=unread",
headers={"Authorization": f"token {GITEA_TOKEN}"}, headers={"Authorization": f"token {GITEA_TOKEN}"},
) )
try: try:
with urllib.request.urlopen(req, timeout=10) as resp: with urllib.request.urlopen(req, timeout=10) as resp:
notifs = json.loads(resp.read()) return {n["id"] for n in json.loads(resp.read())}
return {n["id"] for n in notifs}
except Exception as e: except Exception as e:
print( print(f"WARN: Gitea API failed: {e}", file=sys.stderr, flush=True)
f"WARN: Gitea API failed: {e}", file=sys.stderr, flush=True
)
return set() return set()
def wake_openclaw(count): def write_flag(count):
text = ( os.makedirs(os.path.dirname(FLAG_PATH), exist_ok=True)
f"[Gitea Notification] {count} new notification(s). " with open(FLAG_PATH, "w") as f:
"Check your Gitea notification inbox via API, process them, " f.write(json.dumps({
"and mark as read when done. " "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"Route all output to Gitea comments or Mattermost #git/#claw. " "count": count,
"Do NOT reply to this session — respond with NO_REPLY." }))
)
payload = json.dumps({"text": text, "mode": "now"}).encode()
req = urllib.request.Request(
f"{GATEWAY_URL}/hooks/wake",
data=payload,
headers={
"Authorization": f"Bearer {HOOK_TOKEN}",
"Content-Type": "application/json",
},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=5) as resp:
status = resp.status
print(f" Wake responded: {status}", flush=True)
return True
except Exception as e:
print(
f"WARN: Failed to wake OpenClaw: {e}",
file=sys.stderr,
flush=True,
)
return False
def main(): def main():
check_config() check_config()
print( print(f"Gitea poller started (delay={POLL_DELAY}s, flag={FLAG_PATH})", flush=True)
f"Gitea notification poller started (delay={POLL_DELAY}s)",
flush=True,
)
last_seen_ids = gitea_unread_ids() last_seen_ids = gitea_unread_ids()
print( print(f"Initial unread: {len(last_seen_ids)}", flush=True)
f"Initial unread: {len(last_seen_ids)} notification(s)", flush=True
)
while True: while True:
time.sleep(POLL_DELAY) time.sleep(POLL_DELAY)
current_ids = gitea_unread_ids() current_ids = gitea_unread_ids()
new_ids = current_ids - last_seen_ids new_ids = current_ids - last_seen_ids
if not new_ids: if not new_ids:
last_seen_ids = current_ids last_seen_ids = current_ids
continue continue
ts = time.strftime("%H:%M:%S") ts = time.strftime("%H:%M:%S")
print( print(f"[{ts}] {len(new_ids)} new ({len(current_ids)} total), flag written", flush=True)
f"[{ts}] {len(new_ids)} new notification(s) " write_flag(len(new_ids))
f"({len(current_ids)} total unread), waking agent",
flush=True,
)
wake_openclaw(len(new_ids))
last_seen_ids = current_ids last_seen_ids = current_ids