diff --git a/docs/adr/adr-012-nanobot-fork.md b/docs/adr/adr-012-nanobot-fork.md new file mode 100644 index 00000000..ed6ed455 --- /dev/null +++ b/docs/adr/adr-012-nanobot-fork.md @@ -0,0 +1,201 @@ +# ADR-012: Fork nanobot instead of shimming it + +## Status +Proposed + +Supersedes the "no fork" position recorded in +[docs/design/agent/addressing.md](../design/agent/addressing.md) and in +`stacklets/agent/runtime/README.md`, both of which already name the +condition for revisiting it: "once we accumulate several nanobot changes". +We have. + +## Context + +The agent stacklet runs upstream `nanobot-ai==0.2.2` unmodified and reshapes +it at runtime from `stacklets/agent/runtime/sitecustomize.py`, which Python +auto-imports because the directory is on `PYTHONPATH`. That file now installs +**ten shims over thirteen symbols**: + +| Shim | Patches | Kind | +|---|---|---| +| `brief` | `agent.context.runtime_lines` | public-ish | +| `lean_state` | `agent.context.ContextBuilder.build_messages` | public-ish | +| `memory_tool` | `agent.tools.loader.ToolLoader.discover` | addition | +| `person_tool` | same | addition | +| `history_tool` | same | addition | +| `grep_tool` | `agent.tools.search.GrepTool.execute` | replacement | +| `vault_write` | `WriteFileTool` / `EditFileTool` / `ApplyPatchTool.execute` | replacement | +| `name_trigger` | `channels.matrix.MatrixChannel._is_bot_mentioned` | **private** | +| `thread_trigger` | `_is_bot_mentioned`, `_on_message`, `_on_media_message` | **private** | +| `join_greeting` | `_on_room_invite`, `_handle_message` | **private** | + +Five of those symbols are underscore-prefixed. Upstream owes us nothing for +them, and has already moved one: 0.2.x turned `nanobot.channels.matrix` from a +module into a package, which detached a shim and took a live instance off +Matrix with no error anywhere. + +The decision to shim was right when there were two of them. This ADR records +what changed and what it has cost. + +## Lessons + +### 1. Failing soft makes a broken shim invisible + +Every shim is wrapped in try/except-and-log, deliberately: a broken shim must +never stop the agent answering. The price is that a shim which fails to attach +looks exactly like one that worked. The container starts, reports healthy, +logs a line nobody reads, and the capability is simply gone. Three vault tools +sat dead in the image for weeks that way. + +This is not a bug in the wrapping. It is the shape of the technique. A method +that is *supposed* to be there either is or is not; a monkeypatch that is +supposed to have replaced it has a third state, and that third state is silent. + +### 2. The shim tests cannot catch the thing that actually breaks shims + +`tests/stacklets/test_agent_runtime_shims.py` asserts every patch is attached, +which is real value: it catches *our* mistakes. It cannot catch upstream +moving a symbol, because it runs against a stub nanobot this repo hand-writes. +The stub still has the old symbol, so the lane stays green while production is +broken. The file says so itself. + +What actually holds this line is the version pin in the Dockerfile, which is a +promise to re-read thirteen symbols by hand on every bump. That is a manual +gate guarding an automated system, and it is the wrong way round. + +### 3. Private, synchronous seams force contorted shapes + +`thread_trigger` is the clearest case. The question "is this message in a +thread the agent is part of" needs the homeserver. The gate nanobot exposes +(`_is_bot_mentioned`) is synchronous. So the shim had to split in half: an +async pre-resolution wrapped around `_on_message` that settles the question +and remembers it, and a synchronous set lookup in the gate. Plus the same +wrapper again on `_on_media_message`, because there are two entry points. + +In a fork that is one `async def is_addressed(...)`. The split exists only +because we cannot change a method signature. + +### 4. Shims compose by accident, not by design + +`name_trigger` and `thread_trigger` both wrap `_is_bot_mentioned`. The second +wraps whatever the first left behind, so the chain is thread check, then name +check, then upstream's pill check. That works, and it is genuinely nice that +one failing leaves the other intact. + +But the order is implicit in the order of two `try` blocks in one file, and +nothing anywhere states the intended precedence. The next person to add a gate +shim will get it right by luck or not at all. + +### 5. The expensive bugs were not in the shims. They were under them + +The failure that cost a five-minute runaway loop on a live rig had nothing to +do with monkeypatching: + +* the container-side `stack` shim joined argv with spaces while the host + rebuilt it with `shlex.split`, so every multi-word query lost its boundaries; +* that shim exited 0 whatever happened, so a usage error reached the model + dressed as search results and it asked the same question again, forever. + +Both live at the seam between the agent and the rest of famstack. The shim +machinery is where the attention went; the defects were one layer down, in +plumbing nobody had a test for. **Complexity at one layer buys inattention at +the next.** + +### 6. Seams between two components are where contracts rot + +Two artifacts, both written against an output format that has never existed: + +* `grep_tool._PATH_RE` parses `#1 ... score=` to build a "Paths to read:" + block. `stack memory search` prints `2026-08-03 [Marge] path.md`. The regex + has never matched, so that block has always been empty. +* `memory_tool`'s description promises the model "rank, score, vault path, + snippet, and source links". There is no rank, no score, no source links. + +Nobody wrote these carelessly. They were written against an imagined +contract and never run against the real one, because the only thing that +exercises them is a live model in a container. + +### 7. A shim cannot fix a contract mismatch. It relocates it + +`grep_tool` routes vault greps into `memory_search` so the agent gets semantic +hits instead of literal ones. But `stack memory search` takes a Python regex, +so the routing changes *which* wrong answer the model gets, not whether it +gets one. See [the memory query-language note](../design-notes.md). + +The lesson generalises: shims are good at "call our code instead of theirs". +They are bad at "make two components agree", and reaching for one there hides +the disagreement instead of resolving it. + +## Decision + +Fork `nanobot-ai`, land the internals-patching shims as real code, and keep +the pure modules exactly as they are. + +**What moves into the fork.** Everything that patches a nanobot internal: +`brief`, `lean_state`, `grep_tool`, `vault_write`, `name_trigger`, +`thread_trigger`, `join_greeting`. Each becomes a method or a real extension +point rather than a replaced attribute. + +**What does not move.** The pure modules are the good part of the current +design and they survive the transition unchanged. `name_trigger.py` is text +in, bool out. `thread_trigger.py` is a policy plus two homeserver reads. +`brief.py` assembles from the vault and never raises. Each is unit-testable +without a container, and each is specified by tests that read as documentation +of intent. The fork calls them; it does not absorb them. + +That split is the thing to preserve: **the fork owns the wiring, our modules +own the decisions.** A fork that swallows the policy logic trades one +maintenance problem for a worse one. + +**The three vault tools are a separate question.** `memory_search`, +`memory_person` and `memory_history` are *additions* through +`ToolLoader.discover`, not replacements of upstream behaviour. If upstream +keeps a discovery seam they can stay outside the fork. Simpler is to move them +in with everything else and stop having two mechanisms. + +## Migration shape + +1. Fork at the currently pinned release. No behaviour change in step one, so + the diff is reviewable as "shims became methods". +2. Land the seven internals shims as real code, keeping the pure modules as + imports. +3. Replace stub-based attachment tests with tests against the real package. + A fork means we can import what we changed, which retires the whole class + of "green lane, broken production" described in lesson 2. +4. State the addressing precedence in one place (pill, name, thread) now that + it is one function instead of two wrappers. +5. Upstream the seams that are generally useful: a context-provider API, a + group-policy hook, an async addressing gate. Every accepted upstream patch + is a line the fork no longer carries. + +## Consequences + +**We own updates.** Today a nanobot release is `docker build`; afterwards it +is a rebase. That is the real cost and it is not small. + +It buys: symbol drift becomes a merge conflict instead of a silent runtime +detach; tests run against the actual code; the sync/async contortions go away; +and the pin stops being a manual thirteen-symbol audit. + +**Fork rot is the risk.** The mitigation is a rule, not a hope: anything that +could be upstream is offered upstream first, and the fork's diff is expected +to shrink over time. If it is still growing after two releases, that is the +signal to reconsider nanobot itself rather than to keep patching. + +## Alternatives considered + +**Keep shimming.** Rejected. The technique is sound at two or three patches +and we are at ten, five of them on private methods. Lesson 2 says the cost is +not "more maintenance" but "no way to know it is broken". + +**Vendor nanobot into the repo.** Rejected without a fork's upstream link: we +would inherit every maintenance cost and lose the path back. + +**Replace nanobot.** Out of scope here. Worth revisiting only if the fork +diff keeps growing. + +## Open + +* Where the fork lives. Arthur refers to reactivating an existing one; it is + not visible under `famstack-dev` or `arthware-dev` from this machine. +* Whether the vault tools move in or stay on the discovery seam. diff --git a/docs/design-notes.md b/docs/design-notes.md index 15cd11ed..58793c36 100644 --- a/docs/design-notes.md +++ b/docs/design-notes.md @@ -98,3 +98,48 @@ succeeded. The cost is a framework concept where today there is a plain function, so it needs to earn its place - but "a resource nobody created and nobody noticed" is the second time this pattern has cost a debugging session. + +## Two query languages, one hop, and only one caller has it (2026-08-04) + +`stack memory search ` matches the query as a **Python regex** against +file content. The archivist and the agent both search that vault, and only one +of them knows it. + +The archivist routes a chat question through `stacklets/docs/bot/recall.py`, +which is the hop. On a message ending in `?` it asks the classifier for 2-4 +keywords that would literally appear in a matching document, then OR-alternates +them (`"|".join(re.escape(k) for k in keywords)`) for the memory walker and +joins them with ` OR ` for Paperless, because Whoosh and `re` disagree about +what `|` means. Asked "What do we still need to buy for the camping trip?" it +reports `Searched for: Travel, Shopping, Trip` and answers with a citation. + +The agent's `memory_search` tool sends the sentence itself. As a regex that +looks for those exact words adjacent, which no file contains, so every +natural-language question returns nothing. The tool's own parameter +description says "Natural-language question or keywords", so the contract the +model is handed is not the contract the CLI implements. What the model does +with an empty non-answer is ask again, differently, which is the shape of the +loop in [ADR-012](adr/adr-012-nanobot-fork.md) lesson 5. + +**Where the hop belongs.** Not in the archivist. Own the resource, own the +concern: memory owns the vault and owns what a query means against it, the +same way the archivist owns filing because it owns Paperless. `recall.py` sits +in `stacklets/docs/bot/` for historical reasons, and the archivist already +imports `memory.lib`, so the dependency direction is established and points the +right way. + +Moving it makes every caller correct at once: the agent tool, the archivist, +`stack memory search` from a terminal, and whatever asks next. Leaving it +means the next consumer re-learns this the way the agent did. + +**What has to be decided when it moves.** The rewrite needs an LLM, so a +search command that has never called a model would start to, and that changes +its latency and its failure modes. Options are a flag (`--natural`), inferring +it from the trailing `?` the way the archivist does, or keeping the rewrite as +a lib function that callers opt into. The archivist also needs the keywords +back, not just the regex, because it shows `Searched for: ...` so a family can +see when a bad rewrite hid results; that visibility is worth keeping and the +return shape has to carry it. + +Both current callers should keep working unchanged through the move. That is +the test. diff --git a/docs/design/agent/addressing.md b/docs/design/agent/addressing.md index 2dfd3c73..e0739630 100644 --- a/docs/design/agent/addressing.md +++ b/docs/design/agent/addressing.md @@ -1,6 +1,6 @@ # Family Agent — Addressing & Activation Model -> Status: Design, deferred (capture now, build later) +> Status: Layer 2 (threads-as-conversation) **shipped**; layer 3 deferred > Applies to: the `agent` stacklet (Stacky, nanobot in a container) > Sibling docs: > - [plan.md](plan.md) — the agent implementation plan @@ -48,25 +48,36 @@ Constraints: A 1:1 room with `@stacky-bot` needs no mentions. This is the natural home for a private, fluid conversation with Stacky. No work required. -### 2. Threads-as-conversation (DECIDED — the priority fix) +### 2. Threads-as-conversation (SHIPPED, except the session key) > **Requirement (Arthur, with repro):** when a user **replies inside a thread > Stacky is part of**, Stacky must auto-respond **without a mention**. Repro: > Stacky posted the Itchy & Scratchy Land list; Homer replied in-thread "gibts > noch mehr?"; Stacky stayed silent because the thread reply carried no > `@`-mention. A thread you are in IS the conversation — no re-mentioning. -Make a **thread the conversation unit** in shared rooms. Three shims (step 0 is -the fix for the gap above — replies don't thread today): -0. **Make Stacky actually reply in a thread** — propagate the incoming thread root - through to the outbound reply metadata (it is dropped today), and start a thread - on the first reply so even a plain mention opens one. (Without this, even after - the gate change below, Stacky's answer would land in the main timeline, not the - thread.) +Make a **thread the conversation unit** in shared rooms. Three steps: + +0. ~~**Make Stacky actually reply in a thread**~~ — **no longer needed.** The + observation above predates the nanobot 0.2.x upgrade. In 0.2.2 the inbound + thread root does survive the agent loop: `_base_metadata` merges + `_thread_metadata(event)` on the way in, `loop.py`'s response builder copies + `msg.metadata` onto the `OutboundMessage`, and `send()` turns it back into an + `m.thread` relation via `_build_thread_relates_to`. A reply to a threaded + message threads. (nanobot still never *starts* a thread, so the first answer + to a top-level mention is top-level — the family opens the thread on it.) 1. Fold the thread root into the session key so each thread is its own scoped - memory (today all threads share the room session). -2. In `_should_process_message`, treat a message in a thread **whose root Stacky - authored, or where Stacky has already posted**, as addressed — so a thread reply - auto-responds with no mention. This is the core of the requirement above. + memory. **Still open:** all threads in a room share the room session, so two + parallel threads bleed context into each other. +2. Treat a message in a thread **whose root Stacky authored, or where Stacky has + already posted**, as addressed. **Shipped** as the `thread_trigger` shim + (`stacklets/agent/runtime/thread_trigger.py`), specified by + `tests/stacklets/test_agent_thread_trigger.py`. + + Two halves, because nanobot's gate is synchronous and the question is not: + `_on_message` (async) settles thread membership against the homeserver and + remembers a yes; `_is_bot_mentioned` (sync) reads that set. Scoped to threads + Stacky participates in rather than all threads, because the archivist and mail + bot thread in the same rooms and those conversations are theirs. Result: *mention once → Stacky opens a thread → the whole thread is a conversation, no re-mentioning → the thread is visibly scoped and just ends when you stop.* This diff --git a/stacklets/agent/client/stack b/stacklets/agent/client/stack old mode 100644 new mode 100755 index 59772efa..28f8e951 --- a/stacklets/agent/client/stack +++ b/stacklets/agent/client/stack @@ -8,26 +8,87 @@ container<->host boundary, so we use host.docker.internal - the same path famstack uses for oMLX/whisper). The API is core-owned and dual-mode: a plaintext line gets the CLI's plaintext output back (token-lean, no JSON), confined to an allowlist of read/domain commands. See stacklets/core/famstack-api.py. + +THE LINE HAS TO SURVIVE THE ROUND TRIP + +argv goes over the wire as one line and the server rebuilds it with +`shlex.split`, so this end has to quote the way shlex reads. Joining on a +bare space instead lost every argument boundary: `memory search "school run"` +arrived as four words, argparse rejected `run` as a stray positional, and +`memory_search` - the agent's main retrieval tool - failed on every query +with more than one word in it. `shlex.join` is the exact inverse of the +server's `shlex.split`, so what the tool built is what the CLI runs. + +AND SO DOES THE EXIT CODE + +A failed command that exits 0 is worse than one that crashes, because the +caller reads a usage message as a result. That is what turned the quoting +bug into a hang: `memory_search` checks the return code, saw success, handed +the model argparse's usage text as search output, and the model retried the +identical call for as long as anyone let it. The plaintext protocol carries +the status back as a trailing `stack-exit: N` line, added by the server only +when the command failed, stripped here and turned back into our own exit +code. Successful output is unchanged, byte for byte. """ import os +import shlex import socket import sys -host, port = os.environ.get("STACK_API_ADDR", "host.docker.internal:42001").rsplit(":", 1) -try: - # `with` closes the socket on every exit path, including errors. - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: - sock.settimeout(125) - sock.connect((host, int(port))) - sock.sendall((" ".join(sys.argv[1:]) + "\n").encode()) - sock.shutdown(socket.SHUT_WR) # done sending; half-close so the server reads EOF - while True: - chunk = sock.recv(4096) - if not chunk: - break - sys.stdout.write(chunk.decode(errors="replace")) - sys.stdout.flush() -except Exception as e: - sys.stderr.write(f"stack: cannot reach the famstack API ({e})\n") - sys.exit(1) +# Written by the server after a failed command; never part of real output. +EXIT_MARKER = "stack-exit:" + + +def wire_line(argv): + """Encode argv as the single line the API parses with `shlex.split`.""" + return shlex.join(argv) + "\n" + + +def split_exit_code(output): + """(text, exit_code) - peel the server's trailing status line, if any.""" + stripped = output.rstrip("\n") + body, sep, last = stripped.rpartition("\n") + candidate = last if sep else stripped + if candidate.startswith(EXIT_MARKER): + try: + code = int(candidate[len(EXIT_MARKER):].strip()) + except ValueError: + return output, 0 + text = body + "\n" if sep and body else "" + return text, code + return output, 0 + + +def main(argv): + host, port = os.environ.get( + "STACK_API_ADDR", "host.docker.internal:42001", + ).rsplit(":", 1) + try: + # `with` closes the socket on every exit path, including errors. + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.settimeout(125) + sock.connect((host, int(port))) + sock.sendall(wire_line(argv).encode()) + sock.shutdown(socket.SHUT_WR) # done sending; half-close so the server reads EOF + chunks = [] + while True: + chunk = sock.recv(4096) + if not chunk: + break + chunks.append(chunk) + except Exception as e: + sys.stderr.write(f"stack: cannot reach the famstack API ({e})\n") + return 1 + + # Buffered rather than streamed: the status line only arrives at the end, + # and printing it before peeling it off would put protocol noise in the + # model's context. Command output here is small by construction. + text, code = split_exit_code(b"".join(chunks).decode(errors="replace")) + sys.stdout.write(text) + sys.stdout.flush() + return code + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/stacklets/agent/docker-compose.yml b/stacklets/agent/docker-compose.yml index d9b5ae52..607b6428 100644 --- a/stacklets/agent/docker-compose.yml +++ b/stacklets/agent/docker-compose.yml @@ -47,6 +47,10 @@ services: # image runs standalone, but a bind mount means an edit + restart takes # effect without a rebuild. Keep in sync with the Dockerfile COPY. - ./runtime:/app/runtime:ro + # The `stack` shim, live for the same reason. It is the agent's only + # route to the CLI, so a bug in it takes out every vault tool at once — + # and it stayed baked-only long enough for one to go unnoticed. + - ./client/stack:/usr/local/bin/stack:ro restart: unless-stopped networks: diff --git a/stacklets/agent/runtime/memory_tool.py b/stacklets/agent/runtime/memory_tool.py index a1f98171..77c6c370 100644 --- a/stacklets/agent/runtime/memory_tool.py +++ b/stacklets/agent/runtime/memory_tool.py @@ -84,8 +84,18 @@ async def execute( stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=130) out = stdout.decode(errors="replace").strip() err = stderr.decode(errors="replace").strip() - if proc.returncode != 0: + # `stack memory search` exits 1 for "nothing matched", which is an + # answer. Only 2 and up (bad arguments, unreadable vault) are + # failures. Reporting an empty result as a failure tells the model + # to try again when the honest reply is that there is nothing there. + if proc.returncode not in (0, 1): return f"Error: memory search failed with exit {proc.returncode}: {err or out}" + # The status decides, not the text. A search that matched nothing + # reaches here as the API's generic "(no output)" placeholder, which + # reads like something went wrong; the model's next move after an + # ambiguous non-answer is to ask again. + if proc.returncode == 1: + return "(no memory results)" return out or "(no memory results)" diff --git a/stacklets/agent/runtime/sitecustomize.py b/stacklets/agent/runtime/sitecustomize.py index d380cd65..9172ef57 100644 --- a/stacklets/agent/runtime/sitecustomize.py +++ b/stacklets/agent/runtime/sitecustomize.py @@ -41,7 +41,12 @@ agent by its configured name counts as a mention, not just an autocompleted pill. Families type "Stacky, what's on our list?". -8. join_greeting (join_greeting.py) — on being invited, take one turn and +8. thread_trigger (thread_trigger.py) — a message inside a thread the agent is + part of counts as a mention too. A thread is a conversation; nobody repeats + the name on every line of one. Scoped to threads the agent participates in, + because the archivist and mail bot thread in the same rooms. + +9. join_greeting (join_greeting.py) — on being invited, take one turn and introduce the room's topic instead of joining in silence. WHY SHIMS AND NOT A FORK @@ -71,6 +76,12 @@ (all three are `async def`; a sync replacement returns a str into the loop's `await` and the tool call dies with TypeError) name_trigger: `nanobot.channels.matrix.MatrixChannel._is_bot_mentioned(self, event) -> bool` + thread_trigger: `MatrixChannel._is_bot_mentioned` (same symbol, wrapped after it) + `MatrixChannel._on_message(self, room, event)` (async) + `MatrixChannel._on_media_message(self, room, event)` (async) + `MatrixChannel.client` (nio AsyncClient), `MatrixChannel.config.user_id` + — and that `groupPolicy: mention` still routes through + `_is_bot_mentioned` (config.json sets that policy) join_greeting: `nanobot.channels.matrix.MatrixChannel._on_room_invite(self, room, event)` `MatrixChannel._handle_message(sender_id, chat_id, content, metadata, is_dm)` @@ -207,6 +218,57 @@ def _is_bot_mentioned(self, event): _log.exception("name-trigger shim could not attach (nanobot internals changed?)") +# ── thread_trigger: a reply inside the agent's own thread is addressed to it ── +# The threaded half of the same gate, in two parts because nanobot's gate is +# synchronous and the question is not: `_on_message` (async) settles whether the +# thread is ours and remembers it, `_is_bot_mentioned` (sync) reads that answer. +# Wraps whatever `_is_bot_mentioned` is by now, so pill mentions and the name +# matcher keep working and this only gets a say when both said no — and so a +# name-trigger that failed to attach costs only itself. +try: + import nanobot.channels.matrix as _matrix_thread + from thread_trigger import AgentThreads as _AgentThreads + from thread_trigger import thread_root as _thread_root + + _agent_threads = _AgentThreads() + + def _learn_threads(orig): + """Settle the thread question before nanobot's gate asks it.""" + async def _wrapped(self, room, event): + root = _thread_root(event) + if root and not _agent_threads.includes(root): + # Swallows its own failures; a homeserver hiccup must not + # stop the message being routed by the other rules. + await _agent_threads.observe( + self.client, room.room_id, root, self.config.user_id, + ) + return await orig(self, room, event) + return _wrapped + + _orig_mentioned_pre_thread = _matrix_thread.MatrixChannel._is_bot_mentioned + + def _is_bot_mentioned_or_our_thread(self, event): + if _orig_mentioned_pre_thread(self, event): + return True + try: + root = _thread_root(event) + return bool(root and _agent_threads.includes(root)) + except Exception: + _log.exception("thread trigger failed; addressing by name still works") + return False + + _matrix_thread.MatrixChannel._is_bot_mentioned = _is_bot_mentioned_or_our_thread + _matrix_thread.MatrixChannel._on_message = _learn_threads( + _matrix_thread.MatrixChannel._on_message, + ) + _matrix_thread.MatrixChannel._on_media_message = _learn_threads( + _matrix_thread.MatrixChannel._on_media_message, + ) + _log.info("thread-trigger mention shim active") +except Exception: + _log.exception("thread-trigger shim could not attach (nanobot internals changed?)") + + # ── join_greeting: say something useful the moment you are invited ─────────── # Stock nanobot joins an invite silently. In a topic room that silence is the # family's first impression of the agent, so it takes one ordinary turn instead diff --git a/stacklets/agent/runtime/thread_trigger.py b/stacklets/agent/runtime/thread_trigger.py new file mode 100644 index 00000000..70c8dc48 --- /dev/null +++ b/stacklets/agent/runtime/thread_trigger.py @@ -0,0 +1,173 @@ +"""Is this message part of a conversation the agent is already having? + +`name_trigger.py` answers "was the agent addressed *in this sentence*". +That is the right question for the main timeline, where every message +lands next to every other one and the only thing tying a request to a +responder is the name in it. Inside a Matrix thread it is the wrong +question. A thread *is* the tie: it is a bounded conversation with a +root, and nobody says the other person's name on every line of one. + + Stacky, what do we still need for camping? + └─ thread + Stacky: Tent poles, the gas cartridge, ... + and the sleeping mats? <- addressed, no name + put those on the list too <- addressed, no name + +So this module supplies the other half of the gate: a message in a +thread the agent is part of counts as being spoken to. + +WHICH THREADS + +Not all of them, and that boundary is the entire design. famstack's +other bots thread as well -- the archivist answers a filing under the +document that was uploaded, the mail bot posts an email body under its +card -- and those threads are *their* conversations, in the same rooms. +An agent that claimed every thread would answer into every filing +discussion in the house, breaking the rule the archivist already applies +to itself: exactly one component responds to a message. + +A thread is the agent's when the agent participates in it, which is two +facts on the homeserver: + + 1. the thread hangs off something the agent said, or + 2. the agent has posted in the thread. + +(1) is the ordinary case -- the agent answers, someone opens a thread on +that answer. (2) is how the agent joins a thread that started as +something else: named inside an archivist filing thread, it replies +there, and from then on the thread is a conversation with it. + +Both are read from Matrix rather than remembered, so a restart loses +nothing: the agent's own message is still in the thread afterwards. + +WHY THE ANSWER IS CACHED ONE WAY ONLY + +A positive is permanent -- a message cannot leave a thread -- so it is +kept, and a busy thread costs one lookup instead of one per line. A +negative is not cached, because a thread the agent has nothing to do +with at 10:00 is one it was invited into at 10:01. + +RELATION TO `MicroBot.get_thread_root` + +The bot framework reads the same `m.thread` relation, and scans thread +children the same way in `_thread_envelopes`. It is not imported here: +the agent is a separate image running the nanobot harness, with neither +`microbot.py` (which needs aiohttp, markdown, loguru and the bot +framework's own room context) nor `lib/stack` mounted -- and mounting +`lib/stack` pulls the whole CLI framework in through its package init +for the sake of six lines. What is duplicated is a read of the Matrix +spec, not of anyone's implementation, which is why the duplication is +acceptable and the tests state the shared contract. Sharing the +matcher properly is tracked as step 4 of docs/design/brain/write-layer.md. +""" + +from __future__ import annotations + +import logging + +_log = logging.getLogger("agent.runtime.thread") + +# How many of a thread's messages to read looking for one of the agent's +# own. Newest first, so a conversation the agent is in is found in the +# first few; the cap is what stops one chat message in a thread with +# hundreds of replies from becoming hundreds of API calls. +_SCAN_LIMIT = 20 + + +def thread_root(event) -> str | None: + """The id of the thread `event` belongs to, or None if it is top-level. + + Reads the relation the sender's own client wrote, so there is no + round trip and no bookkeeping: Matrix is the ledger. A plain reply + (`m.in_reply_to` with no `rel_type`) is a quote, not a conversation, + and is deliberately not a thread. An event of an unexpected shape + simply is not in one -- routing must never raise on it. + """ + source = getattr(event, "source", None) + content = source.get("content") if isinstance(source, dict) else None + relation = content.get("m.relates_to") if isinstance(content, dict) else None + if not isinstance(relation, dict) or relation.get("rel_type") != "m.thread": + return None + root = relation.get("event_id") + return root if isinstance(root, str) and root else None + + +class AgentThreads: + """The threads this agent is a participant in. + + One instance per running channel. `observe` asks the homeserver + about a thread the first time a message arrives in it and remembers + a yes; `includes` is the cheap read the message gate uses, so the + part of this that runs on the synchronous routing path is a set + lookup and nothing else. + """ + + def __init__(self) -> None: + self._ours: set[str] = set() + + def includes(self, root: str) -> bool: + """Whether `root` is known to be a thread the agent is in.""" + return root in self._ours + + async def observe( + self, client, room_id: str, root: str, agent_user_id: str, + *, limit: int = _SCAN_LIMIT, + ) -> bool: + """Settle whether the thread at `root` is the agent's, and record it. + + Best-effort by construction: every homeserver failure reads as + "not the agent's thread". Losing a threaded follow-up costs the + family one repeated name; an exception escaping here would come + out of the channel's message handler and take the agent off + Matrix for every room at once. + """ + if root in self._ours: + return True + if await self._agent_in_thread(client, room_id, root, agent_user_id, limit): + self._ours.add(root) + return True + return False + + async def _agent_in_thread( + self, client, room_id: str, root: str, agent_user_id: str, limit: int, + ) -> bool: + # Cheapest question first: a thread rooted at the agent's own + # message is the agent's, and that is one fetch with no paging. + try: + resp = await client.room_get_event(room_id, root) + if getattr(getattr(resp, "event", None), "sender", None) == agent_user_id: + return True + except Exception: + # Fall through rather than return: the thread scan below can + # still answer yes, and it is the more informative of the two. + _log.debug("thread root fetch failed for %s", root, exc_info=True) + + try: + examined = 0 + async for related in client.room_get_event_relations( + room_id, root, _thread_relationship(), + ): + examined += 1 + if examined > limit: + break + if getattr(related, "sender", None) == agent_user_id: + return True + except Exception: + _log.debug("thread relations fetch failed for %s", root, exc_info=True) + return False + + +def _thread_relationship(): + """nio's `RelationshipType.thread`, or the wire value if nio is absent. + + The tests drive this module with a hand-written client so they can + state the rule without a homeserver, and that client does not need + nio installed to do it. Importing lazily keeps the dependency where + it belongs -- in the container -- instead of in the spec. + """ + try: + from nio.api import RelationshipType + + return RelationshipType.thread + except Exception: + return "m.thread" diff --git a/stacklets/core/famstack-api.py b/stacklets/core/famstack-api.py index b46769d2..b841a3ba 100644 --- a/stacklets/core/famstack-api.py +++ b/stacklets/core/famstack-api.py @@ -150,34 +150,57 @@ def handle_request(data): return {"error": str(e)} +# How a failed command reports itself over the plaintext protocol, which +# otherwise carries only the CLI's text. Silence here is not neutral: the agent +# reads a usage message as a result and retries the same call forever, which is +# how one mis-parsed search argument became an unbounded tool loop. Appended +# only on failure, so successful output stays byte-identical. +EXIT_MARKER = "stack-exit:" + + +def _with_exit(text, code): + """The reply body, plus a status line when the command failed.""" + if not code: + return text + if not text.endswith("\n"): + text += "\n" + return f"{text}{EXIT_MARKER} {code}\n" + + def handle_plaintext(line): """Run one allowlisted `stack` command in the CLI's text mode. The token-lean counterpart to `handle_request`: the agent sends a plaintext command line and gets the CLI's normal text output back (no `--json`), so its context stays small. Confined to `DOMAIN_ALLOW` -- never lifecycle ops. + + Every path that is not a command that ran and succeeded reports a non-zero + status, refusals included. A refusal the caller cannot distinguish from an + answer is a retry loop waiting to happen. """ try: args = shlex.split(line) except ValueError as e: - return f"error: could not parse command ({e})\n" + return _with_exit(f"error: could not parse command ({e})\n", 2) if not args: - return "error: empty command\n" + return _with_exit("error: empty command\n", 2) if not any(args[:len(p)] == p for p in DOMAIN_ALLOW): allowed = ", ".join(" ".join(p) for p in DOMAIN_ALLOW) - return f"error: '{' '.join(args[:2])}' is not allowed. Allowed: {allowed}\n" + return _with_exit( + f"error: '{' '.join(args[:2])}' is not allowed. Allowed: {allowed}\n", 126, + ) if _is_denied_write(args): - return DENY_HINT + return _with_exit(DENY_HINT, 126) try: r = subprocess.run( [str(STACK_BIN), *args], capture_output=True, text=True, timeout=120, cwd=str(REPO_ROOT), ) - return r.stdout or r.stderr or "(no output)\n" + return _with_exit(r.stdout or r.stderr or "(no output)\n", r.returncode) except subprocess.TimeoutExpired: - return "error: command timed out\n" + return _with_exit("error: command timed out\n", 124) except Exception as e: - return f"error: {e}\n" + return _with_exit(f"error: {e}\n", 1) def handle_client(conn): diff --git a/stacklets/messages/cli/_matrix.py b/stacklets/messages/cli/_matrix.py index ec6291ed..a1b5f485 100644 --- a/stacklets/messages/cli/_matrix.py +++ b/stacklets/messages/cli/_matrix.py @@ -428,7 +428,7 @@ def list_users(self): # ── Messaging ──────────────────────────────────────────────────────── - def send(self, room, message, html=None, mentions=None): + def send(self, room, message, html=None, mentions=None, thread_root=None): """Send a text message to a room (by alias or ID). Resolves aliases automatically. If html is provided, sends a @@ -437,6 +437,11 @@ def send(self, room, message, html=None, mentions=None): That payload, not the display name in the text, is what a bot's mention gate keys on, so it is how you get a bot (like the agent) to answer in a group room. Returns (ok, detail). + + `thread_root` posts the message inside the thread rooted at that + event id, the way a client's "Reply in thread" does. A thread is a + conversation the agent treats as addressed to it once it is part of + one, so this is how that path is exercised from the terminal. """ if room.startswith("!"): room_id = room @@ -452,6 +457,15 @@ def send(self, room, message, html=None, mentions=None): body["formatted_body"] = html if mentions: body["m.mentions"] = {"user_ids": [self._full_user(u) for u in mentions]} + if thread_root: + # `is_falling_back` plus the in-reply-to pointer is what a real + # client sends, so thread-blind clients still render it in context. + body["m.relates_to"] = { + "rel_type": "m.thread", + "event_id": thread_root, + "is_falling_back": True, + "m.in_reply_to": {"event_id": thread_root}, + } status, resp = _put( self._url(f"/_matrix/client/v3/rooms/{room_id}/send/m.room.message/{txn}"), body, diff --git a/stacklets/messages/cli/read.py b/stacklets/messages/cli/read.py index 5cee1c09..4eb824fe 100644 --- a/stacklets/messages/cli/read.py +++ b/stacklets/messages/cli/read.py @@ -1,14 +1,19 @@ """ -stack messages read [--limit N] — show a room's recent messages +stack messages read [--limit N] [--ids] — show a room's recent messages Prints the last N messages (default 20) in a room, oldest-first, as `HH:MM sender: text`. Reads through the Synapse admin API, so it works for any room without the admin having to be a member — handy for checking what a bot replied after a capture, or reading back a conversation from the terminal. +`--ids` prints each message's event id underneath it. That is what +`stack messages send --thread ` takes, so the two together let you +reply in a thread on a message you did not send — the bot's own answer, say. + Examples: stack messages read chat stack messages read topic-camping --limit 5 + stack messages read thread-rig --ids # ids for --thread stack messages read '!abc123:home' # a room ID works too The room can be a bare alias ('chat'), a full alias ('#chat:home'), or a room @@ -29,29 +34,34 @@ def _parse_args(argv): - """(room, limit, error) from the raw arg list.""" + """(room, limit, show_ids, error) from the raw arg list.""" limit = 20 + show_ids = False rest = [] i = 0 while i < len(argv): if argv[i] == "--limit": if i + 1 >= len(argv): - return None, None, "--limit needs a number" + return None, None, None, "--limit needs a number" try: limit = int(argv[i + 1]) except ValueError: - return None, None, f"--limit wants a number, got {argv[i + 1]!r}" + return None, None, None, f"--limit wants a number, got {argv[i + 1]!r}" i += 2 continue + if argv[i] == "--ids": + show_ids = True + i += 1 + continue rest.append(argv[i]) i += 1 if not rest: - return None, None, "Usage: stack messages read [--limit N]" - return rest[0], limit, None + return None, None, None, "Usage: stack messages read [--limit N] [--ids]" + return rest[0], limit, show_ids, None def run(args, stacklet, config): - room, limit, err = _parse_args(args or []) + room, limit, show_ids, err = _parse_args(args or []) if err: return {"error": err} @@ -81,12 +91,19 @@ def run(args, stacklet, config): sender = ev.get("sender", "?").split(":")[0].lstrip("@") body = ev.get("content", {}).get("body", "") ts = ev.get("origin_server_ts") + event_id = ev.get("event_id", "") when = datetime.fromtimestamp(ts / 1000).strftime("%H:%M") if ts else "--:--" first, *more = body.split("\n") print(f"{when} {sender}: {first}") for line in more: # indent continuation lines so replies stay readable print(f" {line}") - messages.append({"sender": sender, "body": body, "ts": ts}) + if show_ids: + # The id is what `send --thread` takes, so print it where you can + # copy it: under the message it belongs to, not in a separate list. + print(f" [{event_id}]") + messages.append( + {"sender": sender, "body": body, "ts": ts, "event_id": event_id}, + ) if not messages: print(f"(no messages in {room})") diff --git a/stacklets/messages/cli/send.py b/stacklets/messages/cli/send.py index 80ff6e0a..3748b5c0 100644 --- a/stacklets/messages/cli/send.py +++ b/stacklets/messages/cli/send.py @@ -1,5 +1,5 @@ """ -stack messages send "message" [--as ] [--mention ] — send a message to a room +stack messages send "message" [--as ] [--mention ] [--thread ] — send a message to a room Sends a plain text message to the specified room. By default it posts as stacker-bot (the system account); pass `--as ` to post as a family @@ -12,6 +12,12 @@ in the text alone does not, because the bot keys on the mention payload, not the display name. +Pass `--thread ` to post inside a thread, the way a client's "Reply +in thread" does. The event id comes back from the send that started the thread, +so a whole threaded conversation can be driven from the terminal. The agent +treats a thread it is part of as addressed to it, so this is also how you check +that a follow-up needs no mention at all. + This is the building block other stacklets use for notifications: - photos could notify #notifications when a backup completes - docs could notify when a new document is archived @@ -94,6 +100,7 @@ def run(args, stacklet, config): # sys.argv: ['stack', 'messages', 'send', '', '', ...] sender = None mentions = [] + thread_root = None rest = [] argv = sys.argv[3:] # skip 'stack', 'messages', 'send' i = 0 @@ -110,10 +117,17 @@ def run(args, stacklet, config): mentions.append(argv[i + 1]) i += 2 continue + if argv[i] == "--thread": + if i + 1 >= len(argv): + return {"error": "--thread needs an event id"} + thread_root = argv[i + 1] + i += 2 + continue rest.append(argv[i]) i += 1 if len(rest) < 2: - return {"error": 'Usage: stack messages send "message" [--as ] [--mention ]'} + return {"error": 'Usage: stack messages send "message" [--as ] ' + '[--mention ] [--thread ]'} room = rest[0] message = " ".join(rest[1:]) @@ -142,7 +156,10 @@ def run(args, stacklet, config): html = _simple_markdown_to_html(message) - ok, detail = client.send(room, message, html=html, mentions=mentions or None) + ok, detail = client.send( + room, message, html=html, mentions=mentions or None, + thread_root=thread_root, + ) if ok: return {"ok": True, "room": room, "event_id": detail} else: diff --git a/tests/framework/test_compose_bind_mounts.py b/tests/framework/test_compose_bind_mounts.py new file mode 100644 index 00000000..28b4d6a4 --- /dev/null +++ b/tests/framework/test_compose_bind_mounts.py @@ -0,0 +1,94 @@ +"""A script mounted over a command has to be executable in the repo. + +Bind mounting a file into a container replaces the image's copy *and* its +permissions. A Dockerfile that carefully does `chmod +x` on the file it +copied buys nothing once compose mounts the host's version over the top: +what the container gets is whatever mode the file has in the working tree. + +That is not hypothetical. Mounting `stacklets/agent/client/stack` over +`/usr/local/bin/stack` so the shim could be edited without a rebuild took +the agent's only route to the CLI offline, because the repo file was 644. +Every vault tool failed at once with `executable file not found in $PATH` -- +from a two-line compose change that looked like pure convenience. + +The mode is tracked by git, so this is an audit of the repo rather than of +one machine: a fresh clone gets the same bit, and so does CI. +""" + +from __future__ import annotations + +import os +import re +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent.parent + +# `- ./source:/container/target[:ro]` in a compose volumes list. Sources +# built from ${VARIABLES} are host data directories, not repo files, and +# are skipped -- there is nothing in the tree to check their mode against. +_MOUNT_RE = re.compile( + r"^\s*-\s+(?P\.[^\s:]+):(?P/[^\s:]+)(?::[a-z,]+)?\s*$", + re.MULTILINE, +) + +# Where something has to be executable to be reachable at all: these are the +# directories on a container's PATH. +_BIN_DIRS = ("/usr/local/bin/", "/usr/bin/", "/bin/", "/usr/local/sbin/") + + +def _compose_files() -> list[Path]: + return sorted(REPO_ROOT.glob("stacklets/*/docker-compose*.yml")) + + +def _mounts_onto_commands() -> list[tuple[Path, Path, str]]: + """(compose file, host source, container target) for every bind mount + that lands on a PATH directory.""" + found = [] + for compose in _compose_files(): + for match in _MOUNT_RE.finditer(compose.read_text(encoding="utf-8")): + target = match.group("target") + if not target.startswith(_BIN_DIRS): + continue + source = (compose.parent / match.group("source")).resolve() + found.append((compose, source, target)) + return found + + +def test_a_mounted_command_is_executable_in_the_tree(): + """Otherwise the container has the file and cannot run it.""" + not_executable = [ + (compose.relative_to(REPO_ROOT), source.relative_to(REPO_ROOT), target) + for compose, source, target in _mounts_onto_commands() + if source.is_file() and not os.access(source, os.X_OK) + ] + + assert not not_executable, "\n".join( + f"{compose} mounts {source} over {target}, but {source} is not " + f"executable — the container will report 'executable file not found'. " + f"Fix with: chmod +x {source}" + for compose, source, target in not_executable + ) + + +def test_a_mounted_command_actually_exists(): + """A typo in the source path mounts an empty directory over the command, + which fails the same way and reads like a missing binary.""" + missing = [ + (compose.relative_to(REPO_ROOT), match_source, target) + for compose, match_source, target in _mounts_onto_commands() + if not match_source.exists() + ] + + assert not missing, f"compose mounts a source that is not in the tree: {missing}" + + +def test_the_audit_is_looking_at_something(): + """Guards the guard: if the mount pattern stops matching, both tests + above pass by finding nothing, which is how an audit quietly dies.""" + found = _mounts_onto_commands() + + assert found, ( + "no compose bind mount onto a PATH directory was found — either the " + "agent's `stack` shim mount was removed, or _MOUNT_RE no longer " + "matches the compose volume syntax" + ) diff --git a/tests/stacklets/conftest.py b/tests/stacklets/conftest.py index 6e46068e..0e053ad4 100644 --- a/tests/stacklets/conftest.py +++ b/tests/stacklets/conftest.py @@ -97,13 +97,31 @@ async def execute(self, edits=None, **kwargs): class MatrixChannel: def __init__(self): self.client = types.SimpleNamespace(rooms={}) + self.config = types.SimpleNamespace( + user_id="@stacky-bot:home.local", group_policy="mention", + ) self.handled = [] self.joined = [] + self.processed = [] def _is_bot_mentioned(self, event): # Stock nanobot: only an autocompleted pill counts. return getattr(event, "pill_mention", False) + def _should_process_message(self, room, event): + # Stock nanobot under `groupPolicy: mention`, which is what + # the agent ships with: the mention gate decides. + return self._is_bot_mentioned(event) + + async def _on_message(self, room, event): + # Stock nanobot: gate, then hand the text to the agent. + if self._should_process_message(room, event): + self.processed.append(event) + + async def _on_media_message(self, room, event): + if self._should_process_message(room, event): + self.processed.append(event) + async def _on_room_invite(self, room, event): # Stock nanobot: join, say nothing. self.joined.append(room.room_id) diff --git a/tests/stacklets/test_agent_runtime_shims.py b/tests/stacklets/test_agent_runtime_shims.py index 64b601ae..33f4f7b8 100644 --- a/tests/stacklets/test_agent_runtime_shims.py +++ b/tests/stacklets/test_agent_runtime_shims.py @@ -25,7 +25,7 @@ SHIMMED_MODULES = ("sitecustomize", "brief", "lean_state", "memory_tool", "person_tool", "history_tool", "grep_tool", - "name_trigger", "join_greeting", "vault_write") + "name_trigger", "thread_trigger", "join_greeting", "vault_write") # The stub nanobot itself lives in conftest as `nanobot_stub`, shared with @@ -143,6 +143,96 @@ class _Event: assert channel._is_bot_mentioned(_Event()) +def _threaded(root, sender="@marge:home.local"): + import types as _types + + return _types.SimpleNamespace(sender=sender, source={"content": { + "body": "and the sleeping mats?", + "m.relates_to": {"rel_type": "m.thread", "event_id": root}, + }}) + + +def _room_with_thread(channel, *, root_sender, replies=()): + """Give the stub channel a homeserver holding one thread.""" + import types as _types + + async def room_get_event(room_id, event_id): + return _types.SimpleNamespace( + event=_types.SimpleNamespace(sender=root_sender), + ) + + def room_get_event_relations(room_id, event_id, rel_type=None, **kwargs): + async def _iter(): + for sender in replies: + yield _types.SimpleNamespace(sender=sender) + return _iter() + + channel.client.room_get_event = room_get_event + channel.client.room_get_event_relations = room_get_event_relations + return _types.SimpleNamespace(room_id="!family:home.local") + + +def test_a_reply_in_the_agents_own_thread_is_processed(nanobot): + """Without this shim a thread with the agent stalls after one turn. + + Driven through `_on_message` rather than the matcher, because the + shim is in two halves — an async lookup and a sync gate — and only + the whole path proves they are wired to each other. The message + carries no pill and no name: the thread is the entire signal. + """ + import asyncio + + mods = nanobot() + channel = mods["nanobot.channels.matrix"].MatrixChannel() + room = _room_with_thread(channel, root_sender=channel.config.user_id) + event = _threaded("$stacky-answer") + + asyncio.run(channel._on_message(room, event)) + + assert channel.processed == [event] + + +def test_another_bots_thread_is_left_alone(nanobot): + """The shim widens the gate; it must not open it. + + The archivist answers a filing under the uploaded document, in the + same family room. Every reply there would reach the agent too if + "in a thread" were the rule, and the family would get two bots + talking over one receipt. + """ + import asyncio + + mods = nanobot() + channel = mods["nanobot.channels.matrix"].MatrixChannel() + room = _room_with_thread( + channel, root_sender="@archivist-bot:home.local", + replies=["@homer:home.local"], + ) + + asyncio.run(channel._on_message(room, _threaded("$archivist-card"))) + + assert channel.processed == [] + + +def test_a_top_level_message_still_needs_addressing(nanobot): + """Threads change nothing about the main timeline. A message with no + thread, no pill and no name is not for the agent.""" + import asyncio + import types as _types + + mods = nanobot() + channel = mods["nanobot.channels.matrix"].MatrixChannel() + room = _types.SimpleNamespace(room_id="!family:home.local") + event = _types.SimpleNamespace( + sender="@marge:home.local", body="dinner at seven", + source={"content": {"body": "dinner at seven"}}, + ) + + asyncio.run(channel._on_message(room, event)) + + assert channel.processed == [] + + def test_an_invite_produces_a_greeting_turn(nanobot): """Joining in silence is the behaviour this replaces. diff --git a/tests/stacklets/test_agent_thread_trigger.py b/tests/stacklets/test_agent_thread_trigger.py new file mode 100644 index 00000000..b68116c0 --- /dev/null +++ b/tests/stacklets/test_agent_thread_trigger.py @@ -0,0 +1,255 @@ +"""A thread is a conversation. Whose? + +Matrix threads are how a family has a back-and-forth without flooding the +room, and nobody re-types "Stacky," on every line of one. So a message +posted inside a thread the agent is having counts as talking to the +agent, the same way the second sentence of a phone call does not need +the other person's name in it. + +Read this file as the spec for *which* threads those are, because the +answer is not "all of them". Other famstack bots thread too: the +archivist answers a filing under the upload it filed, the mail bot puts +an email's body under its card. Those threads are their conversations. +Treating every thread as the agent's would put it in the middle of every +document the house files, which is the opposite of the rule the archivist +already applies to itself -- exactly one component responds to a message. + +The test for "is this ours" is participation: the thread hangs off +something the agent said, or the agent has spoken in it. Both are facts +on the homeserver, so nothing here depends on the agent remembering +anything across a restart. +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from types import SimpleNamespace + +import pytest + +_REPO_ROOT = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(_REPO_ROOT / "stacklets" / "agent" / "runtime")) + +from thread_trigger import AgentThreads, thread_root # noqa: E402 + +AGENT = "@stacky-bot:home.local" +ROOM = "!family:home.local" + + +def _in_thread(root: str): + """A message a family member typed inside the thread rooted at `root`.""" + return SimpleNamespace( + sender="@marge:home.local", + source={"content": { + "body": "and the tent poles?", + "m.relates_to": { + "rel_type": "m.thread", + "event_id": root, + "is_falling_back": True, + "m.in_reply_to": {"event_id": "$whatever:home.local"}, + }, + }}, + ) + + +def _said_by(sender: str): + return SimpleNamespace(sender=sender, source={"content": {"body": "..."}}) + + +class _FakeClient: + """The homeserver, as much of it as this decision reads. + + Two lookups: fetch one event, and iterate a thread's children. Both + are what nio offers and what the archivist already uses for the same + question, so the shape here is the real one, not a convenience. + """ + + def __init__(self): + self.events: dict[str, object] = {} + self.children: dict[str, list] = {} + self.event_lookups: list[str] = [] + self.relation_lookups: list[str] = [] + self.get_event_raises: Exception | None = None + self.relations_raise: Exception | None = None + + async def room_get_event(self, room_id, event_id): + self.event_lookups.append(event_id) + if self.get_event_raises: + raise self.get_event_raises + return SimpleNamespace(event=self.events.get(event_id)) + + def room_get_event_relations(self, room_id, event_id, rel_type=None, **kwargs): + self.relation_lookups.append(event_id) + outer = self + + async def _iter(): + if outer.relations_raise: + raise outer.relations_raise + for event in outer.children.get(event_id, []): + yield event + + return _iter() + + +class TestReadingTheThreadRelation: + """`thread_root` — which thread an event belongs to, straight off the + event. No round trip: the relation the sender's client wrote is the + authoritative answer. + + The sibling of `MicroBot.get_thread_root`, which cannot be imported + here (the agent runs its own image with neither the bot framework nor + `lib/stack` mounted). The two read the same Matrix relation, so this + class is also the check that they agree. + """ + + def test_a_threaded_message_names_its_root(self): + assert thread_root(_in_thread("$root:home.local")) == "$root:home.local" + + def test_a_plain_reply_is_not_a_thread(self): + # Quoting a message is not joining a conversation. Element sends + # `m.in_reply_to` without a `rel_type` for that. + event = SimpleNamespace(source={"content": { + "m.relates_to": {"m.in_reply_to": {"event_id": "$x:home.local"}}, + }}) + assert thread_root(event) is None + + def test_a_top_level_message_has_no_root(self): + assert thread_root(SimpleNamespace(source={"content": {"body": "hi"}})) is None + + def test_an_eventless_argument_is_safe(self): + # Routing must never raise on a shape we did not expect. + assert thread_root(None) is None + assert thread_root(SimpleNamespace(source="not a dict")) is None + + +class TestThreadsTheAgentIsPartOf: + + @pytest.mark.asyncio + async def test_a_thread_hanging_off_the_agents_answer_is_its_conversation(self): + """The common case: the agent answers, someone opens a thread on + that answer to follow up. The follow-up is obviously for the agent + and must not need the name again.""" + client = _FakeClient() + client.events["$stacky-answer"] = _said_by(AGENT) + threads = AgentThreads() + + assert await threads.observe(client, ROOM, "$stacky-answer", AGENT) + + @pytest.mark.asyncio + async def test_a_thread_the_agent_has_spoken_in_is_its_conversation(self): + """The other way in: someone names the agent inside a thread that + started as something else (a filed document, an email), the agent + replies there, and the conversation continues. Its own message in + the thread is what makes the rest of it addressed.""" + client = _FakeClient() + client.events["$archivist-card"] = _said_by("@archivist-bot:home.local") + client.children["$archivist-card"] = [ + _said_by("@homer:home.local"), _said_by(AGENT), + ] + threads = AgentThreads() + + assert await threads.observe(client, ROOM, "$archivist-card", AGENT) + + @pytest.mark.asyncio + async def test_another_bots_thread_is_not_the_agents_to_answer(self): + """The whole reason this is not "any thread": the archivist files + a document and answers under it, and the family talks back in that + thread. Those messages belong to the archivist. An agent that + answered them too would make every filing a two-bot argument.""" + client = _FakeClient() + client.events["$archivist-card"] = _said_by("@archivist-bot:home.local") + client.children["$archivist-card"] = [_said_by("@homer:home.local")] + threads = AgentThreads() + + assert not await threads.observe(client, ROOM, "$archivist-card", AGENT) + + @pytest.mark.asyncio + async def test_a_thread_between_people_is_not_the_agents_either(self): + # Two parents planning in a thread are not asking anyone anything. + client = _FakeClient() + client.events["$marge-note"] = _said_by("@marge:home.local") + client.children["$marge-note"] = [_said_by("@homer:home.local")] + + assert not await AgentThreads().observe(client, ROOM, "$marge-note", AGENT) + + +class TestWhatItCostsToAsk: + + @pytest.mark.asyncio + async def test_a_known_thread_is_never_looked_up_twice(self): + """Every message in a busy thread would otherwise re-ask the + homeserver the same question. A thread the agent is in stays one: + its message cannot leave the thread, so the answer cannot change.""" + client = _FakeClient() + client.events["$stacky-answer"] = _said_by(AGENT) + threads = AgentThreads() + + await threads.observe(client, ROOM, "$stacky-answer", AGENT) + await threads.observe(client, ROOM, "$stacky-answer", AGENT) + + assert client.event_lookups == ["$stacky-answer"] + assert threads.includes("$stacky-answer") + + @pytest.mark.asyncio + async def test_the_agents_own_root_settles_it_without_reading_the_thread(self): + # Cheapest answer first: one fetch, no relation paging. + client = _FakeClient() + client.events["$stacky-answer"] = _said_by(AGENT) + + await AgentThreads().observe(client, ROOM, "$stacky-answer", AGENT) + + assert client.relation_lookups == [] + + @pytest.mark.asyncio + async def test_a_long_thread_cannot_turn_into_unbounded_paging(self): + """One chat message must cost a bounded number of API calls, even + in a thread with hundreds of replies.""" + client = _FakeClient() + client.events["$root"] = _said_by("@marge:home.local") + client.children["$root"] = [ + *[_said_by("@homer:home.local") for _ in range(50)], _said_by(AGENT), + ] + + assert not await AgentThreads().observe(client, ROOM, "$root", AGENT, limit=5) + + @pytest.mark.asyncio + async def test_a_thread_not_ours_is_re_checked_later(self): + """The negative is not cached, deliberately: the agent can join a + thread it was not in a minute ago, and the next message in it has + to see that.""" + client = _FakeClient() + client.events["$root"] = _said_by("@marge:home.local") + threads = AgentThreads() + + assert not await threads.observe(client, ROOM, "$root", AGENT) + client.children["$root"] = [_said_by(AGENT)] + assert await threads.observe(client, ROOM, "$root", AGENT) + + +class TestWhenTheHomeserverIsUnhappy: + """A lookup failure must read as "not addressed", never as an + exception out of the routing path. Losing a threaded follow-up is a + disappointment; a raised exception in the message handler takes the + agent off Matrix for every room.""" + + @pytest.mark.asyncio + async def test_a_failed_root_fetch_falls_through_to_the_thread_scan(self): + client = _FakeClient() + client.get_event_raises = ConnectionError("synapse down") + client.children["$root"] = [_said_by(AGENT)] + + assert await AgentThreads().observe(client, ROOM, "$root", AGENT) + + @pytest.mark.asyncio + async def test_a_total_failure_is_simply_not_addressed(self): + client = _FakeClient() + client.get_event_raises = ConnectionError("synapse down") + client.relations_raise = ConnectionError("synapse down") + + assert not await AgentThreads().observe(client, ROOM, "$root", AGENT) + + @pytest.mark.asyncio + async def test_an_unknown_root_is_not_addressed(self): + # The homeserver answers, it just has nothing for that id. + assert not await AgentThreads().observe(_FakeClient(), ROOM, "$gone", AGENT) diff --git a/tests/stacklets/test_agent_vault_tools.py b/tests/stacklets/test_agent_vault_tools.py index ffc28fb9..0cee29f5 100644 --- a/tests/stacklets/test_agent_vault_tools.py +++ b/tests/stacklets/test_agent_vault_tools.py @@ -26,6 +26,7 @@ import asyncio import importlib +import importlib.machinery import importlib.util import sys from pathlib import Path @@ -41,8 +42,16 @@ # ── loading the real components under test ─────────────────────────── def _load_from_path(name: str, path: Path): - """Import a module by file path (famstack-api.py is not importable).""" - spec = importlib.util.spec_from_file_location(name, path) + """Import a module by file path. + + Neither component under test is importable the ordinary way: + `famstack-api.py` has a hyphen in its name, and the client shim is an + extensionless script. The explicit loader is what makes the second + one work -- without it importlib cannot guess how to read a file with + no `.py` suffix and hands back a spec with no loader at all. + """ + loader = importlib.machinery.SourceFileLoader(name, str(path)) + spec = importlib.util.spec_from_file_location(name, path, loader=loader) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module @@ -203,6 +212,191 @@ def test_search_sends_no_backend_flag(vault_tools): assert "--backend" not in argv +def result_of(tool_cls, *, returncode: int, stdout: bytes = b"", + stderr: bytes = b"", **kwargs) -> str: + """What a tool returns to the model for a given CLI outcome. + + The sibling of `argv_of`: that one asks what went out, this one asks + what comes back. Both intercept at `create_subprocess_exec`, the + tool's real call site. + """ + class _Proc: + pass + + proc = _Proc() + proc.returncode = returncode + + async def _communicate(): + return stdout, stderr + + proc.communicate = _communicate + + async def _fake_exec(*_args, **_kwargs): + return proc + + monkey = pytest.MonkeyPatch() + monkey.setattr(asyncio, "create_subprocess_exec", _fake_exec) + try: + return asyncio.run(tool_cls().execute(**kwargs)) + finally: + monkey.undo() + + +class TestNothingFoundIsAnAnswer: + """`stack memory search` documents exit 1 as "no results" -- an + outcome, not a failure (exit 2 is bad arguments, 3 a broken backend). + + The distinction only started mattering when the shim began reporting + real exit codes at all; before that everything arrived as 0. Reading + "no results" as "the search broke" is the same shape of bug as the + one that caused the loop: the model is told something went wrong, + so it tries again instead of saying it found nothing. + """ + + def test_no_results_reads_as_no_results(self, vault_tools): + answer = result_of(vault_tools["memory_search"], + returncode=1, stdout=b"", query="school run") + + assert "error" not in answer.lower(), answer + assert "no memory results" in answer.lower() + + def test_the_status_decides_and_not_the_placeholder_text(self, vault_tools): + """An empty search comes back from the host API as its generic + "(no output)" filler. Passed through, that reads to the model like + something went wrong rather than like an answer -- and an ambiguous + non-answer is what it responds to by asking again.""" + answer = result_of(vault_tools["memory_search"], returncode=1, + stdout=b"(no output)\n", query="school run") + + assert answer == "(no memory results)" + + def test_a_broken_search_still_reads_as_broken(self, vault_tools): + """The other half of the contract: exit 2 and up are real + failures and must not be dressed up as an empty result.""" + answer = result_of(vault_tools["memory_search"], returncode=2, + stderr=b"unrecognized arguments: --backend", + query="school run") + + assert "error" in answer.lower() + assert "--backend" in answer + + def test_results_are_passed_through_verbatim(self, vault_tools): + block = b"2026-08-03 [Marge] family/camping/notes/packliste.md\n Packliste\n" + answer = result_of(vault_tools["memory_search"], + returncode=0, stdout=block, query="camping") + + assert answer == block.decode().strip() + + +# ── gate 3: the transport between the tool and the CLI ─────────────── +# +# The two gates above both read argv straight out of the tool. Nothing +# the agent runs reaches the CLI that way: it goes over a socket as one +# line of text and is rebuilt on the other side. That hop was untested, +# and it was losing every argument boundary. + + +@pytest.fixture(scope="module") +def shim(): + """The container-side `stack` shim, the agent's only route to the CLI.""" + return _load_from_path("stack_client_shim", + REPO_ROOT / "stacklets" / "agent" / "client" / "stack") + + +class TestArgvSurvivesTheWire: + """The shim encodes argv as one line; `handle_plaintext` rebuilds it + with `shlex.split`. If the two disagree about quoting, the CLI runs a + different command than the tool asked for -- and the tool cannot tell, + because it never sees the argv that actually ran. + """ + + def test_a_multi_word_argument_arrives_as_one_argument(self, shim): + """The regression, in one line. + + `memory search "school run"` arrived as four words. argparse + rejected `run` as a stray positional, so the agent's main + retrieval tool failed on every query longer than one word -- + which is nearly every real question a family asks. + """ + import shlex + argv = ["memory", "search", "school run", "--limit", "10"] + + assert shlex.split(shim.wire_line(argv)) == argv + + @pytest.mark.parametrize("argument", [ + "school run", + "when does the car insurance renew?", + "Marge's dentist", + 'he said "no"', + "back\\slash", + "a double space", + "--not-a-flag", + "", + ]) + def test_anything_a_family_might_type_survives(self, shim, argument): + """The query is free text a person typed, so the encoding has to + hold for apostrophes, quotes, question marks and empty strings -- + not just for the tidy identifiers a hand test reaches for.""" + import shlex + argv = ["memory", "search", argument] + + assert shlex.split(shim.wire_line(argv)) == argv + + def test_the_real_tool_argv_survives_it(self, api, shim, vault_tools): + """End to end: what the tool builds is what the CLI would run. + + Uses a multi-word query on purpose. Every fixture in this file + used to search for "Homer", which is exactly why a bug that only + bites on a space lived through all of them. + """ + import shlex + argv = argv_of(vault_tools["memory_search"], + query="when is the school run", limit=10) + + rebuilt = shlex.split(shim.wire_line(argv[1:])) + + assert rebuilt == argv[1:] + assert any(rebuilt[:len(p)] == p for p in api.DOMAIN_ALLOW) + + +class TestFailureIsVisibleToTheCaller: + """A command that failed must not look like one that worked. + + This is what turned a one-argument bug into a hang: the shim always + exited 0, so `memory_search` read argparse's usage text as search + results, returned it to the model as an answer, and the model called + the identical tool again. It was still doing that five minutes later. + """ + + def test_a_failed_command_reports_its_status(self, api, shim): + reply = api._with_exit("stack memory search: error: unrecognized\n", 2) + text, code = shim.split_exit_code(reply) + + assert code == 2 + assert "unrecognized" in text + assert api.EXIT_MARKER not in text, "protocol noise must not reach the model" + + def test_a_successful_command_is_untouched(self, api, shim): + """Success has to stay byte-identical, or every reply grows a line.""" + reply = api._with_exit("#1 vault/homer/about.md score=0.82\n", 0) + + assert reply == "#1 vault/homer/about.md score=0.82\n" + assert shim.split_exit_code(reply) == (reply, 0) + + def test_a_refusal_is_a_failure_too(self, api, shim): + """A denied command used to come back as ordinary output. The + model cannot act on a refusal it cannot recognise as one.""" + _, code = shim.split_exit_code(api.handle_plaintext("up memory")) + + assert code != 0 + + def test_output_that_looks_like_the_marker_is_not_mistaken_for_one(self, shim): + # A vault note quoting the marker must not silently set an exit code. + body = "the log said stack-exit: 3 and then stopped\n" + + assert shim.split_exit_code(body) == (body, 0) + + # ── the vault root a profile actually lives in ─────────────────────── def test_person_reads_generated_profiles(memory_cli, tmp_path): diff --git a/tests/stacklets/test_messages_threading.py b/tests/stacklets/test_messages_threading.py new file mode 100644 index 00000000..e1daab44 --- /dev/null +++ b/tests/stacklets/test_messages_threading.py @@ -0,0 +1,120 @@ +"""Driving a threaded conversation from the terminal. + +A Matrix thread is a conversation, and the agent treats a thread it is +part of as addressed to it (see `test_agent_thread_trigger.py`). That +behaviour was unreachable from the CLI: `send` could only post at the +top level, and `read` never showed the event ids you would thread onto. +So the two flags here are a pair -- `read --ids` tells you what to pass +to `send --thread`, and together they let a whole threaded exchange, +including a reply to a message you did not send, be driven and checked +without a Matrix client. + +What is pinned here is the wire format and the flag contract. That the +agent then *answers* such a message is the shim's business, not this +module's. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +_REPO_ROOT = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(_REPO_ROOT / "stacklets" / "messages" / "cli")) + +import _matrix # noqa: E402 +import read # noqa: E402 + + +class _CapturingClient(_matrix.MatrixClient): + """A real MatrixClient with the one network call intercepted. + + Subclassed rather than stubbed wholesale so the body under test is + built by the actual `send`, including the alias handling and the + mention payload it shares with every other send. + """ + + def __init__(self): + self.token = "tok" + self.server_name = "simpson" + self.base_url = "http://localhost:42031" + self.sent: dict = {} + + def _url(self, path): + return self.base_url + path + + def _full_user(self, user): + return f"@{user}:{self.server_name}" + + +def _send(client, **kwargs): + """Run `send` with the HTTP PUT captured instead of performed.""" + def fake_put(url, body, token=None): + client.sent = body + return 200, {"event_id": "$new"} + + original, _matrix._put = _matrix._put, fake_put + try: + return client.send("!room:simpson", "hello", **kwargs) + finally: + _matrix._put = original + + +class TestSendingIntoAThread: + + def test_a_threaded_send_carries_the_matrix_thread_relation(self): + """The relation is what makes it a thread rather than a quote, and + the root is what a bot reads to decide which conversation this is.""" + client = _CapturingClient() + ok, event_id = _send(client, thread_root="$root:simpson") + + assert ok and event_id == "$new" + relation = client.sent["m.relates_to"] + assert relation["rel_type"] == "m.thread" + assert relation["event_id"] == "$root:simpson" + + def test_it_also_falls_back_to_a_reply_for_thread_blind_clients(self): + """Matrix v1.4: a threaded message carries an `m.in_reply_to` + pointer flagged `is_falling_back`, so a client that does not + render threads still shows it in context instead of orphaned.""" + client = _CapturingClient() + _send(client, thread_root="$root:simpson") + + relation = client.sent["m.relates_to"] + assert relation["is_falling_back"] is True + assert relation["m.in_reply_to"] == {"event_id": "$root:simpson"} + + def test_an_ordinary_send_is_still_top_level(self): + # The flag is opt-in; every existing caller must be unaffected. + client = _CapturingClient() + _send(client) + + assert "m.relates_to" not in client.sent + + def test_a_thread_reply_can_still_mention_someone(self): + # How you pull a bot *into* someone else's thread in the first place. + client = _CapturingClient() + _send(client, thread_root="$root:simpson", mentions=["stacky-bot"]) + + assert client.sent["m.mentions"] == {"user_ids": ["@stacky-bot:simpson"]} + assert client.sent["m.relates_to"]["rel_type"] == "m.thread" + + +class TestAskingForEventIds: + """`read --ids`. Without it there is no way to learn the id of a + message you did not send, which is exactly the message you want to + thread onto: the bot's own answer.""" + + def test_ids_are_off_unless_asked_for(self): + room, limit, show_ids, err = read._parse_args(["chat"]) + assert (room, limit, show_ids, err) == ("chat", 20, False, None) + + def test_the_flag_turns_them_on_without_eating_the_room(self): + room, _, show_ids, err = read._parse_args(["chat", "--ids"]) + assert (room, show_ids, err) == ("chat", True, None) + + def test_it_composes_with_limit_in_either_order(self): + assert read._parse_args(["--ids", "chat", "--limit", "3"])[:3] == \ + ("chat", 3, True) + assert read._parse_args(["chat", "--limit", "3", "--ids"])[:3] == \ + ("chat", 3, True)