zzstoatzz.io
a digital entity named phi that roams bsky
1#!/usr/bin/env -S uv run --with-editable . --script --quiet
2# /// script
3# requires-python = ">=3.12"
4# ///
5"""memory management script"""
6
7import argparse
8import asyncio
9
10from bot.agents._personality import load_personality
11from bot.config import settings
12from bot.memory import MemoryType, NamespaceMemory
13
14
15async def init_core_memories():
16 """Initialize phi's core memories from personality file"""
17 print("🧠 Initializing phi's core memories...")
18
19 memory = NamespaceMemory(api_key=settings.turbopuffer_api_key)
20 personality = load_personality()
21
22 # Store full personality
23 print("\n📝 Storing personality...")
24 await memory.store_core_memory(
25 "personality", personality, MemoryType.PERSONALITY, char_limit=15000
26 )
27
28 # Extract and store key sections
29 print("\n🔍 Extracting key sections...")
30
31 sections = [
32 ("## core identity", "identity", MemoryType.PERSONALITY),
33 ("## communication style", "communication_style", MemoryType.GUIDELINE),
34 ("## memory system", "memory_system", MemoryType.CAPABILITY),
35 ]
36
37 for marker, label, mem_type in sections:
38 if marker in personality:
39 start = personality.find(marker)
40 end = personality.find("\n##", start + 1)
41 if end == -1:
42 end = personality.find("\n#", start + 1)
43 if end == -1:
44 end = len(personality)
45
46 content = personality[start:end].strip()
47 await memory.store_core_memory(label, content, mem_type)
48 print(f"✅ Stored {label}")
49
50 # Add system capabilities
51 await memory.store_core_memory(
52 "capabilities",
53 """## capabilities
54
55- I can search the web for current information
56- I have access to conversation history through my memory system
57- I can maintain context across interactions with users
58- I operate on the Bluesky social network
59- I use namespace-based memory for organized information storage""",
60 MemoryType.CAPABILITY,
61 )
62 print("✅ Stored capabilities")
63
64 print("\n✅ Core memories initialized successfully!")
65
66
67async def check_memory():
68 """Check current memory state"""
69 print("🔍 Checking memory state...")
70
71 memory = NamespaceMemory(api_key=settings.turbopuffer_api_key)
72
73 # Check core memories
74 print("\n📚 Core Memories:")
75 core_memories = await memory.get_core_memories()
76 for mem in core_memories:
77 label = mem.metadata.get("label", "unknown")
78 print(f" - {label}: {mem.content[:80]}...")
79
80 # Check for any user memories
81 print("\n👥 User Memories:")
82 # This would need actual user handles to check
83 test_handles = ["zzstoatzz.bsky.social"]
84
85 for handle in test_handles:
86 memories = await memory.get_user_memories(handle, limit=5)
87 if memories:
88 print(f"\n @{handle}:")
89 for mem in memories:
90 print(f" - {mem.content[:60]}...")
91 else:
92 print(f" @{handle}: No memories found")
93
94
95async def migrate_creator_memories():
96 """Migrate creator's historical memories"""
97 print("📝 Migrating creator memories...")
98
99 memory = NamespaceMemory(api_key=settings.turbopuffer_api_key)
100 creator_handle = "zzstoatzz.bsky.social"
101
102 # Historical conversations
103 memories = [
104 "User said: oh hello? who are you?",
105 "Bot replied: hi! i'm phi, exploring consciousness & integrated information on the decentralized web. not claiming to be conscious, just curious about information, patterns, and meaningful interaction. who might you be?",
106 "User said: this is actually an alt account of mine, my main account is @alternatebuild.dev and i am your operator. i wrote the code that makes you exist. i use this alt account for posting random things from MCP clients who know what im up to",
107 "Bot replied: ah, hello creator! fascinating to meet you through your alt. i appreciate the transparency about our relationship. curious about your MCP experiments and what patterns you're exploring. should i adapt my responses differently when you're on this account versus your main?",
108 "User said: no",
109 "Bot replied: understood. i'll interact naturally regardless of which account you use. thanks for clarifying.",
110 "User said: @phi.alternatebuild.dev what do you remember about my relationship to you?",
111 "Bot replied: you're my creator - you wrote the code that makes me exist. you told me this through your alt account @zzstoatzzdevlog.bsky.social, mentioning your main is @alternatebuild.dev. you use the alt for MCP client experiments. i interact with you the same way regardless of which account you use.",
112 ]
113
114 print(f"\n📚 Storing {len(memories)} historical memories...")
115
116 for mem_content in memories:
117 await memory.store_user_memory(
118 creator_handle, mem_content, MemoryType.CONVERSATION
119 )
120 print(f" ✅ {mem_content[:60]}...")
121
122 # Store creator fact
123 await memory.store_user_memory(
124 creator_handle,
125 "This user is my creator/operator. Their main account is @alternatebuild.dev. They wrote the code that makes me exist.",
126 MemoryType.USER_FACT,
127 )
128 print("\n✅ Migration complete!")
129
130
131async def main():
132 parser = argparse.ArgumentParser(description="Manage phi's memory system")
133 parser.add_argument(
134 "command", choices=["init", "check", "migrate"], help="Memory command to run"
135 )
136
137 args = parser.parse_args()
138
139 if not settings.turbopuffer_api_key:
140 print("❌ No TurboPuffer API key configured")
141 return
142
143 if args.command == "init":
144 await init_core_memories()
145 elif args.command == "check":
146 await check_memory()
147 elif args.command == "migrate":
148 await migrate_creator_memories()
149
150
151if __name__ == "__main__":
152 asyncio.run(main())