Houmeres/LLeMbas
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

The composer decides what a chat is, and the topbar stops trying

Browse Source 
The mode select in the topbar posted with hx-post against a route that only
answers PATCH, so every change returned 405 and the mode never moved. htmx
shows nothing when a request fails, so the control looked like it worked: the
select stayed where you put it and the server ignored you. It has never worked.

Two more of the same kind. A mode could not be chosen at all until the chat
existed, so reaching Plan meant sending something in Manual first and letting
the model answer under the wrong rules. And the project directory box was real
and submitted, but unlabelled and squeezed to a few characters by the select
beside it, so it read as broken -- which is how it was reported.

So the kind, the connection, the directory and the mode move out of the strip
above the text and into one toolbar row beneath it, where attach and send
already are. The directory becomes a button that opens a browser over SFTP,
because a path is something you would rather find than spell. `scan_dir` is new
beside `list_dir`: a picker has to tell a directory from a file before it can
draw the row, and `list_dir` backs a tool whose contract is a list of names and
must not change under a model mid-conversation.

Browsing is a person clicking, not a model calling, so it does not pass through
policy.py -- the same argument the terminal panel rests on. It does mean Manual
mode has a second exception now.

Also: .chip was two components with one name, and the attachment card won, so
the Chat/Agent pills silently wore its padding. --radius-md was used twice and
declared nowhere, so both fell back to 0. .btn.is-active has been set by
syncToggles since the terminal landed and styled by nothing. Enter-to-send
ignored isComposing, so committing an IME candidate sent the message. The
terminal had five colours of a sixteen-colour palette, with fallbacks from a
palette that no longer exists.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jaroslav Beneš
2026-08-02 16:44:57 +02:00
parent 621e95d2e3
commit 803d808723
22 changed files with 1449 additions and 276 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_agent_browse.py
+256
View File
@@ -0,0 +1,256 @@
"""Walking a directory on the far side, to choose where a chat works.
Everything here goes through a real SFTP server on 127.0.0.1. A stub would
prove nothing worth proving: the whole point of `scan_dir` is that it reads
`SFTPName.attrs` to tell a directory from a file, and a stub that returns
whatever shape the test wants cannot be wrong about that.
"""
from __future__ import annotations
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select
from lembas.db.models import SshProfile, User
from lembas.services.agent import ssh as ssh_service
asyncssh = pytest.importorskip("asyncssh")
class _Server(asyncssh.SSHServer):
def begin_auth(self, username: str) -> bool:
return False
@pytest.fixture
async def tree(tmp_path):
"""A real sshd with SFTP, over a directory laid out to be walked."""
root = tmp_path / "project"
(root / "src" / "deep").mkdir(parents=True)
(root / "docs").mkdir()
(root / "README.md").write_text("hello")
(root / "src" / "main.py").write_text("print()")
(root / ".hidden").mkdir()
server = await asyncssh.create_server(
_Server,
"127.0.0.1",
0,
server_host_keys=[asyncssh.generate_private_key("ssh-ed25519")],
sftp_factory=True,
)
port = next(iter(server.sockets)).getsockname()[1]
line, fingerprint = await ssh_service.capture_host_key("127.0.0.1", port)
try:
yield {"port": port, "host_key": line, "fingerprint": fingerprint, "root": str(root)}
finally:
server.close()
await server.wait_closed()
def _spec(tree: dict) -> dict:
return {
"host": "127.0.0.1",
"port": tree["port"],
"username": "tester",
"auth": "key",
"password": "",
"private_key": "",
"key_passphrase": "",
"host_key": tree["host_key"],
"connect_timeout": 15,
}
# --- scan_dir ----------------------------------------------------------------
async def test_a_listing_says_which_rows_are_directories(tree):
"""The whole reason this exists beside `list_dir`, which returns names."""
executor = ssh_service.SshExecutor(_spec(tree), tree["root"])
entries = await executor.scan_dir("")
kinds = {entry.name: entry.is_dir for entry in entries}
assert kinds["src"] is True
assert kinds["docs"] is True
assert kinds["README.md"] is False
async def test_directories_sort_before_files(tree):
"""The order somebody navigating expects: the things you can walk into."""
executor = ssh_service.SshExecutor(_spec(tree), tree["root"])
names = [entry.name for entry in await executor.scan_dir("")]
assert names.index("src") < names.index("README.md")
assert names.index("docs") < names.index("README.md")
async def test_files_carry_a_size(tree):
executor = ssh_service.SshExecutor(_spec(tree), tree["root"])
entries = {entry.name: entry for entry in await executor.scan_dir("")}
assert entries["README.md"].size == len("hello")
async def test_dotfiles_are_listed(tree):
"""Shown, not filtered. A picker that hides `.config` is one somebody has
to work around, and the account on the far side is the boundary here --
not a taste for tidy listings."""
executor = ssh_service.SshExecutor(_spec(tree), tree["root"])
names = [entry.name for entry in await executor.scan_dir("")]
assert ".hidden" in names
assert "." not in names and ".." not in names
async def test_a_relative_path_is_measured_from_the_project_directory(tree):
executor = ssh_service.SshExecutor(_spec(tree), tree["root"])
names = [entry.name for entry in await executor.scan_dir("src")]
assert names == ["deep", "main.py"]
async def test_an_absolute_path_walks_out_of_the_project_directory(tree):
"""Deliberate, and matching `_resolve`'s own docstring.
Containment is the far-side account's job, not this layer's: `shell_run`
could leave the project directory in one line, so a picker that refused to
would be a comfort rather than a control. The browser opens *at* the
project directory; it does not fence it.
"""
executor = ssh_service.SshExecutor(_spec(tree), tree["root"])
names = [entry.name for entry in await executor.scan_dir("/")]
assert names # the real filesystem root, listed without complaint
async def test_a_missing_directory_says_so_in_words(tree):
executor = ssh_service.SshExecutor(_spec(tree), tree["root"])
with pytest.raises(ssh_service.ExecError) as caught:
await executor.scan_dir("nowhere")
assert "no directory" in caught.value.message.lower()
# --- The route ---------------------------------------------------------------
@pytest.fixture
def served_tree(tmp_path):
"""The same tree, but with the server on a thread and a loop of its own.
Its own loop matters: the tests below drive the app through the synchronous
TestClient, so a server sharing the test's loop could never accept the
connection the route is trying to make -- `client.get(...)` is still
blocking it. The async fixture above is fine for calling `scan_dir`
directly, and useless here.
"""
import asyncio
import threading
root = tmp_path / "served"
(root / "src").mkdir(parents=True)
(root / "README.md").write_text("hello")
loop = asyncio.new_event_loop()
thread = threading.Thread(target=loop.run_forever, daemon=True)
thread.start()
async def start():
server = await asyncssh.create_server(
_Server,
"127.0.0.1",
0,
server_host_keys=[asyncssh.generate_private_key("ssh-ed25519")],
sftp_factory=True,
)
port = next(iter(server.sockets)).getsockname()[1]
line, fingerprint = await ssh_service.capture_host_key("127.0.0.1", port)
return server, port, line, fingerprint
server, port, line, fingerprint = asyncio.run_coroutine_threadsafe(start(), loop).result(10)
try:
yield {"port": port, "host_key": line, "fingerprint": fingerprint, "root": str(root)}
finally:
async def stop():
server.close()
await server.wait_closed()
asyncio.run_coroutine_threadsafe(stop(), loop).result(10)
loop.call_soon_threadsafe(loop.stop)
thread.join(timeout=5)
def _profile(db, tree: dict, *, host_key: str | None = None) -> SshProfile:
user = db.scalars(select(User)).first()
profile = SshProfile(
owner_id=user.id,
name="Test box",
host="127.0.0.1",
port=tree["port"],
username="tester",
host_key=tree["host_key"] if host_key is None else host_key,
host_fingerprint=tree["fingerprint"],
default_dir=tree["root"],
)
db.add(profile)
db.commit()
return profile
def test_browsing_lists_the_project_directory(client: TestClient, db, registered, served_tree):
profile = _profile(db, served_tree)
body = client.get(f"/api/agents/{profile.id}/browse").text
assert "src" in body
assert "README.md" in body
assert served_tree["root"] in body
def test_browsing_offers_a_way_back_up(client: TestClient, db, registered, served_tree):
profile = _profile(db, served_tree)
body = client.get(f"/api/agents/{profile.id}/browse", params={"path": served_tree["root"]}).text
assert "Up a level" in body
def test_a_host_key_that_was_never_confirmed_is_refused_in_words(
client: TestClient, db, registered, served_tree
):
"""Rather than the known_hosts prose `connect_kwargs` would raise, which is
accurate and means nothing to somebody looking at a directory picker."""
profile = _profile(db, served_tree, host_key="")
body = client.get(f"/api/agents/{profile.id}/browse").text
assert "has not been confirmed" in body
def test_somebody_elses_connection_is_not_browsable(
client: TestClient, db, registered, served_tree
):
"""These are credentials to somebody's machine, so ownership is the whole
authorisation -- `sharing.py` grants reading, and a host you can read is a
host you can log in to.
The permission is granted to everybody first, so what is being tested is
ownership and not the `agent.ssh` gate in front of it. Without that the
second account is refused before the question is even asked, and the test
would pass whether or not the ownership check existed.
"""
from lembas.services import settings_store
profile = _profile(db, served_tree)
settings_store.update(db, {"default_permissions": {"agent.ssh": True}})
client.post("/auth/logout", follow_redirects=False)
client.post(
"/auth/register",
data={"name": "Sam", "email": "sam@shire.test", "password": "potatoes-po-ta-toes"},
follow_redirects=False,
)
# 404 and not 403: whether that connection exists is not this endpoint's to
# reveal to somebody who does not own it.
assert client.get(f"/api/agents/{profile.id}/browse").status_code == 404
tests/test_agent_mode.py
+227
View File
@@ -0,0 +1,227 @@
"""Choosing the approval mode: after a chat exists, and before one does.
The regression test at the top is the one that was missing. The mode select in
the topbar posted with `hx-post` against a route that only answers `PATCH`, so
every change returned 405 and the mode never moved -- and htmx surfaces nothing
on a failed request, so the control looked like it had worked. A control wired
to a method the route does not serve fails exactly this quietly, which is why
the assertion below reads the row rather than the response.
"""
from __future__ import annotations
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select
from lembas.db.models import KIND_AGENT, Chat, Connection, Model, SshProfile, User
from lembas.services import settings_store
from lembas.services.agent import policy
def _agent_chat(db, *, mode: str = policy.MODE_MANUAL) -> Chat:
"""An agent chat pointed at a profile that is never actually connected to.
Nothing here opens a connection: changing the mode is a database write, and
a real sshd would only make the test slower and flakier.
A model is needed even though nothing generates: with none, `index.html`
renders the "no models available" screen *instead of* the thread and the
composer, so a test asserting on composer markup would pass against a page
that does not contain a composer at all. That is not hypothetical -- it is
how the first version of the assertion below came to be vacuous.
"""
settings_store.update(db, {"enabled": True}, key=settings_store.AGENTS)
user = db.scalars(select(User)).first()
assert user is not None
connection = Connection(name="c", base_url="http://127.0.0.1:1", api_key_encrypted="")
db.add(connection)
db.commit()
db.add(Model(connection_id=connection.id, model_id="m"))
db.commit()
profile = SshProfile(
owner_id=user.id,
name="Test box",
host="127.0.0.1",
port=22,
username="tester",
host_key="host key",
host_fingerprint="SHA256:x",
default_dir="/project",
)
db.add(profile)
db.commit()
chat = Chat(
user_id=user.id,
model_id="m",
connection_id=connection.id,
kind=KIND_AGENT,
ssh_profile_id=profile.id,
project_dir="/project",
agent_mode=mode,
)
db.add(chat)
db.commit()
return chat
def test_patching_the_mode_changes_it(client: TestClient, db, registered):
chat = _agent_chat(db)
response = client.patch(f"/api/chats/{chat.id}", data={"agent_mode": policy.MODE_PLAN})
assert response.status_code == 204, response.text
db.refresh(chat)
assert chat.agent_mode == policy.MODE_PLAN
def test_the_mode_form_uses_a_method_the_route_serves(client: TestClient, db, registered):
"""The bug itself, stated as a test rather than as a comment.
POST is not merely unhandled here, it is *silently* unhandled: htmx swallows
the 405 and the select keeps showing whatever was clicked. Asserting the
method is refused is what stops somebody reintroducing `hx-post` and finding
the mode unchangeable again with nothing in the logs.
"""
chat = _agent_chat(db)
refused = client.post(f"/api/chats/{chat.id}", data={"agent_mode": policy.MODE_AUTO})
assert refused.status_code == 405
db.refresh(chat)
assert chat.agent_mode == policy.MODE_MANUAL
def test_the_rendered_form_patches(client: TestClient, db, registered):
"""And that the template actually carries it, since that is where it broke.
Pinned to the mode form by its id rather than to any `hx-patch` on the
page: the model picker and the system-prompt box patch the same URL, so a
looser assertion would have passed throughout the entire life of the bug.
"""
chat = _agent_chat(db)
body = client.get(f"/chat/{chat.id}").text
assert 'id="agent-mode-form"' in body
form = body[body.index('id="agent-mode-form"') :][:200]
assert f'hx-patch="/api/chats/{chat.id}"' in form
assert "hx-post" not in form
# And that the select outside it is actually submitted by it.
assert 'form="agent-mode-form"' in body
def test_the_mode_is_offered_beside_the_composer_not_in_the_topbar(
client: TestClient, db, registered
):
"""Where it is matters: it belongs where the message is written.
Asserted on order rather than on a class name so it survives a restyle --
what is being pinned is that the control comes after the thread, not what
it looks like.
"""
chat = _agent_chat(db)
body = client.get(f"/chat/{chat.id}").text
assert body.index('id="thread"') < body.index('id="agent-mode-form"')
# --- Chosen before the first word --------------------------------------------
def _profile_for(db) -> SshProfile:
"""A usable connection, and the feature switched on, with no chat yet."""
settings_store.update(db, {"enabled": True}, key=settings_store.AGENTS)
user = db.scalars(select(User)).first()
assert user is not None
connection = Connection(name="c", base_url="http://127.0.0.1:1", api_key_encrypted="")
db.add(connection)
db.commit()
db.add(Model(connection_id=connection.id, model_id="m"))
profile = SshProfile(
owner_id=user.id,
name="Test box",
host="127.0.0.1",
port=22,
username="tester",
host_key="host key",
host_fingerprint="SHA256:x",
default_dir="/project",
)
db.add(profile)
db.commit()
return profile
def test_a_chat_can_start_in_plan_mode(client: TestClient, db, registered):
"""The whole point: reaching Plan without first sending something in Manual."""
profile = _profile_for(db)
client.post(
"/api/chats/start",
data={
"content": "have a look around",
"kind": "agent",
"ssh_profile_id": profile.id,
"project_dir": "/project",
"agent_mode": policy.MODE_PLAN,
},
)
chat = db.scalars(select(Chat)).one()
assert chat.kind == KIND_AGENT
assert chat.agent_mode == policy.MODE_PLAN
def test_a_chat_started_with_no_mode_is_manual(client: TestClient, db, registered):
"""The column default still governs, so nobody's habits change."""
profile = _profile_for(db)
client.post(
"/api/chats/start",
data={"content": "hello", "kind": "agent", "ssh_profile_id": profile.id},
)
assert db.scalars(select(Chat)).one().agent_mode == policy.MODE_MANUAL
def test_an_unrecognised_mode_at_creation_is_ignored(client: TestClient, db, registered):
profile = _profile_for(db)
client.post(
"/api/chats/start",
data={
"content": "hello",
"kind": "agent",
"ssh_profile_id": profile.id,
"agent_mode": "root",
},
)
assert db.scalars(select(Chat)).one().agent_mode == policy.MODE_MANUAL
def test_a_mode_on_a_plain_chat_is_ignored(client: TestClient, db, registered):
"""A plain chat has no approval loop, so a mode on one means nothing.
It must not silently become an agent chat either: the connection is what
decides that, and there is none here.
"""
_profile_for(db)
client.post("/api/chats/start", data={"content": "hello", "agent_mode": policy.MODE_AUTO})
chat = db.scalars(select(Chat)).one()
assert chat.kind != KIND_AGENT
assert chat.agent_mode == policy.MODE_MANUAL
@pytest.mark.parametrize("wanted", ["", "sudo", "PLAN", "edit;auto"])
def test_an_unrecognised_mode_is_ignored(client: TestClient, db, registered, wanted):
"""Ignored rather than refused: an unknown mode is a bug in the sender, and
failing the whole request would leave the reader with a chat they cannot
change. The safe outcome is that nothing moves."""
chat = _agent_chat(db, mode=policy.MODE_EDIT)
client.patch(f"/api/chats/{chat.id}", data={"agent_mode": wanted})
db.refresh(chat)
assert chat.agent_mode == policy.MODE_EDIT