143 lines
5.0 KiB
Python
143 lines
5.0 KiB
Python
import tempfile
|
|
from pathlib import Path
|
|
|
|
from beaver_gateway.agents.claude import ClaudeAgent, Prompts
|
|
from beaver_gateway.conversations.context import (
|
|
bash_paths,
|
|
compose,
|
|
files_touched,
|
|
granules,
|
|
)
|
|
from beaver_gateway.vault.links import LinkIndex
|
|
|
|
|
|
def entry(name: str, tool_input: dict, stamp: str = "2026-09-02T10:00:00Z") -> dict:
|
|
return {
|
|
"type": "assistant",
|
|
"timestamp": stamp,
|
|
"message": {
|
|
"content": [{"type": "tool_use", "name": name, "input": tool_input}]
|
|
},
|
|
}
|
|
|
|
|
|
def test_files_touched_groups_by_path_relative_to_cwd() -> None:
|
|
cwd = Path("/vault")
|
|
files, counts = files_touched(
|
|
[
|
|
entry("Read", {"file_path": "/vault/a.md"}, "2026-09-02T10:00:00Z"),
|
|
entry("Edit", {"file_path": "/vault/a.md"}, "2026-09-02T10:05:00Z"),
|
|
entry("Grep", {"pattern": "x", "path": "/vault"}),
|
|
entry("Bash", {"command": "ls"}),
|
|
entry("Read", {"file_path": "/elsewhere/b.md"}, "2026-09-02T09:00:00Z"),
|
|
],
|
|
cwd=cwd,
|
|
)
|
|
assert [f["path"] for f in files] == ["a.md", "/elsewhere/b.md"]
|
|
assert files[0] == {
|
|
"path": "a.md",
|
|
"reads": 1,
|
|
"writes": 1,
|
|
"other": 0,
|
|
"last_at": "2026-09-02T10:05:00Z",
|
|
}
|
|
assert counts == {"Read": 2, "Edit": 1, "Grep": 1, "Bash": 1}
|
|
|
|
|
|
def test_bash_paths_follow_cd_and_quotes() -> None:
|
|
paths, here, writes = bash_paths(
|
|
"cd /vault/мета && cat 'заметка.md' | head; sed -n 1,5p \"📆 доски/работа.md\"",
|
|
Path("/elsewhere"),
|
|
)
|
|
assert paths == ["/vault/мета/заметка.md", "/vault/мета/📆 доски/работа.md"]
|
|
assert here == Path("/vault/мета")
|
|
assert writes is False
|
|
_, _, writes = bash_paths(
|
|
'python3 - <<PY\np = pathlib.Path("поправки.md")\np.write_text(s)\nPY',
|
|
Path("/vault"),
|
|
)
|
|
assert writes is True
|
|
assert bash_paths("ls *.md; grep -c x notes.md", None)[0] == ["notes.md"]
|
|
|
|
|
|
def test_files_touched_reads_shell_commands_with_a_sticky_cwd() -> None:
|
|
cwd = Path("/vault")
|
|
files, counts = files_touched(
|
|
[
|
|
entry("Bash", {"command": "cd /vault/мета && cat a.md"}, "1"),
|
|
entry("Bash", {"command": "python3 - <<PY\nopen('b.md','w')\nPY"}, "2"),
|
|
entry("Bash", {"command": "ls"}, "3"),
|
|
],
|
|
cwd=cwd,
|
|
)
|
|
assert [(f["path"], f["reads"], f["writes"]) for f in files] == [
|
|
("мета/b.md", 0, 1),
|
|
("мета/a.md", 1, 0),
|
|
]
|
|
assert counts == {"Bash": 3}
|
|
|
|
|
|
def test_compose_reads_granules_and_skills() -> None:
|
|
root = Path(tempfile.mkdtemp(prefix="beaver-ctx-"))
|
|
(root / "voice.md").write_text("be brief " * 40)
|
|
(root / "env.md").write_text("you are in the master thread")
|
|
skills = root / "skills" / "common"
|
|
(skills / "ops").mkdir(parents=True)
|
|
(skills / "ops" / "SKILL.md").write_text(
|
|
"---\nname: ops\ndescription: servers\n---\n"
|
|
)
|
|
agent = ClaudeAgent(
|
|
name="a",
|
|
model="m",
|
|
cwd=root,
|
|
prompts=Prompts(master=(("voice", root / "voice.md"), root / "env.md")),
|
|
skill_sets=(skills,),
|
|
gateway_tools=("say",),
|
|
)
|
|
out = compose(
|
|
agent,
|
|
"master",
|
|
[entry("Read", {"file_path": str(root / "env.md")})],
|
|
context_tokens=50_000,
|
|
)
|
|
assert [g["path"] for g in out["prompt"]["granules"]] == ["voice.md", "env.md"]
|
|
assert out["prompt"]["granules"][0]["tag"] == "voice"
|
|
assert out["prompt"]["tokens_est"] > 0
|
|
assert out["skills"] == [
|
|
{
|
|
"set": "common",
|
|
"path": "skills/common",
|
|
"skills": [
|
|
{"name": "ops", "description": "servers", "path": "skills/common/ops"}
|
|
],
|
|
}
|
|
]
|
|
assert out["tools"]["gateway"] == ["say"]
|
|
assert out["files"][0]["path"] == "env.md"
|
|
assert out["history_tokens_est"] < 50_000
|
|
assert granules(agent, "deep")["granules"] == []
|
|
|
|
|
|
def test_link_index_resolves_stems_and_neighbours() -> None:
|
|
root = Path(tempfile.mkdtemp(prefix="beaver-links-"))
|
|
(root / "people").mkdir()
|
|
(root / "people" / "Marta.md").write_text("works at [[Studio]] with [[Ilya|him]]")
|
|
(root / "Studio.md").write_text("see [[people/Marta]] and [[nowhere]]")
|
|
(root / "Ilya.md").write_text("plain")
|
|
(root / ".obsidian").mkdir()
|
|
(root / ".obsidian" / "x.md").write_text("[[Studio]]")
|
|
index = LinkIndex(root)
|
|
graph = index.neighbours([str(root / "people" / "Marta.md")])
|
|
nodes = {n["path"]: n for n in graph["nodes"]}
|
|
assert nodes["people/Marta.md"]["touched"] is True
|
|
assert set(nodes) == {"people/Marta.md", "Studio.md", "Ilya.md"}
|
|
assert {(e["from"], e["to"]) for e in graph["edges"]} == {
|
|
("people/Marta.md", "Studio.md"),
|
|
("people/Marta.md", "Ilya.md"),
|
|
("Studio.md", "people/Marta.md"),
|
|
}
|
|
missing = index.neighbours(["gone.md"])
|
|
assert missing["nodes"] == [
|
|
{"path": "gone.md", "title": "gone", "touched": True, "exists": False}
|
|
]
|