diff --git a/stacklets/agent/config.json b/stacklets/agent/config.json index 8de1a671..1de17f02 100644 --- a/stacklets/agent/config.json +++ b/stacklets/agent/config.json @@ -37,7 +37,7 @@ }, "channels": { "sendProgress": false, - "sendToolHints": false, + "sendToolHints": true, "showReasoning": false, "matrix": { "enabled": true, diff --git a/stacklets/agent/runtime/sitecustomize.py b/stacklets/agent/runtime/sitecustomize.py index 0ad4d1b3..eeec8687 100644 --- a/stacklets/agent/runtime/sitecustomize.py +++ b/stacklets/agent/runtime/sitecustomize.py @@ -56,6 +56,10 @@ is a question, the asked person's next message there reaches it without its name. +11. status_trace (status_trace.py) — a slow turn gets one status notice that + grows a line as the agent moves on (looking, looking closer, writing it + down); the answer comes as its own message, the notice stays as a trace. + WHY SHIMS AND NOT A FORK nanobot 0.3 has a seam for per-turn context (runtime-context providers) and keeps a session per Matrix thread, which retired the two shims that needed @@ -99,6 +103,11 @@ `MatrixChannel.send(self, msg: OutboundMessage)` (async, a dataclass whose `metadata["event_id"]` is the message being answered) open_question: the voice_trigger symbols, plus `room.room_id` and `event.sender` in the gate + status_trace: `nanobot.bus.outbound_events.ProgressEvent` (`tool_hint`, `tool_events[].name`), + `nanobot.events.ContextCompactionEvent` (`phase`) reaching `MatrixChannel.send`, + reaching `MatrixChannel.send` when `sendToolHints` is on, + `MatrixChannel._send_room_content(room_id, content)` returning a response + with `event_id`, `MatrixChannel._build_thread_relates_to(metadata)` (static) join_greeting: `nanobot.channels.matrix.runtime.MatrixChannel._on_room_invite(self, room, event)` `MatrixChannel._handle_message(sender_id, chat_id, content, metadata, is_dm)` @@ -152,6 +161,15 @@ def _loop_init(self, *args, **kwargs): _log.exception("brief shim could not attach (nanobot internals changed?)") +def _is_progress(msg) -> bool: + """True for nanobot's progress messages (tool hints, progress text), not an answer.""" + try: + from nanobot.bus.outbound_events import ProgressEvent + except Exception: + return False + return isinstance(getattr(msg, "event", None), ProgressEvent) + + # ── state_log: the message list per turn, for debugging (opt-in) ───────────── # AGENT_STATE_LOG=1 writes what nanobot built for each call to llm-state.log, # one line per message, to compare against the Matrix chat. Off by default: @@ -299,6 +317,96 @@ def _is_bot_mentioned_or_our_thread(self, event): _log.exception("thread-trigger shim could not attach (nanobot internals changed?)") +# ── status_trace: a slow turn shows what the agent is doing ────────────────── +# One status notice per slow turn, a line appended as the agent moves on; the +# answer comes as its own message and the notice stays as a trace. nanobot's +# tool hints (config.json `sendToolHints`) are the signal and never reach the +# chat themselves. Installed before the voice and question gates, so a message +# they drop never starts a status. See status_trace.py. +try: + import asyncio as _asyncio_trace + import os as _os_trace + import time as _time_trace + + import nanobot.channels.matrix.runtime as _matrix_trace + import status_trace as _status_trace + from nanobot.events import ContextCompactionEvent as _Compaction + + # How often a running trace looks whether its status should change. + _TRACE_TICK = 1.0 + _traces: dict[tuple[str, str], "_Running"] = {} + + class _Running: + """One turn's trace: its status notice in Matrix, and the loop keeping it current.""" + + def __init__(self, channel, room_id: str, metadata: dict) -> None: + self.trace = _status_trace.Trace(started=_time_trace.monotonic(), + lang=_os_trace.environ.get("AGENT_LANGUAGE", "")) + self.channel, self.room_id, self.metadata = channel, room_id, metadata + self.notice_id = "" + self.task = _asyncio_trace.create_task(self._keep_current()) + + async def _keep_current(self) -> None: + while True: + await _asyncio_trace.sleep(_TRACE_TICK) + lines = self.trace.due(at=_time_trace.monotonic()) + if lines: + await self._show("\n".join(lines)) + + async def _show(self, text: str) -> None: + notice = {"msgtype": "m.notice", "body": text} + if self.notice_id: + content = {**notice, "body": f"* {text}", "m.new_content": notice, + "m.relates_to": {"rel_type": "m.replace", "event_id": self.notice_id}} + else: + thread = self.channel._build_thread_relates_to(self.metadata) + content = {**notice, "m.relates_to": thread} if thread else notice + try: + response = await self.channel._send_room_content(self.room_id, content) + except Exception: + _log.debug("status notice not sent", exc_info=True) + return + self.notice_id = self.notice_id or getattr(response, "event_id", "") or "" + + def _trace_key(chat_id: str, metadata) -> tuple[str, str]: + return chat_id, (metadata or {}).get("thread_root_event_id", "") or "" + + _orig_handle_trace = _matrix_trace.MatrixChannel._handle_message + _orig_send_trace = _matrix_trace.MatrixChannel.send + + async def _handle_and_trace(self, sender_id, chat_id, content, *args, **kwargs): + metadata = kwargs.get("metadata") or {} + key = _trace_key(chat_id, metadata) + if key in _traces: + _traces.pop(key).task.cancel() + _traces[key] = _Running(self, chat_id, metadata) + return await _orig_handle_trace(self, sender_id, chat_id, content, *args, **kwargs) + + async def _send_traced(self, msg): + key = _trace_key(msg.chat_id, msg.metadata) + running = _traces.get(key) + if _is_progress(msg) and msg.event.tool_hint: + # The signal, not a message: what the agent started doing. + if running: + running.trace.tools([e.get("name", "") for e in msg.event.tool_events or []]) + return None + if isinstance(msg.event, _Compaction): + # A line in the family's words while someone waits; between turns + # nobody does, and a notice out of nowhere only puzzles. + if running and msg.event.phase == "started": + running.trace.tidying() + return None + if running and msg.content and not _is_progress(msg): + _traces.pop(key).task.cancel() + return await _orig_send_trace(self, msg) + + _matrix_trace.MatrixChannel._handle_message = _handle_and_trace + _matrix_trace.MatrixChannel.send = _send_traced + _log.info("status-trace shim active") +except Exception: + _log.exception("status-trace shim could not attach (nanobot internals changed?)") + + # ── voice_trigger: a voice message is addressed by what was said ───────────── # nanobot's gate runs before transcription, and a voice message's text is # "Voice message", so in a shared room it never reached the agent. Let it @@ -343,9 +451,10 @@ async def _handle_message(self, sender_id, chat_id, content, *args, **kwargs): async def _send(self, msg): # The answer carries the voice message's event id for threading; the - # first one with text gets what was heard in front of it. + # first one with text gets what was heard in front of it. Progress + # (a tool hint, see status_trace) is not the answer. heard = _pending_voice.answering((msg.metadata or {}).get("event_id", "")) \ - if msg.content else "" + if msg.content and not _is_progress(msg) else "" if heard: msg = _dataclasses_voice.replace(msg, content=_quoted(heard, msg.content)) return await _orig_send_voice(self, msg) @@ -387,7 +496,7 @@ async def _handle_and_remember(self, sender_id, chat_id, content, *args, **kwarg return await _orig_handle_q(self, sender_id, chat_id, content, *args, **kwargs) async def _send_and_remember(self, msg): - if msg.content: + if msg.content and not _is_progress(msg): if _asks(msg.content): _open_questions.asked(msg.chat_id, _talking_to.get(msg.chat_id, ""), at=_time_question.monotonic()) diff --git a/stacklets/agent/runtime/status_trace.py b/stacklets/agent/runtime/status_trace.py new file mode 100644 index 00000000..3a2750b6 --- /dev/null +++ b/stacklets/agent/runtime/status_trace.py @@ -0,0 +1,135 @@ +"""What the agent is doing, in one status message that grows while it works. + +A turn that needs a search, a page and a list takes Stacky ten to thirty +seconds. A typing indicator alone reads as "stuck" after a few of them. +So a turn that is not answered quickly gets one status message, and +while the agent works, a short line is appended to it now and then: + + Moment, ich schau mal nach … + Hab was gefunden, ich prüf es noch genauer … + Ich trag das ein … + +The answer comes as a message of its own; the status stays above it as a +trace of what the agent did. It is a notice (`m.notice`), which Matrix's +default push rules do not notify about and Element shows greyed out, so +the trace never buzzes a phone. Only the answer does. + +The lines are ours, not the model's: nanobot's raw tool hints read like a +shell (`$ stack memory list show einkaufsliste`), and the model's own +progress text is unpredictable. When nanobot compacts the conversation +during a turn ("Compressing context…"), that is a line too, in the +family's words. Each tool round is mapped to a phase +(looking, looking again, writing down) and a phrase for it is picked at +random from a few, in the household's language, never the same twice in +one trace. + +`Trace` decides what the status says and when it changes, from the times +it is given; the caller posts and edits the message. + + trace = Trace(started=0.0, lang="de") + trace.tools(["memory_search"]) + trace.due(at=5.0) # None: answered this fast, there is no status + trace.due(at=6.0) # ["Moment, ich schau mal nach …"] +""" + +from __future__ import annotations + +import random +from collections.abc import Callable, Sequence + +# A turn answered within this many seconds needs no status at all. +FIRST_AFTER = 6.0 +# At most one new line this often, so the status does not chatter. +GAP = 10.0 +# With nothing new for this long, one line saying it is still at it. +STILL_AFTER = 20.0 +# A status longer than this stops growing; the answer is what matters. +MAX_LINES = 5 + +LOOK, CHECK, WRITE, THINK, STILL, TIDY = "look", "check", "write", "think", "still", "tidy" + +# Tools that change a list or a page; everything else reads. +_WRITERS = {"list_edit", "write_file", "edit_file", "apply_patch"} + +PHRASES: dict[str, dict[str, tuple[str, ...]]] = { + "de": { + LOOK: ("Moment, ich schau mal nach …", "Lass mich kurz nachsehen …", + "Ich such das gerade raus …"), + CHECK: ("Hab was gefunden, ich prüf es noch genauer …", + "Da ist was, ich schau es mir genauer an …", + "Ich seh noch an einer Stelle nach …"), + WRITE: ("Ich trag das ein …", "Ich kümmer mich drum …", "Ich schreib das auf …"), + THINK: ("Moment …", "Ich denk kurz nach …"), + STILL: ("Dauert noch einen Moment …", "Gleich hab ich's …"), + TIDY: ("Moment, ich sortier kurz meine Gedanken und Notizen …", + "Ich bring kurz meine Gedanken und Notizen in Ordnung …"), + }, + "en": { + LOOK: ("One moment, let me check …", "Let me look that up …", + "Looking for it …"), + CHECK: ("Found something, checking the details …", + "There is something, taking a closer look …", + "Checking one more place …"), + WRITE: ("Writing that down …", "On it …", "Putting it on the list …"), + THINK: ("One moment …", "Thinking it over …"), + STILL: ("Still on it …", "Almost there …"), + TIDY: ("Hang on, let me sort my thoughts and notes …", + "Tidying up my thoughts and notes for a second …"), + }, +} + + +def phase_of(tools: Sequence[str], *, looked_before: bool) -> str: + """The phase a round of tool calls belongs to.""" + if any(name in _WRITERS for name in tools): + return WRITE + return CHECK if looked_before else LOOK + + +class Trace: + """The lines of one turn's status, and when they change. + + `tools` records what the agent started doing; `due` answers, at a given + time, the whole status to show if it should change now, or None. + """ + + def __init__(self, *, started: float, lang: str, + pick: Callable[[Sequence[str]], str] = random.choice) -> None: + self._started = started + self._phrases = PHRASES.get((lang or "")[:2].lower(), PHRASES["en"]) + self._pick = pick + self._lines: list[str] = [] + self._phases: list[str] = [] + self._pending = THINK + self._looked = False + self._last_at = started + + def tools(self, names: Sequence[str]) -> None: + phase = phase_of(names, looked_before=self._looked) + self._looked = self._looked or phase in (LOOK, CHECK) + self._pending = phase + + def tidying(self) -> None: + """nanobot started compacting the conversation, which can take a while.""" + self._pending = TIDY + + def due(self, *, at: float) -> list[str] | None: + if at - self._started < FIRST_AFTER or len(self._lines) >= MAX_LINES: + return None + if not self._lines: + return self._add(self._pending, at) + if at - self._last_at < GAP: + return None + if self._pending != self._phases[-1]: + return self._add(self._pending, at) + if at - self._last_at >= STILL_AFTER and self._phases[-1] != STILL: + return self._add(STILL, at) + return None + + def _add(self, phase: str, at: float) -> list[str]: + fresh = [p for p in self._phrases[phase] if p not in self._lines] + self._lines.append(self._pick(fresh or self._phrases[phase])) + self._phases.append(phase) + self._pending = phase + self._last_at = at + return list(self._lines) diff --git a/tests/unit/stacklets/conftest.py b/tests/unit/stacklets/conftest.py index ec6cee1e..5fc6f79b 100644 --- a/tests/unit/stacklets/conftest.py +++ b/tests/unit/stacklets/conftest.py @@ -144,12 +144,24 @@ def __init__(self): self.handled = [] self.typing_stopped = [] self.sent = [] + self.room_contents = [] self.joined = [] self.processed = [] def is_allowed(self, sender): return True + async def _send_room_content(self, room_id, content): + # nanobot 0.3: m.room.message with E2EE handled; the response + # carries the new event's id. + self.room_contents.append((room_id, content)) + return types.SimpleNamespace(event_id=f"$notice{len(self.room_contents)}") + + @staticmethod + def _build_thread_relates_to(metadata): + root = (metadata or {}).get("thread_root_event_id") + return {"rel_type": "m.thread", "event_id": root} if root else None + async def send(self, msg): # nanobot 0.3 channels/matrix/runtime.py: an OutboundMessage, # whose metadata carries the event id of the message answered. @@ -203,6 +215,24 @@ def mod(name, **attrs): mod("nanobot.agent") mod("nanobot.agent.context", ContextBuilder=ContextBuilder) mod("nanobot.agent.loop", AgentLoop=AgentLoop) + + @dataclasses.dataclass(frozen=True) + class ProgressEvent: + # nanobot 0.3 bus/outbound_events.py: a tool hint carries the + # calls it announces in `tool_events`. + content: str = "" + tool_hint: bool = False + tool_events: list | None = None + + @dataclasses.dataclass(frozen=True) + class ContextCompactionEvent: + # nanobot 0.3 events.py; a channel shows it as "Compressing context…". + compaction_id: str = "" + phase: str = "started" + + mod("nanobot.events", ContextCompactionEvent=ContextCompactionEvent) + mod("nanobot.bus") + mod("nanobot.bus.outbound_events", ProgressEvent=ProgressEvent) mod("nanobot.runtime_context", RuntimeContextBlock=RuntimeContextBlock, wrap_runtime_context_lines=wrap_runtime_context_lines) mod("nanobot.agent.tools") diff --git a/tests/unit/stacklets/test_agent_runtime_shims.py b/tests/unit/stacklets/test_agent_runtime_shims.py index 1a44018c..31b232fc 100644 --- a/tests/unit/stacklets/test_agent_runtime_shims.py +++ b/tests/unit/stacklets/test_agent_runtime_shims.py @@ -26,7 +26,8 @@ SHIMMED_MODULES = ("sitecustomize", "brief", "state_log", "memory_tool", "person_tool", "history_tool", "name_trigger", "thread_trigger", "join_greeting", "vault_write", - "list_tool", "tool_trim", "voice_trigger", "open_question") + "list_tool", "tool_trim", "voice_trigger", "open_question", + "status_trace") # The stub nanobot itself lives in conftest as `nanobot_stub`, shared with @@ -356,6 +357,7 @@ class _Outbound: chat_id: str content: str metadata: dict + event: object = None monkeypatch.setenv("AGENT_NAME", "Stacky") channel = nanobot()["nanobot.channels.matrix.runtime"].MatrixChannel() @@ -372,6 +374,24 @@ class _Outbound: "Sonst noch was?"] +def test_a_tool_hint_before_the_answer_does_not_take_the_quote(nanobot, monkeypatch): + import asyncio + + monkeypatch.setenv("AGENT_NAME", "Stacky") + mods = nanobot() + channel = mods["nanobot.channels.matrix.runtime"].MatrixChannel() + channel._should_process_message(None, _voice()) + _hand_over(channel, "Stacky, wir brauchen Milch.") + + asyncio.run(channel.send(_outbound("$ list_edit", event=_hint(mods, "list_edit"), + metadata={"event_id": "$voice"}))) + asyncio.run(channel.send(_outbound("Milch steht jetzt auf der Einkaufsliste.", + metadata={"event_id": "$voice"}))) + + assert [m.content for m in channel.sent] == [ + "> Stacky, wir brauchen Milch.\n\nMilch steht jetzt auf der Einkaufsliste."] + + def test_a_voice_message_for_someone_else_gets_no_answer(nanobot, monkeypatch): """The same rule as typed text: no name, no answer, and no typing left on.""" monkeypatch.setenv("AGENT_NAME", "Stacky") @@ -411,6 +431,7 @@ class _Outbound: chat_id: str content: str metadata: dict + event: object = None class _Text: sender = "@homer:simpson" @@ -431,6 +452,91 @@ class _Text: assert not channel._should_process_message(kitchen, _Text()), "only the first answer" +# ── a slow turn shows what the agent is doing ──────────────────────────── + +def _outbound(content, *, event=None, metadata=None): + import dataclasses + + @dataclasses.dataclass + class _Outbound: + chat_id: str + content: str + metadata: dict + event: object = None + event: object = None + + return _Outbound("!kitchen", content, metadata or {}, event) + + +def _hint(mods, *names): + progress = mods["nanobot.bus.outbound_events"].ProgressEvent + return progress(content="$ stack ...", tool_hint=True, + tool_events=[{"name": n} for n in names]) + + +def _slow_turn(nanobot, monkeypatch, steps): + """Run a turn whose status may change at once, step by step; return the channel.""" + import asyncio + import sys as _sys + + mods = nanobot() + monkeypatch.setattr(_sys.modules["status_trace"], "FIRST_AFTER", 0.0) + monkeypatch.setattr(_sys.modules["status_trace"], "GAP", 0.0) + monkeypatch.setattr(_sys.modules["sitecustomize"], "_TRACE_TICK", 0.01) + monkeypatch.setenv("AGENT_LANGUAGE", "de") + channel = mods["nanobot.channels.matrix.runtime"].MatrixChannel() + + async def _run(): + await channel._handle_message(sender_id="@homer:simpson", chat_id="!kitchen", + content="Stacky, was fehlt noch?", metadata={}) + for step in steps(mods): + await asyncio.sleep(0.05) + await channel.send(step) + await asyncio.sleep(0.05) + + asyncio.run(_run()) + return channel + + +def test_a_slow_turn_shows_one_status_that_grows(nanobot, monkeypatch): + channel = _slow_turn(nanobot, monkeypatch, lambda mods: [ + _outbound("$ memory_search", event=_hint(mods, "memory_search")), + _outbound("$ list_edit", event=_hint(mods, "list_edit")), + _outbound("Milch steht jetzt auf der Einkaufsliste."), + ]) + + first, *edits = [content for _, content in channel.room_contents] + assert first["msgtype"] == "m.notice", "a notice: Matrix does not notify about it" + assert all(e["m.relates_to"] == {"rel_type": "m.replace", "event_id": "$notice1"} + for e in edits), "every change edits the one status" + assert len(edits[-1]["m.new_content"]["body"].splitlines()) >= 2 + assert [m.content for m in channel.sent] == ["Milch steht jetzt auf der Einkaufsliste."], \ + "tool hints never reach the chat; the answer is a message of its own" + + +def test_compacting_shows_as_a_status_line_not_as_nanobots_notice(nanobot, monkeypatch): + def steps(mods): + compaction = mods["nanobot.events"].ContextCompactionEvent + return [_outbound("Compressing context…", event=compaction(phase="started")), + _outbound("Context compacted.", event=compaction(phase="succeeded")), + _outbound("Erledigt.")] + channel = _slow_turn(nanobot, monkeypatch, steps) + + shown = channel.room_contents[-1][1] + body = (shown.get("m.new_content") or shown)["body"] + assert "Notizen" in body + assert [m.content for m in channel.sent] == ["Erledigt."] + + +def test_the_status_stops_with_the_answer(nanobot, monkeypatch): + import asyncio + + channel = _slow_turn(nanobot, monkeypatch, lambda mods: [_outbound("Erledigt.")]) + shown = len(channel.room_contents) + asyncio.run(asyncio.sleep(0.05)) + assert len(channel.room_contents) == shown + + # ── failure is contained, and visible ──────────────────────────────────── def test_a_moved_symbol_does_not_take_the_others_down(nanobot): diff --git a/tests/unit/stacklets/test_agent_status_trace.py b/tests/unit/stacklets/test_agent_status_trace.py new file mode 100644 index 00000000..73f6d7c7 --- /dev/null +++ b/tests/unit/stacklets/test_agent_status_trace.py @@ -0,0 +1,100 @@ +"""The status a slow turn shows: when it appears, and what it says. + +A turn answered quickly gets no status at all. A slow one gets one +status message, a line appended as the agent moves on (looking, looking +closer, writing down), never faster than one line every ten seconds, and +one "still on it" when nothing new happened for a while. The lines are +picked from a few per phase so the same turn never says the same thing +twice. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[3] / "stacklets" / "agent" / "runtime")) + +from status_trace import ( # noqa: E402 + FIRST_AFTER, GAP, PHRASES, STILL_AFTER, Trace, phase_of, +) + +DE = PHRASES["de"] + + +def _first(options): + return options[0] + + +def _trace() -> Trace: + return Trace(started=0.0, lang="de", pick=_first) + + +def test_a_quick_answer_gets_no_status(): + trace = _trace() + trace.tools(["memory_search"]) + assert trace.due(at=FIRST_AFTER - 0.5) is None + + +def test_a_slow_turn_says_what_the_agent_is_doing(): + trace = _trace() + trace.tools(["memory_search"]) + assert trace.due(at=FIRST_AFTER) == [DE["look"][0]] + + +def test_looking_again_after_a_look_reads_as_checking_closer(): + trace = _trace() + trace.tools(["memory_search"]) + trace.due(at=FIRST_AFTER) + trace.tools(["read_file"]) + assert trace.due(at=FIRST_AFTER + 1) is None, "not faster than one line in ten seconds" + assert trace.due(at=FIRST_AFTER + GAP) == [DE["look"][0], DE["check"][0]] + + +def test_writing_to_a_list_is_its_own_line(): + trace = _trace() + trace.tools(["exec"]) + trace.due(at=FIRST_AFTER) + trace.tools(["list_edit"]) + assert trace.due(at=FIRST_AFTER + GAP)[-1] == DE["write"][0] + + +def test_nothing_new_for_a_while_says_it_is_still_at_it_once(): + trace = _trace() + trace.tools(["memory_search"]) + trace.due(at=FIRST_AFTER) + assert trace.due(at=FIRST_AFTER + GAP) is None, "nothing new yet" + assert trace.due(at=FIRST_AFTER + STILL_AFTER)[-1] == DE["still"][0] + assert trace.due(at=FIRST_AFTER + STILL_AFTER + GAP * 3) is None + + +def test_a_turn_still_thinking_before_any_tool_says_so(): + assert _trace().due(at=FIRST_AFTER) == [DE["think"][0]] + + +def test_the_same_phase_twice_never_repeats_a_line(): + trace = _trace() + trace.tools(["memory_search"]) + trace.due(at=FIRST_AFTER) + trace.tools(["list_edit"]) + trace.due(at=FIRST_AFTER + GAP) + trace.tools(["read_file"]) + trace.due(at=FIRST_AFTER + 2 * GAP) + trace.tools(["list_edit"]) + lines = trace.due(at=FIRST_AFTER + 3 * GAP) + assert lines[-1] == DE["write"][1] and len(set(lines)) == len(lines) + + +def test_an_unknown_household_language_reads_english(): + trace = Trace(started=0.0, lang="fr", pick=_first) + assert trace.due(at=FIRST_AFTER) == [PHRASES["en"]["think"][0]] + + +def test_a_round_that_writes_is_writing_even_with_a_read_in_it(): + assert phase_of(["read_file", "list_edit"], looked_before=False) == "write" + + +def test_compacting_the_conversation_is_a_line_in_the_familys_words(): + trace = _trace() + trace.tidying() + assert trace.due(at=FIRST_AFTER) == [DE["tidy"][0]]