mirror of
https://github.com/0xWheatyz/handler.git
synced 2026-08-30 21:56:25 +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
55 lines
2.1 KiB
Python
55 lines
2.1 KiB
Python
"""DAL read/write functions and the answer backfill."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from handler.db import repository as repo
|
|
|
|
|
|
def test_project_and_agent_crud(conn):
|
|
repo.create_project(conn, "proj", "/tmp/proj", git_remote="git@x:proj.git")
|
|
assert repo.get_project(conn, "proj")["root_dir"] == "/tmp/proj"
|
|
assert [p["id"] for p in repo.list_projects(conn)] == ["proj"]
|
|
|
|
a = repo.create_agent(conn, "proj", "api", "/tmp/proj/api")
|
|
assert a["status"] == "working"
|
|
assert repo.get_agent_by_name(conn, "proj", "api")["id"] == a["id"]
|
|
assert repo.get_agent_by_name(conn, "proj", "missing") is None
|
|
|
|
|
|
def test_log_insert_and_answer_backfill(conn):
|
|
repo.create_project(conn, "p", "/tmp/p")
|
|
a = repo.create_agent(conn, "p", "a", "/tmp/p/a")
|
|
|
|
log_id = repo.insert_log_entry(
|
|
conn, a["id"], status="paused_for_input", question="Which DB?"
|
|
)
|
|
open_q = repo.get_latest_open_question(conn, a["id"])
|
|
assert open_q["id"] == log_id
|
|
|
|
assert repo.update_log_answer(conn, log_id, "Postgres") is True
|
|
# Once answered, it is no longer an open question.
|
|
assert repo.get_latest_open_question(conn, a["id"]) is None
|
|
assert repo.get_log(conn, a["id"])[0]["answer"] == "Postgres"
|
|
|
|
|
|
def test_shared_context_upsert(conn):
|
|
repo.create_project(conn, "p", "/tmp/p")
|
|
a = repo.create_agent(conn, "p", "a", "/tmp/p/a")
|
|
|
|
repo.set_shared_context(conn, "staging_url", "https://a", a["id"])
|
|
assert repo.get_shared_context_key(conn, "staging_url")["value"] == "https://a"
|
|
repo.set_shared_context(conn, "staging_url", "https://b", a["id"])
|
|
assert repo.get_shared_context_key(conn, "staging_url")["value"] == "https://b"
|
|
assert len(repo.get_shared_context(conn)) == 1
|
|
|
|
|
|
def test_shared_log_only_global(conn):
|
|
repo.create_project(conn, "p", "/tmp/p")
|
|
a = repo.create_agent(conn, "p", "a", "/tmp/p/a")
|
|
repo.insert_log_entry(conn, a["id"], status="working", summary="private")
|
|
repo.insert_log_entry(
|
|
conn, a["id"], status="working", summary="shared", visibility="global"
|
|
)
|
|
shared = repo.get_shared_log(conn)
|
|
assert [e["summary"] for e in shared] == ["shared"]
|