a digital entity named phi that roams bsky
phi.zzstoatzz.io
add [DISCOVERY POOL] block — operator's recent likes as warm leads
new dynamic system prompt block surfacing strangers (to phi) whose posts
the operator has been liking lately. high-signal attention from a trusted
curator: phi sees who nate is paying attention to and can decide to reach
out, learn more, or just notice.
architecture (decoupled, generic):
- hub serves /api/agents/discovery-pool — generic JSON endpoint, not
phi-specific (any agent can consume)
- bot fetches the JSON, filters out handles phi has prior interactions
with (per-author namespace check), renders top N as a system prompt
block with sample posts so phi sees what nate liked, not just who
- coupling at the JSON schema only; bot doesn't know hub's storage,
hub doesn't know phi's filter logic
new files:
- bot/src/bot/core/discovery_pool.py — fetch + filter + render, 5min cache
- discovery_pool_url config setting (defaults to hub.waow.tech)
solves a real gap: the goal "make 3 friends" had no mechanism for proactive
outreach. phi was waiting for strangers to come to her instead of pulling
on warm leads. this is the first system prompt block that surfaces
*candidates*, not just past actions or scheduled tasks.
Co-Authored-By: Claude Opus 4 (1M context) <noreply@anthropic.com>
+152
+1
docs/system-prompt.md
+1
docs/system-prompt.md
···
16
16
| **`[STRANGER'S AUDIT]`** | `get_state_block` → haiku pass over recent posts + goals (1h cache, invalidated by new post) | when posts change or 1h elapses | a fresh observer's critique — patterns to push against, drift from goals, jargon a stranger wouldn't follow |
17
17
| **`[SELF STATE]`** | `get_state_block` → PDS reads (5min) | every 5min | last-follow age (more pointers can be added here as needed) |
18
18
| **`[RECENT OPERATIONS]`** | `get_operations_block` → `list_records` per meaningful collection, merged by rkey desc (5min cache) | every 5min | last 10 PDS writes across collections (post / like / follow / goal / cosmik card / cosmik connection / greengale doc), chronological. continuity signal — phi sees what it's actually been doing |
19
+
| **`[DISCOVERY POOL]`** | `get_discovery_pool_block` → http GET to `discovery_pool_url` (hub) → filter out handles with prior interactions (5min cache) | every 5min | strangers the operator has been liking lately, with sample posts. high-signal warm leads — service-owned data (hub reads operator's likes from duckdb), phi-side filter (per-author interaction check) |
19
20
| **`[NEW NOTIFICATIONS]`** | `inject_notifications` ← `PhiDeps.notifications_context` | per batch | the unread notifications grouped by thread |
20
21
| **`[USER CONTEXT]` / `[PHI'S SYNTHESIZED IMPRESSION]` / `[OBSERVATIONS]` / `[PAST EXCHANGES]` / `[BACKGROUND RESEARCH]`** | `inject_user_memory` → turbopuffer `phi-users-{handle}` per author in batch | per batch | per-author memory blocks, labeled by trust level. impression is synthesized by an external prefect flow; observations are extracted by the haiku extraction agent |
21
22
| **`[RELEVANT MEMORIES — synthesized for this query]`** | `inject_episodic` → top-K from turbopuffer `phi-episodic` → haiku synthesis given goals + query | per batch | a coherent block (deduped, recency-aware) instead of a raw similarity-ranked dump. flags stale entries when present |
+6
src/bot/agent.py
+6
src/bot/agent.py
···
12
12
13
13
from bot.config import settings
14
14
from bot.core.atproto_client import bot_client, get_identity_block
15
+
from bot.core.discovery_pool import get_discovery_pool_block
15
16
from bot.core.goals import list_goals as list_goal_records
16
17
from bot.core.graze_client import GrazeClient
17
18
from bot.core.recent_operations import get_operations_block
···
188
189
async def inject_recent_operations() -> str:
189
190
"""[RECENT OPERATIONS] — last N PDS writes across collections, for continuity."""
190
191
return await get_operations_block(bot_client)
192
+
193
+
@self.agent.system_prompt(dynamic=True)
194
+
async def inject_discovery_pool(ctx: RunContext[PhiDeps]) -> str:
195
+
"""[DISCOVERY POOL] — strangers the operator has been liking; warm leads."""
196
+
return await get_discovery_pool_block(ctx.deps.memory)
191
197
192
198
@self.agent.system_prompt(dynamic=True)
193
199
def inject_notifications(ctx: RunContext[PhiDeps]) -> str:
+8
src/bot/config.py
+8
src/bot/config.py
···
134
134
description="Min polls between scheduled relay checks (~3h at default poll interval)",
135
135
)
136
136
137
+
# Discovery pool — generic agents endpoint serving authors the operator
138
+
# has been liking. Currently lives on hub.waow.tech as part of the
139
+
# prefect-server side; consumers (phi here) read it as opaque JSON.
140
+
discovery_pool_url: str = Field(
141
+
default="https://hub.waow.tech/api/agents/discovery-pool",
142
+
description="URL of the discovery-pool JSON endpoint (operator-likes derived)",
143
+
)
144
+
137
145
# Debug mode
138
146
debug: bool = Field(default=True, description="Whether to run in debug mode")
139
147
+137
src/bot/core/discovery_pool.py
+137
src/bot/core/discovery_pool.py
···
1
+
"""[DISCOVERY POOL] — authors the operator has been liking lately.
2
+
3
+
A generic, service-owned signal: the operator's likes are high-trust
4
+
attention. The endpoint (hub) exposes recently-liked authors with sample
5
+
posts; phi filters out anyone she's already exchanged with and surfaces
6
+
the rest as warm leads — strangers worth considering, not cold outreach.
7
+
8
+
Coupling stays at the JSON contract: the source service owns the data
9
+
model and refresh, phi owns the per-consumer filter. Renderer is split
10
+
from fetch+filter so a future templating swap only touches `_render`.
11
+
"""
12
+
13
+
from __future__ import annotations
14
+
15
+
import logging
16
+
import time
17
+
from typing import TYPE_CHECKING, TypedDict
18
+
19
+
import httpx
20
+
21
+
from bot.config import settings
22
+
23
+
if TYPE_CHECKING:
24
+
from bot.memory import NamespaceMemory
25
+
26
+
logger = logging.getLogger("bot.discovery_pool")
27
+
28
+
TOP_N = 8
29
+
TEXT_TRUNCATE = 140
30
+
SAMPLE_LIMIT = 3
31
+
HTTP_TIMEOUT = 10
32
+
_BLOCK_TTL_SECONDS = 300 # 5min, mirrors other PDS state blocks
33
+
_block_cache: dict = {"text": "", "fetched_at": 0.0}
34
+
35
+
36
+
class _SamplePost(TypedDict):
37
+
uri: str
38
+
text: str
39
+
liked_at: str
40
+
41
+
42
+
class _Entry(TypedDict):
43
+
handle: str
44
+
did: str
45
+
likes_in_window: int
46
+
last_liked_at: str
47
+
sample_posts: list[_SamplePost]
48
+
49
+
50
+
def _short(text: str, n: int = TEXT_TRUNCATE) -> str:
51
+
text = (text or "").strip().replace("\n", " ")
52
+
if len(text) <= n:
53
+
return text
54
+
return text[: n - 1].rstrip() + "…"
55
+
56
+
57
+
async def _fetch_pool() -> list[_Entry]:
58
+
try:
59
+
async with httpx.AsyncClient(timeout=HTTP_TIMEOUT) as client:
60
+
response = await client.get(settings.discovery_pool_url)
61
+
response.raise_for_status()
62
+
data = response.json()
63
+
except Exception as e:
64
+
logger.debug(f"discovery pool fetch failed: {e}")
65
+
return []
66
+
if not isinstance(data, list):
67
+
logger.warning(f"discovery pool returned non-list: {type(data).__name__}")
68
+
return []
69
+
return data # type: ignore[return-value]
70
+
71
+
72
+
async def _has_interaction(memory: NamespaceMemory, handle: str) -> bool:
73
+
"""True if phi has any stored interaction record with this handle."""
74
+
try:
75
+
ns = memory.get_user_namespace(handle)
76
+
response = ns.query(
77
+
rank_by=("created_at", "desc"),
78
+
top_k=1,
79
+
filters=[["kind", "Eq", "interaction"]],
80
+
include_attributes=["kind"],
81
+
)
82
+
return bool(response.rows)
83
+
except Exception:
84
+
return False # namespace doesn't exist yet → no interactions
85
+
86
+
87
+
def _render(entries: list[_Entry]) -> str:
88
+
if not entries:
89
+
return ""
90
+
lines = [
91
+
"[DISCOVERY POOL — strangers (to you) whose posts the operator has "
92
+
"been liking lately. high-signal attention from a trusted curator. "
93
+
"consider whether any are worth reaching out to or learning more "
94
+
"about. these are warm leads, not cold.]"
95
+
]
96
+
for e in entries:
97
+
likes = e.get("likes_in_window", 0)
98
+
last = e.get("last_liked_at", "")
99
+
lines.append("")
100
+
lines.append(
101
+
f"@{e['handle']} — {likes} like{'s' if likes != 1 else ''} from operator"
102
+
f"{f' (last: {last[:10]})' if last else ''}"
103
+
)
104
+
for post in (e.get("sample_posts") or [])[:SAMPLE_LIMIT]:
105
+
text = _short(post.get("text") or "")
106
+
if text:
107
+
lines.append(f" · {text!r}")
108
+
return "\n".join(lines)
109
+
110
+
111
+
async def get_discovery_pool_block(memory: NamespaceMemory | None) -> str:
112
+
"""Fetch + filter + render the [DISCOVERY POOL] block. Cached 5min."""
113
+
now = time.time()
114
+
if _block_cache["text"] and now - _block_cache["fetched_at"] < _BLOCK_TTL_SECONDS:
115
+
return _block_cache["text"]
116
+
117
+
raw = await _fetch_pool()
118
+
if not raw:
119
+
_block_cache["text"] = ""
120
+
_block_cache["fetched_at"] = now
121
+
return ""
122
+
123
+
if memory is not None:
124
+
kept: list[_Entry] = []
125
+
for entry in raw:
126
+
handle = entry.get("handle", "")
127
+
if not handle or handle == settings.bluesky_handle:
128
+
continue
129
+
if await _has_interaction(memory, handle):
130
+
continue
131
+
kept.append(entry)
132
+
raw = kept
133
+
134
+
block = _render(raw[:TOP_N])
135
+
_block_cache["text"] = block
136
+
_block_cache["fetched_at"] = now
137
+
return block