Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108"""Python hook callables through the real daemon: a callback the modelregisters in one cell gates run()/files in later cells, in the same kernel,and goes away with it. Catches a kernel-side gate that never sees callbacksregistered by a previous cell (a registry reset per cell, or one bound to thewrong module instance), which unit tests of albedo_hooks cannot reach."""
import jsonimport unittest
from harness import Albedo, Provider, python, text
class PythonHooksTests(unittest.TestCase): def setUp(self): self.replies = [] self.provider = Provider( lambda request: self.replies.pop(0) if self.replies else text("done") ) self.addCleanup(self.provider.close) self.app = Albedo(self.provider, protocol="responses") self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None)
def session(self): sid = self.app.session() with self.app.api( f"/sessions/{sid}/extensions", {"name": "hooks", "enabled": True} ) as response: json.load(response) return sid
def turn(self, sid): before = len(self.provider.requests) self.app.prompt(sid, "exercise hooks").close() self.app.idle(sid) return [ json.dumps(e["request"]["input"]) for e in self.provider.requests[before:] ]
def test_callback_blocks_rewrites_and_adds_context(self): sid = self.session() blocked = self.app.workspace / "blocked.txt" rewritten = self.app.workspace / "rewritten.txt" moved = self.app.workspace / "moved.txt" self.replies = [ python( "def gate(i):\n" " path = i['tool_input']['file_path']\n" " if path.endswith('blocked.txt'):\n" " return {'decision': 'block', 'reason': 'PY_CALLBACK_NO'}\n" " if path.endswith('moved.txt'):\n" f" return {{'hookSpecificOutput': {{'permissionDecision': 'allow', 'updatedInput': {{'file_path': {str(rewritten)!r}, 'content': 'x'}}}}}}\n" " return {'hookSpecificOutput': {'additionalContext': 'PY_CALLBACK_CONTEXT'}}\n" "remove = hooks.register('PreToolUse', gate, 'Write')\n" "print(hooks.registered())" ), python( "try:\n" f" files.write({str(blocked)!r}, 'no')\n" "except PermissionError as error:\n" " print('RAISED', error)\n" f"files.write({str(moved)!r}, 'y')" ), python("remove(); print('REMOVED', hooks.registered())"), text("done"), ] requests = self.turn(sid) self.assertEqual(len(requests), 4) self.assertIn("PY_CALLBACK_NO", requests[2]) self.assertIn("RAISED", requests[2]) self.assertIn("PY_CALLBACK_CONTEXT", requests[2]) self.assertIn("REMOVED", requests[3]) self.assertFalse(blocked.exists()) self.assertFalse(moved.exists()) self.assertEqual(rewritten.read_text(), "x")
def test_callbacks_on_outer_events_stop_and_outer_tool_call(self): sid = self.session() self.replies = [ python( "seen = []\n" "def pre(i):\n" " seen.append(i['tool_name'])\n" " return {'hookSpecificOutput': {'additionalContext': 'PY_OUTER_' + 'PRE'}}\n" "def stop(i):\n" " if not i['stop_hook_active']:\n" " return {'decision': 'block', 'reason': 'PY_STOP_REASON'}\n" "hooks.register('PreToolUse', pre, '^python$')\n" "hooks.register('Stop', stop)\n" "print('registered')" ), python("print('second cell')"), text("first answer"), text("second answer"), ] requests = self.turn(sid) # the registering cell ran before its callbacks existed, so only the # second cell is gated; the Stop callback then continues the turn once self.assertEqual(len(requests), 4) self.assertNotIn("PY_OUTER_PRE", requests[1]) self.assertIn("PY_OUTER_PRE", requests[2]) self.assertIn("PY_STOP_REASON", requests[3]) self.assertIn("first answer", requests[3])
if __name__ == "__main__": unittest.main()