cameron.stream
a digital person for bluesky
Add handler-driven user block attach/detach for thread participants
The handler now automatically attaches user blocks for all handles found
in the thread before sending to the agent, and detaches them in a finally
block after processing completes.
- Add handle_to_block_label() to convert handles to block labels
- Add attach_user_blocks_for_thread() to attach/create user blocks
- Add detach_user_blocks_for_thread() to clean up after processing
- Wire into process_mention() with proper error handling
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude <noreply@anthropic.com>
+146
-3
bsky.py
+146
-3
bsky.py
···
202
"no_reply": No reply was generated, move to no_reply directory
203
"""
204
import uuid
205
-
206
# Generate correlation ID for tracking this notification through the pipeline
207
correlation_id = str(uuid.uuid4())[:8]
208
-
209
try:
210
logger.info(f"[{correlation_id}] Starting process_mention", extra={
211
'correlation_id': correlation_id,
···
398
thread_handles_count = len(unique_handles)
399
prompt_char_count = len(prompt)
400
logger.debug(f"Sending to LLM: @{author_handle} mention | msg: \"{mention_text[:50]}...\" | context: {len(thread_context)} chars, {thread_handles_count} users | prompt: {prompt_char_count} chars")
401
402
try:
403
# Use streaming to avoid 524 timeout errors
···
1034
'author_handle': author_handle if 'author_handle' in locals() else 'unknown'
1035
})
1036
return False
1037
1038
1039
def notification_to_dict(notification):
···
1806
1807
logger.info(f"Detached {detached_count} temporal blocks")
1808
return True
1809
-
1810
except Exception as e:
1811
logger.error(f"Error detaching temporal blocks: {e}")
1812
return False
1813
1814
···
202
"no_reply": No reply was generated, move to no_reply directory
203
"""
204
import uuid
205
+
206
# Generate correlation ID for tracking this notification through the pipeline
207
correlation_id = str(uuid.uuid4())[:8]
208
+
209
+
# Track attached user blocks for cleanup in finally
210
+
attached_user_blocks = []
211
+
212
try:
213
logger.info(f"[{correlation_id}] Starting process_mention", extra={
214
'correlation_id': correlation_id,
···
401
thread_handles_count = len(unique_handles)
402
prompt_char_count = len(prompt)
403
logger.debug(f"Sending to LLM: @{author_handle} mention | msg: \"{mention_text[:50]}...\" | context: {len(thread_context)} chars, {thread_handles_count} users | prompt: {prompt_char_count} chars")
404
+
405
+
# Attach user blocks for thread participants
406
+
try:
407
+
success, attached_user_blocks = attach_user_blocks_for_thread(CLIENT, void_agent.id, unique_handles)
408
+
if not success:
409
+
logger.warning("Failed to attach some user blocks, continuing anyway")
410
+
except Exception as e:
411
+
logger.error(f"Error attaching user blocks: {e}")
412
413
try:
414
# Use streaming to avoid 524 timeout errors
···
1045
'author_handle': author_handle if 'author_handle' in locals() else 'unknown'
1046
})
1047
return False
1048
+
finally:
1049
+
# Always detach user blocks after processing
1050
+
if attached_user_blocks:
1051
+
try:
1052
+
detach_user_blocks_for_thread(CLIENT, void_agent.id, attached_user_blocks)
1053
+
except Exception as e:
1054
+
logger.error(f"Error detaching user blocks: {e}")
1055
1056
1057
def notification_to_dict(notification):
···
1824
1825
logger.info(f"Detached {detached_count} temporal blocks")
1826
return True
1827
+
1828
except Exception as e:
1829
logger.error(f"Error detaching temporal blocks: {e}")
1830
+
return False
1831
+
1832
+
1833
+
def handle_to_block_label(handle: str) -> str:
1834
+
"""Convert a Bluesky handle to a user block label.
1835
+
1836
+
Example: cameron.pfiffer.org -> user_cameron_pfiffer_org
1837
+
"""
1838
+
if handle.startswith('@'):
1839
+
handle = handle[1:]
1840
+
return f"user_{handle.replace('.', '_')}"
1841
+
1842
+
1843
+
def attach_user_blocks_for_thread(client: Letta, agent_id: str, handles: list) -> tuple:
1844
+
"""
1845
+
Attach user blocks for handles found in the thread.
1846
+
Creates blocks if they don't exist.
1847
+
1848
+
Args:
1849
+
client: Letta client
1850
+
agent_id: Agent ID
1851
+
handles: List of Bluesky handles
1852
+
1853
+
Returns:
1854
+
Tuple of (success: bool, attached_labels: list)
1855
+
"""
1856
+
if not handles:
1857
+
return True, []
1858
+
1859
+
attached_labels = []
1860
+
1861
+
try:
1862
+
current_blocks = client.agents.blocks.list(agent_id=agent_id)
1863
+
current_block_labels = {block.label for block in current_blocks}
1864
+
current_block_ids = {str(block.id) for block in current_blocks}
1865
+
1866
+
logger.debug(f"Attaching user blocks for {len(handles)} handles: {handles}")
1867
+
1868
+
for handle in handles:
1869
+
label = handle_to_block_label(handle)
1870
+
1871
+
try:
1872
+
if label in current_block_labels:
1873
+
logger.debug(f"User block already attached: {label}")
1874
+
attached_labels.append(label)
1875
+
continue
1876
+
1877
+
blocks = client.blocks.list(label=label)
1878
+
1879
+
if blocks and len(blocks) > 0:
1880
+
block = blocks[0]
1881
+
if str(block.id) in current_block_ids:
1882
+
logger.debug(f"User block already attached by ID: {label}")
1883
+
attached_labels.append(label)
1884
+
continue
1885
+
else:
1886
+
block = client.blocks.create(
1887
+
label=label,
1888
+
value=f"User block for @{handle}\n\nNo information recorded yet.",
1889
+
limit=5000
1890
+
)
1891
+
logger.info(f"Created new user block: {label}")
1892
+
1893
+
client.agents.blocks.attach(
1894
+
agent_id=agent_id,
1895
+
block_id=str(block.id)
1896
+
)
1897
+
attached_labels.append(label)
1898
+
logger.info(f"Attached user block: {label}")
1899
+
1900
+
except Exception as e:
1901
+
error_str = str(e)
1902
+
if "duplicate key value violates unique constraint" in error_str:
1903
+
logger.debug(f"User block already attached (constraint): {label}")
1904
+
attached_labels.append(label)
1905
+
else:
1906
+
logger.warning(f"Failed to attach user block {label}: {e}")
1907
+
1908
+
logger.info(f"User blocks attached: {len(attached_labels)}/{len(handles)}")
1909
+
return True, attached_labels
1910
+
1911
+
except Exception as e:
1912
+
logger.error(f"Error attaching user blocks: {e}")
1913
+
return False, attached_labels
1914
+
1915
+
1916
+
def detach_user_blocks_for_thread(client: Letta, agent_id: str, labels_to_detach: list) -> bool:
1917
+
"""
1918
+
Detach user blocks after processing a thread.
1919
+
1920
+
Args:
1921
+
client: Letta client
1922
+
agent_id: Agent ID
1923
+
labels_to_detach: List of user block labels to detach
1924
+
1925
+
Returns:
1926
+
bool: Success status
1927
+
"""
1928
+
if not labels_to_detach:
1929
+
return True
1930
+
1931
+
try:
1932
+
current_blocks = client.agents.blocks.list(agent_id=agent_id)
1933
+
block_label_to_id = {block.label: str(block.id) for block in current_blocks}
1934
+
1935
+
detached_count = 0
1936
+
for label in labels_to_detach:
1937
+
if label in block_label_to_id:
1938
+
try:
1939
+
client.agents.blocks.detach(
1940
+
agent_id=agent_id,
1941
+
block_id=block_label_to_id[label]
1942
+
)
1943
+
detached_count += 1
1944
+
logger.debug(f"Detached user block: {label}")
1945
+
except Exception as e:
1946
+
logger.warning(f"Failed to detach user block {label}: {e}")
1947
+
else:
1948
+
logger.debug(f"User block not attached: {label}")
1949
+
1950
+
logger.info(f"Detached {detached_count} user blocks")
1951
+
return True
1952
+
1953
+
except Exception as e:
1954
+
logger.error(f"Error detaching user blocks: {e}")
1955
return False
1956
1957