mirror of
https://github.com/0xWheatyz/handler.git
synced 2026-08-30 12:26:24 +00:00
0xWheatyz
6fb26115ce
Phase 2 configures forge for the agents (operator only sets a credential_ref + optional version pin) and lets them drive a junior→senior→deploy workflow: - Credential resolution/injection (control/credentials.py): credential_ref pointers (env:/file:/cmd:) resolved only at spawn, injected as FORGE_TOKEN + host var, with a forge-host-scoped git credential helper reading the token from env (never on disk / in the DB). Resolution is a fail-fast spawn gate. - Role-based forge skills committed into the managed repo (control/skills_gen.py, `handler forge-init`): forge-junior/senior/deploy + a workflow overview. - Hard approval gate (hooks/gate.py, approvals table, migration 0002): merge/deploy — and direct pushes to protected branches — are denied unless a DIFFERENT agent has an `approved` record for the branch, pinned to the reviewed commit (approved_sha). Senior records verdicts via `handler approve`/`reject`. - forge/git seams (control/forge.py, control/gitops.py) matching the Phase 1 seam pattern. - CI status poller (control/poller.py, `handler poll-ci [--watch]`) backfilling ci_status/ci_checked_at via `forge ci list`. - Fix: migrations/env.py commits explicitly after run_migrations — pysqlite on Py 3.12+ was rolling back the final migration's DDL + alembic_version stamp (latent in Phase 1). Reviewed via a separate code-reviewer pass; gate-bypass and credential-scoping findings addressed. 106 tests, ruff clean, verified end-to-end against real git + migrations.
68 lines
2.9 KiB
Python
68 lines
2.9 KiB
Python
"""Phase 2 DAL: agent role, approvals, and CI backfill helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from handler.db import repository as repo
|
|
|
|
|
|
def test_agent_role_is_stored(conn):
|
|
repo.create_project(conn, "p", "/tmp/p")
|
|
a = repo.create_agent(conn, "p", "senior", "/tmp/p/senior", role="senior")
|
|
assert a["role"] == "senior"
|
|
assert repo.get_agent_by_id(conn, a["id"])["role"] == "senior"
|
|
|
|
|
|
def test_approval_record_and_latest(conn):
|
|
repo.create_project(conn, "p", "/tmp/p")
|
|
junior = repo.create_agent(conn, "p", "junior", "/tmp/p/j", role="junior")
|
|
senior = repo.create_agent(conn, "p", "senior", "/tmp/p/s", role="senior")
|
|
|
|
assert repo.get_latest_approval(conn, "p", "feat/x") is None
|
|
|
|
repo.record_approval(conn, "p", "feat/x", "rejected", junior["id"], note="nit")
|
|
latest = repo.record_approval(conn, "p", "feat/x", "approved", senior["id"], pr_ref="7")
|
|
got = repo.get_latest_approval(conn, "p", "feat/x")
|
|
# Latest wins (by insertion order).
|
|
assert got["id"] == latest["id"]
|
|
assert got["status"] == "approved"
|
|
assert got["approved_by_agent_id"] == senior["id"]
|
|
assert got["pr_ref"] == "7"
|
|
|
|
|
|
def test_approval_scoped_by_project_and_branch(conn):
|
|
repo.create_project(conn, "p", "/tmp/p")
|
|
a = repo.create_agent(conn, "p", "s", "/tmp/p/s")
|
|
repo.record_approval(conn, "p", "feat/x", "approved", a["id"])
|
|
assert repo.get_latest_approval(conn, "p", "feat/other") is None
|
|
|
|
|
|
def test_pending_ci_entries_and_backfill(conn):
|
|
repo.create_project(conn, "p", "/tmp/p")
|
|
a = repo.create_agent(conn, "p", "a", "/tmp/p/a")
|
|
# A push-recording entry (pending) and a normal entry (not_applicable).
|
|
pending_id = repo.insert_log_entry(
|
|
conn, a["id"], status="working", push_sha="deadbeef", ci_status="pending"
|
|
)
|
|
repo.insert_log_entry(conn, a["id"], status="working", summary="no push")
|
|
|
|
entries = repo.get_pending_ci_entries(conn)
|
|
assert [e["id"] for e in entries] == [pending_id]
|
|
assert entries[0]["push_sha"] == "deadbeef"
|
|
assert entries[0]["project_id"] == "p"
|
|
assert entries[0]["working_dir"] == "/tmp/p/a"
|
|
|
|
assert repo.update_ci_status(conn, pending_id, "pass") is True
|
|
# No longer pending once resolved.
|
|
assert repo.get_pending_ci_entries(conn) == []
|
|
assert repo.get_log(conn, a["id"])[-1]["ci_status"] in ("pass", "not_applicable")
|
|
|
|
|
|
def test_pending_ci_entries_scoped_to_project(conn):
|
|
repo.create_project(conn, "p1", "/tmp/p1")
|
|
repo.create_project(conn, "p2", "/tmp/p2")
|
|
a1 = repo.create_agent(conn, "p1", "a", "/tmp/p1/a")
|
|
a2 = repo.create_agent(conn, "p2", "a", "/tmp/p2/a")
|
|
repo.insert_log_entry(conn, a1["id"], status="working", push_sha="s1", ci_status="pending")
|
|
repo.insert_log_entry(conn, a2["id"], status="working", push_sha="s2", ci_status="pending")
|
|
assert [e["project_id"] for e in repo.get_pending_ci_entries(conn, project_id="p1")] == ["p1"]
|