Skip to content

fix: harden the curator loop and how bots reach their Matrix client #63

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
arthware-dev merged 3 commits into from
Aug 4, 2026
Merged

fix: harden the curator loop and how bots reach their Matrix client #63

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
98dd6c4
fix(memory): keep the wiki curator alive when a rebuild misbehaves
arthware-dev Aug 3, 2026
8068bb0
refactor: bind the conditionals the type checker could not follow
arthware-dev Aug 3, 2026
b3502a0
refactor(bots): say plainly that a running bot has a Matrix client
arthware-dev Aug 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 5 additions & 4 deletions lib/stack/cli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1394,6 +1394,7 @@ def main():

# Stacklet CLI plugins
stacklet_cmds = {}
stck = None
if repo_root:
stck = create_stack(repo_root, instance_dir)
stacklet_cmds = _load_stacklet_commands(stck)
Expand All @@ -1414,13 +1415,13 @@ def main():
args, _remaining = parser.parse_known_args()

if args.version:
name = stck.product_name() if repo_root else "stack"
sha = stck._git_commit() if repo_root else "unknown"
name = stck.product_name() if stck else "stack"
sha = stck._git_commit() if stck else "unknown"
print(f"{name} {VERSION} ({sha})")
return

if args.help:
name = stck.product_name() if repo_root else "stack"
name = stck.product_name() if stck else "stack"
print_help(name, stacklet_cmds or None)
sys.exit(0)

Expand All @@ -1436,7 +1437,7 @@ def main():
else:
print_status(result)
return
name = stck.product_name() if repo_root else "stack"
name = stck.product_name() if stck else "stack"
print_help(name, stacklet_cmds or None)
sys.exit(0)

Expand Down
92 changes: 55 additions & 37 deletions stacklets/core/bot-runner/microbot.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,24 @@ def __init__(self, homeserver: str, user_id: str, password: str, session_dir: st
self._http: aiohttp.ClientSession | None = None
self._running = False

@property
def client(self) -> AsyncClient:
"""The nio client, from `start()` onwards.

`_client` does not exist until `start()` builds it, which makes
every use after that point read as "might be None" even though
the bot cannot be running without one. Going through here says
that out loud: connected is the precondition, and a caller that
got it wrong hears so, instead of an `AttributeError: 'NoneType'
object has no attribute 'room_send'` from somewhere downstream.

Teardown and the paths that degrade gracefully when there is no
connection still read `_client` directly and check it.
"""
if self._client is None:
raise RuntimeError(f"[{self.name}] not connected to Matrix yet")
return self._client

async def start(self) -> None:
"""Start the bot: login -> initial sync -> register callbacks -> sync loop."""
store_path = str(self._session_dir / f"{self.name}_crypto")
Expand Down Expand Up @@ -185,13 +203,13 @@ async def start(self) -> None:
async def on_invite(room, event):
if isinstance(event, InviteMemberEvent) and event.state_key == self.user_id:
logger.info("[{}] Invited to {} by {}", self.name, room.room_id, event.sender)
resp = await self._client.join(room.room_id)
resp = await self.client.join(room.room_id)
logger.info("[{}] Join result: {}", self.name, resp)
if isinstance(resp, JoinResponse):
self._pending_room_joins.add(room.room_id)
self._anchor_cursor_on_join(room.room_id)

self._client.add_event_callback(on_invite, InviteMemberEvent)
self.client.add_event_callback(on_invite, InviteMemberEvent)

# ── Initial sync ─────────────────────────────────────────────
logger.info("[{}] Initial sync...", self.name)
Expand Down Expand Up @@ -224,7 +242,7 @@ async def on_encrypted(room, event):
if room.room_id in decrypt_notified:
return
decrypt_notified.add(room.room_id)
await self._client.room_send(
await self.client.room_send(
room_id=room.room_id,
message_type="m.room.message",
content={
Expand Down Expand Up @@ -310,7 +328,7 @@ async def _password_login(self, retries=30, interval=10) -> bool:
return False

for attempt in range(1, retries + 1):
resp = await self._client.login(self.password)
resp = await self.client.login(self.password)
if isinstance(resp, LoginResponse):
logger.info("[{}] Logged in (device {})", self.name, resp.device_id)
self._save_session()
Expand Down Expand Up @@ -417,9 +435,9 @@ async def _drain_room(self, room_id: str) -> None:
# Page backward from the live sync position, collecting events newer
# than the cursor, until we cross it. Then process oldest-first.
pending: list = []
start = self._client.next_batch
start = self.client.next_batch
for _ in range(self.MAX_DRAIN_PAGES):
resp = await self._client.room_messages(
resp = await self.client.room_messages(
room_id, start=start, direction=MessageDirection.back,
limit=self.DRAIN_PAGE_SIZE,
)
Expand Down Expand Up @@ -469,14 +487,14 @@ async def _set_read_receipt(self, room_id: str, event_id: str | None) -> None:
if not event_id:
return
try:
await self._client.update_receipt_marker(room_id, event_id)
await self.client.update_receipt_marker(room_id, event_id)
except Exception as e:
logger.debug("[{}] read receipt update failed in {}: {}",
self.name, room_id, e)

async def _dispatch(self, room_id: str, event) -> None:
"""Invoke every handler whose registered type matches the event."""
room = self._client.rooms.get(room_id)
room = self.client.rooms.get(room_id)
if room is None:
return
for event_type, handler in self._handlers:
Expand All @@ -499,7 +517,7 @@ async def _set_typing(self, room_id: str, on: bool = True) -> None:
"""
logger.info("[{}] typing -> {} in {}", self.name, "on" if on else "off", room_id)
try:
resp = await self._client.room_typing(
resp = await self.client.room_typing(
room_id, typing_state=on, timeout=300000,
)
logger.info("[{}] typing response: {}", self.name, type(resp).__name__)
Expand All @@ -512,14 +530,14 @@ async def _room_send(
content: dict,
message_type: str = "m.room.message",
) -> None:
"""Thin wrapper around ``self._client.room_send``.
"""Thin wrapper around ``self.client.room_send``.

Exists so subclasses can route every send through a single
framework-owned method — useful when we want to add cross-
cutting behavior (audit logging, retries, etc.) without
touching every call site. Today it's a passthrough.
"""
await self._client.room_send(
await self.client.room_send(
room_id=room_id, message_type=message_type, content=content,
)

Expand Down Expand Up @@ -656,7 +674,7 @@ async def _answer(
return
root = source_event
try:
resp = await self._client.room_get_event(room_id, source_event)
resp = await self.client.room_get_event(room_id, source_event)
existing = self.get_thread_root(getattr(resp, "event", None))
if existing:
root = existing
Expand Down Expand Up @@ -745,7 +763,7 @@ async def post_source_message(
"m.in_reply_to": {"event_id": thread_root_event_id},
}
try:
resp = await self._client.room_send(
resp = await self.client.room_send(
room_id=room_id, message_type="m.room.message", content=content,
)
except Exception as e:
Expand Down Expand Up @@ -776,7 +794,7 @@ async def _reply_parent_envelope(self, room_id: str, event) -> dict | None:
if not in_reply_to:
return None
try:
resp = await self._client.room_get_event(room_id, in_reply_to)
resp = await self.client.room_get_event(room_id, in_reply_to)
except Exception as e:
logger.debug("[{}] reply parent fetch failed: {}", self.name, e)
return None
Expand Down Expand Up @@ -815,7 +833,7 @@ async def _thread_envelopes(
envelopes: list[tuple[str, dict]] = []
try:
examined = 0
async for related in self._client.room_get_event_relations(
async for related in self.client.room_get_event_relations(
room_id, root_event_id, RelationshipType.thread,
):
examined += 1
Expand Down Expand Up @@ -864,10 +882,10 @@ async def _sync_display_name(self) -> None:
if not self.display_name:
return
try:
resp = await self._client.get_displayname(self.user_id)
resp = await self.client.get_displayname(self.user_id)
current = getattr(resp, "displayname", None)
if current != self.display_name:
await self._client.set_displayname(self.display_name)
await self.client.set_displayname(self.display_name)
logger.info(
"[{}] Display name set to {!r}", self.name, self.display_name,
)
Expand All @@ -891,7 +909,7 @@ async def _download_media(self, mxc_url: str) -> bytes | None:
session = self._ensure_http()
async with session.get(
download_url,
headers={"Authorization": f"Bearer {self._client.access_token}"},
headers={"Authorization": f"Bearer {self.client.access_token}"},
) as resp:
if resp.status == 200:
return await resp.read()
Expand All @@ -918,7 +936,7 @@ async def _upload_media(
params={"filename": filename},
data=data,
headers={
"Authorization": f"Bearer {self._client.access_token}",
"Authorization": f"Bearer {self.client.access_token}",
"Content-Type": content_type or "application/octet-stream",
},
) as resp:
Expand Down Expand Up @@ -973,7 +991,7 @@ async def send_file(
if metadata:
content.update(metadata)
try:
resp = await self._client.room_send(
resp = await self.client.room_send(
room_id=room_id, message_type="m.room.message", content=content,
)
except Exception as e:
Expand Down Expand Up @@ -1010,7 +1028,7 @@ async def _send_error(self, room_id: str, event, exc: BaseException) -> None:
content["m.relates_to"] = {
"m.in_reply_to": {"event_id": reply_to},
}
await self._client.room_send(
await self.client.room_send(
room_id=room_id,
message_type="m.room.message",
content=content,
Expand All @@ -1032,7 +1050,7 @@ async def on_room_joined(self, room_id: str) -> None:

Hook for join-time behaviour like posting a welcome message.
Default is a no-op so subclasses opt in explicitly. Contract:
``self._client.rooms[room_id]`` is guaranteed to exist and to
``self.client.rooms[room_id]`` is guaranteed to exist and to
carry members + name when this fires. The framework wires
invite-accept to a sync-response callback to enforce that
contract so subclasses do not need to poll or defer.
Expand Down Expand Up @@ -1189,7 +1207,7 @@ def _room_config_url(self, room_id: str) -> str:
)

def _auth_headers(self) -> dict:
return {"Authorization": f"Bearer {self._client.access_token}"}
return {"Authorization": f"Bearer {self.client.access_token}"}

async def get_room_config(self, room_id: str) -> dict:
"""The bot's config for this room (its ``dev.famstack.room`` room
Expand Down Expand Up @@ -1370,7 +1388,7 @@ async def emit_event(self, room_id: str, event_type: str, body: dict) -> bool:
event shouldn't take down the caller's main path.
"""
try:
await self._client.room_send(
await self.client.room_send(
room_id=room_id,
message_type=event_type,
content=body,
Expand All @@ -1386,13 +1404,13 @@ async def stop(self) -> None:

def _trust_all_devices(self) -> None:
"""Mark all known devices as trusted without interactive verification."""
if not self._client.olm:
if not self.client.olm:
return
try:
for user_id in self._client.device_store.users:
for device in self._client.device_store.active_user_devices(user_id):
if not self._client.olm.is_device_verified(device):
self._client.verify_device(device)
for user_id in self.client.device_store.users:
for device in self.client.device_store.active_user_devices(user_id):
if not self.client.olm.is_device_verified(device):
self.client.verify_device(device)
logger.info("[{}] Trusted device {} of {}", self.name, device.device_id, user_id)
except Exception as e:
logger.debug("[{}] Trust devices: {}", self.name, e)
Expand All @@ -1403,9 +1421,9 @@ def _restore_session(self) -> bool:
return False
try:
data = json.loads(self.session_file.read_text())
self._client.access_token = data["access_token"]
self._client.user_id = data["user_id"]
self._client.device_id = data["device_id"]
self.client.access_token = data["access_token"]
self.client.user_id = data["user_id"]
self.client.device_id = data["device_id"]
logger.info("[{}] Restored session (device {})", self.name, data["device_id"])
return True
except (json.JSONDecodeError, KeyError) as e:
Expand All @@ -1414,8 +1432,8 @@ def _restore_session(self) -> bool:

def _clear_session(self):
"""Wipe saved session and in-memory credentials."""
self._client.access_token = ""
self._client.device_id = ""
self.client.access_token = ""
self.client.device_id = ""
if self.session_file.exists():
self.session_file.unlink()
logger.info("[{}] Deleted stale session file", self.name)
Expand All @@ -1424,9 +1442,9 @@ def _save_session(self) -> None:
"""Persist the current session so it survives container restarts."""
self.session_file.parent.mkdir(parents=True, exist_ok=True)
self.session_file.write_text(json.dumps({
"access_token": self._client.access_token,
"user_id": self._client.user_id,
"device_id": self._client.device_id,
"access_token": self.client.access_token,
"user_id": self.client.user_id,
"device_id": self.client.device_id,
}))

# ── Message cursor ───────────────────────────────────────────────
Expand Down
22 changes: 15 additions & 7 deletions stacklets/docs/bot/archivist.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -186,6 +186,14 @@ def _t(lang: str, key: str, **kwargs) -> str:

# ── Helpers ──────────────────────────────────────────────────────────────────

# Keyed by the doc type `google_docs_export_url` reports, for the "downloading
# a Google Doc" line and the filename it lands under.
_GOOGLE_TYPE_LABELS = {
"document": "Google Doc",
"spreadsheets": "Google Sheet",
"presentation": "Google Slides",
}

_MIME_BY_EXT = {
"pdf": "application/pdf",
"jpg": "image/jpeg", "jpeg": "image/jpeg",
Expand Down Expand Up @@ -577,8 +585,8 @@ async def on_first_sync(self) -> None:
``_send_room_welcome_if_needed`` orchestrator.
"""

for room_id in list(self._client.rooms):
room = self._client.rooms[room_id]
for room_id in list(self.client.rooms):
room = self.client.rooms[room_id]
ctx = self._room_context(room)
try:
await self._send_room_welcome_if_needed(room, ctx)
Expand Down Expand Up @@ -1224,7 +1232,7 @@ async def _fetch_event(self, room_id: str, event_id: str | None):
if not event_id:
return None
try:
resp = await self._client.room_get_event(room_id, event_id)
resp = await self.client.room_get_event(room_id, event_id)
except Exception as e:
logger.debug("[archivist] event fetch failed for {}: {}", event_id, e)
return None
Expand Down Expand Up @@ -1780,7 +1788,7 @@ async def _on_reaction(self, room, event) -> None:
if not target_id:
return
try:
resp = await self._client.room_get_event(room.room_id, target_id)
resp = await self.client.room_get_event(room.room_id, target_id)
except Exception as e:
logger.debug("[archivist] reaction target fetch failed: {}", e)
return
Expand Down Expand Up @@ -2378,10 +2386,10 @@ async def _handle_url(
*, date_filed: str | None = None, submitter_mxid: str | None = None,
):
google_export = _google_docs_export_url(url)
doc_type = ""
if google_export:
download_url, doc_type = google_export
type_labels = {"document": "Google Doc", "spreadsheets": "Google Sheet", "presentation": "Google Slides"}
await self._send(room_id, self.t("downloading_google", type=type_labels.get(doc_type, "Google Doc")), reply_to)
await self._send(room_id, self.t("downloading_google", type=_GOOGLE_TYPE_LABELS.get(doc_type, "Google Doc")), reply_to)
else:
download_url = url
await self._send(room_id, self.t("downloading_url"), reply_to)
Expand All @@ -2407,7 +2415,7 @@ async def _handle_url(
# Determine filename
if google_export:
filename = f"google-{doc_type}.pdf"
display_name = type_labels.get(doc_type, "Google Doc")
display_name = _GOOGLE_TYPE_LABELS.get(doc_type, "Google Doc")
elif "pdf" in content_type or url.lower().endswith(".pdf"):
url_path = url.split("?")[0].split("#")[0]
filename = url_path.rsplit("/", 1)[-1] if "/" in url_path else "document.pdf"
Expand Down
Loading
Loading