from __future__ import annotations
import asyncio
import inspect
import json
import sqlite3
import struct
import tempfile
import unittest
from pathlib import Path
from klbr.hooks import AttentionDecision, DeliveryDecision, describe
from klbr_runtime.release import fingerprint
from klbr_runtime.wire import MAX_FRAME, ProtocolError, encode, read_frame
from peer import ROOT, behavior_copy
class FrameTests(unittest.IsolatedAsyncioTestCase):
async def read(self,data):
reader=asyncio.StreamReader();reader.feed_data(data);reader.feed_eof()
return await read_frame(reader)
async def test_length_prefix_roundtrip_unicode(self):
value={"test":"hello 🌙"}
frame=encode(value)
self.assertEqual(struct.unpack('>I',frame[:4])[0],len(frame)-4)
self.assertEqual(await self.read(frame),value)
async def test_truncated_frames_fail(self):
for frame in [b'\x00\x00',struct.pack('>I',20)+b'{}']:
with self.assertRaises(asyncio.IncompleteReadError): await self.read(frame)
async def test_oversize_rejected_before_payload_read(self):
with self.assertRaises(ProtocolError): await self.read(struct.pack('>I',MAX_FRAME+1))
with self.assertRaises(ProtocolError): encode({"large":"x"*MAX_FRAME})
async def test_duplicate_keys_and_nonfinite_values_rejected(self):
for raw in [b'{"a":1,"a":2}',b'{"x":NaN}',b'{"x":Infinity}',b'\xff',b'{']:
with self.subTest(raw=raw),self.assertRaises(ProtocolError):
await self.read(struct.pack('>I',len(raw))+raw)
async def test_shared_golden_frames(self):
fixtures=json.loads((ROOT/"tests/runtime-fixtures/envelopes.json").read_text())
for fixture in fixtures:
with self.subTest(kind=fixture["message"]["kind"]):
self.assertEqual(await self.read(encode(fixture)),fixture)
class ValueTests(unittest.TestCase):
def test_decision_shapes_are_explicit(self):
# A mistyped public SDK slot must fail locally, not kill a workbench by
# sending an invalid typed wire variant to the host.
with self.assertRaises(ValueError):
asyncio.run(describe("misspelled.slot"))
self.assertEqual(DeliveryDecision.allow().to_wire(),{"action":"allow"})
self.assertEqual(AttentionDecision.defer(8).to_wire(),{"action":"defer","seconds":8})
for decision in [DeliveryDecision("allow","invalid"),DeliveryDecision.hold(""),
AttentionDecision.defer(float("nan")),AttentionDecision.defer(-1)]:
with self.subTest(decision=decision),self.assertRaises(ValueError): decision.to_wire()
def test_behavior_identity_changes_with_source(self):
with tempfile.TemporaryDirectory() as directory:
root=behavior_copy(Path(directory)/"candidate")
original=fingerprint(root)
self.assertEqual(original,fingerprint(ROOT/"behavior"))
with (root/"klbr_hooks/defaults.py").open("a") as file: file.write("\n# a change\n")
self.assertNotEqual(original,fingerprint(root))
def test_behavior_symlinks_rejected(self):
with tempfile.TemporaryDirectory() as directory:
root=behavior_copy(Path(directory)/"candidate")
(root/"outside.py").symlink_to('/etc/passwd')
with self.assertRaises(ValueError): fingerprint(root)
def test_behavior_root_symlinks_rejected(self):
with tempfile.TemporaryDirectory() as directory:
root=Path(directory)/"alias"
root.symlink_to(ROOT/"behavior",target_is_directory=True)
with self.assertRaises(ValueError): fingerprint(root)
def test_oversized_behavior_file_rejected(self):
with tempfile.TemporaryDirectory() as directory:
root=behavior_copy(Path(directory)/"candidate")
with (root/"oversized.py").open("wb") as file:
file.truncate(2_097_153)
with self.assertRaises(ValueError): fingerprint(root)
def test_shared_behavior_digest(self):
fixture=json.loads((ROOT/"tests/runtime-fixtures/behavior.json").read_text())
self.assertEqual(fingerprint(ROOT/"behavior"),fixture["sha256"])
class SchemaTests(unittest.TestCase):
"""Executes the exact shipped migration, not Rust Store methods.
Python's linked SQLite is only used for these serial schema tests. The Rust
runtime separately enforces the documented WAL-reset fixed-version floor.
"""
def setUp(self):
self.db=sqlite3.connect(':memory:')
self.db.execute('PRAGMA foreign_keys=ON')
self.db.executescript((ROOT/'klbr-runtime/migrations/001_runtime.sql').read_text())
self.db.execute("INSERT INTO sessions(id) VALUES('root')")
self.db.execute("INSERT INTO sessions(id) VALUES('child')")
self.db.execute("INSERT INTO releases(id,path) VALUES('revision','/test')")
self.db.commit()
def tearDown(self): self.db.close()
def event(self,key=None,session='root',kind='input'):
return self.db.execute("INSERT INTO events(session_id,kind,source_key,payload,created_ms) VALUES(?,?,?,?,0)",
(session,kind,key,'{}')).lastrowid
def execution(self,identity='cell',session='root'):
event=self.event(session=session,kind='execution.started')
self.db.execute("INSERT INTO executions(id,session_id,generation,hook_revision,state,started_event) VALUES(?,?,'g','revision','running',?)",
(identity,session,event))
def test_schema_identity(self):
self.assertEqual(self.db.execute('PRAGMA user_version').fetchone()[0],1)
self.assertEqual(self.db.execute('PRAGMA application_id').fetchone()[0],1263288882)
def test_originals_cannot_be_updated_or_deleted(self):
identity=self.event()
for sql in ['UPDATE events SET created_ms=1 WHERE seq=?','DELETE FROM events WHERE seq=?']:
with self.assertRaises(sqlite3.IntegrityError): self.db.execute(sql,(identity,))
def test_source_identity_is_unique_per_session(self):
self.event('source-1')
with self.assertRaises(sqlite3.IntegrityError): self.event('source-1')
self.event('source-1',session='child')
def test_burst_has_one_inbox_row_per_input(self):
for i in range(500):
event=self.event(f'input-{i}')
self.db.execute('INSERT INTO inbox(event_id) VALUES(?)',(event,))
self.assertEqual(self.db.execute("SELECT count(*) FROM inbox WHERE status='pending'").fetchone()[0],500)
def test_foreign_keys_prevent_orphaned_inbox(self):
with self.assertRaises(sqlite3.IntegrityError): self.db.execute('INSERT INTO inbox(event_id) VALUES(999)')
def test_unfinished_execution_unique_with_independent_sessions(self):
self.execution()
with self.assertRaises(sqlite3.IntegrityError): self.execution('second')
self.execution('child-cell',session='child')
def test_yield_does_not_open_a_second_foreground_slot(self):
self.execution()
self.db.execute("UPDATE executions SET state='yielded' WHERE id='cell'")
with self.assertRaises(sqlite3.IntegrityError): self.execution('second')
def test_confirmed_effect_requires_receipt(self):
self.execution()
with self.assertRaises(sqlite3.IntegrityError):
self.db.execute("INSERT INTO effects(id,execution_id,request_id,text,status) VALUES('e','cell','r','hello','confirmed')")
def test_publication_transaction_rolls_back_receipt_and_event_together(self):
self.execution();self.db.commit()
before=self.db.execute('SELECT count(*) FROM events').fetchone()[0]
self.db.execute('BEGIN IMMEDIATE')
event=self.event(kind='local.sent')
self.db.execute("INSERT INTO effects(id,execution_id,request_id,text,status,event_id) VALUES('e','cell','r','hello','confirmed',?)",(event,))
self.db.rollback()
self.assertEqual(self.db.execute('SELECT count(*) FROM effects').fetchone()[0],0)
self.assertEqual(self.db.execute('SELECT count(*) FROM events').fetchone()[0],before)
def test_dedup_is_by_request_not_message_text(self):
self.execution()
for identity in ['a','b']:
self.db.execute("INSERT INTO effects(id,execution_id,request_id,text,status) VALUES(?,'cell',?,'same','proposed')",(identity,identity))
with self.assertRaises(sqlite3.IntegrityError):
self.db.execute("INSERT INTO effects(id,execution_id,request_id,text,status) VALUES('c','cell','a','other','proposed')")
self.assertEqual(self.db.execute('SELECT count(*) FROM effects').fetchone()[0],2)
def test_impossible_ready_with_deadline_is_rejected(self):
with self.assertRaises(sqlite3.IntegrityError):
self.db.execute("UPDATE sessions SET state='ready',wake_at_ms=100 WHERE id='root'")
class PeerSurfaceTests(unittest.TestCase):
"""The peer surface names its subject once; the duplicated functional layer stays gone."""
def test_no_callable_repeats_peer_id(self):
import klbr.peer as peer
for name in ["send", "get", "stop"]:
signature = inspect.signature(getattr(peer.PeerSession, name))
self.assertNotIn("peer_id", signature.parameters, name)
for name in ["result", "cancel"]:
signature = inspect.signature(getattr(peer.PeerRequestHandle, name))
self.assertNotIn("peer_id", signature.parameters, name)
def test_subjectless_verbs_are_the_whole_module_surface(self):
import klbr.peer as peer
self.assertEqual(sorted(peer.__all__), ["PeerRequestHandle", "PeerSession", "answer", "open", "peers"])
for gone in ["send", "result", "cancel", "get", "stop", "list_peers", "submit_result"]:
self.assertFalse(hasattr(peer, gone), f"peer.{gone} repeats a subject the handle already carries")
def test_the_agents_alias_vocabulary_is_gone(self):
with self.assertRaises(ModuleNotFoundError):
import klbr.agents # noqa: F401