skill: add portable read-only learning-data readers
This commit is contained in:
amosblomqvist
@@ -0,0 +1,18 @@
|
||||
# LearnHarness data skill tests
|
||||
|
||||
Run independently of the application, with Python 3.10+ and no dependencies:
|
||||
|
||||
```sh
|
||||
python3 -B -m unittest discover -s tests/skills -v
|
||||
```
|
||||
|
||||
These tests exercise the self-contained readers in `skills/learnharness/` against
|
||||
small synthetic records and the native golden data in `tests/fixtures/golden-data/`.
|
||||
They check preservation of learning evidence, graph selection/merging, unsupported
|
||||
formats, path boundaries, and unchanged file bytes/mtimes. A portability test copies
|
||||
the skill and fixture data outside the repository and invokes both output formats
|
||||
from an unrelated working directory, checking that no cache or data files appear.
|
||||
|
||||
No real installed-user profile, provider keys, app server, LLM, or network is used.
|
||||
The test command is intentionally separate from the app's Node test suite; the skill
|
||||
has no app-runtime, release, or website integration.
|
||||
@@ -0,0 +1,169 @@
|
||||
"""Filesystem boundary and portability tests for the standalone skill."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import contextlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.dont_write_bytecode = True
|
||||
SCRIPTS = Path(__file__).resolve().parents[2] / "skills" / "learnharness" / "scripts"
|
||||
sys.path.insert(0, str(SCRIPTS))
|
||||
import _data
|
||||
|
||||
|
||||
class DataHelpersTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.temp = tempfile.TemporaryDirectory()
|
||||
self.base = Path(self.temp.name)
|
||||
self.root = self.base / "data"
|
||||
self.root.mkdir()
|
||||
|
||||
def tearDown(self):
|
||||
self.temp.cleanup()
|
||||
|
||||
def record(self, **changes):
|
||||
value = {
|
||||
"id": "chat_one", "title": "A lesson", "createdAt": "2026-10-01T00:00:00Z",
|
||||
"updatedAt": "2026-10-02T00:00:00Z", "messages": [],
|
||||
}
|
||||
value.update(changes)
|
||||
return value
|
||||
|
||||
def save_chat(self, **changes):
|
||||
(self.root / "chats").mkdir(exist_ok=True)
|
||||
(self.root / "chats" / "chat_one.json").write_text(json.dumps(self.record(**changes)), encoding="utf-8")
|
||||
|
||||
def test_explicit_existing_root(self):
|
||||
self.assertEqual(_data.data_root(str(self.root)), self.root.resolve())
|
||||
|
||||
def test_missing_root_is_not_created(self):
|
||||
target = self.base / "not-created"
|
||||
with self.assertRaisesRegex(_data.DataError, "nothing was created"):
|
||||
_data.data_root(str(target))
|
||||
self.assertFalse(target.exists())
|
||||
|
||||
def test_mac_standard_root(self):
|
||||
expected = self.base / "Library" / "Application Support" / "LearnHarness" / "data"
|
||||
expected.mkdir(parents=True)
|
||||
with patch.object(_data.sys, "platform", "darwin"), patch.object(Path, "home", return_value=self.base):
|
||||
self.assertEqual(_data.data_root(), expected.resolve())
|
||||
|
||||
def test_windows_standard_root(self):
|
||||
expected = self.base / "roaming" / "LearnHarness" / "data"
|
||||
expected.mkdir(parents=True)
|
||||
with patch.object(_data.sys, "platform", "win32"), patch.dict(os.environ, {"APPDATA": str(self.base / "roaming")}):
|
||||
self.assertEqual(_data.data_root(), expected.resolve())
|
||||
|
||||
def test_windows_missing_appdata_requires_explicit_root(self):
|
||||
with patch.object(_data.sys, "platform", "win32"), patch.dict(os.environ, {}, clear=True):
|
||||
with self.assertRaisesRegex(_data.DataError, "--data-dir"):
|
||||
_data.data_root()
|
||||
|
||||
def test_linux_xdg_and_fallback(self):
|
||||
xdg = self.base / "config"
|
||||
expected = xdg / "LearnHarness" / "data"
|
||||
expected.mkdir(parents=True)
|
||||
with patch.object(_data.sys, "platform", "linux"), patch.dict(os.environ, {"XDG_CONFIG_HOME": str(xdg)}):
|
||||
self.assertEqual(_data.data_root(), expected.resolve())
|
||||
fallback = self.base / ".config" / "LearnHarness" / "data"
|
||||
fallback.mkdir(parents=True)
|
||||
with patch.object(_data.sys, "platform", "linux"), patch.object(Path, "home", return_value=self.base), patch.dict(os.environ, {}, clear=True):
|
||||
self.assertEqual(_data.data_root(), fallback.resolve())
|
||||
|
||||
def test_relative_xdg_is_explicit_error(self):
|
||||
with patch.object(_data.sys, "platform", "linux"), patch.dict(os.environ, {"XDG_CONFIG_HOME": "relative"}):
|
||||
with self.assertRaisesRegex(_data.DataError, "relative"):
|
||||
_data.data_root()
|
||||
|
||||
def test_explicit_root_works_on_unknown_os(self):
|
||||
with patch.object(_data.sys, "platform", "unknown"):
|
||||
self.assertEqual(_data.data_root(str(self.root)), self.root.resolve())
|
||||
with self.assertRaisesRegex(_data.DataError, "--data-dir"):
|
||||
_data.data_root()
|
||||
|
||||
def test_safe_path_rejects_traversal_and_absolute_paths(self):
|
||||
self.assertEqual(_data.safe_path(self.root, "chats/chat_one.json"), self.root.resolve() / "chats" / "chat_one.json")
|
||||
for relative in ("../auth.json", self.base / "outside.json"):
|
||||
with self.subTest(path=relative), self.assertRaises(_data.DataError):
|
||||
_data.safe_path(self.root, relative)
|
||||
|
||||
def test_safe_path_rejects_escaping_symlink(self):
|
||||
outside = self.base / "private.json"
|
||||
outside.write_text('{"unrelated": true}', encoding="utf-8")
|
||||
try:
|
||||
(self.root / "link.json").symlink_to(outside)
|
||||
except (OSError, NotImplementedError):
|
||||
self.skipTest("Symlinks unavailable for this user")
|
||||
with self.assertRaisesRegex(_data.DataError, "outside"):
|
||||
_data.read_json(self.root, "link.json")
|
||||
|
||||
def test_json_read_missing_and_malformed_without_writes(self):
|
||||
self.assertEqual(_data.read_json(self.root, "optional.json", missing=[]), [])
|
||||
target = self.root / "bad.json"
|
||||
target.write_text("{", encoding="utf-8")
|
||||
with self.assertRaisesRegex(_data.DataError, "Cannot read"):
|
||||
_data.read_json(self.root, "bad.json")
|
||||
self.assertEqual(target.read_text(encoding="utf-8"), "{")
|
||||
self.assertEqual([p.name for p in self.root.iterdir()], ["bad.json"])
|
||||
|
||||
def test_non_json_constants_are_not_accepted(self):
|
||||
(self.root / "bad.json").write_text('{"value": NaN}', encoding="utf-8")
|
||||
with self.assertRaises(_data.DataError):
|
||||
_data.read_json(self.root, "bad.json")
|
||||
|
||||
def test_unstamped_native_chat_is_read_without_stamping(self):
|
||||
self.save_chat()
|
||||
target = self.root / "chats" / "chat_one.json"
|
||||
before = target.read_bytes()
|
||||
self.assertEqual(_data.read_chat(self.root, "chat_one"), self.record())
|
||||
self.assertEqual(target.read_bytes(), before)
|
||||
self.assertEqual(sorted(p.relative_to(self.root).as_posix() for p in self.root.rglob("*")), ["chats", "chats/chat_one.json"])
|
||||
|
||||
def test_future_chat_version_is_refused(self):
|
||||
self.save_chat(version=2)
|
||||
with self.assertRaisesRegex(_data.DataError, "Update the skill"):
|
||||
_data.read_chat(self.root, "chat_one")
|
||||
self.save_chat(version=True)
|
||||
with self.assertRaisesRegex(_data.DataError, "invalid version"):
|
||||
_data.read_chat(self.root, "chat_one")
|
||||
|
||||
def test_pre_native_parts_dialect_is_not_silently_empty(self):
|
||||
self.save_chat(messages=[{"role": "user", "parts": [{"type": "text", "text": "old shape"}]}])
|
||||
with self.assertRaisesRegex(_data.DataError, "older UIMessage"):
|
||||
_data.read_chat(self.root, "chat_one")
|
||||
|
||||
def test_read_chat_rejects_missing_and_mismatched_records(self):
|
||||
with self.assertRaisesRegex(_data.DataError, "not found"):
|
||||
_data.read_chat(self.root, "chat_one")
|
||||
self.save_chat(id="chat_other")
|
||||
with self.assertRaisesRegex(_data.DataError, "mismatch"):
|
||||
_data.read_chat(self.root, "chat_one")
|
||||
|
||||
def test_read_chat_rejects_path_in_place_of_id(self):
|
||||
for value in ("../auth", "..", "", "chats/chat_one", "chats\\chat_one"):
|
||||
with self.subTest(value=value), self.assertRaises(_data.DataError):
|
||||
_data.read_chat(self.root, value)
|
||||
|
||||
def test_common_flags_and_unicode_json(self):
|
||||
parser = argparse.ArgumentParser()
|
||||
_data.add_common_arguments(parser)
|
||||
args = parser.parse_args(["--data-dir", str(self.root), "--format", "json"])
|
||||
self.assertEqual(args.data_dir, str(self.root))
|
||||
self.assertEqual(args.format, "json")
|
||||
out = io.StringIO()
|
||||
with contextlib.redirect_stdout(out):
|
||||
_data.print_json({"statement": "∇ and café"})
|
||||
self.assertIn("∇", out.getvalue())
|
||||
self.assertEqual(json.loads(out.getvalue()), {"statement": "∇ and café"})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,351 @@
|
||||
"""Portable, read-only graph CLI tests; no application-store imports."""
|
||||
|
||||
import contextlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
sys.dont_write_bytecode = True
|
||||
SCRIPTS = Path(__file__).resolve().parents[2] / "skills" / "learnharness" / "scripts"
|
||||
sys.path.insert(0, str(SCRIPTS))
|
||||
import graph
|
||||
|
||||
GOLDEN = Path(__file__).resolve().parents[1] / "fixtures" / "golden-data"
|
||||
|
||||
|
||||
def snapshot(root):
|
||||
"""Notice both changed bytes/mtimes and newly created files/directories."""
|
||||
return {
|
||||
str(path.relative_to(root)): (path.read_bytes(), path.stat().st_mtime_ns) if path.is_file() else None
|
||||
for path in root.rglob("*")
|
||||
}
|
||||
|
||||
|
||||
def point(chat, message=""):
|
||||
return {"chatId": chat, "messageId": message}
|
||||
|
||||
|
||||
def node(identifier, session=None, **updates):
|
||||
value = {
|
||||
"id": identifier, "name": f"Name {identifier}",
|
||||
"statement": f"Full statement {identifier}", "aliases": [],
|
||||
"createdAt": "2026-01-01T00:00:00Z",
|
||||
"provenance": [] if session is None else [point(session)],
|
||||
}
|
||||
value.update(updates)
|
||||
return value
|
||||
|
||||
|
||||
def chain(identifier, premises, conclusion, session, **updates):
|
||||
value = {
|
||||
"id": identifier, "premises": premises, "conclusion": conclusion,
|
||||
"createdAt": "2026-01-02T00:00:00Z", "provenance": [point(session)],
|
||||
}
|
||||
value.update(updates)
|
||||
return value
|
||||
|
||||
|
||||
class GraphTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.temp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.temp.cleanup)
|
||||
self.root = Path(self.temp.name)
|
||||
|
||||
def write_json(self, relative, value):
|
||||
path = self.root / relative
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")
|
||||
|
||||
def write_ledger(self, name, records):
|
||||
path = self.root / "graph" / f"{name}.jsonl"
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text("".join(json.dumps(record, ensure_ascii=False) + "\n" for record in records), encoding="utf-8")
|
||||
|
||||
def invoke(self, *arguments, format="json", root=None):
|
||||
out, err = io.StringIO(), io.StringIO()
|
||||
with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
|
||||
status = graph.main(["--data-dir", str(root or self.root), "--format", format, *arguments])
|
||||
return status, out.getvalue(), err.getvalue()
|
||||
|
||||
def result(self, *arguments, root=None):
|
||||
status, output, diagnostics = self.invoke(*arguments, root=root)
|
||||
self.assertEqual(status, 0, diagnostics)
|
||||
self.assertEqual(diagnostics, "")
|
||||
payload = json.loads(output)
|
||||
self.assertEqual(payload["data_dir"], str((root or self.root).resolve()))
|
||||
return payload["records"]
|
||||
|
||||
def fixture(self):
|
||||
self.write_ledger("nodes", [
|
||||
node("n_b", "chat_old", name="Identity stands", statement="First precise statement", aliases=["Original wording"], provenance=[point("chat_old", "m1")], extra="preserved"),
|
||||
node("n_a", "src_notes", name="Café", statement="A separate proposition.", aliases=["Literal [brackets]?", "two\nlines"]),
|
||||
node("n_b", "chat_new", name="Name must not replace", statement="Later restatement", aliases=["Original wording", "Another phrasing", ""], createdAt="2099-01-01", provenance=[point("chat_old", "m1"), point("chat_new", "m2"), point("chat_new", "m2")]),
|
||||
node("n_empty"),
|
||||
node("n_surviving", "chat_new"),
|
||||
node("n_unknown", "missing_record"),
|
||||
])
|
||||
self.write_ledger("chains", [
|
||||
chain("arg_first", ["n_b", "n_a"], "n_surviving", "chat_old", extra="preserved"),
|
||||
chain("arg_other_id", ["n_a", "n_b"], "n_surviving", "chat_new", createdAt="2099-01-01"),
|
||||
chain("arg_single", ["n_surviving"], "n_a", "src_notes"),
|
||||
chain("arg_dangling", ["n_missing"], "n_a", "chat_old"),
|
||||
])
|
||||
for identifier, title, updated in [
|
||||
("chat_old", "Old conversation", "2026-02-01T00:00:00Z"),
|
||||
("chat_new", "Recent conversation", "2026-03-01T00:00:00Z"),
|
||||
("chat_without_nodes", "Not a birth session", "2099-01-01"),
|
||||
]:
|
||||
self.write_json(f"chats/{identifier}.json", {
|
||||
"id": identifier, "title": title, "createdAt": "2026-01-01T00:00:00Z",
|
||||
"updatedAt": updated, "messages": [],
|
||||
})
|
||||
self.write_json("graph/sources.json", {
|
||||
"src_notes": {"id": "src_notes", "filePath": "/never/open/these notes.md", "importedAt": "2026-02-15T00:00:00Z"},
|
||||
})
|
||||
|
||||
def test_merge_nodes_first_identity_and_union(self):
|
||||
self.fixture()
|
||||
before = snapshot(self.root)
|
||||
merged = self.result("propositions")
|
||||
self.assertEqual([item["id"] for item in merged], ["n_a", "n_b", "n_empty", "n_surviving", "n_unknown"])
|
||||
value = next(item for item in merged if item["id"] == "n_b")
|
||||
self.assertEqual(value["name"], "Identity stands")
|
||||
self.assertEqual(value["statement"], "First precise statement")
|
||||
self.assertEqual(value["createdAt"], "2026-01-01T00:00:00Z")
|
||||
self.assertEqual(value["aliases"], ["Original wording", "Later restatement", "Another phrasing"])
|
||||
self.assertEqual(value["provenance"], [point("chat_old", "m1"), point("chat_new", "m2")])
|
||||
self.assertEqual(value["extra"], "preserved")
|
||||
self.assertEqual(snapshot(self.root), before)
|
||||
|
||||
def test_chain_key_identity_provenance_and_endpoint_filter(self):
|
||||
self.fixture()
|
||||
merged = self.result("chains")
|
||||
self.assertEqual([item["id"] for item in merged], ["arg_first", "arg_single", "arg_dangling"])
|
||||
self.assertEqual(merged[0]["premises"], ["n_b", "n_a"])
|
||||
self.assertEqual(merged[0]["createdAt"], "2026-01-02T00:00:00Z")
|
||||
self.assertEqual(merged[0]["extra"], "preserved")
|
||||
self.assertEqual(merged[0]["provenance"], [point("chat_old"), point("chat_new")])
|
||||
self.assertEqual(self.result("chains", "--node", "n_b"), merged[:1])
|
||||
self.assertEqual(len(self.result("chains", "--node", "n_a")), 3)
|
||||
self.assertEqual(self.result("chains", "--node", "no-such-node"), [])
|
||||
|
||||
def test_birth_index_metadata_and_filter(self):
|
||||
self.fixture()
|
||||
sessions = self.result("sessions")
|
||||
self.assertEqual([item["id"] for item in sessions], ["chat_new", "src_notes", "chat_old", "missing_record"])
|
||||
self.assertEqual([item["propositions"] for item in sessions], [1, 1, 1, 1])
|
||||
self.assertEqual(sessions[0]["kind"], "chat")
|
||||
self.assertEqual(sessions[0]["name"], "Recent conversation")
|
||||
self.assertEqual(sessions[1]["name"], "/never/open/these notes.md")
|
||||
self.assertEqual(sessions[-1], {"id": "missing_record", "name": "missing_record", "kind": "file", "createdAt": None, "updatedAt": None, "propositions": 1})
|
||||
self.assertEqual([item["id"] for item in self.result("propositions", "--session", "chat_new")], ["n_surviving"])
|
||||
self.assertEqual([item["id"] for item in self.result("propositions", "--session", "chat_old")], ["n_b"])
|
||||
self.assertEqual(self.result("propositions", "--session", "nonexistent"), [])
|
||||
|
||||
def test_bad_chat_metadata_is_isolated_and_diagnosed(self):
|
||||
self.fixture()
|
||||
corrupted = self.root / "chats" / "chat_old.json"
|
||||
corrupted.write_text("{torn", encoding="utf-8")
|
||||
self.write_json("chats/chat_future.json", {
|
||||
"id": "chat_future", "title": "Future metadata", "createdAt": "2026-01-01",
|
||||
"updatedAt": "2026-01-01", "version": 2, "messages": [],
|
||||
})
|
||||
self.write_json("chats/src_notes.json", {
|
||||
"id": "src_notes", "title": "Older parts metadata", "createdAt": "2026-01-01",
|
||||
"updatedAt": "2026-01-01", "messages": [{"role": "assistant", "parts": []}],
|
||||
})
|
||||
self.write_json("chats/chat_mismatch.json", {
|
||||
"id": "another_id", "title": "Wrong filename", "createdAt": "2026-01-01",
|
||||
"updatedAt": "2026-01-01", "messages": [],
|
||||
})
|
||||
before = snapshot(self.root)
|
||||
for arguments, format in [(("sessions",), "json"), (("propositions",), "markdown")]:
|
||||
with self.subTest(arguments=arguments):
|
||||
status, output, diagnostics = self.invoke(*arguments, format=format)
|
||||
self.assertEqual(status, 0, diagnostics)
|
||||
for name in ("chat_old.json", "chat_future.json", "src_notes.json", "chat_mismatch.json"):
|
||||
self.assertIn(name, diagnostics)
|
||||
for reason in ("format 2", "older UIMessage/parts", "filename/ID mismatch", "skipped"):
|
||||
self.assertIn(reason, diagnostics)
|
||||
if format == "json":
|
||||
sessions = {item["id"]: item for item in json.loads(output)["records"]}
|
||||
self.assertEqual(sessions["chat_old"]["name"], "chat_old")
|
||||
self.assertIsNone(sessions["chat_old"]["createdAt"])
|
||||
self.assertEqual(sessions["chat_new"]["name"], "Recent conversation")
|
||||
self.assertEqual(sessions["src_notes"]["name"], "/never/open/these notes.md")
|
||||
else:
|
||||
self.assertIn("First precise statement", output)
|
||||
self.assertIn("Recent conversation", output)
|
||||
self.assertIn("/never/open/these notes.md", output)
|
||||
self.assertEqual(snapshot(self.root), before)
|
||||
|
||||
def test_explicit_empty_query_and_ids_rejected_but_whitespace_accepted(self):
|
||||
for arguments in (("find", ""), ("propositions", "--session", ""), ("chains", "--node", "")):
|
||||
with self.subTest(arguments=arguments):
|
||||
out, err = io.StringIO(), io.StringIO()
|
||||
with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
|
||||
with self.assertRaises(SystemExit) as raised:
|
||||
graph.main(["--data-dir", str(self.root), *arguments])
|
||||
self.assertEqual(raised.exception.code, 2)
|
||||
self.assertEqual(out.getvalue(), "")
|
||||
self.assertIn("must not be empty", err.getvalue())
|
||||
for arguments in (("find", " "), ("propositions", "--session", " "), ("chains", "--node", " ")):
|
||||
self.assertEqual(self.result(*arguments), [])
|
||||
self.assertEqual(list(self.root.iterdir()), [])
|
||||
|
||||
def test_empty_first_provenance_acquires_birth_from_surviving_link(self):
|
||||
self.write_ledger("nodes", [node("n_x"), node("n_x", "survivor")])
|
||||
self.assertEqual([item["id"] for item in self.result("propositions", "--session", "survivor")], ["n_x"])
|
||||
self.assertEqual(self.result("sessions")[0]["id"], "survivor")
|
||||
|
||||
def test_session_order_timestamp_ties_then_stable_id(self):
|
||||
self.write_ledger("nodes", [node("n_b", "b"), node("n_a", "a"), node("n_c", "c")])
|
||||
self.assertEqual([item["id"] for item in self.result("sessions")], ["a", "b", "c"])
|
||||
|
||||
def test_find_matches_names_statements_aliases_and_literal_punctuation(self):
|
||||
self.fixture()
|
||||
for query in ("IDENTITY", "precise", "later RESTATEMENT", "another phrasing"):
|
||||
self.assertEqual([item["id"] for item in self.result("find", query)], ["n_b"])
|
||||
self.assertEqual([item["id"] for item in self.result("find", "[brackets]?")], ["n_a"])
|
||||
self.assertEqual([item["id"] for item in self.result("find", "Café")], ["n_a"])
|
||||
self.assertEqual(self.result("find", "n_b"), []) # IDs are not searchable fields.
|
||||
self.assertEqual(self.result("find", "Name must not replace"), [])
|
||||
self.assertEqual(self.result("find", " no such text "), [])
|
||||
|
||||
def test_find_case_matching_agrees_with_app_js_regex(self):
|
||||
self.write_ledger("nodes", [node("n_case", name="Exact label", statement="Exact statement", aliases=["K", "ſ", "ı", "İ", "ς", "𐐨", "Café"])])
|
||||
for query in ("k", "s", "i", "𐐀"):
|
||||
# Explicit stars avoid unrelated substring matches in the name.
|
||||
self.assertEqual(self.result("find", query + "*"), [], query)
|
||||
for query in ("K*", "ſ*", "ı*", "İ*", "σ*", "𐐨*", "CAFÉ*"):
|
||||
self.assertEqual([item["id"] for item in self.result("find", query)], ["n_case"], query)
|
||||
|
||||
def test_find_star_glob_newlines_and_ledger_order(self):
|
||||
self.fixture()
|
||||
self.assertEqual([item["id"] for item in self.result("find", "*statement*")], ["n_b", "n_empty", "n_surviving", "n_unknown"])
|
||||
self.assertEqual([item["id"] for item in self.result("find", "two*lines")], ["n_a"])
|
||||
self.assertEqual(self.result("find", "precise*"), [])
|
||||
self.assertEqual(len(self.result("find", "*")), 5)
|
||||
|
||||
def test_missing_ledgers_empty_graph_and_no_directory_creation(self):
|
||||
for command in ("sessions", "propositions", "chains"):
|
||||
self.assertEqual(self.result(command), [])
|
||||
self.assertEqual(self.result("find", "anything"), [])
|
||||
self.assertEqual(list(self.root.iterdir()), [])
|
||||
status, output, diagnostics = self.invoke("propositions", format="markdown")
|
||||
self.assertEqual(status, 0, diagnostics)
|
||||
self.assertIn("No matching records", output)
|
||||
self.assertEqual(list(self.root.iterdir()), [])
|
||||
|
||||
def test_malformed_and_wrong_shape_lines_warn_instead_of_silent_loss(self):
|
||||
self.write_ledger("nodes", [node("n_ok", "source")])
|
||||
path = self.root / "graph" / "nodes.jsonl"
|
||||
with path.open("a", encoding="utf-8") as handle:
|
||||
handle.write('\n{"torn":\n{"id":"bad"}\nnull\n')
|
||||
self.write_ledger("chains", [chain("arg_bad", [], "n_ok", "source")])
|
||||
before = snapshot(self.root)
|
||||
status, output, diagnostics = self.invoke("propositions")
|
||||
self.assertEqual(status, 0)
|
||||
self.assertEqual([item["id"] for item in json.loads(output)["records"]], ["n_ok"])
|
||||
self.assertIn("nodes.jsonl:3: malformed JSON", diagnostics)
|
||||
self.assertIn("nodes.jsonl:4: invalid record shape", diagnostics)
|
||||
self.assertIn("nodes.jsonl:5: invalid record shape", diagnostics)
|
||||
self.assertIn("chains.jsonl:1: invalid record shape", diagnostics)
|
||||
self.assertIn("incomplete", diagnostics)
|
||||
self.assertEqual(snapshot(self.root), before)
|
||||
|
||||
def test_future_format_refuses_without_partial_output_or_writes(self):
|
||||
self.fixture()
|
||||
self.write_json("graph/format.json", {"version": 2})
|
||||
before = snapshot(self.root)
|
||||
for command in ("sessions", "propositions", "chains"):
|
||||
status, output, diagnostics = self.invoke(command)
|
||||
self.assertNotEqual(status, 0)
|
||||
self.assertEqual(output, "")
|
||||
self.assertIn("newer", diagnostics)
|
||||
self.assertEqual(snapshot(self.root), before)
|
||||
|
||||
def test_current_format_works_and_malformed_json_refuses(self):
|
||||
self.write_json("graph/format.json", {"version": 1})
|
||||
self.assertEqual(self.result("propositions"), [])
|
||||
(self.root / "graph" / "format.json").write_text("{torn", encoding="utf-8")
|
||||
status, output, diagnostics = self.invoke("propositions")
|
||||
self.assertNotEqual(status, 0)
|
||||
self.assertEqual(output, "")
|
||||
self.assertIn("format.json", diagnostics)
|
||||
|
||||
def test_source_paths_are_only_references_and_markdown_is_full_text(self):
|
||||
self.fixture()
|
||||
before = snapshot(self.root)
|
||||
status, output, diagnostics = self.invoke("propositions", format="markdown")
|
||||
self.assertEqual(status, 0, diagnostics)
|
||||
for text in ("First precise statement", "Original wording", "Later restatement", "Another phrasing", "chat_old", "m1", "Recent conversation", "Café", "/never/open/these notes.md"):
|
||||
self.assertIn(text, output)
|
||||
self.assertIn(f"Data directory: {self.root.resolve()}", output)
|
||||
self.assertIn("not mastery or guaranteed truth", output)
|
||||
status, output, diagnostics = self.invoke("chains", "--node", "n_a", format="markdown")
|
||||
self.assertEqual(status, 0, diagnostics)
|
||||
self.assertIn("First precise statement", output)
|
||||
self.assertIn("missing proposition", output)
|
||||
self.assertEqual(snapshot(self.root), before)
|
||||
|
||||
@unittest.skipUnless(hasattr(os, "symlink"), "symlinks unavailable")
|
||||
def test_ledger_symlink_escape_is_rejected(self):
|
||||
with tempfile.TemporaryDirectory() as external:
|
||||
path = Path(external) / "nodes.jsonl"
|
||||
path.write_text(json.dumps(node("n_external")), encoding="utf-8")
|
||||
(self.root / "graph").mkdir()
|
||||
try:
|
||||
(self.root / "graph" / "nodes.jsonl").symlink_to(path)
|
||||
except OSError as exc:
|
||||
self.skipTest(f"symlink permission unavailable: {exc}")
|
||||
status, output, diagnostics = self.invoke("propositions")
|
||||
self.assertNotEqual(status, 0)
|
||||
self.assertEqual(output, "")
|
||||
self.assertTrue(diagnostics)
|
||||
self.assertEqual(path.read_text(encoding="utf-8"), json.dumps(node("n_external")))
|
||||
|
||||
def test_cli_root_options_order_missing_root_and_no_bytecode(self):
|
||||
env = {**os.environ, "LH_DATA_DIR": str(GOLDEN)}
|
||||
env.pop("PYTHONDONTWRITEBYTECODE", None)
|
||||
before = snapshot(self.root)
|
||||
caches_before = set(SCRIPTS.rglob("*.pyc"))
|
||||
command = [sys.executable, str(SCRIPTS / "graph.py")]
|
||||
completed = subprocess.run([*command, "--data-dir", str(self.root), "--format", "json", "sessions"], env=env, capture_output=True, text=True, check=False)
|
||||
self.assertEqual(completed.returncode, 0, completed.stderr)
|
||||
self.assertEqual(json.loads(completed.stdout), {"data_dir": str(self.root.resolve()), "records": []}) # Never falls back to repo/LH_DATA_DIR.
|
||||
bad = subprocess.run([*command, "sessions", "--data-dir", str(self.root)], env=env, capture_output=True, text=True, check=False)
|
||||
self.assertNotEqual(bad.returncode, 0)
|
||||
absent = subprocess.run([*command, "--data-dir", str(self.root / "missing"), "sessions"], env=env, capture_output=True, text=True, check=False)
|
||||
self.assertNotEqual(absent.returncode, 0)
|
||||
self.assertEqual(absent.stdout, "")
|
||||
self.assertEqual(snapshot(self.root), before)
|
||||
self.assertEqual(set(SCRIPTS.rglob("*.pyc")), caches_before)
|
||||
|
||||
def test_golden_data_read_only_broad_graph(self):
|
||||
before = snapshot(GOLDEN)
|
||||
nodes = self.result("propositions", root=GOLDEN)
|
||||
chains = self.result("chains", root=GOLDEN)
|
||||
sessions = self.result("sessions", root=GOLDEN)
|
||||
self.assertGreaterEqual(len(nodes), 250)
|
||||
self.assertGreaterEqual(len(chains), 150)
|
||||
self.assertEqual(len({item["id"] for item in nodes}), len(nodes))
|
||||
self.assertEqual(sum(item["propositions"] for item in sessions), sum(bool(item["provenance"]) for item in nodes))
|
||||
for session in sessions:
|
||||
born = self.result("propositions", "--session", session["id"], root=GOLDEN)
|
||||
self.assertEqual(len(born), session["propositions"])
|
||||
self.assertTrue(all(item["provenance"][0]["chatId"] == session["id"] for item in born))
|
||||
self.assertTrue(any(item["kind"] == "file" for item in sessions))
|
||||
# Golden chat files are not the original graph birth sessions; their
|
||||
# missing historical metadata must remain honestly undated.
|
||||
self.assertTrue(any(item["createdAt"] is None for item in sessions))
|
||||
self.assertEqual(snapshot(GOLDEN), before)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,393 @@
|
||||
"""Read-only evidence-preservation tests for the native session readers."""
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
sys.dont_write_bytecode = True
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
SCRIPTS = ROOT / "skills" / "learnharness" / "scripts"
|
||||
GOLDEN = ROOT / "tests" / "fixtures" / "golden-data"
|
||||
sys.path.insert(0, str(SCRIPTS))
|
||||
spec = importlib.util.spec_from_file_location("lh_sessions_reader", SCRIPTS / "sessions.py")
|
||||
sessions = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(sessions)
|
||||
|
||||
|
||||
def record(messages=None, session_id="sample", **extra):
|
||||
return {
|
||||
"id": session_id, "title": "Evidence session", "createdAt": "2026-10-10T10:00:00Z",
|
||||
"updatedAt": "2026-10-10T11:00:00Z", "messages": messages or [], **extra,
|
||||
}
|
||||
|
||||
|
||||
def assistant(*blocks, message_id="teacher", **extra):
|
||||
return {"id": message_id, "role": "assistant", "content": list(blocks), **extra}
|
||||
|
||||
|
||||
def call(name, arguments, call_id="call-one"):
|
||||
return {"type": "toolCall", "id": call_id, "name": name, "arguments": arguments}
|
||||
|
||||
|
||||
def result(name, details=None, call_id="call-one", content=None, **extra):
|
||||
value = {"id": "answer-result", "role": "toolResult", "toolName": name,
|
||||
"toolCallId": call_id, "isError": False,
|
||||
"content": content if content is not None else [{"type": "text", "text": '{"kind":"quiz"}'}], **extra}
|
||||
if details is not None:
|
||||
value["details"] = details
|
||||
return value
|
||||
|
||||
|
||||
def marker(kind, content, details, marker_id="marker"):
|
||||
return {"id": marker_id, "role": "custom", "customType": kind,
|
||||
"content": content, "details": details, "display": True}
|
||||
|
||||
|
||||
def snapshot(folder):
|
||||
return {str(p.relative_to(folder)): hashlib.sha256(p.read_bytes()).hexdigest()
|
||||
for p in folder.rglob("*") if p.is_file()}
|
||||
|
||||
|
||||
class SessionsTest(unittest.TestCase):
|
||||
def cli(self, root, *args):
|
||||
return subprocess.run([sys.executable, "-B", str(SCRIPTS / "sessions.py"),
|
||||
"--data-dir", str(root), *args], capture_output=True,
|
||||
text=True, env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"}, check=False)
|
||||
|
||||
def save(self, root, value, filename=None):
|
||||
folder = root / "chats"
|
||||
folder.mkdir(exist_ok=True)
|
||||
path = folder / (filename or f"{value['id']}.json")
|
||||
path.write_text(json.dumps(value), encoding="utf-8")
|
||||
return path
|
||||
|
||||
def render(self, value, **flags):
|
||||
return sessions.render_markdown(sessions.selected_record(value, **flags))
|
||||
|
||||
def test_golden_default_preserves_learning_and_omits_noise(self):
|
||||
before = snapshot(GOLDEN)
|
||||
listing = self.cli(GOLDEN, "--format", "json", "list")
|
||||
self.assertEqual(listing.returncode, 0, listing.stderr)
|
||||
index = json.loads(listing.stdout)
|
||||
self.assertEqual(len(index["sessions"]), 2)
|
||||
self.assertEqual(index["skipped_count"], 0)
|
||||
self.assertEqual(index["data_dir"], str(GOLDEN.resolve()))
|
||||
self.assertEqual({s["kind"] for s in index["sessions"]}, {"teach", "practice"})
|
||||
for summary in index["sessions"]:
|
||||
response = self.cli(GOLDEN, "read", summary["id"])
|
||||
self.assertEqual(response.returncode, 0, response.stderr)
|
||||
raw = json.loads((GOLDEN / "chats" / f"{summary['id']}.json").read_text())
|
||||
text = response.stdout
|
||||
for message in raw["messages"]:
|
||||
self.assertIn(message["id"], text)
|
||||
if message["role"] == "user":
|
||||
self.assertIn(sessions._text(message["content"]), text)
|
||||
if message["role"] == "custom":
|
||||
self.assertIn(message["customType"], text)
|
||||
if message["role"] == "toolResult" and message.get("toolName") in sessions.QUESTION_TOOLS:
|
||||
self.assertIn(message["toolCallId"], text)
|
||||
details = message.get("details", {})
|
||||
for key in ("thinking", "note", "readingThinking", "committed", "text", "explanation", "other"):
|
||||
if key in details:
|
||||
self.assertIn(sessions._text(details[key]), text)
|
||||
if message["role"] == "assistant":
|
||||
for block in message["content"]:
|
||||
if block["type"] == "text":
|
||||
self.assertIn(sessions._text(block["text"]), text)
|
||||
self.assertIn("Explicit omissions", text)
|
||||
self.assertIn(f"Data directory: {GOLDEN.resolve()}", text)
|
||||
self.assertNotIn('"thinkingSignature"', text)
|
||||
self.assertEqual(before, snapshot(GOLDEN))
|
||||
|
||||
def test_answer_details_are_full_truth_and_indexes_are_not_joined(self):
|
||||
answer = {
|
||||
"kind": "quiz", "questionText": "Q snapshot", "options": [
|
||||
{"index": 1, "label": "Displayed B"}, {"index": 2, "label": "Displayed A"}],
|
||||
"answers": [{"index": 1, "label": "Displayed B", "value": "b"}],
|
||||
"correctIndices": [2], "correct": False, "confidence": 0, "dontKnow": False,
|
||||
"thinking": "Thinking exactly", "note": "Legacy note separately",
|
||||
"readingThinking": "Reading exactly", "explanation": "Teacher resolution",
|
||||
}
|
||||
source = record([assistant(call("quiz", {"question": "Raw Q", "options": [
|
||||
{"label": "Authored A", "value": "a"}, {"label": "Authored B", "value": "b"}],
|
||||
"correctAnswer": "a", "explanation": "Call explanation"})),
|
||||
result("quiz", answer, content=[{"type": "text", "text": json.dumps({
|
||||
"kind": "quiz", "confidence": 0, "thinking": "Thinking exactly",
|
||||
"thinkingWhileReading": "Reading exactly",
|
||||
"answers": [{"index": 1, "label": "Displayed B"}],
|
||||
})}])])
|
||||
before = copy.deepcopy(source)
|
||||
selected = sessions.selected_record(source)
|
||||
self.assertEqual(selected["record"]["messages"][1]["details"], answer)
|
||||
self.assertNotIn("content", selected["record"]["messages"][1])
|
||||
text = sessions.render_markdown(selected)
|
||||
for phrase in ("Displayed B", "Authored A", "Thinking exactly", "Legacy note separately", "Reading exactly", "Teacher resolution", "confidence (raw 0–100)**: 0", "Teacher-authored", "Reveal status", "snapshot display indexes"):
|
||||
self.assertIn(phrase, text)
|
||||
self.assertNotIn("99", text)
|
||||
self.assertEqual(source, before)
|
||||
|
||||
def test_extra_or_conflicting_answer_content_is_not_silently_discarded(self):
|
||||
cases = [
|
||||
({}, {"kind": "explain", "text": "only surviving learner answer"}),
|
||||
({"kind": "quiz", "confidence": 0}, {"kind": "quiz", "confidence": 99}),
|
||||
({"kind": "quiz"}, {"kind": "quiz", "thinking": "extra learner reasoning"}),
|
||||
]
|
||||
for details, compact in cases:
|
||||
with self.subTest(details=details, compact=compact):
|
||||
message = result("quiz", details, content=[{"type": "text", "text": json.dumps(compact)}])
|
||||
selected = sessions.selected_record(record([message]))
|
||||
self.assertEqual(selected["record"]["messages"][0]["content"], message["content"])
|
||||
text = sessions.render_markdown(selected)
|
||||
self.assertIn("may differ from the primary details snapshot", text)
|
||||
self.assertIn(sessions._text(json.dumps(compact)), text)
|
||||
message = result("quiz", {"kind": "quiz", "note": "Legacy thought", "readingThinking": "Reading thought"},
|
||||
content=[{"type": "text", "text": '{"kind":"quiz","thinking":"Legacy thought","thinkingWhileReading":"Reading thought"}'}])
|
||||
selected = sessions.selected_record(record([message]))
|
||||
self.assertNotIn("content", selected["record"]["messages"][0])
|
||||
self.assertIn("verified redundant", selected["omissions"][0]["reason"])
|
||||
diagnostic = sessions.selected_record(record([message]), include_operations=True)
|
||||
self.assertEqual(diagnostic["record"]["messages"][0]["content"], message["content"])
|
||||
non_json = result("explain", content=[{"type": "text", "text": '{"text": NaN}'}])
|
||||
self.assertIn("unparsed; full snapshot unavailable", self.render(record([non_json])))
|
||||
|
||||
def test_only_exact_learning_acknowledgments_are_noise(self):
|
||||
for tool, ack in sessions.ACK_TEXT.items():
|
||||
with self.subTest(tool=tool):
|
||||
message = result(tool, content=[{"type": "text", "text": ack}])
|
||||
default = sessions.selected_record(record([message]))
|
||||
self.assertNotIn("content", default["record"]["messages"][0])
|
||||
full = sessions.selected_record(record([message]), include_operations=True)
|
||||
self.assertEqual(full["record"]["messages"][0]["content"], message["content"])
|
||||
message["content"] = [{"type": "text", "text": "Historical teaching/assessment, not an acknowledgment"}]
|
||||
self.assertIn("Historical teaching/assessment", self.render(record([message])))
|
||||
dispatched = result("research", "Dispatched (job example).", content=[{"type": "text", "text": "Full historical research brief"}])
|
||||
self.assertIn("Full historical research brief", self.render(record([dispatched])))
|
||||
|
||||
def test_all_answer_modes_and_zero_false_empty_values(self):
|
||||
payloads = [
|
||||
("probe", {"kind": "quiz", "dontKnow": True, "answers": [], "thinking": "don't know reasoning"}),
|
||||
("ask_user_question", {"kind": "ask", "other": "Other exact text", "answers": [], "text": "typed answer"}),
|
||||
("explain", {"kind": "explain", "text": "free recall answer", "committed": "full committed solution", "readingThinking": "reading channel"}),
|
||||
("recall", {"kind": "explain", "text": "recalled verbatim", "committed": "recall key"}),
|
||||
("quiz", {"kind": "quiz", "dismissed": True, "answers": [], "dontKnow": False}),
|
||||
]
|
||||
text = self.render(record([result(name, payload) for name, payload in payloads]))
|
||||
for _, payload in payloads:
|
||||
for value in payload.values():
|
||||
if isinstance(value, str):
|
||||
self.assertIn(sessions._text(value), text)
|
||||
self.assertIn("Learner don't know**: true", text)
|
||||
self.assertIn("Skipped/dismissed**: true", text)
|
||||
self.assertIn("Learner answers (stored display labels/indexes)**: []", text)
|
||||
|
||||
def test_json_content_fallback_when_details_absent_or_null(self):
|
||||
payload = {"kind": "explain", "text": "fallback answer", "note": "old note", "thinkingWhileReading": "fallback reading", "confidence": 0}
|
||||
for details in ({}, {"details": None}):
|
||||
with self.subTest(details=details):
|
||||
message = result("recall", content=[{"type": "text", "text": json.dumps(payload)}])
|
||||
message.update(details)
|
||||
text = self.render(record([message]))
|
||||
self.assertIn("JSON fallback", text)
|
||||
self.assertIn("fallback answer", text)
|
||||
self.assertIn("old note", text)
|
||||
self.assertIn("fallback reading", text)
|
||||
self.assertIn("confidence (raw 0–100)**: 0", text)
|
||||
text = self.render(record([result("probe", content=[{"type": "text", "text": "not JSON"}])]))
|
||||
self.assertIn("unparsed; full snapshot unavailable", text)
|
||||
self.assertIn("not JSON", text)
|
||||
# Parsing one text block must not hide sibling images/extra fields or
|
||||
# misattribute a later block to content[0].text.
|
||||
richer = result("recall", content=[
|
||||
{"type": "image", "mimeType": "image/png", "data": "BINARY-NOT-PRINTED"},
|
||||
{"type": "text", "text": json.dumps(payload)},
|
||||
])
|
||||
text = self.render(record([richer]))
|
||||
self.assertIn("image/png", text)
|
||||
self.assertIn("fallback answer", text)
|
||||
self.assertNotIn("BINARY-NOT-PRINTED", text)
|
||||
self.assertNotIn("content[0].text (JSON fallback", text)
|
||||
richer["content"] = [{"type": "text", "text": json.dumps(payload), "futureField": "retain block metadata"}]
|
||||
self.assertIn("retain block metadata", self.render(record([richer])))
|
||||
|
||||
def test_pending_failed_and_legacy_duplicate_call_ids(self):
|
||||
messages = [
|
||||
assistant(call("quiz", {"question": "First", "correctAnswer": "first key"}), message_id="first"),
|
||||
result("quiz", {"kind": "quiz", "text": "first answer"}),
|
||||
assistant(call("quiz", {"question": "Failed", "correctAnswer": "failed key"}), message_id="second"),
|
||||
result("quiz", {"problems": ["validation failure"]}, isError=True,
|
||||
content=[{"type": "text", "text": "failure text"}]),
|
||||
assistant(call("problem", {"statement": "Pending", "modelSolution": "pending full model solution"}), message_id="third"),
|
||||
]
|
||||
statuses = sessions._call_statuses(messages)
|
||||
self.assertIn("result recorded", statuses[(0, 0)])
|
||||
self.assertIn("FAILED", statuses[(2, 0)])
|
||||
self.assertIn("PENDING", statuses[(4, 0)])
|
||||
text = self.render(record(messages))
|
||||
for word in ("first key", "failed key", "pending full model solution", "validation failure", "failure text", "not successful", "PENDING", "even pending"):
|
||||
self.assertIn(word, text)
|
||||
orphan = result("quiz", {"kind": "quiz", "text": "orphan answer"}, call_id="later")
|
||||
statuses = sessions._call_statuses([orphan, assistant(call("quiz", {}, call_id="later"))])
|
||||
self.assertIn("PENDING", statuses[(1, 0)])
|
||||
|
||||
def test_flags_are_separate_and_do_not_hide_learner_thinking(self):
|
||||
source = record([
|
||||
assistant({"type": "thinking", "thinking": "PRIVATE AGENT THOUGHT", "thinkingSignature": "SIGNATURE"},
|
||||
{"type": "text", "text": "Teacher text"}, call("read", {"path": "NO FILE OPEN"}),
|
||||
usage={"tokens": 123}, provider="vendor", responseId="provider-response"),
|
||||
result("read", "routine read result", content=[{"type": "text", "text": "routine read result"}]),
|
||||
result("quiz", {"thinking": "LEARNER THOUGHT", "readingThinking": "LEARNER READING"}),
|
||||
marker("practice-thinking", "practice content", {"text": "PRACTICE THOUGHT"}),
|
||||
], turns=[{"usage": 42}], extractions=[{"cost": 23}])
|
||||
default = self.render(source)
|
||||
for flag, value in (("agent thinking", "PRIVATE AGENT THOUGHT"), ("metadata", "provider-response"), ("routine", "routine read result")):
|
||||
self.assertNotIn(value, default, flag)
|
||||
for flags in ({}, {"include_agent_thinking": True}, {"include_operations": True},
|
||||
{"include_operations": True, "include_agent_thinking": True}):
|
||||
text = self.render(source, **flags)
|
||||
for learner in ("LEARNER THOUGHT", "LEARNER READING", "PRACTICE THOUGHT"):
|
||||
self.assertIn(learner, text)
|
||||
self.assertEqual("PRIVATE AGENT THOUGHT" in text, flags.get("include_agent_thinking", False))
|
||||
self.assertEqual("routine read result" in text, flags.get("include_operations", False))
|
||||
self.assertEqual("SIGNATURE" in text, flags.get("include_agent_thinking", False) and flags.get("include_operations", False))
|
||||
|
||||
def test_full_artifacts_refs_and_no_payload_embedding(self):
|
||||
brief = "FULL BRIEF START\n" + "long evidence paragraph.\n" * 800 + "FULL BRIEF END"
|
||||
messages = [
|
||||
marker("subagent-steer", "Delivered research", {"jobId": "job-research", "kind": "research", "toolCallId": "research-call", "result": {"brief": brief, "sources": [{"title": "Source", "url": "https://example.invalid/source"}]}}),
|
||||
marker("subagent-steer", "PDF delivery", {"jobId": "job-pdf", "kind": "pdf", "toolCallId": "pdf-call", "result": {"brief": brief, "path": "/never/open.pdf", "pagesRead": [1, 4], "pagesConsulted": [2]}}),
|
||||
marker("subagent-steer", "Demo reference", {"kind": "demo", "result": {"title": "Demo title", "html": "<script>PAYLOAD-HTML</script>", "tested": True, "notes": "Demo notes"}}),
|
||||
marker("subagent-steer", "Visual reference", {"kind": "visual", "result": {"svg": "<svg>PAYLOAD-SVG</svg>", "notes": "Visual notes"}}),
|
||||
{"id": "image-user", "role": "user", "content": [{"type": "image", "mimeType": "image/png", "data": "PAYLOAD-BASE64"}, {"type": "text", "text": "Describe this image"}]},
|
||||
]
|
||||
for flags in ({}, {"include_operations": True, "include_agent_thinking": True}):
|
||||
selected = sessions.selected_record(record(messages), **flags)
|
||||
text = sessions.render_markdown(selected)
|
||||
output = json.dumps(selected)
|
||||
self.assertIn(brief, text)
|
||||
for phrase in ("job-research", "research-call", "job-pdf", "pdf-call", "/never/open.pdf", "pagesRead", "pagesConsulted", "Demo title", "Demo notes", "Visual notes", "image/png"):
|
||||
self.assertIn(phrase, text)
|
||||
for payload in ("PAYLOAD-HTML", "PAYLOAD-SVG", "PAYLOAD-BASE64"):
|
||||
self.assertNotIn(payload, text)
|
||||
self.assertNotIn(payload, output)
|
||||
self.assertEqual(selected["omitted_field_count"], 3)
|
||||
# A persisted full artifact result is not a routine dispatch acknowledgment.
|
||||
direct = self.render(record([result("research", {"brief": brief}), result("pdf", {"brief": brief, "pagesRead": [7]})]))
|
||||
self.assertEqual(direct.count(brief), 2)
|
||||
self.assertIn("pagesRead", direct)
|
||||
|
||||
def test_graph_context_compact_default_and_full_operations(self):
|
||||
context = marker("graph-context", "VERBOSE INJECTED GRAPH PREAMBLE", {
|
||||
"propositions": [{"id": "node-a", "name": "Node A", "statement": "VERBOSE GRAPH STATEMENT"},
|
||||
{"id": "node-b", "name": "Node B", "statement": "SECOND GRAPH STATEMENT"}],
|
||||
"chains": ["chain-a", "chain-b"], "futureContext": "unknown preserved context",
|
||||
}, marker_id="context-id")
|
||||
source = record([context,
|
||||
{"id": "learner", "role": "user", "content": "VERBOSE GRAPH STATEMENT"},
|
||||
assistant({"type": "text", "text": "Teacher speech preserved"}),
|
||||
marker("practice-thinking", "Learner practice speech", {"text": "Learner trace"}),
|
||||
marker("subagent-steer", "Artifact delivery", {"kind": "research", "result": {"brief": "Full artifact speech"}})])
|
||||
original = copy.deepcopy(source)
|
||||
compact = sessions.selected_record(source)
|
||||
selected_context = compact["record"]["messages"][0]
|
||||
self.assertNotIn("content", selected_context)
|
||||
self.assertNotIn("statement", selected_context["details"]["propositions"][0])
|
||||
self.assertEqual(selected_context["details"]["chains"], ["chain-a", "chain-b"])
|
||||
self.assertEqual({o["source"] for o in compact["omissions"]}, {
|
||||
"messages[0].content", "messages[0].details.propositions[0].statement",
|
||||
"messages[0].details.propositions[1].statement"})
|
||||
text = sessions.render_markdown(compact)
|
||||
for phrase in ("context-id", "node-a", "Node A", "node-b", "Node B", "chain-a", "chain-b", "unknown preserved context",
|
||||
"VERBOSE GRAPH STATEMENT", "Teacher speech preserved", "Learner practice speech", "Learner trace", "Full artifact speech"):
|
||||
self.assertIn(phrase, text)
|
||||
self.assertNotIn("VERBOSE INJECTED GRAPH PREAMBLE", text)
|
||||
self.assertNotIn("SECOND GRAPH STATEMENT", text)
|
||||
self.assertNotIn("Unsupported/unrendered native content shape", text)
|
||||
full = sessions.selected_record(source, include_operations=True)
|
||||
self.assertEqual(full["record"]["messages"][0], context)
|
||||
self.assertIn("VERBOSE INJECTED GRAPH PREAMBLE", sessions.render_markdown(full))
|
||||
self.assertIn("SECOND GRAPH STATEMENT", sessions.render_markdown(full))
|
||||
self.assertEqual(source, original)
|
||||
# Unknown context shapes cannot safely be compacted into references.
|
||||
unknown = marker("graph-context", "only stored context", {"future": "shape"})
|
||||
self.assertIn("only stored context", self.render(record([unknown])))
|
||||
|
||||
def test_unknown_shapes_markers_and_content_are_explicit(self):
|
||||
source = record([
|
||||
marker("new-marker", "unknown content", {"futureField": "opaque payload"}),
|
||||
assistant({"type": "future-block", "opaque": "block evidence"}, call("new-tool", {"note": "future tool"})),
|
||||
{"id": "new-role", "role": "alien", "content": {"unknown": "role evidence"}},
|
||||
result("new-tool", {"future": "result evidence"}),
|
||||
])
|
||||
text = self.render(source)
|
||||
self.assertGreaterEqual(text.lower().count("unsupported/unrendered"), 5)
|
||||
self.assertIn("Stored role: alien (unsupported semantics; fields preserved).", text)
|
||||
for value in ("unknown content", "opaque payload", "block evidence", "future tool", "role evidence", "result evidence"):
|
||||
self.assertIn(value, text)
|
||||
# HTML in ordinary learner text stays verbatim in selected JSON, but is inert markdown.
|
||||
source = record([{"role": "user", "id": "u", "content": '<script>alert("untrusted")</script>'}])
|
||||
self.assertEqual(sessions.selected_record(source)["record"]["messages"][0]["content"], source["messages"][0]["content"])
|
||||
self.assertNotIn("<script>", self.render(source))
|
||||
self.assertIn("<script>", self.render(source))
|
||||
|
||||
def test_cli_errors_isolated_list_and_missing_defaults(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
self.save(root, record())
|
||||
self.save(root, record(session_id="future", version=2))
|
||||
self.save(root, record(session_id="old", messages=[{"role": "assistant", "parts": []}]))
|
||||
(root / "chats" / "broken.json").write_text("{bad", encoding="utf-8")
|
||||
before = snapshot(root)
|
||||
output = self.cli(root, "--format", "json", "list")
|
||||
self.assertEqual(output.returncode, 0, output.stderr)
|
||||
value = json.loads(output.stdout)
|
||||
self.assertEqual(value["sessions"][0]["kind"], "teach")
|
||||
self.assertEqual(value["skipped_count"], 3)
|
||||
self.assertEqual({s["id"] for s in value["skipped"]}, {"future", "old", "broken"})
|
||||
self.assertEqual(output.stderr.count("Warning: skipped session"), 3)
|
||||
plain = self.cli(root, "list")
|
||||
self.assertIn("Skipped/unreadable sessions: 3", plain.stdout)
|
||||
for session_id, phrase in (("future", "Update the skill"), ("old", "UIMessage"), ("broken", "Cannot read"), ("missing", "Session not found"), ("../escape", "session ID")):
|
||||
result_ = self.cli(root, "read", session_id)
|
||||
self.assertNotEqual(result_.returncode, 0)
|
||||
self.assertIn(phrase, result_.stderr)
|
||||
self.assertNotIn("Traceback", result_.stderr)
|
||||
self.assertEqual(result_.stdout, "")
|
||||
good = self.cli(root, "--format", "json", "read", "sample", "--include-agent-thinking", "--include-operations")
|
||||
self.assertEqual(good.returncode, 0, good.stderr)
|
||||
selected = json.loads(good.stdout)
|
||||
self.assertIn("record", selected)
|
||||
self.assertEqual(selected["data_dir"], str(root.resolve()))
|
||||
self.assertTrue(selected["reading_notes"])
|
||||
self.assertNotEqual(self.cli(root, "read", "sample", "--format", "json").returncode, 0)
|
||||
self.assertEqual(before, snapshot(root))
|
||||
absent = root / "not-created"
|
||||
missing = self.cli(absent, "list")
|
||||
self.assertNotEqual(missing.returncode, 0)
|
||||
self.assertFalse(absent.exists())
|
||||
|
||||
def test_symlink_escape_refused_without_opening_external_record(self):
|
||||
with tempfile.TemporaryDirectory() as directory, tempfile.TemporaryDirectory() as outside:
|
||||
root = Path(directory)
|
||||
target = self.save(Path(outside), record(session_id="escape"))
|
||||
(root / "chats").mkdir()
|
||||
try:
|
||||
(root / "chats" / "escape.json").symlink_to(target)
|
||||
except OSError as exc:
|
||||
self.skipTest(str(exc))
|
||||
result_ = self.cli(root, "read", "escape")
|
||||
self.assertNotEqual(result_.returncode, 0)
|
||||
self.assertIn("outside", result_.stderr)
|
||||
listing = self.cli(root, "--format", "json", "list")
|
||||
self.assertEqual(json.loads(listing.stdout)["skipped_count"], 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,100 @@
|
||||
"""The skill remains a portable folder, not an integration with the app."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
sys.dont_write_bytecode = True
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
SKILL = REPO / "skills" / "learnharness"
|
||||
GOLDEN = REPO / "tests" / "fixtures" / "golden-data"
|
||||
|
||||
|
||||
def snapshot(root):
|
||||
return {
|
||||
path.relative_to(root).as_posix(): (path.read_bytes(), path.stat().st_mtime_ns) if path.is_file() else None
|
||||
for path in root.rglob("*")
|
||||
}
|
||||
|
||||
|
||||
class SkillFolderTest(unittest.TestCase):
|
||||
def test_portable_frontmatter_and_local_reference(self):
|
||||
text = (SKILL / "SKILL.md").read_text(encoding="utf-8")
|
||||
frontmatter = text.split("---", 2)[1]
|
||||
fields = dict(line.split(": ", 1) for line in frontmatter.strip().splitlines())
|
||||
self.assertEqual(fields["name"], SKILL.name)
|
||||
self.assertTrue(0 < len(fields["description"]) <= 1024)
|
||||
self.assertTrue(0 < len(fields["compatibility"]) <= 500)
|
||||
self.assertEqual(fields["license"], "AGPL-3.0-only")
|
||||
self.assertLess(len(text.splitlines()), 500)
|
||||
self.assertIn("references/data-format.md", text)
|
||||
self.assertTrue((SKILL / "references" / "data-format.md").is_file())
|
||||
self.assertIn("Never modify LearnHarness's data", text)
|
||||
self.assertIn("archival view", text)
|
||||
self.assertNotIn("/Users/amos/", text)
|
||||
|
||||
def test_license_is_bundled_for_a_standalone_copy(self):
|
||||
self.assertEqual((SKILL / "LICENSE").read_bytes(), (REPO / "LICENSE").read_bytes())
|
||||
self.assertIn("GNU AFFERO GENERAL PUBLIC LICENSE", (SKILL / "LICENSE").read_text(encoding="utf-8"))
|
||||
|
||||
def test_python_syntax_supports_declared_minimum(self):
|
||||
for script in (SKILL / "scripts").glob("*.py"):
|
||||
with self.subTest(script=script.name):
|
||||
ast.parse(script.read_text(encoding="utf-8"), filename=str(script), feature_version=(3, 10))
|
||||
|
||||
def test_copied_skill_runs_outside_repository_and_creates_no_cache_or_data(self):
|
||||
with tempfile.TemporaryDirectory() as temporary:
|
||||
base = Path(temporary)
|
||||
skill = base / "standalone" / "learnharness"
|
||||
data = base / "learning-data"
|
||||
elsewhere = base / "unrelated-working-directory"
|
||||
elsewhere.mkdir()
|
||||
shutil.copytree(SKILL, skill)
|
||||
shutil.copytree(GOLDEN, data)
|
||||
before_skill = snapshot(skill)
|
||||
before_data = snapshot(data)
|
||||
env = dict(os.environ)
|
||||
env.pop("PYTHONDONTWRITEBYTECODE", None)
|
||||
env.pop("PYTHONPATH", None)
|
||||
commands = [
|
||||
("sessions.py", ["list"]),
|
||||
("sessions.py", ["read", "chat_mv27rrra_feff23b6"]),
|
||||
("sessions.py", ["read", "chat_mv293wv5_10656952"]),
|
||||
("sessions.py", ["read", "chat_mv293wv5_10656952", "--include-agent-thinking", "--include-operations"]),
|
||||
("graph.py", ["sessions"]),
|
||||
("graph.py", ["propositions"]),
|
||||
("graph.py", ["find", "*"]),
|
||||
("graph.py", ["chains"]),
|
||||
("graph.py", ["chains", "--node", "n_nonexistent"]),
|
||||
]
|
||||
for filename, arguments in commands:
|
||||
for output_format in ("markdown", "json"):
|
||||
with self.subTest(script=filename, arguments=arguments, format=output_format):
|
||||
# Deliberately omit -B here: the entrypoints must disable
|
||||
# bytecode before importing their helper on their own.
|
||||
completed = subprocess.run(
|
||||
[sys.executable, str(skill / "scripts" / filename), "--data-dir", str(data), "--format", output_format, *arguments],
|
||||
cwd=elsewhere, env=env, capture_output=True, text=True, encoding="utf-8", timeout=30,
|
||||
)
|
||||
self.assertEqual(completed.returncode, 0, completed.stderr)
|
||||
self.assertTrue(completed.stdout.strip())
|
||||
if output_format == "json":
|
||||
result = json.loads(completed.stdout)
|
||||
self.assertEqual(result["data_dir"], str(data.resolve()))
|
||||
else:
|
||||
self.assertIn(str(data.resolve()), completed.stdout)
|
||||
self.assertEqual(snapshot(skill), before_skill)
|
||||
self.assertEqual(snapshot(data), before_data)
|
||||
self.assertEqual(list(elsewhere.iterdir()), [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user