Jaroslav Beneš
1eba860d39
Four places a model can reach for, differing in who writes a record and how it gets in front of the model. **Knowledge** is uploaded by a person and searched by the model. It goes through `services/files.py:prepare` — the same pipeline as a chat attachment — so the same PDF produces the same text whichever way it arrived, and `Document` carries the same content columns as `Attachment` for the same reason. **Notes** are written by the model and edited by you. Too long to inject, so they are searched. **Memory** is short facts, and every one of them goes into every request. That single decision is where the rest of its design comes from: records are capped short, the block has a budget, there is no search tool because the model is already looking at them, and they are not shareable — a record about a person is not content to hand round. **Skills** are saved procedures. Only the name and description are injected; the body is fetched when the model decides one applies, which is what makes a hundred skills affordable. A model may write and revise its own — the safety story is not a gate but a record: every revision is kept, attributed and revertible. A model that has just read a hostile page can save a skill that outlives the conversation, and the honest mitigation is that it is visible and undoable rather than that it was prevented. **The harness** is why any of it gets used. A model handed a tools array ignores it and answers from recall, because nothing in the request suggests otherwise. `services/harness.py` assembles a preamble from what this chat actually has: when to reach for each tool, the memories, the skill index. This is an exception to "system prompts are precedence, not concatenation", and a deliberate one. That rule governs the three *authored* layers and is untouched — exactly one still wins. The harness is a different axis: it describes the machinery rather than the behaviour, nobody authored it, and there is nothing for it to disagree with. It is prepended to whichever authored prompt won, in one system message, since several endpoints reject a second. Supporting changes: - **Sharing**, in one helper. `visible_to()` is the only definition of who can see a library item and every listing and tool goes through it. Sharing grants *reading*; two people editing one note with no history and no merge is worse than copying it. **Administrators do not bypass this** — they bypass permissions elsewhere because an admin can grant themselves those anyway, but reading somebody's private notes is a different act. - **FTS5**, created by `db/migrations.py:ensure_fts` with the triggers an external-content index needs. Idempotent, like the column sync beside it. Terms are ANDed and then ORed: the caller is usually a model writing a whole question, and requiring every word loses the match on one absent term. - **The attach button is a menu** — file, image, a web page, or a document from the library. Attaching a document copies it, because history must not change when a document is edited later. - **A URL fetcher with an SSRF guard.** This server can reach the router, the other services on the box and LLeMbas itself, and the address can come from a model. Private ranges are refused *after resolution* and redirects are followed by hand so every hop is checked. An admin can open it deliberately. - **Model capabilities split** into protocol support and a toggle per built-in tool. Rows predating the split have no `tool_*` keys, and absent counts as on when `tools` is on — otherwise an upgrade silently takes web search away from every model already configured for it. Also fixes the test fixture, which built the schema with `create_all` and so ran against a database without the FTS tables production has; it now runs `sync_schema`, the same path startup takes. 430 tests, ruff clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
174 lines
6.5 KiB
Python
174 lines
6.5 KiB
Python
"""The operational preamble, and how it sits beside the authored prompt."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from lembas.db.models import Chat, Connection, Model, User
|
|
from lembas.security.passwords import hash_password
|
|
from lembas.services import chat as chat_service
|
|
from lembas.services import harness, settings_store
|
|
from lembas.services import tools as tools_service
|
|
from lembas.services.library import memories as memories_service
|
|
from lembas.services.library import skills as skills_service
|
|
|
|
|
|
@pytest.fixture
|
|
def owner(db):
|
|
user = User(name="Frodo", email="f@shire.test", password_hash=hash_password("x"))
|
|
db.add(user)
|
|
db.commit()
|
|
return user
|
|
|
|
|
|
def _tools(*names):
|
|
return [tools_service.REGISTRY[name].schema for name in names]
|
|
|
|
|
|
# --- Composition -------------------------------------------------------------
|
|
def test_no_tools_means_no_harness(db, owner):
|
|
"""An empty harness is worse than none: tokens that say only that there is
|
|
nothing to say."""
|
|
assert harness.compose(db, owner, []) == ""
|
|
assert harness.compose(db, owner, None) == ""
|
|
|
|
|
|
def test_only_the_guidance_for_offered_tools_appears(db, owner):
|
|
text = harness.compose(db, owner, _tools("web_search"))
|
|
assert "Look things up" in text
|
|
assert "You keep notes" not in text
|
|
assert "Skills are procedures" not in text
|
|
|
|
|
|
def test_the_memory_block_is_included_when_memory_is_offered(db, owner):
|
|
memories_service.add(db, owner=owner, content="Prefers metric units.")
|
|
text = harness.compose(db, owner, _tools("memory_add"))
|
|
assert "What you know about this person" in text
|
|
assert "Prefers metric units." in text
|
|
|
|
|
|
def test_memories_are_absent_without_the_memory_tool(db, owner):
|
|
"""A model not given the memory tool has no business being told them."""
|
|
memories_service.add(db, owner=owner, content="Prefers metric units.")
|
|
text = harness.compose(db, owner, _tools("web_search"))
|
|
assert "Prefers metric units." not in text
|
|
|
|
|
|
def test_the_skill_index_is_names_and_descriptions_only(db, owner):
|
|
skills_service.create(
|
|
db, owner=owner, name="weekly-report", description="When asked.", body="SECRET"
|
|
)
|
|
text = harness.compose(db, owner, _tools("skill_get"))
|
|
assert "weekly-report: When asked." in text
|
|
assert "SECRET" not in text
|
|
|
|
|
|
def test_an_empty_store_contributes_no_heading(db, owner):
|
|
"""And the guidance must not point at a heading that is not there: telling a
|
|
model to consult an absent section is a good way to make it invent one."""
|
|
text = harness.compose(db, owner, _tools("memory_add", "skill_get"))
|
|
assert "### What you know about this person" not in text
|
|
assert "### Skills available" not in text
|
|
assert "was remembered earlier" not in text
|
|
assert "You can remember durable facts" in text
|
|
|
|
|
|
def test_the_harness_is_capped(db, owner, monkeypatch):
|
|
monkeypatch.setattr(harness, "MAX_HARNESS_CHARS", 200)
|
|
for index in range(50):
|
|
skills_service.create(
|
|
db, owner=owner, name=f"skill-{index}", description="x" * 200, body="y"
|
|
)
|
|
assert len(harness.compose(db, owner, _tools("skill_get"))) <= 202
|
|
|
|
|
|
# --- Joining -----------------------------------------------------------------
|
|
def test_the_authored_prompt_comes_last():
|
|
"""It is closest to the conversation, and it is what the user actually
|
|
wrote."""
|
|
joined = harness.join("HARNESS", "AUTHORED")
|
|
assert joined.index("HARNESS") < joined.index("AUTHORED")
|
|
|
|
|
|
def test_either_half_alone_is_returned_unchanged():
|
|
assert harness.join("", "AUTHORED") == "AUTHORED"
|
|
assert harness.join("HARNESS", "") == "HARNESS"
|
|
assert harness.join("", "") == ""
|
|
|
|
|
|
# --- Through build_request ---------------------------------------------------
|
|
def _chat(db, owner, *, capabilities, model_prompt="", chat_prompt=""):
|
|
connection = Connection(name="c", base_url="http://h", api_key_encrypted="")
|
|
db.add(connection)
|
|
db.commit()
|
|
db.add(
|
|
Model(
|
|
connection_id=connection.id,
|
|
model_id="m",
|
|
capabilities_json=capabilities,
|
|
system_prompt=model_prompt,
|
|
)
|
|
)
|
|
db.commit()
|
|
chat = Chat(
|
|
user_id=owner.id, model_id="m", connection_id=connection.id, system_prompt=chat_prompt
|
|
)
|
|
db.add(chat)
|
|
db.commit()
|
|
return chat
|
|
|
|
|
|
def _system(body):
|
|
first = body["messages"][0] if body["messages"] else {}
|
|
return first.get("content", "") if first.get("role") == "system" else ""
|
|
|
|
|
|
def test_a_request_without_tools_has_no_harness_and_no_tools_key(db, owner):
|
|
chat = _chat(db, owner, capabilities={})
|
|
body = chat_service.build_request(db, chat, tools=[], user=owner)
|
|
assert "tools" not in body
|
|
assert "How to work" not in _system(body)
|
|
|
|
|
|
def test_the_harness_precedes_the_authored_prompt(db, owner):
|
|
settings_store.update(db, {"enabled": True}, key=settings_store.SEARCH)
|
|
chat = _chat(db, owner, capabilities={"tools": True}, chat_prompt="Speak as Gandalf.")
|
|
offered = tools_service.enabled_tools(db, chat, owner)
|
|
|
|
system = _system(chat_service.build_request(db, chat, tools=offered, user=owner))
|
|
assert system.index("How to work") < system.index("Speak as Gandalf.")
|
|
|
|
|
|
def test_precedence_between_the_authored_layers_is_untouched(db, owner):
|
|
"""The harness is a different axis. Exactly one authored layer still wins,
|
|
and it is still the most specific one."""
|
|
settings_store.update(db, {"system_prompt": "Instance."})
|
|
settings_store.update(db, {"enabled": True}, key=settings_store.SEARCH)
|
|
|
|
chat = _chat(
|
|
db, owner, capabilities={"tools": True}, model_prompt="Model.", chat_prompt="Chat."
|
|
)
|
|
offered = tools_service.enabled_tools(db, chat, owner)
|
|
system = _system(chat_service.build_request(db, chat, tools=offered, user=owner))
|
|
|
|
assert "Chat." in system
|
|
assert "Model." not in system
|
|
assert "Instance." not in system
|
|
# And the resolver on its own is unchanged.
|
|
assert chat_service.effective_system_prompt(db, chat) == "Chat."
|
|
|
|
|
|
def test_a_model_prompt_wins_when_the_chat_has_none(db, owner):
|
|
settings_store.update(db, {"system_prompt": "Instance."})
|
|
chat = _chat(db, owner, capabilities={}, model_prompt="Model.")
|
|
system = _system(chat_service.build_request(db, chat, user=owner))
|
|
assert system == "Model."
|
|
|
|
|
|
def test_the_tools_array_rides_along(db, owner):
|
|
settings_store.update(db, {"enabled": True}, key=settings_store.SEARCH)
|
|
chat = _chat(db, owner, capabilities={"tools": True})
|
|
offered = tools_service.enabled_tools(db, chat, owner)
|
|
body = chat_service.build_request(db, chat, tools=offered, user=owner)
|
|
assert body["tools"] == offered
|