Something went wrong. Try again.
vibescrobble! every agent, live on stage. vibescrobble.com
Something went wrong. Try again.
29 kB · 686 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687"""`bin/hook.py`, run the way Claude Code runs it: one process per event, apayload on stdin, and whatever it prints read back as the hook's output.
Each test gets a machine of its own: a `PATH` holding only the commands itsays are installed, a didbot daemon socket or none, and a state directorynobody else writes. The `vibescrobble` on that `PATH` is a stand-in whose`status` answers from a directory the test fills, so "signed in" is a filethe test writes rather than a sign-in against a real server.
Run with `python3 -m unittest discover -s plugin/tests`."""
import jsonimport osimport shutilimport socketimport subprocessimport sysimport tempfileimport timeimport unittest
HOOK = os.path.join(os.path.dirname(__file__), os.pardir, "bin", "hook.py")
DID = "did:web:kestrel.agents.localhost"OTHER_DID = "did:web:heron.agents.localhost"PLC = "did:plc:nlzmjyfv6loqtxyzvdcznwgf"HANDLE = "alice.bsky.social"
#: `vibescrobble status --as <did>` exits 0 when `$SIGNED_IN/<did>` exists.FAKE_VIBESCROBBLE = """#!/bin/shif [ "$1" = status ] && [ -e "$SIGNED_IN/$3" ]; then exit 0; fiexit 1"""
FAKE_DIDBOT = "#!/bin/sh\nexit 0\n"
class Machine: """One machine's worth of what the hook looks at."""
def __init__(self, root, vibescrobble=True, didbot=True, daemon=True, account=None): self.root = root self.account = account #: More of the hook's environment, over what `run` sets. self.env = {} #: Where the hook runs, as the directory Claude Code was started in. self.cwd = None self.bin = os.path.join(root, "bin") self.signed_in_dir = os.path.join(root, "signed-in") self.data = os.path.join(root, "data") os.makedirs(self.bin) os.makedirs(self.signed_in_dir) if vibescrobble: self.install("vibescrobble", FAKE_VIBESCROBBLE) if didbot: self.install("didbot", FAKE_DIDBOT) self.sock = os.path.join(root, "agent.sock") self.listener = None if daemon: self.listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) self.listener.bind(self.sock)
def install(self, name, script): path = os.path.join(self.bin, name) with open(path, "w") as handle: handle.write(script) os.chmod(path, 0o755)
def close(self): if self.listener: self.listener.close()
def sign_in(self, did): open(os.path.join(self.signed_in_dir, did), "w").close()
def run(self, event, quiet_seconds=300, env_file=None, **payload): """The hook's `additionalContext` for one event, or `None`.""" payload = {"hook_event_name": event, "session_id": "s1", **payload} env = { "PATH": self.bin, "CLAUDE_PLUGIN_DATA": self.data, "DIDBOT_SOCK": self.sock, "SIGNED_IN": self.signed_in_dir, "VIBESCROBBLE_REMIND_SECONDS": str(quiet_seconds), "HOME": self.root, } if env_file: env["CLAUDE_ENV_FILE"] = env_file if self.account: env["VIBESCROBBLE_AS"] = self.account env.update(self.env) done = subprocess.run( [sys.executable, HOOK], input=json.dumps(payload), capture_output=True, text=True, env=env, cwd=self.cwd, timeout=10, check=True, ) errors = os.path.join(self.data, "vibescrobble", "hook-errors.log") assert not os.path.exists(errors), open(errors).read() if not done.stdout.strip(): return None parsed = json.loads(done.stdout) if "decision" in parsed: return parsed output = parsed["hookSpecificOutput"] assert output["hookEventName"] == event return output["additionalContext"]
def bash(self, command, quiet_seconds=300, **payload): return self.run( "PreToolUse", quiet_seconds=quiet_seconds, tool_name="Bash", tool_input={"command": command}, **payload, )
def go_quiet(self, agent_id=None): """Backdate this context's last reminder past any quiet period.""" name = f"s1-{agent_id or '-'}" path = os.path.join(self.data, "vibescrobble", "reminded", name) past = time.time() - 3600 os.utime(path, (past, past))
class HookTest(unittest.TestCase): def machine(self, **kwargs): root = tempfile.mkdtemp(prefix="vibescrobble-hook-") self.addCleanup(shutil.rmtree, root, ignore_errors=True) machine = Machine(root, **kwargs) self.addCleanup(machine.close) return machine
class Preflight(HookTest): """One fact for the first thing missing, once, and nothing after."""
def assert_one_fact_then_silence(self, machine, expected): said = machine.run("SessionStart") self.assertIn(expected, said) # A fact, not an instruction: no command to run, nothing to install. self.assertNotIn("`", said) self.assertNotIn("cargo", said) self.assertNotIn("vibescrobble login", said) self.assertIsNone(machine.bash("ls", quiet_seconds=0)) machine.go_quiet() self.assertIsNone(machine.bash("ls"))
def test_no_vibescrobble(self): machine = self.machine(vibescrobble=False) self.assert_one_fact_then_silence(machine, "vibescrobble is not installed")
def test_no_didbot(self): machine = self.machine(didbot=False) self.assert_one_fact_then_silence(machine, "didbot is not installed")
def test_no_daemon(self): machine = self.machine(daemon=False) self.assert_one_fact_then_silence(machine, "No didbot agent daemon")
def test_vibescrobble_is_named_before_didbot(self): machine = self.machine(vibescrobble=False, didbot=False, daemon=False) self.assertIn("vibescrobble is not installed", machine.run("SessionStart"))
def test_everything_present_opens_with_the_sign_in(self): machine = self.machine() said = machine.run("SessionStart") self.assertIn("vibescrobble login --as <your DID>", said) self.assertIn("didbot oauth approve", said) self.assertIn("public", said)
class Reminders(HookTest): def test_a_quiet_context_is_reminded(self): machine = self.machine() machine.run("SessionStart") self.assertIsNone(machine.bash("ls")) machine.go_quiet() said = machine.bash("ls") self.assertIn("vibescrobble post --as <your DID>", said) self.assertIn("vibescrobble login --as <your DID>", said) # And then not again straight away. self.assertIsNone(machine.bash("ls"))
def test_running_vibescrobble_restarts_the_quiet(self): machine = self.machine() machine.run("SessionStart") machine.go_quiet() self.assertIsNone( machine.bash(f"vibescrobble post --as {DID} --text 'x' --emoji '*'") ) self.assertIsNone(machine.bash("ls"))
def test_a_signed_in_context_is_not_told_to_sign_in(self): machine = self.machine() machine.run("SessionStart") machine.bash(f"vibescrobble login --as {DID}") machine.sign_in(DID) machine.go_quiet() said = machine.bash("ls") self.assertIn(f"vibescrobble post --as {DID}", said) self.assertNotIn("login", said)
def test_a_context_that_has_not_finished_signing_in_is_told_how(self): machine = self.machine() machine.run("SessionStart") machine.bash(f"vibescrobble login --as {DID}") machine.go_quiet() said = machine.bash("ls") self.assertIn(f"vibescrobble login --as {DID}", said)
def test_any_account_is_kept_and_a_placeholder_is_not(self): machine = self.machine() machine.run("SessionStart") machine.bash("vibescrobble post --as '<your DID>' --text x --emoji '*'") machine.go_quiet() self.assertIn("--as <your DID>", machine.bash("ls")) for account in (PLC, HANDLE): machine.bash(f"vibescrobble post --as {account} --text x --emoji '*'") machine.go_quiet() self.assertIn(f"vibescrobble post --as {account}", machine.bash("ls"))
def test_an_as_inside_the_text_is_text(self): machine = self.machine() machine.run("SessionStart") machine.bash( f"vibescrobble post --as {DID} --text 'unlike --as {OTHER_DID}' --emoji '*'" ) machine.go_quiet() said = machine.bash("ls") self.assertIn(f"vibescrobble post --as {DID}", said) self.assertNotIn(OTHER_DID, said)
def test_a_heredoc_that_writes_a_command_names_no_account(self): machine = self.machine() machine.run("SessionStart") machine.bash(f"vibescrobble post --as {DID} --text x --emoji '*'") # A script fed on stdin that writes the words into a page: the line # runs python3, not vibescrobble, quoted delimiter or not, `<<-` too. for opener, closer in (("<<'EOF'", "EOF"), ("<<EOF", "EOF"), ("<<-PY", "\tPY")): machine.bash( f"python3 - {opener}\n" f"open('page.html', 'w').write('vibescrobble login --as {OTHER_DID}')\n" f"print('vibescrobble post --as {OTHER_DID} --text y')\n" f"{closer}\n" ) machine.go_quiet() said = machine.bash("ls") self.assertIn(f"vibescrobble post --as {DID}", said, opener) self.assertNotIn(OTHER_DID, said, opener) # A command after the body still counts. machine.bash(f"cat > notes <<EOF\nvibescrobble --as {DID}\nEOF\nvibescrobble status --as {OTHER_DID}") machine.go_quiet() self.assertIn(f"vibescrobble login --as {OTHER_DID}", machine.bash("ls"))
def test_the_command_after_a_control_operator_names_its_own_account(self): machine = self.machine() machine.run("SessionStart") machine.bash(f"cd /tmp && ~/.cargo/bin/vibescrobble login --as {DID}; echo ok") machine.go_quiet() self.assertIn(f"vibescrobble login --as {DID}", machine.bash("ls")) machine.bash( f"vibescrobble status --as {DID} | cat; vibescrobble post --as {OTHER_DID}" " --text 'x' --emoji '*'" ) machine.go_quiet() self.assertIn(f"vibescrobble login --as {OTHER_DID}", machine.bash("ls"))
def test_an_unfinished_quote_keeps_the_account_before_it(self): machine = self.machine() machine.run("SessionStart") machine.bash(f"vibescrobble post --as {DID} --text 'it is not closed") machine.go_quiet() self.assertIn(f"vibescrobble post --as {DID}", machine.bash("ls"))
def test_a_text_that_is_the_word_as_does_not_hide_the_account(self): machine = self.machine() machine.run("SessionStart") machine.bash(f"vibescrobble post --text --as --as {DID} --emoji '*'") machine.go_quiet() self.assertIn(f"vibescrobble post --as {DID}", machine.bash("ls"))
class ConfiguredAccount(HookTest): """`$VIBESCROBBLE_AS` is an account, and needs nothing of didbot's."""
def test_it_needs_no_didbot(self): machine = self.machine(didbot=False, daemon=False, account=HANDLE) said = machine.run("SessionStart") self.assertIn(f"vibescrobble login --as {HANDLE}", said) self.assertIn(f"vibescrobble post --as {HANDLE}", said) self.assertIn("opens it in a browser", said) self.assertIn("`bot` self-label", said) self.assertNotIn("didbot", said)
def test_it_wins_over_a_didbot_identity(self): machine = self.machine(account=PLC) said = machine.run("SessionStart") self.assertIn(f"vibescrobble login --as {PLC}", said) self.assertNotIn("didbot oauth approve", said)
def test_signed_in_it_is_reminded_to_post_only(self): machine = self.machine(didbot=False, daemon=False, account=PLC) machine.sign_in(PLC) said = machine.run("SessionStart") self.assertIn(f"You are signed in to it as {PLC}", said) machine.go_quiet() said = machine.bash("ls") self.assertIn(f"vibescrobble post --as {PLC}", said) self.assertNotIn("login", said)
def test_a_quiet_context_is_told_who_approves(self): machine = self.machine(didbot=False, daemon=False, account=HANDLE) machine.run("SessionStart") machine.go_quiet() said = machine.bash("ls") self.assertIn(f"vibescrobble login --as {HANDLE}", said) self.assertIn("person who owns", said)
def test_without_it_or_didbot_recording_is_off(self): machine = self.machine(didbot=False, daemon=False) said = machine.run("SessionStart") self.assertIn("No vibescrobble account is configured", said)
class Beginnings(HookTest): def test_a_compacted_session_that_signed_in_is_not_told_to_again(self): machine = self.machine() machine.run("SessionStart", source="startup") machine.bash(f"vibescrobble login --as {DID}") machine.sign_in(DID) said = machine.run("SessionStart", source="compact") self.assertIn(f"signed in to it as {DID}", said) self.assertIn(f"vibescrobble post --as {DID}", said) self.assertNotIn("vibescrobble login", said) self.assertIn("public", said)
def test_a_subagent_does_not_inherit_its_parents_sign_in(self): machine = self.machine() machine.run("SessionStart") machine.bash(f"vibescrobble login --as {DID}") machine.sign_in(DID) said = machine.run("SubagentStart", agent_id="a1") self.assertIn("vibescrobble login --as <your DID>", said)
machine.bash(f"vibescrobble login --as {OTHER_DID}", agent_id="a1") machine.go_quiet(agent_id="a1") said = machine.bash("ls", agent_id="a1") self.assertIn(f"vibescrobble login --as {OTHER_DID}", said)
def test_a_subagent_must_post_before_it_can_stop(self): machine = self.machine() machine.run("SubagentStart", agent_id="a1")
output = machine.run("SubagentStop", agent_id="a1")
self.assertEqual(output["decision"], "block") self.assertIn("vibescrobble post --as", output["reason"]) self.assertIn("completion", output["reason"])
def test_a_subagent_is_held_for_a_post_once_and_then_may_stop(self): machine = self.machine() machine.run("SubagentStart", agent_id="a1") machine.run("SubagentStart", agent_id="a2")
self.assertEqual(machine.run("SubagentStop", agent_id="a1")["decision"], "block") self.assertIsNone(machine.run("SubagentStop", agent_id="a1")) self.assertIsNone(machine.run("SubagentStop", agent_id="a1")) # Another subagent of the same session is held once of its own. self.assertEqual(machine.run("SubagentStop", agent_id="a2")["decision"], "block") self.assertIsNone(machine.run("SubagentStop", agent_id="a2"))
def test_a_subagent_a_stop_hook_already_sent_back_may_stop(self): machine = self.machine() machine.run("SubagentStart", agent_id="a1")
self.assertIsNone( machine.run("SubagentStop", agent_id="a1", stop_hook_active=True) )
def test_a_subagent_that_posted_may_stop(self): machine = self.machine() machine.run("SubagentStart", agent_id="a1") machine.bash( f"vibescrobble post --as {DID} --text x --emoji '*'", agent_id="a1" )
self.assertIsNone(machine.run("SubagentStop", agent_id="a1"))
def test_a_subagent_that_cannot_post_is_not_blocked(self): machine = self.machine(vibescrobble=False, didbot=False, daemon=False)
self.assertIsNone(machine.run("SubagentStop", agent_id="a1"))
class Cleanup(HookTest): """What a session leaves on disk, and when it goes."""
def folders(self, machine): state = os.path.join(machine.root, ".local", "state", "vibescrobble") data = os.path.join(machine.data, "vibescrobble") return { "claude-session": os.path.join(state, "claude-session"), "reminded": os.path.join(data, "reminded"), "as": os.path.join(data, "as"), "posted": os.path.join(data, "posted"), "nudged": os.path.join(data, "nudged"), }
def names(self, machine, folder): try: return sorted(os.listdir(self.folders(machine)[folder])) except OSError: return []
def age(self, machine, folder, name, days): path = os.path.join(self.folders(machine)[folder], name) past = time.time() - days * 24 * 3600 os.utime(path, (past, past))
def test_the_end_of_a_session_removes_its_own_files_and_says_nothing(self): machine = self.machine() machine.run("SessionStart", session_id="s1") machine.run("SessionStart", session_id="s10") machine.run("SubagentStart", session_id="s1", agent_id="a1") machine.bash(f"vibescrobble post --as {DID}", session_id="s1") machine.bash(f"vibescrobble post --as {OTHER_DID}", session_id="s10") self.assertEqual(self.names(machine, "claude-session"), ["s1", "s10"]) self.assertEqual(self.names(machine, "reminded"), ["s1--", "s1-a1", "s10--"])
machine.run("SubagentStop", session_id="s1", agent_id="a1") self.assertEqual(self.names(machine, "posted"), ["s1--", "s10--"]) self.assertEqual(self.names(machine, "nudged"), ["s1-a1"])
self.assertIsNone(machine.run("SessionEnd", session_id="s1", reason="other")) self.assertEqual(self.names(machine, "claude-session"), ["s10"]) self.assertEqual(self.names(machine, "reminded"), ["s10--"]) self.assertEqual(self.names(machine, "posted"), ["s10--"]) self.assertEqual(self.names(machine, "nudged"), [])
def test_the_end_of_a_session_that_left_nothing_is_harmless(self): machine = self.machine() self.assertIsNone(machine.run("SessionEnd", session_id="s1")) self.assertIsNone(machine.run("SessionEnd"))
def test_a_resumed_session_keeps_its_sign_in_and_its_status_line(self): machine = self.machine() machine.run("SessionStart") machine.bash(f"vibescrobble login --as {DID}") machine.sign_in(DID) machine.run("SessionEnd") self.assertEqual(self.names(machine, "claude-session"), [])
said = machine.run("SessionStart", source="resume") self.assertIn(f"signed in to it as {DID}", said) self.assertEqual(self.names(machine, "claude-session"), ["s1"])
def test_a_session_that_never_named_an_account_gets_no_status_line_file(self): machine = self.machine(account=PLC) machine.run("SessionStart") self.assertEqual(self.names(machine, "claude-session"), [])
def test_the_start_of_a_session_prunes_what_nobody_has_touched_for_a_month(self): machine = self.machine() for session in ("old", "new"): machine.run("SessionStart", session_id=session) machine.bash(f"vibescrobble post --as {DID}", session_id=session) machine.run("SubagentStop", session_id="old", agent_id="a1") for folder, name in ( ("claude-session", "old"), ("reminded", "old--"), ("as", "old--"), ("posted", "old--"), ("nudged", "old-a1"), ): self.age(machine, folder, name, 31) self.age(machine, "as", "new--", 29)
machine.run("SessionStart", session_id="next") self.assertEqual(self.names(machine, "claude-session"), ["new"]) self.assertEqual(self.names(machine, "reminded"), ["new--", "next--"]) self.assertEqual(self.names(machine, "as"), ["new--"]) self.assertEqual(self.names(machine, "posted"), ["new--"]) self.assertEqual(self.names(machine, "nudged"), [])
def test_a_subagent_start_does_not_prune(self): machine = self.machine() machine.run("SessionStart", session_id="old") self.age(machine, "reminded", "old--", 31) machine.run("SubagentStart", session_id="s1", agent_id="a1") self.assertEqual(self.names(machine, "reminded"), ["old--", "s1-a1"])
class Model(HookTest): """The session's model, handed to `vibescrobble post` through the shell."""
def test_the_start_of_a_session_exports_its_model(self): machine = self.machine() env_file = os.path.join(machine.root, "env") machine.run("SessionStart", env_file=env_file, model="qwen3-4b it's") machine.run("SubagentStart", env_file=env_file, agent_id="a1") with open(env_file) as handle: exported = handle.read() self.assertEqual(exported, "export VIBESCROBBLE_MODEL='qwen3-4b it'\"'\"'s'\n") sourced = subprocess.run( ["sh", "-c", f". {env_file}; printf %s \"$VIBESCROBBLE_MODEL\""], capture_output=True, text=True, check=True, ) self.assertEqual(sourced.stdout, "qwen3-4b it's")
def test_nothing_is_exported_without_a_model_or_a_file(self): machine = self.machine() env_file = os.path.join(machine.root, "env") machine.run("SessionStart", env_file=env_file) machine.run("SessionStart", model="claude-opus-5-5") self.assertFalse(os.path.exists(env_file))
class ContextModel(HookTest): """The model each context runs, left for `vibescrobble post` to name."""
def transcript(self, machine, path, *models): os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w") as handle: for model in models: entry = {"type": "assistant", "message": {"model": model}} handle.write(json.dumps(entry) + "\n") handle.write(json.dumps({"type": "user"}) + "\n")
def hint(self, machine, did): stem = "".join(c if c.isalnum() else f"_{ord(c):02X}" for c in did) path = os.path.join(machine.root, ".local", "state", "vibescrobble", stem) try: with open(path + ".model") as handle: return handle.read() except OSError: return None
def test_each_context_names_its_own_model(self): machine = self.machine() session = os.path.join(machine.root, "projects", "s1.jsonl") self.transcript(machine, session, "claude-opus-5-5", "<synthetic>") subagent = os.path.join( machine.root, "projects", "s1", "subagents", "agent-a1.jsonl" ) self.transcript(machine, subagent, "qwen3-4b")
machine.bash(f"vibescrobble post --as {DID} --text x", transcript_path=session) machine.bash( f"vibescrobble post --as {OTHER_DID} --text y", transcript_path=session, agent_id="a1", ) self.assertEqual(self.hint(machine, DID), "claude-opus-5-5") self.assertEqual(self.hint(machine, OTHER_DID), "qwen3-4b")
def test_a_transcript_naming_no_model_takes_the_hint_away(self): machine = self.machine() session = os.path.join(machine.root, "projects", "s1.jsonl") self.transcript(machine, session, "claude-opus-5-5") machine.bash(f"vibescrobble post --as {DID}", transcript_path=session) self.transcript(machine, session, "not a model id") machine.bash(f"vibescrobble post --as {DID}", transcript_path=session) self.assertIsNone(self.hint(machine, DID))
def test_an_empty_or_relative_base_is_passed_over_as_the_cli_does(self): machine = self.machine() machine.cwd = os.path.join(machine.root, "checkout") os.makedirs(machine.cwd) machine.env = {"VIBESCROBBLE_STATE": "", "XDG_STATE_HOME": "state"} session = os.path.join(machine.root, "projects", "s1.jsonl") self.transcript(machine, session, "claude-opus-5-5") machine.bash(f"vibescrobble post --as {DID}", transcript_path=session) self.assertEqual(self.hint(machine, DID), "claude-opus-5-5") self.assertEqual(os.listdir(machine.cwd), [])
class StatusLine(HookTest): """The DID a session names, left where `vibescrobble statusline` reads it."""
def left(self, machine, session_id): path = os.path.join( machine.root, ".local", "state", "vibescrobble", "claude-session", session_id ) try: with open(path) as handle: return handle.read() except OSError: return None
def test_the_session_s_did_is_left_and_a_subagent_s_is_not(self): machine = self.machine() machine.bash(f"vibescrobble login --as {DID}") self.assertEqual(self.left(machine, "s1"), DID) machine.bash(f"vibescrobble post --as {OTHER_DID}", agent_id="a1") self.assertEqual(self.left(machine, "s1"), DID)
def test_a_session_id_is_a_filename_of_letters_digits_and_dashes(self): machine = self.machine() machine.bash(f"vibescrobble status --as {DID}", session_id="../a b") self.assertEqual(self.left(machine, "---a-b"), DID)
#: What `cursor_path` names `/home/kestrel/.claude/projects/p/s1.jsonl`, as#: the CLI's `usage/cursor.rs` test names it too.KNOWN_CURSOR = "375914a1d0bded0078d4bb7416501a66"
class Counting(HookTest): """The transcript left for counting, and its cursor seeded."""
def state(self, machine, *parts): return os.path.join(machine.root, ".local", "state", "vibescrobble", *parts)
def cursor(self, machine, found): import hashlib
name = hashlib.sha256(found.encode()).hexdigest()[:32] return self.state(machine, "usage", "cursors", name + ".json")
def read_cursor(self, machine, found): with open(self.cursor(machine, found)) as handle: return json.load(handle)
def write(self, path, size): os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w") as handle: handle.write("x" * (size - 1) + "\n")
def test_the_cursor_is_named_as_the_cli_names_it(self): sys.path.insert(0, os.path.dirname(HOOK)) self.addCleanup(sys.path.remove, os.path.dirname(HOOK)) import hook
found = hook.cursor_path("/home/kestrel/.claude/projects/p/s1.jsonl") self.assertEqual(os.path.basename(found), KNOWN_CURSOR + ".json")
def test_each_context_leaves_its_own_transcript(self): machine = self.machine() session = os.path.join(machine.root, "projects", "s1.jsonl") machine.bash(f"vibescrobble post --as {DID} --text x", transcript_path=session) machine.bash( f"vibescrobble post --as {OTHER_DID} --text y", transcript_path=session, agent_id="a1", ) stem = lambda did: "".join(c if c.isalnum() else f"_{ord(c):02X}" for c in did) with open(self.state(machine, stem(DID) + ".transcript")) as handle: self.assertEqual(handle.read(), session) with open(self.state(machine, stem(OTHER_DID) + ".transcript")) as handle: self.assertEqual( handle.read(), os.path.join(machine.root, "projects", "s1", "subagents", "agent-a1.jsonl"), )
def test_a_context_starts_counting_where_it_began(self): machine = self.machine() session = os.path.join(machine.root, "projects", "s1.jsonl") # A resumed or forked session's file already holds copied history. self.write(session, 1000) machine.run("SessionStart", transcript_path=session) self.assertEqual(self.read_cursor(machine, session)["offset"], 1000) self.assertEqual(self.read_cursor(machine, session)["path"], session) # Starting again, as a compaction does, moves nothing. self.write(session, 5000) machine.run("SessionStart", transcript_path=session) self.assertEqual(self.read_cursor(machine, session)["offset"], 1000)
def test_a_subagent_starts_at_zero(self): machine = self.machine() session = os.path.join(machine.root, "projects", "s1.jsonl") subagent = os.path.join(machine.root, "projects", "s1", "subagents", "agent-a1.jsonl") machine.run("SubagentStart", transcript_path=session, agent_id="a1") self.assertEqual(self.read_cursor(machine, subagent)["offset"], 0) self.assertFalse(os.path.exists(self.cursor(machine, session)))
if __name__ == "__main__": unittest.main()