a digital entity named phi that roams bsky
phi.zzstoatzz.io
move exploration scripts to sandbox
register_webhook.py: one-off wisp.place webhook tool
test_memory_smoke.py: integration tests needing live credentials
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
+94
+80
sandbox/register_webhook.py
+80
sandbox/register_webhook.py
···
1
+
"""Register a wisp.place webhook for mention backlinks.
2
+
3
+
Usage:
4
+
uv run python scripts/register_webhook.py <webhook-url>
5
+
uv run python scripts/register_webhook.py --list
6
+
uv run python scripts/register_webhook.py --delete <rkey>
7
+
"""
8
+
9
+
import sys
10
+
from datetime import datetime, timezone
11
+
12
+
from atproto import Client
13
+
14
+
from bot.config import settings
15
+
16
+
17
+
def main():
18
+
client = Client(base_url=settings.bluesky_service)
19
+
client.login(settings.bluesky_handle, settings.bluesky_password)
20
+
did = client.me.did
21
+
print(f"authenticated as {settings.bluesky_handle} ({did})")
22
+
23
+
if len(sys.argv) > 1 and sys.argv[1] == "--list":
24
+
result = client.com.atproto.repo.list_records(
25
+
params={"repo": did, "collection": "place.wisp.v2.wh", "limit": 50}
26
+
)
27
+
if not result.records:
28
+
print("no webhooks registered")
29
+
return
30
+
for rec in result.records:
31
+
rkey = rec.uri.split("/")[-1]
32
+
val = rec.value if isinstance(rec.value, dict) else vars(rec.value) if hasattr(rec.value, '__dict__') else str(rec.value)
33
+
print(f" [{rkey}] {val}")
34
+
return
35
+
36
+
if len(sys.argv) > 2 and sys.argv[1] == "--delete":
37
+
rkey = sys.argv[2]
38
+
client.com.atproto.repo.delete_record(
39
+
data={"repo": did, "collection": "place.wisp.v2.wh", "rkey": rkey}
40
+
)
41
+
print(f"deleted webhook {rkey}")
42
+
return
43
+
44
+
url = sys.argv[1] if len(sys.argv) > 1 else None
45
+
if not url:
46
+
print(__doc__)
47
+
sys.exit(1)
48
+
49
+
record = {
50
+
"$type": "place.wisp.v2.wh",
51
+
"scope": {
52
+
"aturi": f"at://{did}",
53
+
"backlinks": True,
54
+
},
55
+
"url": url,
56
+
"events": ["create"],
57
+
"enabled": True,
58
+
"createdAt": datetime.now(timezone.utc).isoformat(),
59
+
}
60
+
61
+
if settings.wisp_webhook_secret:
62
+
record["secret"] = settings.wisp_webhook_secret
63
+
print("using HMAC secret from WISP_WEBHOOK_SECRET")
64
+
65
+
result = client.com.atproto.repo.create_record(
66
+
data={
67
+
"repo": did,
68
+
"collection": "place.wisp.v2.wh",
69
+
"record": record,
70
+
}
71
+
)
72
+
73
+
print(f"webhook registered: {result.uri}")
74
+
print(f" url: {url}")
75
+
print(f" scope: at://{did} (backlinks)")
76
+
print(f" events: create")
77
+
78
+
79
+
if __name__ == "__main__":
80
+
main()
+14
tests/test_memory_smoke.py
sandbox/test_memory_smoke.py
+14
tests/test_memory_smoke.py
sandbox/test_memory_smoke.py
···
43
43
results = await memory.search("zzstoatzzdevlog.bsky.social", "hello", top_k=3)
44
44
print(f"\n--- search results ---\n{results}\n---")
45
45
assert isinstance(results, list)
46
+
47
+
48
+
async def test_search_unified(memory):
49
+
"""search_unified returns a list from both user + episodic namespaces."""
50
+
results = await memory.search_unified("zzstoatzzdevlog.bsky.social", "hello", top_k=3)
51
+
print(f"\n--- unified results ---\n{results}\n---")
52
+
assert isinstance(results, list)
53
+
54
+
55
+
async def test_search_unified_missing_user(memory):
56
+
"""search_unified works when user namespace doesn't exist (episodic-only)."""
57
+
results = await memory.search_unified("nonexistent-user-12345.example", "hello", top_k=3)
58
+
print(f"\n--- unified (missing user) ---\n{results}\n---")
59
+
assert isinstance(results, list)