mirror of
https://github.com/0xWheatyz/handler.git
synced 2026-08-30 12:26:24 +00:00
Claude
eba0e19ec9
Implements the Phase 1 MVP from the README: a stateless control layer + HTTP API over a centralized database, with hook-enforced test/push gates. - DB layer: SQLAlchemy Core, one schema rendering both Postgres (BIGSERIAL / TIMESTAMPTZ / JSONB) and SQLite (INTEGER PK / TEXT / JSON) via portable types; native ON CONFLICT DO UPDATE checkmark upsert on both dialects. - Alembic dual-dialect migrations (render_as_batch for SQLite); tests run a real `alembic upgrade head`. - FastAPI: projects/agents/checkmark/log/answer/resume + shared log/context routes, single global bearer token, higher-trust token gating shared-context writes, project isolation on every route. - Hooks (`python -m handler.hooks <event>`): Stop test gate (block on red), PreToolUse AskUserQuestion defer + `git push` gate (tests then throwaway build), Notification generic webhook (no-op without WEBHOOK_URL). Identity via env injected at spawn; verify is the mock seam. - Control CLI: spawn/list/attach/kill, hard `.mise.toml [tasks.test]` gate, generated per-agent settings.json, identity + DATABASE_URL injected via tmux; tmux is the mock seam. - 45 tests (SQLite), ruff clean. Live claude/tmux/mise spawning deferred behind the mocked seams. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01W5ZuS5pV1NS6eKsRZHXonY
66 lines
2.4 KiB
Python
66 lines
2.4 KiB
Python
"""Stop / SessionEnd checkpoint gate."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from handler.db import repository as repo
|
|
from handler.hooks import checkpoint, verify
|
|
from handler.hooks.context import HookInput, Identity
|
|
|
|
|
|
def _seed(conn):
|
|
repo.create_project(conn, "p", "/tmp/p")
|
|
a = repo.create_agent(conn, "p", "a", "/tmp/p/a")
|
|
return Identity(a["id"], "p", "a", "/tmp/p/a")
|
|
|
|
|
|
def test_stop_blocks_on_failing_tests(conn, monkeypatch):
|
|
ident = _seed(conn)
|
|
monkeypatch.setattr(verify, "run_test", lambda cwd: (False, "1 failed"))
|
|
|
|
result = checkpoint.handle_stop(conn, ident, HookInput({"session_id": "s1"}, "stop"))
|
|
assert result["decision"] == "block"
|
|
assert "test gate failed" in result["reason"]
|
|
|
|
cm = repo.get_checkmark(conn, ident.agent_id)
|
|
assert cm["tests_status"] == "fail"
|
|
assert cm["status"] == "blocked"
|
|
# A blocked turn never records "done".
|
|
assert repo.get_agent_by_name(conn, "p", "a")["status"] == "blocked"
|
|
|
|
|
|
def test_stop_allows_done_on_passing_tests(conn, monkeypatch):
|
|
ident = _seed(conn)
|
|
monkeypatch.setattr(verify, "run_test", lambda cwd: (True, "ok"))
|
|
|
|
result = checkpoint.handle_stop(conn, ident, HookInput({"session_id": "s1"}, "stop"))
|
|
assert result == {} # no block
|
|
|
|
cm = repo.get_checkmark(conn, ident.agent_id)
|
|
assert cm["tests_status"] == "pass"
|
|
assert cm["status"] == "done"
|
|
assert cm["log_entry_id"] is not None
|
|
|
|
|
|
def test_stop_does_not_reblock_when_already_active(conn, monkeypatch):
|
|
ident = _seed(conn)
|
|
monkeypatch.setattr(verify, "run_test", lambda cwd: (False, "still failing"))
|
|
hi = HookInput({"session_id": "s1", "stop_hook_active": True}, "stop")
|
|
result = checkpoint.handle_stop(conn, ident, hi)
|
|
assert result == {} # recorded, but not an infinite block
|
|
assert repo.get_checkmark(conn, ident.agent_id)["tests_status"] == "fail"
|
|
|
|
|
|
def test_session_end_records_without_gate(conn, monkeypatch):
|
|
ident = _seed(conn)
|
|
# Even if tests would fail, SessionEnd must not run the gate or block.
|
|
monkeypatch.setattr(
|
|
verify, "run_test", lambda cwd: (_ for _ in ()).throw(AssertionError("gate ran"))
|
|
)
|
|
result = checkpoint.handle_session_end(
|
|
conn, ident, HookInput({"reason": "clear"}, "session_end")
|
|
)
|
|
assert result == {}
|
|
assert repo.get_checkmark(conn, ident.agent_id)["where_it_stopped"].startswith(
|
|
"session ended"
|
|
)
|