From 98dd6c4705cdbccc115c492760bc392d61613250 Mon Sep 17 00:00:00 2001 From: Arthur Date: Mon, 3 Aug 2026 23:10:35 +0200 Subject: [PATCH 1/3] fix(memory): keep the wiki curator alive when a rebuild misbehaves A rebuild that wedges is killed, and a child that cannot start at all is reported instead of taking the sidecar's loop down with it. Pins both in tests, since the loop calling it has no exception handling of its own. --- stacklets/memory/bot/curator.py | 10 +++++- tests/stacklets/test_memory_curator.py | 48 ++++++++++++++++++++++++++ 2 files changed, 57 insertions(+), 1 deletion(-) diff --git a/stacklets/memory/bot/curator.py b/stacklets/memory/bot/curator.py index 0c31f4c3..3fbf16f7 100644 --- a/stacklets/memory/bot/curator.py +++ b/stacklets/memory/bot/curator.py @@ -745,12 +745,20 @@ async def rebuild(selection: list[str]) -> bool: call dies with the child instead of inside this loop.""" label = " ".join(selection) if selection else "(full sweep)" logger.info("[curator] rebuilding wiki: {}", label) + # Starting the child and waiting on it are separate failure modes, and + # only the second one has a child to kill. Keeping them in one block + # left `proc.kill()` reachable on a path where `proc` was never bound. try: proc = await asyncio.create_subprocess_exec( sys.executable, ENTRYPOINT, "wiki", *selection, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, ) + except Exception as e: + logger.warning("[curator] rebuild failed to start: {}", e) + return False + + try: out_bytes, _ = await asyncio.wait_for( proc.communicate(), timeout=REBUILD_TIMEOUT_SECS, ) @@ -759,7 +767,7 @@ async def rebuild(selection: list[str]) -> bool: logger.warning("[curator] rebuild timed out after {}s", REBUILD_TIMEOUT_SECS) return False except Exception as e: - logger.warning("[curator] rebuild failed to start: {}", e) + logger.warning("[curator] rebuild failed: {}", e) return False output = out_bytes.decode(errors="replace").strip() diff --git a/tests/stacklets/test_memory_curator.py b/tests/stacklets/test_memory_curator.py index ad66b4ca..84139960 100644 --- a/tests/stacklets/test_memory_curator.py +++ b/tests/stacklets/test_memory_curator.py @@ -8,10 +8,16 @@ and the nightly once-per-day gate. All pure and tested here; the end-to-end path rides the integration rig like the rest of the wiki. + +The one exception is `rebuild`, the single subprocess call. It is not +pure, but the loop that calls it has no exception handling of its own, +so what it does when the child misbehaves is the sidecar's survival and +is pinned here too. """ from __future__ import annotations +import asyncio import sys import time from pathlib import Path @@ -20,6 +26,7 @@ sys.path.insert(0, str(_REPO_ROOT / "stacklets" / "memory" / "bot")) sys.path.insert(0, str(_REPO_ROOT / "stacklets" / "memory" / "bot" / "cli")) +import curator # noqa: E402 from curator import ( # noqa: E402 Debounce, diff_to_fileops, @@ -391,3 +398,44 @@ async def drop(): await drop_task assert woke is True assert time.monotonic() - start < 1.0 + + +# ── rebuild: the one subprocess call ───────────────────────────────────── +# +# `main()` calls this straight from its `while True` with nothing wrapped +# around it, so `rebuild` owes the loop a bool no matter how the child +# behaves. Two ways it can fail on the way in: the generation wedges (an +# LLM call that never returns), or the child never starts at all. Both +# have to come back False with the sidecar still running, and a wedged +# child has to actually die rather than be left behind holding the vault. + + +class TestRebuild: + async def test_a_wedged_generation_is_killed_not_abandoned( + self, tmp_path, monkeypatch, + ): + marker = tmp_path / "child-ran-to-completion" + script = tmp_path / "hang.py" + script.write_text( + "import time\n" + "time.sleep(1.0)\n" + f"open({str(marker)!r}, 'w').close()\n", + encoding="utf-8", + ) + monkeypatch.setattr(curator, "ENTRYPOINT", str(script)) + monkeypatch.setattr(curator, "REBUILD_TIMEOUT_SECS", 0.3) + + start = time.monotonic() + assert await curator.rebuild([]) is False + # Back on the timeout, not on the child's own schedule. + assert time.monotonic() - start < 1.0 + + # Past when the child would have finished had it survived the kill. + await asyncio.sleep(1.2) + assert not marker.exists() + + async def test_a_child_that_cannot_start_is_false_not_a_crash( + self, monkeypatch, + ): + monkeypatch.setattr(curator.sys, "executable", "/nonexistent/python") + assert await curator.rebuild([]) is False From 8068bb02df6a79c7541b2f8a65748ef6f9d045f3 Mon Sep 17 00:00:00 2001 From: Arthur Date: Mon, 3 Aug 2026 23:10:42 +0200 Subject: [PATCH 2/3] refactor: bind the conditionals the type checker could not follow Each of these read a variable that is only assigned under an earlier branch. Safe today because the same flag guards both ends, one edit away from a NameError. No behaviour change. --- lib/stack/cli.py | 9 +++++---- stacklets/docs/bot/archivist.py | 14 +++++++++++--- stacklets/memory/cli/write.py | 1 + 3 files changed, 17 insertions(+), 7 deletions(-) diff --git a/lib/stack/cli.py b/lib/stack/cli.py index e855f75a..587d13b0 100644 --- a/lib/stack/cli.py +++ b/lib/stack/cli.py @@ -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) @@ -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) @@ -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) diff --git a/stacklets/docs/bot/archivist.py b/stacklets/docs/bot/archivist.py index 60bfb085..f6b1fa62 100644 --- a/stacklets/docs/bot/archivist.py +++ b/stacklets/docs/bot/archivist.py @@ -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", @@ -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) @@ -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" diff --git a/stacklets/memory/cli/write.py b/stacklets/memory/cli/write.py index d92ef359..52e79ff2 100644 --- a/stacklets/memory/cli/write.py +++ b/stacklets/memory/cli/write.py @@ -105,6 +105,7 @@ def run(args, stacklet, config): actor = actor.strip().split(":")[0].lstrip("@") or "someone" + edits = None if as_patch: try: edits = json.loads(content) From b3502a03949cfcf46a361a9b28a36fb0ea803bdf Mon Sep 17 00:00:00 2001 From: Arthur Date: Tue, 4 Aug 2026 08:17:48 +0200 Subject: [PATCH 3/3] refactor(bots): say plainly that a running bot has a Matrix client Everything after start() went through an optional _client, so each of the 39 uses read as "might be None". A connected() property states the precondition once: use it and a mistake reports itself, instead of an AttributeError on None further downstream. Teardown and the paths that degrade without a connection still check _client themselves. --- stacklets/core/bot-runner/microbot.py | 92 ++++++++++++++++----------- stacklets/docs/bot/archivist.py | 8 +-- 2 files changed, 59 insertions(+), 41 deletions(-) diff --git a/stacklets/core/bot-runner/microbot.py b/stacklets/core/bot-runner/microbot.py index 512df1de..ebe22409 100644 --- a/stacklets/core/bot-runner/microbot.py +++ b/stacklets/core/bot-runner/microbot.py @@ -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") @@ -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) @@ -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={ @@ -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() @@ -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, ) @@ -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: @@ -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__) @@ -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, ) @@ -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 @@ -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: @@ -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 @@ -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 @@ -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, ) @@ -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() @@ -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: @@ -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: @@ -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, @@ -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. @@ -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 @@ -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, @@ -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) @@ -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: @@ -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) @@ -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 ─────────────────────────────────────────────── diff --git a/stacklets/docs/bot/archivist.py b/stacklets/docs/bot/archivist.py index f6b1fa62..e557089c 100644 --- a/stacklets/docs/bot/archivist.py +++ b/stacklets/docs/bot/archivist.py @@ -585,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) @@ -1232,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 @@ -1788,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