feat(web): fully web-managed control plane via a DB command queue

Make credentials/hosts, projects, agents, and approvals manageable from the
dashboard. The API and control layer are separate containers, so the API can't
run control actions directly (no git/tmux/claude, doesn't own the tmux
sessions). Instead the API enqueues a command and a worker in the control
container executes it and writes the result back.

Data model (migration 0003):
- `commands` queue/audit table; `forge_hosts` registry; `approvals` gains a
  nullable approver id + `actor` so operator verdicts are first-class.

Control worker:
- `control/worker.py` claims commands and dispatches to the existing control
  functions (spawn/kill/resume/record_approval/write_skills/poller.sweep),
  plus a periodic CI sweep. New `handler worker` CLI subcommand; it becomes the
  control image's default command (subsumes `poll-ci --watch`).

API:
- `require_admin` gate + `ADMIN_TOKEN`; project GET/PATCH/DELETE; agent
  spawn/kill/delete; resume now enqueues (fixes a cross-container bug where the
  API tried to send tmux keys to a session in the control container); new
  approvals/commands/hosts routes; forge-init and poll-ci enqueue endpoints.

Credentials/hosts:
- host->token-env lookup consults the `forge_hosts` registry first (built-in
  map is the fallback); `resolve()` refactored to a scheme dispatch reserving
  `db:` for a future encrypted store. Web input restricts credential_ref to
  env:/file:/db: (cmd: stays CLI-only — it would run arbitrary commands).

Dashboard:
- New tabs for projects, agents (spawn/kill with live command-status polling),
  approvals, hosts, and an activity/audit view; shared context is now writable.

Tests: +33 (queue atomicity, worker dispatch, CRUD, hosts, admin gating,
cmd: rejection, host-aware credentials, and an API->queue->worker->spawn
end-to-end). README gains a Web management section.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CrhrBToauu4L2qG6jdnuFP
This commit is contained in:
Claude
2026-07-10 16:32:45 +00:00
parent 7b5a5e3c27
commit 4f05d09c2b
31 changed files with 2277 additions and 146 deletions
+6
View File
@@ -12,6 +12,12 @@ AUTH_TOKEN=change-me-to-a-long-random-string
# Falls back to AUTH_TOKEN if unset.
# SHARED_CONTEXT_WRITE_TOKEN=
# Optional admin token gating the web control surface: enqueuing control commands
# (spawn/kill/resume/approve/reject/forge-init/poll-ci), project CRUD, forge-host CRUD,
# and credential-pointer edits. Falls back to AUTH_TOKEN if unset. Give operators this
# token in the dashboard to unlock management actions.
# ADMIN_TOKEN=
# Optional generic webhook target for the Notification hook (ntfy, Pushover, Slack, ...).
# Fully bring-your-own; the Notification hook is a no-op when unset.
# WEBHOOK_URL=https://ntfy.sh/my-topic
+4 -3
View File
@@ -59,7 +59,8 @@ VOLUME /var/lib/handler
HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
CMD handler list >/dev/null 2>&1 || exit 1
# Default to the CI poller — the one long-running control process. Override the command
# for one-shot control operations, e.g. `docker compose run --rm control handler list`.
# Default to the worker: it drains the control-command queue the API enqueues
# (spawn/kill/resume/approve/…) and sweeps CI on an interval (subsuming `poll-ci --watch`).
# Override for one-shot control operations, e.g. `docker compose run --rm control handler list`.
ENTRYPOINT ["docker-entrypoint.sh"]
CMD ["handler", "poll-ci", "--watch"]
CMD ["handler", "worker"]
+49 -4
View File
@@ -101,6 +101,7 @@ Configuration is entirely environment-driven (see [`.env.example`](.env.example)
| `DATABASE_URL` | `sqlite:////abs/path.db` or `postgresql+psycopg://…` | `sqlite:///./handler.db` |
| `AUTH_TOKEN` | Global bearer token gating every API route | *(required for the API)* |
| `SHARED_CONTEXT_WRITE_TOKEN` | Higher-trust token gating `PUT /shared/context/:key` | falls back to `AUTH_TOKEN` |
| `ADMIN_TOKEN` | Gates the web control surface (enqueue commands, project/host CRUD, credential edits) | falls back to `AUTH_TOKEN` |
| `WEBHOOK_URL` | Generic target for the `Notification` hook (ntfy, Slack, …) | unset → no-op |
| `PROJECTS_ROOT` | Base dir for per-project roots / worktrees | `./projects` |
| `CLAUDE_BIN` / `MISE_BIN` / `TMUX_BIN` / `FORGE_BIN` / `GIT_BIN` | Binary overrides | `claude` / `mise` / `tmux` / `forge` / `git` |
@@ -151,27 +152,71 @@ the `/var/lib/handler` data volume:
| Image | Dockerfile | Runs | Workflow |
|---|---|---|---|
| `ghcr.io/0xwheatyz/handler` | [`Dockerfile`](Dockerfile) | the API (`uvicorn`) — also applies migrations on start | [`docker.yml`](.github/workflows/docker.yml) |
| `ghcr.io/0xwheatyz/handler/control` | [`Dockerfile.control`](Dockerfile.control) | the control layer (`handler poll-ci --watch`) | [`docker-control.yml`](.github/workflows/docker-control.yml) |
| `ghcr.io/0xwheatyz/handler/control` | [`Dockerfile.control`](Dockerfile.control) | the control worker (`handler worker`) | [`docker-control.yml`](.github/workflows/docker-control.yml) |
The control image bakes in `git` + `tmux`; the `claude` and `forge` binaries are
bring-your-own (layer or mount them in for live agent spawning — the CI poller degrades
gracefully without `forge`).
gracefully without `forge`). The **worker** drains the control-command queue the API
enqueues (spawn/kill/resume/approve/reject/forge-init/poll-ci) and sweeps CI on an interval
(subsuming `poll-ci --watch`), so the whole system is drivable from the dashboard — see
[Web management](#web-management).
[`docker-compose.yml`](docker-compose.yml) wires both up with Postgres. The API owns
migrations, so the control service runs with `RUN_MIGRATIONS=false` and waits for the API:
```bash
export AUTH_TOKEN="$(openssl rand -hex 32)"
docker compose up -d # db + api + control (CI poller)
export ADMIN_TOKEN="$(openssl rand -hex 32)" # unlocks management actions in the dashboard
docker compose up -d # db + api + control (worker)
# One-shot control commands run against the same image:
docker compose run --rm control handler list
docker compose run --rm control handler spawn --project leeworks-api --name junior --task "…"
```
## Web management
The dashboard (and the API under it) manages everything — git credentials & hosts,
projects, agents, and approvals — without dropping to the CLI. Because the API and control
layer are **separate containers** (the API has no `git`/`tmux`/`claude` and doesn't own the
tmux sessions), the API can't run control actions directly. Instead it **enqueues a command**
and the worker in the control container executes it and writes the result back:
```
Dashboard ──HTTP──▶ API (read + enqueue) Control container
│ writes a `commands` row │ worker: claim → dispatch → result
▼ ▼
┌─────────────── shared database ───────────────┐
│ projects agents approvals commands hosts │
└────────────────────────────────────────────────┘
```
What the dashboard can now do (all state-changing actions require `ADMIN_TOKEN`):
- **Projects** — create / edit / delete (`root_dir`, `git_remote`, `credential_ref`).
- **Agents** — spawn (name, role, worktree/subdir, task) and kill via the queue; delete the
row; plus the existing checkmark / log / answer-resume views.
- **Approvals** — record an operator verdict per branch (approve/reject); the deploy gate
treats an operator verdict as a genuine second party (no self-approval).
- **Forge hosts** — a registry mapping a host to the token env var to inject at spawn, so
self-hosted forges work without a code change (the built-in host map is the fallback).
- **Credentials** — manage a project's `credential_ref` **pointer**. The DB still never
stores a raw token: web-settable schemes are `env:` / `file:` / `db:` (the `cmd:` scheme
is CLI-only, since it would run an arbitrary command in the control container). `db:` is
reserved for a future encrypted secret store.
- **Activity** — every enqueued command with its status (queued → running → done/failed) —
the audit log of what the dashboard triggered. The UI polls `GET /commands/{id}` for
live status.
The command queue is exposed over HTTP as `POST …/agents/spawn`, `POST …/agents/{n}/kill`,
`POST …/approvals`, `POST …/forge-init`, `POST …/poll-ci`, and `GET /commands[/{id}]`;
hosts as `/hosts`; project mutation as `PATCH`/`DELETE /projects/{id}`. Run the worker with
`handler worker` (the control image's default command).
## Control CLI
The `handler` command manages agent processes (the write side):
The `handler` command manages agent processes directly (an alternative to the queue, for
operators at a shell):
```bash
handler spawn --project leeworks-api --name junior --role junior --worktree feat/auth --task "add login"
+8 -6
View File
@@ -22,12 +22,14 @@ services:
condition: service_healthy
restart: unless-stopped
# Control layer: the `handler` CLI running the CI poller loop. Shares the database and
# the handler-data volume with the API. It waits for the API (which owns migrations),
# so RUN_MIGRATIONS is off here to avoid a startup race. Run one-shot control commands
# against the same image with, e.g., `docker compose run --rm control handler list`.
# Live agent spawning also needs `git`/`tmux` (baked in) plus bring-your-own
# `claude`/`forge` binaries — layer or mount those in.
# Control layer: the `handler` worker. Drains the control-command queue the API enqueues
# (spawn/kill/resume/approve/reject/forge-init/poll-ci) and sweeps CI on an interval.
# Shares the database and the handler-data volume with the API. It waits for the API
# (which owns migrations), so RUN_MIGRATIONS is off here to avoid a startup race. Run
# one-shot control commands against the same image with, e.g.,
# `docker compose run --rm control handler list`. Live agent spawning also needs
# `git`/`tmux` (baked in) plus bring-your-own `claude`/`forge` binaries — layer or mount
# those in.
control:
image: ghcr.io/0xwheatyz/handler/control:latest
build:
+5 -2
View File
@@ -15,7 +15,7 @@ from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from ..config import get_settings
from .routes import agents, interaction, projects, shared
from .routes import agents, approvals, commands, hosts, interaction, projects, shared
_STATIC_DIR = Path(__file__).parent / "static"
@@ -36,6 +36,9 @@ def create_app() -> FastAPI:
app.include_router(projects.router)
app.include_router(agents.router)
app.include_router(interaction.router)
app.include_router(approvals.router)
app.include_router(commands.router)
app.include_router(hosts.router)
app.include_router(shared.router)
# Optional CORS, only for operators who host the UI on a different origin than the
@@ -46,7 +49,7 @@ def create_app() -> FastAPI:
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origin_list,
allow_methods=["GET", "POST", "PUT"],
allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
allow_headers=["Authorization", "Content-Type"],
)
+23 -5
View File
@@ -36,11 +36,13 @@ def require_auth(
settings: Settings = Depends(get_settings),
) -> None:
token = creds.credentials if creds else None
# The shared-context write token is higher-trust, so it also grants normal access;
# a single request carries one bearer, and it should never be rejected for being the
# more privileged one.
valid = _check(token, settings.auth_token) or _check(
token, settings.effective_shared_write_token
# The shared-context write and admin tokens are higher-trust, so they also grant
# normal access; a single request carries one bearer, and it should never be rejected
# for being the more privileged one.
valid = (
_check(token, settings.auth_token)
or _check(token, settings.effective_shared_write_token)
or _check(token, settings.effective_admin_token)
)
if not valid:
raise HTTPException(
@@ -62,3 +64,19 @@ def require_shared_write(
detail="shared-context write requires the shared-context write token",
headers={"WWW-Authenticate": "Bearer"},
)
def require_admin(
creds: HTTPAuthorizationCredentials | None = Depends(_bearer),
settings: Settings = Depends(get_settings),
) -> None:
"""Gate for the web control surface: enqueuing control commands, project/host CRUD,
and credential-pointer edits. Requires specifically the admin token (which defaults to
the global token when ADMIN_TOKEN is unset)."""
token = creds.credentials if creds else None
if not _check(token, settings.effective_admin_token):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="this action requires the admin token",
headers={"WWW-Authenticate": "Bearer"},
)
+54 -6
View File
@@ -1,8 +1,9 @@
"""Agent listing/registration and the read views (checkmark, log).
"""Agent listing/registration, read views (checkmark, log), and lifecycle actions.
The agent *row* is registered here (the API mirror listed in README 3.3); the agent
*process* is spawned by the control CLI. All routes are nested under
``/projects/{project}`` so nothing crosses a project boundary.
The agent *row* is registered here; the agent *process* (tmux + claude) is created by the
control worker, so ``spawn``/``kill`` enqueue a command (admin-gated) rather than acting
in-process — the API container has no tmux/git/claude and does not own the sessions. All
routes are nested under ``/projects/{project}`` so nothing crosses a project boundary.
"""
from __future__ import annotations
@@ -12,8 +13,8 @@ from sqlalchemy import Connection
from sqlalchemy.exc import IntegrityError
from ...db import repository as repo
from ..deps import db_conn, require_auth
from ..schemas import AgentIn, AgentOut, CheckmarkOut, LogEntryOut
from ..deps import db_conn, require_admin, require_auth
from ..schemas import AgentIn, AgentOut, CheckmarkOut, CommandOut, LogEntryOut, SpawnIn
from .common import resolve_agent
router = APIRouter(
@@ -49,11 +50,58 @@ def create_agent(project: str, body: AgentIn, conn: Connection = Depends(db_conn
name=body.name,
working_dir=body.working_dir,
status=body.status,
role=body.role,
)
except IntegrityError as exc: # pragma: no cover - guarded above
raise HTTPException(status.HTTP_409_CONFLICT, detail="agent exists") from exc
@router.post(
"/spawn",
response_model=CommandOut,
status_code=status.HTTP_202_ACCEPTED,
dependencies=[Depends(require_admin)],
)
def enqueue_spawn(project: str, body: SpawnIn, conn: Connection = Depends(db_conn)) -> dict:
"""Enqueue a spawn; the worker creates the agent row + tmux session and reports back."""
_require_project(conn, project)
if repo.get_agent_by_name(conn, project, body.name) is not None:
raise HTTPException(
status.HTTP_409_CONFLICT,
detail=f"agent '{body.name}' already exists in project '{project}'",
)
payload = body.model_dump(exclude={"name"}, exclude_none=True)
return repo.enqueue_command(
conn,
"spawn",
project_id=project,
agent_name=body.name,
payload=payload,
requested_by="operator:web",
)
@router.post(
"/{name}/kill",
response_model=CommandOut,
status_code=status.HTTP_202_ACCEPTED,
dependencies=[Depends(require_admin)],
)
def enqueue_kill(project: str, name: str, conn: Connection = Depends(db_conn)) -> dict:
resolve_agent(conn, project, name)
return repo.enqueue_command(
conn, "kill", project_id=project, agent_name=name, requested_by="operator:web"
)
@router.delete("/{name}", dependencies=[Depends(require_admin)])
def delete_agent(project: str, name: str, conn: Connection = Depends(db_conn)) -> dict:
"""Remove the agent row (does not kill a live session — kill first)."""
resolve_agent(conn, project, name)
repo.delete_agent(conn, project, name)
return {"deleted": name}
@router.get("/{name}/checkmark", response_model=CheckmarkOut)
def get_checkmark(project: str, name: str, conn: Connection = Depends(db_conn)) -> dict:
agent = resolve_agent(conn, project, name)
+65
View File
@@ -0,0 +1,65 @@
"""Branch approvals — read the standing verdicts, enqueue new ones.
Recording a verdict resolves the reviewed HEAD sha (which requires the working tree in the
control container), so ``POST`` enqueues an ``approve``/``reject`` command for the worker.
Operator verdicts set ``actor='operator:web'`` and no acting agent, which the deploy gate
treats as a genuine second party (satisfying the "no self-approval" rule).
"""
from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy import Connection
from ...db import repository as repo
from ..deps import db_conn, require_admin, require_auth
from ..schemas import ApprovalIn, ApprovalOut, CommandOut
router = APIRouter(
prefix="/projects/{project}/approvals",
tags=["approvals"],
dependencies=[Depends(require_auth)],
)
def _require_project(conn: Connection, project: str) -> None:
if repo.get_project(conn, project) is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, detail=f"project '{project}' not found")
@router.get("", response_model=list[ApprovalOut])
def list_approvals(
project: str,
branch: str | None = Query(None),
conn: Connection = Depends(db_conn),
) -> list[dict]:
_require_project(conn, project)
return repo.list_approvals(conn, project, branch=branch)
@router.post(
"",
response_model=CommandOut,
status_code=status.HTTP_202_ACCEPTED,
dependencies=[Depends(require_admin)],
)
def enqueue_approval(
project: str, body: ApprovalIn, conn: Connection = Depends(db_conn)
) -> dict:
_require_project(conn, project)
payload = {
"branch": body.branch,
"sha": body.sha,
"pr": body.pr,
"note": body.note,
}
# Verdict ('approved'/'rejected') -> command type ('approve'/'reject').
command_type = "approve" if body.status == "approved" else "reject"
return repo.enqueue_command(
conn,
command_type,
project_id=project,
agent_name=body.agent_name,
payload={k: v for k, v in payload.items() if v is not None},
requested_by="operator:web",
)
+47
View File
@@ -0,0 +1,47 @@
"""The command queue's read surface + the global poll-ci enqueue.
Every control action the dashboard triggers becomes a ``commands`` row; these routes let
the UI poll a command's status (queued -> running -> done/failed) and show an activity log.
Enqueuing project-scoped actions lives with those resources (agents/projects/approvals);
the one non-scoped action, a global CI sweep, is enqueued here.
"""
from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy import Connection
from ...db import repository as repo
from ..deps import db_conn, require_admin, require_auth
from ..schemas import CommandOut
router = APIRouter(tags=["commands"], dependencies=[Depends(require_auth)])
@router.get("/commands", response_model=list[CommandOut])
def list_commands(
project: str | None = Query(None),
limit: int = Query(100, ge=1, le=500),
offset: int = Query(0, ge=0),
conn: Connection = Depends(db_conn),
) -> list[dict]:
return repo.list_commands(conn, project_id=project, limit=limit, offset=offset)
@router.get("/commands/{command_id}", response_model=CommandOut)
def get_command(command_id: int, conn: Connection = Depends(db_conn)) -> dict:
command = repo.get_command(conn, command_id)
if command is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, detail=f"command {command_id} not found")
return command
@router.post(
"/poll-ci",
response_model=CommandOut,
status_code=status.HTTP_202_ACCEPTED,
dependencies=[Depends(require_admin)],
)
def enqueue_global_poll_ci(conn: Connection = Depends(db_conn)) -> dict:
"""Enqueue a CI sweep across every project (per-project sweep is on the project route)."""
return repo.enqueue_command(conn, "poll_ci", requested_by="operator:web")
+70
View File
@@ -0,0 +1,70 @@
"""Forge-host registry — the web-managed replacement for the hardcoded host->token-env map.
Registering a host lets ``control.credentials`` inject the right per-host token env var
(and scope the git credential helper) for self-hosted forges without a code change. The
registry only holds the env-var *name* and metadata — never a secret (secrets stay behind
``credential_ref`` pointers). Reads take the normal token; writes take the admin token.
"""
from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import Connection
from sqlalchemy.exc import IntegrityError
from ...db import repository as repo
from ..deps import db_conn, require_admin, require_auth
from ..schemas import HostIn, HostOut, HostUpdateIn
router = APIRouter(prefix="/hosts", tags=["hosts"], dependencies=[Depends(require_auth)])
def _get_or_404(conn: Connection, hostname: str) -> dict:
host = repo.get_host(conn, hostname)
if host is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, detail=f"host '{hostname}' not found")
return host
@router.get("", response_model=list[HostOut])
def list_hosts(conn: Connection = Depends(db_conn)) -> list[dict]:
return repo.list_hosts(conn)
@router.get("/{hostname}", response_model=HostOut)
def get_host(hostname: str, conn: Connection = Depends(db_conn)) -> dict:
return _get_or_404(conn, hostname)
@router.post(
"", response_model=HostOut, status_code=status.HTTP_201_CREATED,
dependencies=[Depends(require_admin)],
)
def create_host(body: HostIn, conn: Connection = Depends(db_conn)) -> dict:
if repo.get_host(conn, body.hostname) is not None:
raise HTTPException(status.HTTP_409_CONFLICT, detail=f"host '{body.hostname}' exists")
try:
return repo.create_host(
conn,
hostname=body.hostname,
forge_type=body.forge_type,
token_env_var=body.token_env_var,
base_url=body.base_url,
)
except IntegrityError as exc: # pragma: no cover - guarded above
raise HTTPException(status.HTTP_409_CONFLICT, detail="host exists") from exc
@router.patch("/{hostname}", response_model=HostOut, dependencies=[Depends(require_admin)])
def update_host(
hostname: str, body: HostUpdateIn, conn: Connection = Depends(db_conn)
) -> dict:
_get_or_404(conn, hostname)
return repo.update_host(conn, hostname, **body.model_dump(exclude_unset=True))
@router.delete("/{hostname}", dependencies=[Depends(require_admin)])
def delete_host(hostname: str, conn: Connection = Depends(db_conn)) -> dict:
_get_or_404(conn, hostname)
repo.delete_host(conn, hostname)
return {"deleted": hostname}
+24 -14
View File
@@ -1,10 +1,11 @@
"""Answer + resume — the async replacement for a human sitting at the tmux TTY.
``answer`` writes the operator's reply into the log entry that recorded the question
(the sole API mutation of ``log_entries``). ``resume`` then feeds that answer back to
the agent via ``claude --resume``, routed through the control-layer seam so it stays
mockable and the API/control boundary is explicit. They are two endpoints (README 3.3)
so the operator can answer many questions, then resume.
``answer`` writes the operator's reply into the log entry that recorded the question (the
sole API mutation of ``log_entries``). ``resume`` then enqueues a ``resume`` command: the
control worker — which runs in the control container where the agent's tmux session
actually lives — feeds the answer back via ``claude --resume``. (Doing this in-process in
the API container would fail after the API/control split, since the session isn't here.)
They are two endpoints (README 3.3) so the operator can answer many questions, then resume.
"""
from __future__ import annotations
@@ -12,10 +13,9 @@ from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import Connection
from ...control import spawn
from ...db import repository as repo
from ..deps import db_conn, require_auth
from ..schemas import AnswerIn, AnswerOut, ResumeIn, ResumeOut
from ..deps import db_conn, require_admin, require_auth
from ..schemas import AnswerIn, AnswerOut, CommandOut, ResumeIn
from .common import resolve_agent
router = APIRouter(
@@ -54,15 +54,21 @@ def answer(
return AnswerOut(log_entry_id=log_entry_id, answered=True)
@router.post("/resume", response_model=ResumeOut)
@router.post(
"/resume",
response_model=CommandOut,
status_code=status.HTTP_202_ACCEPTED,
dependencies=[Depends(require_admin)],
)
def resume(
project: str,
name: str,
body: ResumeIn,
conn: Connection = Depends(db_conn),
) -> ResumeOut:
) -> dict:
agent = resolve_agent(conn, project, name)
# Resolve the answer to feed back here (the API has the log); the worker just delivers.
answer_text = body.answer
if answer_text is None:
open_q = repo.get_latest_open_question(conn, agent["id"])
@@ -80,7 +86,11 @@ def resume(
detail="no answer available to resume with; answer first or pass one",
)
ok, detail = spawn.resume(agent, answer_text)
if ok:
repo.set_agent_status(conn, agent["id"], "working")
return ResumeOut(agent=name, resumed=ok, detail=detail)
return repo.enqueue_command(
conn,
"resume",
project_id=project,
agent_name=name,
payload={"answer": answer_text},
requested_by="operator:web",
)
+68 -3
View File
@@ -1,4 +1,9 @@
"""Project registration + listing (control-plane; the process spawn is the CLI's job)."""
"""Project CRUD + project-scoped control actions.
Reads and row registration take the normal token; edits/deletes and the enqueue actions
(forge-init, poll-ci) take the admin token. The agent *process* work (spawn/kill) lives in
``agents.py``; here we cover the project itself and the two project-wide control actions.
"""
from __future__ import annotations
@@ -7,17 +12,29 @@ from sqlalchemy import Connection
from sqlalchemy.exc import IntegrityError
from ...db import repository as repo
from ..deps import db_conn, require_auth
from ..schemas import ProjectIn, ProjectOut
from ..deps import db_conn, require_admin, require_auth
from ..schemas import CommandOut, ProjectIn, ProjectOut, ProjectUpdateIn
router = APIRouter(prefix="/projects", tags=["projects"], dependencies=[Depends(require_auth)])
def _get_or_404(conn: Connection, project_id: str) -> dict:
project = repo.get_project(conn, project_id)
if project is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, detail=f"project '{project_id}' not found")
return project
@router.get("", response_model=list[ProjectOut])
def list_projects(conn: Connection = Depends(db_conn)) -> list[dict]:
return repo.list_projects(conn)
@router.get("/{project_id}", response_model=ProjectOut)
def get_project(project_id: str, conn: Connection = Depends(db_conn)) -> dict:
return _get_or_404(conn, project_id)
@router.post("", response_model=ProjectOut, status_code=status.HTTP_201_CREATED)
def create_project(body: ProjectIn, conn: Connection = Depends(db_conn)) -> dict:
if repo.get_project(conn, body.id) is not None:
@@ -32,3 +49,51 @@ def create_project(body: ProjectIn, conn: Connection = Depends(db_conn)) -> dict
)
except IntegrityError as exc: # pragma: no cover - guarded above
raise HTTPException(status.HTTP_409_CONFLICT, detail="project exists") from exc
@router.patch("/{project_id}", response_model=ProjectOut, dependencies=[Depends(require_admin)])
def update_project(
project_id: str, body: ProjectUpdateIn, conn: Connection = Depends(db_conn)
) -> dict:
_get_or_404(conn, project_id)
fields = body.model_dump(exclude_unset=True)
return repo.update_project(conn, project_id, **fields)
@router.delete("/{project_id}", dependencies=[Depends(require_admin)])
def delete_project(project_id: str, conn: Connection = Depends(db_conn)) -> dict:
_get_or_404(conn, project_id)
repo.delete_project(conn, project_id)
return {"deleted": project_id}
@router.post(
"/{project_id}/forge-init",
response_model=CommandOut,
status_code=status.HTTP_202_ACCEPTED,
dependencies=[Depends(require_admin)],
)
def enqueue_forge_init(
project_id: str, no_commit: bool = False, conn: Connection = Depends(db_conn)
) -> dict:
_get_or_404(conn, project_id)
return repo.enqueue_command(
conn,
"forge_init",
project_id=project_id,
payload={"no_commit": no_commit},
requested_by="operator:web",
)
@router.post(
"/{project_id}/poll-ci",
response_model=CommandOut,
status_code=status.HTTP_202_ACCEPTED,
dependencies=[Depends(require_admin)],
)
def enqueue_poll_ci(project_id: str, conn: Connection = Depends(db_conn)) -> dict:
_get_or_404(conn, project_id)
return repo.enqueue_command(
conn, "poll_ci", project_id=project_id, requested_by="operator:web"
)
+122 -1
View File
@@ -5,8 +5,33 @@ mapping straight in; timestamps serialize as ISO-8601.
from __future__ import annotations
from datetime import datetime
from typing import Literal
from pydantic import BaseModel, ConfigDict
from pydantic import BaseModel, ConfigDict, field_validator
# Roles + forge families mirrored from db.tables; Literal gives clean 422s on bad input.
Role = Literal["junior", "senior", "deploy"]
ForgeType = Literal["github", "gitlab", "gitea", "forgejo", "bitbucket"]
# credential_ref schemes an operator may set over the web. ``cmd:`` is intentionally
# excluded — it would run an arbitrary command in the control container at spawn — so the
# API rejects it even though the CLI/DB path still allows it.
_WEB_CREDENTIAL_SCHEMES = {"env", "file", "db"}
def _validate_web_credential_ref(value: str | None) -> str | None:
if value is None:
return None
value = value.strip()
if not value:
return None
scheme = value.split(":", 1)[0]
if scheme not in _WEB_CREDENTIAL_SCHEMES:
raise ValueError(
f"credential_ref scheme '{scheme}' is not allowed from the API; "
"use env:, file:, or db: (cmd: is CLI-only for safety)"
)
return value
class ProjectIn(BaseModel):
@@ -15,6 +40,24 @@ class ProjectIn(BaseModel):
git_remote: str | None = None
credential_ref: str | None = None
@field_validator("credential_ref")
@classmethod
def _check_credential_ref(cls, v: str | None) -> str | None:
return _validate_web_credential_ref(v)
class ProjectUpdateIn(BaseModel):
"""Editable project columns; omit a field to leave it unchanged."""
root_dir: str | None = None
git_remote: str | None = None
credential_ref: str | None = None
@field_validator("credential_ref")
@classmethod
def _check_credential_ref(cls, v: str | None) -> str | None:
return _validate_web_credential_ref(v)
class ProjectOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
@@ -30,6 +73,7 @@ class AgentIn(BaseModel):
name: str
working_dir: str
status: str = "working"
role: Role | None = None
class AgentOut(BaseModel):
@@ -40,6 +84,83 @@ class AgentOut(BaseModel):
name: str
working_dir: str
status: str
role: Role | None = None
created_at: datetime
class SpawnIn(BaseModel):
"""Enqueue a spawn: the worker creates the agent row + tmux process in the control
container. ``worktree`` and ``subdir`` are mutually exclusive (worktree wins if both)."""
name: str
role: Role | None = None
worktree: str | None = None
subdir: str | None = None
task: str | None = None
class CommandOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
project_id: str | None = None
agent_name: str | None = None
type: str
payload: dict | None = None
status: str
result: dict | None = None
error: str | None = None
requested_by: str | None = None
claimed_by: str | None = None
created_at: datetime
claimed_at: datetime | None = None
finished_at: datetime | None = None
class HostIn(BaseModel):
hostname: str
forge_type: ForgeType
token_env_var: str | None = None
base_url: str | None = None
class HostUpdateIn(BaseModel):
forge_type: ForgeType | None = None
token_env_var: str | None = None
base_url: str | None = None
class HostOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
hostname: str
forge_type: str
token_env_var: str | None = None
base_url: str | None = None
created_at: datetime
class ApprovalIn(BaseModel):
branch: str
status: Literal["approved", "rejected"] = "approved"
agent_name: str | None = None # read HEAD from this agent's working dir when no sha
sha: str | None = None
pr: str | None = None
note: str | None = None
class ApprovalOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
project_id: str
branch: str
approved_sha: str | None = None
pr_ref: str | None = None
status: str
approved_by_agent_id: int | None = None
actor: str | None = None
note: str | None = None
created_at: datetime
+273 -21
View File
@@ -6,6 +6,10 @@
*
* Security: every value from the API is rendered with Alpine `x-text` (textContent)
* in index.html — never x-html — so agent-authored strings can't inject markup.
*
* Control actions (spawn/kill/resume/approve/…) are async: the API enqueues a command
* and the control worker executes it. enqueueAndTrack() posts the command, then polls
* GET /commands/{id} until it reaches done/failed, surfacing the result in a banner.
*/
const TOKEN_KEY = "handler_token";
@@ -35,16 +39,25 @@ function app() {
logLimit: LOG_LIMIT,
logOffset: 0,
shared: { log: [], context: [] },
approvals: [],
hosts: [],
commands: [],
// --- answer form ---
// --- forms ---
answerText: "",
answerBusy: false,
answerMsg: "",
answerError: false,
spawnForm: { name: "", role: "", placement: "worktree", worktree: "", subdir: "", task: "" },
approvalForm: { branch: "", status: "approved", agent_name: "", sha: "", note: "" },
projectForm: { id: "", root_dir: "", git_remote: "", credential_ref: "", _editing: false },
hostForm: { hostname: "", forge_type: "github", token_env_var: "", base_url: "", _editing: false },
sharedForm: { key: "", value: "" },
// --- ui ---
tab: "agents",
lastError: "",
cmd: { text: "", error: false, busy: false },
_poll: null,
get selectedAgent() {
@@ -120,7 +133,10 @@ function app() {
if (!res.ok) {
let detail = `${res.status}`;
try {
detail = (await res.json()).detail || detail;
const body = await res.json();
detail = body.detail || detail;
// Pydantic 422 returns a list of validation errors.
if (Array.isArray(detail)) detail = detail.map((d) => d.msg || JSON.stringify(d)).join("; ");
} catch (_) {}
const err = new Error(detail);
err.status = res.status;
@@ -130,6 +146,50 @@ function app() {
return res.json();
},
_sleep(ms) {
return new Promise((r) => setTimeout(r, ms));
},
/* Post a control action, then poll its command to a terminal state. */
async enqueueAndTrack(path, body, label) {
this.cmd = { text: `${label}: queued…`, error: false, busy: true };
try {
const command = await this.api(path, { method: "POST", body });
return await this._trackCommand(command.id, label);
} catch (e) {
if (e instanceof AuthError) return null;
this.cmd = { text: `${label} failed: ${e.message}`, error: true, busy: false };
return null;
}
},
async _trackCommand(id, label) {
for (let i = 0; i < 40; i++) {
let c;
try {
c = await this.api(`/commands/${id}`);
} catch (e) {
if (e instanceof AuthError) return null;
this.cmd = { text: `${label}: ${e.message}`, error: true, busy: false };
return null;
}
if (c.status === "done" || c.status === "failed") {
const ok = c.status === "done";
const detail = c.error || (c.result ? JSON.stringify(c.result) : "");
this.cmd = {
text: `${label} ${ok ? "done" : "failed"}${detail ? " — " + detail : ""}`,
error: !ok,
busy: false,
};
return c;
}
this.cmd = { text: `${label}: ${c.status}`, error: false, busy: true };
await this._sleep(600);
}
this.cmd = { text: `${label}: still running (see Activity). Is the worker up?`, error: false, busy: false };
return null;
},
// --- lifecycle ---
async start() {
await this.loadProjects();
@@ -154,10 +214,15 @@ function app() {
/* One poll cycle for whatever view is active. Swallows AuthError (already handled). */
async tick() {
try {
if (this.tab === "shared") {
await this.loadShared();
if (this.tab === "shared") return await this.loadShared();
if (this.tab === "activity") return await this.loadCommands();
if (this.tab === "hosts") return await this.loadHosts();
if (this.tab === "projects") return await this.loadProjects();
if (this.tab === "approvals") {
if (this.selectedProjectId) await this.loadApprovals();
return;
}
// agents tab
if (this.selectedProjectId) await this.loadAgents();
if (this.selectedAgentName) {
await this.loadCheckmark();
@@ -172,6 +237,12 @@ function app() {
this.tick();
},
switchTab(tab) {
this.tab = tab;
this.cmd = { text: "", error: false, busy: false };
this.tick();
},
// --- projects ---
async loadProjects() {
try {
@@ -190,14 +261,61 @@ function app() {
this.log = [];
this.logOffset = 0;
await this.loadAgents();
if (this.tab === "approvals") await this.loadApprovals();
},
resetProjectForm() {
this.projectForm = { id: "", root_dir: "", git_remote: "", credential_ref: "", _editing: false };
},
editProject(p) {
this.projectForm = {
id: p.id,
root_dir: p.root_dir,
git_remote: p.git_remote || "",
credential_ref: p.credential_ref || "",
_editing: true,
};
},
async saveProject() {
const f = this.projectForm;
const body = {
root_dir: f.root_dir.trim(),
git_remote: f.git_remote.trim() || null,
credential_ref: f.credential_ref.trim() || null,
};
try {
if (f._editing) {
await this.api(`/projects/${encodeURIComponent(f.id)}`, { method: "PATCH", body });
this.cmd = { text: `project '${f.id}' updated`, error: false, busy: false };
} else {
await this.api("/projects", { method: "POST", body: { id: f.id.trim(), ...body } });
this.cmd = { text: `project '${f.id}' created`, error: false, busy: false };
}
this.resetProjectForm();
await this.loadProjects();
} catch (e) {
if (e instanceof AuthError) return;
this.cmd = { text: e.message, error: true, busy: false };
}
},
async deleteProject(id) {
if (!confirm(`Delete project '${id}'? Its agents/log rows go with it.`)) return;
try {
await this.api(`/projects/${encodeURIComponent(id)}`, { method: "DELETE" });
this.cmd = { text: `project '${id}' deleted`, error: false, busy: false };
if (this.selectedProjectId === id) this.selectedProjectId = "";
await this.loadProjects();
} catch (e) {
if (e instanceof AuthError) return;
this.cmd = { text: e.message, error: true, busy: false };
}
},
// --- agents ---
async loadAgents() {
const p = this.selectedProjectId;
if (!p) return;
const agents = await this.api(`/projects/${encodeURIComponent(p)}/agents`);
this.agents = agents;
this.agents = await this.api(`/projects/${encodeURIComponent(p)}/agents`);
this.lastError = "";
},
@@ -215,6 +333,39 @@ function app() {
return `/projects/${encodeURIComponent(this.selectedProjectId)}/agents/${encodeURIComponent(this.selectedAgentName)}`;
},
async spawnAgent() {
const f = this.spawnForm;
const body = { name: f.name.trim(), role: f.role || null, task: f.task.trim() || null };
if (f.placement === "worktree" && f.worktree.trim()) body.worktree = f.worktree.trim();
if (f.placement === "subdir" && f.subdir.trim()) body.subdir = f.subdir.trim();
const p = encodeURIComponent(this.selectedProjectId);
const final = await this.enqueueAndTrack(`/projects/${p}/agents/spawn`, body, `spawn ${body.name}`);
if (final && final.status === "done") {
this.spawnForm = { name: "", role: "", placement: "worktree", worktree: "", subdir: "", task: "" };
}
await this.loadAgents();
},
async killAgent(name) {
const p = encodeURIComponent(this.selectedProjectId);
await this.enqueueAndTrack(`/projects/${p}/agents/${encodeURIComponent(name)}/kill`, undefined, `kill ${name}`);
await this.loadAgents();
},
async deleteAgent(name) {
if (!confirm(`Delete the agent row '${name}'? (Kill the session first if live.)`)) return;
const p = encodeURIComponent(this.selectedProjectId);
try {
await this.api(`/projects/${p}/agents/${encodeURIComponent(name)}`, { method: "DELETE" });
this.cmd = { text: `agent row '${name}' deleted`, error: false, busy: false };
if (this.selectedAgentName === name) this.selectedAgentName = null;
await this.loadAgents();
} catch (e) {
if (e instanceof AuthError) return;
this.cmd = { text: e.message, error: true, busy: false };
}
},
async loadCheckmark() {
try {
this.checkmark = await this.api(`${this._agentPath()}/checkmark`);
@@ -259,17 +410,12 @@ function app() {
try {
await this.api(`${this._agentPath()}/answer`, { method: "POST", body: { answer: text } });
if (resume) {
const r = await this.api(`${this._agentPath()}/resume`, { method: "POST", body: { answer: text } });
if (r.resumed) {
this.answerMsg = "Answered and resumed.";
this.answerText = "";
await this.tick(); // flip the badge to working without waiting a full interval
} else {
this.answerError = true;
this.answerMsg = `Answer saved, but resume failed: ${r.detail || "unknown error"}`;
await this.loadCheckmark();
await this.loadLog();
}
this.answerMsg = "Answer saved; resume enqueued.";
this.answerText = "";
await this.enqueueAndTrack(`${this._agentPath()}/resume`, { answer: text }, "resume");
await this.loadAgents();
await this.loadCheckmark();
await this.loadLog();
} else {
this.answerMsg = "Answer saved (agent still paused).";
this.answerText = "";
@@ -285,12 +431,104 @@ function app() {
}
},
// --- shared tab ---
switchToShared() {
this.tab = "shared";
this.loadShared();
// --- approvals ---
async loadApprovals() {
if (!this.selectedProjectId) {
this.approvals = [];
return;
}
try {
this.approvals = await this.api(`/projects/${encodeURIComponent(this.selectedProjectId)}/approvals`);
this.lastError = "";
} catch (e) {
if (!(e instanceof AuthError)) this.lastError = e.message;
}
},
async submitApproval() {
const f = this.approvalForm;
const body = {
branch: f.branch.trim(),
status: f.status,
agent_name: f.agent_name.trim() || null,
sha: f.sha.trim() || null,
note: f.note.trim() || null,
};
const p = encodeURIComponent(this.selectedProjectId);
await this.enqueueAndTrack(`/projects/${p}/approvals`, body, `${f.status} ${f.branch}`);
this.approvalForm = { branch: "", status: "approved", agent_name: "", sha: "", note: "" };
await this.loadApprovals();
},
// --- hosts ---
async loadHosts() {
try {
this.hosts = await this.api("/hosts");
this.lastError = "";
} catch (e) {
if (!(e instanceof AuthError)) this.lastError = e.message;
}
},
resetHostForm() {
this.hostForm = { hostname: "", forge_type: "github", token_env_var: "", base_url: "", _editing: false };
},
editHost(h) {
this.hostForm = {
hostname: h.hostname,
forge_type: h.forge_type,
token_env_var: h.token_env_var || "",
base_url: h.base_url || "",
_editing: true,
};
},
async saveHost() {
const f = this.hostForm;
const body = {
forge_type: f.forge_type,
token_env_var: f.token_env_var.trim() || null,
base_url: f.base_url.trim() || null,
};
try {
if (f._editing) {
await this.api(`/hosts/${encodeURIComponent(f.hostname)}`, { method: "PATCH", body });
this.cmd = { text: `host '${f.hostname}' updated`, error: false, busy: false };
} else {
await this.api("/hosts", { method: "POST", body: { hostname: f.hostname.trim(), ...body } });
this.cmd = { text: `host '${f.hostname}' created`, error: false, busy: false };
}
this.resetHostForm();
await this.loadHosts();
} catch (e) {
if (e instanceof AuthError) return;
this.cmd = { text: e.message, error: true, busy: false };
}
},
async deleteHost(hostname) {
if (!confirm(`Delete host '${hostname}'?`)) return;
try {
await this.api(`/hosts/${encodeURIComponent(hostname)}`, { method: "DELETE" });
this.cmd = { text: `host '${hostname}' deleted`, error: false, busy: false };
await this.loadHosts();
} catch (e) {
if (e instanceof AuthError) return;
this.cmd = { text: e.message, error: true, busy: false };
}
},
// --- activity / commands ---
async loadCommands() {
try {
this.commands = await this.api("/commands?limit=50");
this.lastError = "";
} catch (e) {
if (!(e instanceof AuthError)) this.lastError = e.message;
}
},
async pollCiGlobal() {
await this.enqueueAndTrack("/poll-ci", undefined, "poll-ci (all projects)");
await this.loadCommands();
},
// --- shared tab ---
async loadShared() {
try {
const [log, context] = await Promise.all([
@@ -303,6 +541,20 @@ function app() {
if (!(e instanceof AuthError)) this.lastError = e.message;
}
},
async setSharedContext() {
const key = this.sharedForm.key.trim();
const value = this.sharedForm.value.trim();
if (!key || !value) return;
try {
await this.api(`/shared/context/${encodeURIComponent(key)}`, { method: "PUT", body: { value } });
this.cmd = { text: `shared context '${key}' set`, error: false, busy: false };
this.sharedForm = { key: "", value: "" };
await this.loadShared();
} catch (e) {
if (e instanceof AuthError) return;
this.cmd = { text: e.message, error: true, busy: false };
}
},
// --- rendering helpers ---
badgeClass(kind, value) {
+213 -7
View File
@@ -16,8 +16,9 @@
<div class="modal-backdrop" x-show="showTokenModal">
<form class="modal" @submit.prevent="saveToken()">
<h2>Handler</h2>
<p class="muted">Paste your API bearer token to continue.</p>
<input type="password" x-model="tokenInput" placeholder="AUTH_TOKEN"
<p class="muted">Paste your API bearer token to continue. Management actions
(spawn, approve, edit projects/hosts) require the <strong>admin</strong> token.</p>
<input type="password" x-model="tokenInput" placeholder="AUTH_TOKEN / ADMIN_TOKEN"
autocomplete="current-password" x-ref="tokenField" />
<p class="error" x-show="tokenError" x-text="tokenError"></p>
<button type="submit">Save</button>
@@ -37,8 +38,12 @@
</template>
</select>
<nav class="tabs">
<button :class="{ active: tab === 'agents' }" @click="tab = 'agents'">Agents</button>
<button :class="{ active: tab === 'shared' }" @click="switchToShared()">Shared</button>
<button :class="{ active: tab === 'agents' }" @click="switchTab('agents')">Agents</button>
<button :class="{ active: tab === 'approvals' }" @click="switchTab('approvals')">Approvals</button>
<button :class="{ active: tab === 'projects' }" @click="switchTab('projects')">Projects</button>
<button :class="{ active: tab === 'hosts' }" @click="switchTab('hosts')">Hosts</button>
<button :class="{ active: tab === 'activity' }" @click="switchTab('activity')">Activity</button>
<button :class="{ active: tab === 'shared' }" @click="switchTab('shared')">Shared</button>
</nav>
<button class="ghost" @click="refresh()" title="Refresh now"></button>
<button class="ghost" @click="signOut()" title="Sign out / change token">Sign out</button>
@@ -46,18 +51,45 @@
</header>
<p class="banner error" x-show="lastError" x-text="lastError"></p>
<!-- Command feedback: the async result of any enqueued control action. -->
<p class="banner" :class="cmd.error ? 'error' : 'ok'" x-show="cmd.text" x-text="cmd.text"></p>
<!-- AGENTS TAB -->
<main class="layout" x-show="tab === 'agents'">
<!-- Agent list -->
<!-- Agent list + spawn -->
<section class="panel agents">
<h3>Agents <span class="muted" x-show="selectedProjectId" x-text="'· ' + selectedProjectId"></span></h3>
<p class="muted" x-show="selectedProjectId && agents.length === 0">No agents in this project.</p>
<p class="muted" x-show="!selectedProjectId">Select a project.</p>
<div class="card" x-show="selectedProjectId">
<h4>Spawn an agent</h4>
<div class="form-grid">
<input x-model="spawnForm.name" placeholder="name (e.g. junior)" />
<select x-model="spawnForm.role">
<option value="">role — none</option>
<option value="junior">junior</option>
<option value="senior">senior</option>
<option value="deploy">deploy</option>
</select>
<select x-model="spawnForm.placement">
<option value="worktree">git worktree on branch</option>
<option value="subdir">subdir under root</option>
</select>
<input x-show="spawnForm.placement === 'worktree'" x-model="spawnForm.worktree" placeholder="branch (e.g. feat/auth)" />
<input x-show="spawnForm.placement === 'subdir'" x-model="spawnForm.subdir" placeholder="subdir (e.g. api)" />
</div>
<textarea x-model="spawnForm.task" rows="2" placeholder="initial task / prompt (optional)"></textarea>
<div class="toolbar">
<button class="primary" @click="spawnAgent()" :disabled="cmd.busy || !spawnForm.name.trim()">Spawn</button>
</div>
</div>
<p class="muted" x-show="selectedProjectId && agents.length === 0">No agents in this project.</p>
<ul class="agent-list">
<template x-for="a in agents" :key="a.id">
<li :class="{ selected: a.name === selectedAgentName }" @click="selectAgent(a.name)">
<span class="agent-name" x-text="a.name"></span>
<span class="badge" :class="badgeClass('role', a.role || 'none')" x-show="a.role" x-text="a.role"></span>
<span class="badge" :class="badgeClass('status', a.status)" x-text="a.status"></span>
<span class="needs-answer" x-show="a.status === 'paused_for_input'">needs answer</span>
</li>
@@ -71,6 +103,9 @@
<span x-text="selectedAgentName"></span>
<span class="badge" x-show="selectedAgent" :class="badgeClass('status', selectedAgent?.status)"
x-text="selectedAgent?.status"></span>
<span class="spacer"></span>
<button class="ghost small" @click="killAgent(selectedAgentName)" :disabled="cmd.busy">Kill</button>
<button class="ghost small danger" @click="deleteAgent(selectedAgentName)" :disabled="cmd.busy">Delete row</button>
</h3>
<!-- Checkmark -->
@@ -157,6 +192,166 @@
</section>
</main>
<!-- APPROVALS TAB -->
<main class="layout" x-show="tab === 'approvals'">
<section class="panel">
<h3>Approvals <span class="muted" x-show="selectedProjectId" x-text="'· ' + selectedProjectId"></span></h3>
<p class="muted" x-show="!selectedProjectId">Select a project.</p>
<div class="card" x-show="selectedProjectId">
<h4>Record a verdict</h4>
<div class="form-grid">
<input x-model="approvalForm.branch" placeholder="branch (e.g. feat/auth)" />
<select x-model="approvalForm.status">
<option value="approved">approve</option>
<option value="rejected">reject</option>
</select>
<input x-model="approvalForm.agent_name" placeholder="agent (optional — reads its HEAD)" />
<input x-model="approvalForm.sha" placeholder="sha (optional — pins the approval)" />
</div>
<input x-model="approvalForm.note" placeholder="note (optional)" />
<div class="toolbar">
<button class="primary" @click="submitApproval()" :disabled="cmd.busy || !approvalForm.branch.trim()">Enqueue verdict</button>
</div>
</div>
<p class="muted" x-show="selectedProjectId && approvals.length === 0">No approvals recorded.</p>
<div class="table-wrap" x-show="approvals.length">
<table class="log">
<thead><tr><th>When</th><th>Branch</th><th>Verdict</th><th>By</th><th>SHA</th><th>Note</th></tr></thead>
<tbody>
<template x-for="ap in approvals" :key="ap.id">
<tr>
<td class="muted nowrap" x-text="fmt(ap.created_at)"></td>
<td class="mono" x-text="ap.branch"></td>
<td><span class="badge" :class="badgeClass('approval', ap.status)" x-text="ap.status"></span></td>
<td x-text="ap.approved_by_agent_id ? ('agent ' + ap.approved_by_agent_id) : (ap.actor || '—')"></td>
<td class="mono" x-text="ap.approved_sha ? ap.approved_sha.slice(0,7) : '—'"></td>
<td x-text="ap.note || '—'"></td>
</tr>
</template>
</tbody>
</table>
</div>
</section>
</main>
<!-- PROJECTS TAB -->
<main class="layout" x-show="tab === 'projects'">
<section class="panel">
<h3>Projects</h3>
<div class="card">
<h4 x-text="projectForm._editing ? ('Edit project · ' + projectForm.id) : 'Register a project'"></h4>
<div class="form-grid">
<input x-model="projectForm.id" :disabled="projectForm._editing" placeholder="id / slug (e.g. leeworks-api)" />
<input x-model="projectForm.root_dir" placeholder="root_dir (e.g. /var/lib/handler/projects/leeworks)" />
<input x-model="projectForm.git_remote" placeholder="git_remote (optional)" />
<input x-model="projectForm.credential_ref" placeholder="credential_ref (env:VAR / file:/path / db:id)" />
</div>
<p class="muted small">credential_ref is a <em>pointer</em>, never the token. <code>cmd:</code> is CLI-only.</p>
<div class="toolbar">
<button class="primary" @click="saveProject()" :disabled="cmd.busy || !projectForm.id.trim() || !projectForm.root_dir.trim()"
x-text="projectForm._editing ? 'Save changes' : 'Create'"></button>
<button class="ghost" x-show="projectForm._editing" @click="resetProjectForm()">Cancel</button>
</div>
</div>
<div class="table-wrap" x-show="projects.length">
<table class="log">
<thead><tr><th>ID</th><th>root_dir</th><th>git_remote</th><th>credential_ref</th><th></th></tr></thead>
<tbody>
<template x-for="p in projects" :key="p.id">
<tr>
<td class="mono" x-text="p.id"></td>
<td class="mono" x-text="p.root_dir"></td>
<td class="mono" x-text="p.git_remote || '—'"></td>
<td class="mono" x-text="p.credential_ref || '—'"></td>
<td class="nowrap">
<button class="ghost small" @click="editProject(p)">Edit</button>
<button class="ghost small danger" @click="deleteProject(p.id)">Delete</button>
</td>
</tr>
</template>
</tbody>
</table>
</div>
</section>
</main>
<!-- HOSTS TAB -->
<main class="layout" x-show="tab === 'hosts'">
<section class="panel">
<h3>Forge hosts</h3>
<p class="muted small">Maps a git host to the token env var to inject at spawn (and the credential-helper scope).
Holds no secrets — only the env-var name.</p>
<div class="card">
<h4 x-text="hostForm._editing ? ('Edit host · ' + hostForm.hostname) : 'Register a host'"></h4>
<div class="form-grid">
<input x-model="hostForm.hostname" :disabled="hostForm._editing" placeholder="hostname (e.g. git.corp.internal)" />
<select x-model="hostForm.forge_type">
<option value="github">github</option>
<option value="gitlab">gitlab</option>
<option value="gitea">gitea</option>
<option value="forgejo">forgejo</option>
<option value="bitbucket">bitbucket</option>
</select>
<input x-model="hostForm.token_env_var" placeholder="token env var (e.g. GITEA_TOKEN)" />
<input x-model="hostForm.base_url" placeholder="base_url (optional, e.g. https://git.corp.internal)" />
</div>
<div class="toolbar">
<button class="primary" @click="saveHost()" :disabled="cmd.busy || !hostForm.hostname.trim()"
x-text="hostForm._editing ? 'Save changes' : 'Create'"></button>
<button class="ghost" x-show="hostForm._editing" @click="resetHostForm()">Cancel</button>
</div>
</div>
<p class="muted" x-show="hosts.length === 0">No hosts registered (built-in host map still applies).</p>
<div class="table-wrap" x-show="hosts.length">
<table class="log">
<thead><tr><th>Hostname</th><th>Type</th><th>Token env var</th><th>Base URL</th><th></th></tr></thead>
<tbody>
<template x-for="h in hosts" :key="h.hostname">
<tr>
<td class="mono" x-text="h.hostname"></td>
<td x-text="h.forge_type"></td>
<td class="mono" x-text="h.token_env_var || '—'"></td>
<td class="mono" x-text="h.base_url || '—'"></td>
<td class="nowrap">
<button class="ghost small" @click="editHost(h)">Edit</button>
<button class="ghost small danger" @click="deleteHost(h.hostname)">Delete</button>
</td>
</tr>
</template>
</tbody>
</table>
</div>
</section>
</main>
<!-- ACTIVITY TAB -->
<main class="layout" x-show="tab === 'activity'">
<section class="panel">
<h3>Activity <span class="muted">· control commands</span>
<span class="spacer"></span>
<button class="ghost small" @click="pollCiGlobal()" :disabled="cmd.busy">Sweep CI now</button>
</h3>
<p class="muted" x-show="commands.length === 0">No commands yet.</p>
<div class="table-wrap" x-show="commands.length">
<table class="log">
<thead><tr><th>When</th><th>Type</th><th>Project</th><th>Agent</th><th>Status</th><th>Result / Error</th></tr></thead>
<tbody>
<template x-for="c in commands" :key="c.id">
<tr>
<td class="muted nowrap" x-text="fmt(c.created_at)"></td>
<td class="mono" x-text="c.type"></td>
<td class="mono" x-text="c.project_id || '—'"></td>
<td class="mono" x-text="c.agent_name || '—'"></td>
<td><span class="badge" :class="badgeClass('cmd', c.status)" x-text="c.status"></span></td>
<td class="mono small" x-text="c.error || (c.result ? JSON.stringify(c.result) : '—')"></td>
</tr>
</template>
</tbody>
</table>
</div>
</section>
</main>
<!-- SHARED TAB -->
<main class="layout" x-show="tab === 'shared'">
<section class="panel">
@@ -180,7 +375,18 @@
</div>
</section>
<section class="panel">
<h3>Shared context <span class="muted">(read-only)</span></h3>
<h3>Shared context</h3>
<div class="card">
<h4>Set a key</h4>
<div class="form-grid">
<input x-model="sharedForm.key" placeholder="key" />
<input x-model="sharedForm.value" placeholder="value" />
</div>
<p class="muted small">Requires the shared-context write token (or admin/global if unset).</p>
<div class="toolbar">
<button class="primary" @click="setSharedContext()" :disabled="!sharedForm.key.trim() || !sharedForm.value.trim()">Set</button>
</div>
</div>
<p class="muted" x-show="shared.context.length === 0">No shared context keys.</p>
<div class="table-wrap" x-show="shared.context.length">
<table class="log">
+25
View File
@@ -90,6 +90,7 @@ textarea { resize: vertical; }
.tabs button.active { background: var(--accent); border-color: var(--accent); color: #fff; }
.banner { margin: 0; padding: 0.5rem 1rem; background: rgba(229,72,77,0.12); }
.banner.ok { background: rgba(53,194,106,0.12); color: var(--green); }
/* --- layout --- */
.layout {
@@ -177,3 +178,27 @@ table.log th { color: var(--muted); font-weight: 500; font-size: 0.8rem; }
.badge-visibility-project { background: rgba(90,101,114,0.2); color: var(--grey); }
.badge-visibility-global { background: transparent; color: var(--blue); border-color: var(--blue); }
.badge-role-junior { background: rgba(76,141,255,0.14); color: var(--blue); }
.badge-role-senior { background: rgba(232,163,61,0.16); color: var(--amber); }
.badge-role-deploy { background: rgba(53,194,106,0.16); color: var(--green); }
.badge-approval-approved { background: rgba(53,194,106,0.18); color: var(--green); }
.badge-approval-rejected { background: rgba(229,72,77,0.18); color: var(--red); }
.badge-cmd-queued { background: rgba(90,101,114,0.2); color: var(--grey); }
.badge-cmd-running { background: rgba(232,163,61,0.18); color: var(--amber); }
.badge-cmd-done { background: rgba(53,194,106,0.18); color: var(--green); }
.badge-cmd-failed { background: rgba(229,72,77,0.18); color: var(--red); }
/* --- management forms --- */
.form-grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(200px, 1fr)); gap: 0.5rem; margin-bottom: 0.5rem; }
.toolbar { display: flex; gap: 0.5rem; margin-top: 0.6rem; align-items: center; }
.spacer { flex: 1; }
h3 { display: flex; align-items: center; gap: 0.5rem; }
button.small { padding: 0.2rem 0.55rem; font-size: 0.8rem; }
button.danger { color: var(--red); border-color: var(--border); }
button.danger:hover:not(:disabled) { border-color: var(--red); }
.small { font-size: 0.8rem; }
.card > .form-grid + .muted { margin: 0.25rem 0 0; }
.card code { background: var(--bg); padding: 0.05rem 0.3rem; border-radius: 4px; }
+11
View File
@@ -26,6 +26,12 @@ class Settings(BaseSettings):
# auth_token when unset (README 3.4 open question, resolved to "gate it").
shared_context_write_token: str | None = None
# Optional admin token gating the state-changing control surface exposed to the web:
# enqueuing control commands (spawn/kill/resume/approve/…), project CRUD, host CRUD,
# and credential-pointer edits. Falls back to auth_token when unset. A single global
# token, like auth_token — per-user RBAC is future work.
admin_token: str | None = None
# Optional generic webhook target for the Notification hook. No-op when unset.
webhook_url: str | None = None
@@ -70,6 +76,11 @@ class Settings(BaseSettings):
"""Token required to write shared_context; defaults to the global token."""
return self.shared_context_write_token or self.auth_token
@property
def effective_admin_token(self) -> str:
"""Token required for the web control surface; defaults to the global token."""
return self.admin_token or self.auth_token
@lru_cache
def get_settings() -> Settings:
+21 -1
View File
@@ -15,7 +15,7 @@ import sys
from ..db import repository as repo
from ..db.engine import connection
from . import poller, skills_gen, spawn, tmux
from . import poller, skills_gen, spawn, tmux, worker
def _cmd_spawn(args: argparse.Namespace) -> int:
@@ -191,6 +191,15 @@ def _cmd_forge_init(args: argparse.Namespace) -> int:
return 0
def _cmd_worker(args: argparse.Namespace) -> int:
print(
f"worker starting (poll={args.interval}s, ci-sweep={args.ci_interval}s); "
"draining control commands + sweeping CI"
)
worker.run(poll_interval=args.interval, ci_interval=args.ci_interval)
return 0 # pragma: no cover - run loops until interrupted
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(prog="handler", description="Handler control layer")
sub = parser.add_subparsers(dest="command", required=True)
@@ -245,6 +254,17 @@ def build_parser() -> argparse.ArgumentParser:
p_forge.add_argument("--no-commit", action="store_true", help="write but don't git-commit")
p_forge.set_defaults(func=_cmd_forge_init)
p_worker = sub.add_parser(
"worker", help="run the control worker: drain enqueued commands + sweep CI"
)
p_worker.add_argument(
"--interval", type=float, default=2.0, help="seconds to idle when the queue is empty"
)
p_worker.add_argument(
"--ci-interval", type=float, default=30.0, help="seconds between CI sweeps (0 disables)"
)
p_worker.set_defaults(func=_cmd_worker)
return parser
+111 -44
View File
@@ -18,8 +18,12 @@ from __future__ import annotations
import os
import shlex
import subprocess
from typing import TYPE_CHECKING
from urllib.parse import urlsplit
if TYPE_CHECKING:
from sqlalchemy import Connection
# The env var Handler always injects and that the git credential helper reads back.
CANONICAL_TOKEN_ENV = "FORGE_TOKEN"
@@ -42,6 +46,65 @@ class CredentialError(Exception):
"""Raised when a ``credential_ref`` cannot be resolved to a value."""
# Schemes an operator may set from the web. ``cmd:`` is deliberately excluded there — it
# executes an arbitrary command in the control container at spawn — so the API rejects it
# while the CLI/DB path still allows it (see api/schemas.py). ``db:`` is reserved for the
# future encrypted secret store (not yet resolvable).
WEB_SETTABLE_SCHEMES = ("env", "file", "db")
def _resolve_env(rest: str) -> str:
value = os.environ.get(rest)
if value is None:
raise CredentialError(f"credential_ref env var '{rest}' is not set")
return value
def _resolve_file(rest: str) -> str:
try:
with open(os.path.expanduser(rest)) as fh:
return fh.read().strip()
except OSError as exc:
raise CredentialError(f"credential_ref file '{rest}' unreadable: {exc}") from exc
def _resolve_cmd(rest: str) -> str:
try:
result = subprocess.run(
shlex.split(rest),
capture_output=True,
text=True,
timeout=30,
)
except (OSError, subprocess.SubprocessError) as exc:
raise CredentialError(f"credential_ref cmd '{rest}' failed: {exc}") from exc
if result.returncode != 0:
raise CredentialError(
f"credential_ref cmd '{rest}' exited {result.returncode}: "
f"{(result.stderr or '').strip()}"
)
return result.stdout.strip()
def _resolve_db(rest: str) -> str:
# Reserved for the encrypted secret store (a later phase); the ``db:`` scheme and this
# dispatch seam land now so that store is a drop-in without touching every caller.
raise CredentialError(
f"credential_ref 'db:{rest}' scheme is reserved for the encrypted secret store, "
"which is not enabled yet"
)
# Scheme -> resolver. Adding the encrypted store later means wiring _resolve_db to it,
# nothing else here changes.
_RESOLVERS = {
"env": _resolve_env,
"file": _resolve_file,
"cmd": _resolve_cmd,
"db": _resolve_db,
}
def resolve(credential_ref: str | None) -> str | None:
"""Resolve a ``credential_ref`` pointer to an actual secret value.
@@ -57,40 +120,13 @@ def resolve(credential_ref: str | None) -> str | None:
if not rest:
raise CredentialError(f"credential_ref '{ref}' has no value after '{scheme}:'")
if scheme == "env":
value = os.environ.get(rest)
if value is None:
raise CredentialError(f"credential_ref env var '{rest}' is not set")
return value
if scheme == "file":
try:
with open(os.path.expanduser(rest)) as fh:
return fh.read().strip()
except OSError as exc:
raise CredentialError(f"credential_ref file '{rest}' unreadable: {exc}") from exc
if scheme == "cmd":
try:
result = subprocess.run(
shlex.split(rest),
capture_output=True,
text=True,
timeout=30,
)
except (OSError, subprocess.SubprocessError) as exc:
raise CredentialError(f"credential_ref cmd '{rest}' failed: {exc}") from exc
if result.returncode != 0:
raise CredentialError(
f"credential_ref cmd '{rest}' exited {result.returncode}: "
f"{(result.stderr or '').strip()}"
)
return result.stdout.strip()
raise CredentialError(
f"credential_ref '{ref}' has unknown scheme '{scheme}' "
"(expected env:, file:, or cmd:)"
)
resolver = _RESOLVERS.get(scheme)
if resolver is None:
raise CredentialError(
f"credential_ref '{ref}' has unknown scheme '{scheme}' "
"(expected env:, file:, cmd:, or db:)"
)
return resolver(rest)
def remote_host(git_remote: str | None) -> str | None:
@@ -108,10 +144,7 @@ def remote_host(git_remote: str | None) -> str | None:
return None
def _host_token_env(git_remote: str | None) -> str | None:
host = remote_host(git_remote)
if not host:
return None
def _builtin_host_token_env(host: str) -> str | None:
for known, env_var in _HOST_TOKEN_ENV.items():
if host == known or host.endswith("." + known):
return env_var
@@ -121,13 +154,40 @@ def _host_token_env(git_remote: str | None) -> str | None:
return None
def git_credential_config(git_remote: str | None) -> tuple[str, str] | None:
def _host_token_env(git_remote: str | None, conn: Connection | None = None) -> str | None:
"""The per-host token env var for a remote.
Consults the web-managed ``forge_hosts`` registry first (when a ``conn`` is available),
so operators can register self-hosted forges without a code change; falls back to the
built-in map so behaviour is unchanged when no row exists.
"""
host = remote_host(git_remote)
if not host:
return None
if conn is not None:
row = _host_row(conn, host)
if row and row.get("token_env_var"):
return row["token_env_var"]
return _builtin_host_token_env(host)
def _host_row(conn: Connection, host: str) -> dict | None:
# Local import to keep this module import-light and avoid a control<->db import cycle.
from ..db import repository as repo
return repo.get_host(conn, host)
def git_credential_config(
git_remote: str | None, conn: Connection | None = None
) -> tuple[str, str] | None:
"""The scoped git config (key, value) that installs the credential helper.
Scopes the helper to the forge's HTTPS base URL — ``credential.https://host.helper``
— so the injected token is only ever offered to that host, never to an arbitrary
HTTPS URL the agent might touch. Returns ``None`` for ssh/unknown remotes, where no
HTTPS credential helper is needed (ssh uses deploy keys).
HTTPS URL the agent might touch. A registered host's explicit ``base_url`` wins when
set. Returns ``None`` for ssh/unknown remotes, where no HTTPS credential helper is
needed (ssh uses deploy keys).
"""
if not git_remote or "://" not in git_remote:
return None
@@ -135,20 +195,27 @@ def git_credential_config(git_remote: str | None) -> tuple[str, str] | None:
if parts.scheme not in ("https", "http") or not parts.hostname:
return None
base = f"{parts.scheme}://{parts.hostname}"
if conn is not None:
row = _host_row(conn, parts.hostname.lower())
if row and row.get("base_url"):
base = row["base_url"].rstrip("/")
return f"credential.{base}.helper", git_credential_helper_value()
def credential_env(token: str | None, git_remote: str | None) -> dict[str, str]:
def credential_env(
token: str | None, git_remote: str | None, conn: Connection | None = None
) -> dict[str, str]:
"""The environment variables to inject so forge + git both authenticate.
Always sets ``FORGE_TOKEN`` (the generic name forge accepts and our git helper
reads); additionally sets the host-specific var (``GITHUB_TOKEN`` etc.) when the
remote host is recognized, so per-host tooling works with zero extra config.
remote host is recognized — via the ``forge_hosts`` registry when a ``conn`` is given,
otherwise the built-in map — so per-host tooling works with zero extra config.
"""
if not token:
return {}
env = {CANONICAL_TOKEN_ENV: token}
host_var = _host_token_env(git_remote)
host_var = _host_token_env(git_remote, conn)
if host_var:
env[host_var] = token
return env
+10 -5
View File
@@ -52,7 +52,9 @@ def _shell_quote(value: str) -> str:
return "'" + value.replace("'", "'\\''") + "'"
def _install_git_credentials(working_dir: str, git_remote: str | None) -> None:
def _install_git_credentials(
working_dir: str, git_remote: str | None, conn=None
) -> None:
"""Install a repo-local git credential helper that reads the injected token.
The helper hands back ``$FORGE_TOKEN`` from the environment, so the raw value is never
@@ -61,7 +63,7 @@ def _install_git_credentials(working_dir: str, git_remote: str | None) -> None:
A no-op for ssh/unknown remotes (deploy keys handle those). Best-effort — a
working_dir that isn't a git repo yet shouldn't block the spawn.
"""
cfg = credentials.git_credential_config(git_remote)
cfg = credentials.git_credential_config(git_remote, conn)
if cfg is not None:
key, value = cfg
gitops.config_local(working_dir, key, value)
@@ -116,9 +118,12 @@ def spawn(
}
if role:
env["HANDLER_AGENT_ROLE"] = role
env.update(credentials.credential_env(token, project.get("git_remote")))
if token:
_install_git_credentials(working_dir, project.get("git_remote"))
# A short read connection lets credential/host resolution consult the forge_hosts
# registry (falling back to the built-in host map when a host has no row).
with connection() as conn:
env.update(credentials.credential_env(token, project.get("git_remote"), conn))
if token:
_install_git_credentials(working_dir, project.get("git_remote"), conn)
# Verify the pinned forge version, if one is configured. Non-fatal: a version drift
# is recorded as a warning rather than blocking the spawn, since not every agent
+235
View File
@@ -0,0 +1,235 @@
"""The control-container worker: executes commands the API enqueues.
The API (in its own container) has no ``git``/``tmux``/``claude`` and does not own the
tmux sessions, so it cannot run control actions directly. Instead it writes a ``queued``
row to the ``commands`` table; this worker — running in the control container — claims each
row, dispatches it to the *same* control functions the CLI uses (``spawn``/``poller``/
``skills_gen``/``repo.record_approval``), and writes the result or error back. It also runs
the periodic CI sweep, subsuming the old ``poll-ci --watch`` loop.
Every command runs in isolation: one bad command is recorded as ``failed`` and never stops
the loop. ``execute_command`` is the pure dispatch seam (given a claimed command dict,
returns a JSON-safe result or raises); ``drain``/``run`` are the claim+finish plumbing.
"""
from __future__ import annotations
import os
import time
from ..db import repository as repo
from ..db.engine import connection
from . import gitops, poller, skills_gen, spawn
class CommandError(Exception):
"""A command that cannot be executed (bad payload, missing target, …)."""
def _payload(command: dict) -> dict:
return command.get("payload") or {}
def _cmd_spawn(command: dict) -> dict:
p = _payload(command)
name = command.get("agent_name") or p.get("name")
if not command.get("project_id") or not name:
raise CommandError("spawn requires project_id and an agent name")
agent = spawn.spawn(
command["project_id"],
name,
subdir=p.get("subdir") or p.get("dir"),
worktree_branch=p.get("worktree"),
task=p.get("task"),
role=p.get("role"),
)
result = {
"agent_id": agent["id"],
"name": agent["name"],
"working_dir": agent["working_dir"],
}
if agent.get("forge_note"):
result["forge_note"] = agent["forge_note"]
return result
def _cmd_kill(command: dict) -> dict:
name = command.get("agent_name")
if not command.get("project_id") or not name:
raise CommandError("kill requires project_id and agent_name")
spawn.kill(command["project_id"], name)
return {"killed": name}
def _cmd_resume(command: dict) -> dict:
name = command.get("agent_name")
answer = _payload(command).get("answer")
if not command.get("project_id") or not name:
raise CommandError("resume requires project_id and agent_name")
if not answer:
raise CommandError("resume requires an 'answer' in the payload")
with connection() as conn:
agent = repo.get_agent_by_name(conn, command["project_id"], name)
if agent is None:
raise CommandError(f"agent '{name}' not found in project '{command['project_id']}'")
ok, detail = spawn.resume(agent, answer)
if ok:
with connection() as conn:
repo.set_agent_status(conn, agent["id"], "working")
return {"resumed": ok, "detail": detail}
def _record_verdict(command: dict, status: str) -> dict:
p = _payload(command)
branch = p.get("branch")
if not command.get("project_id") or not branch:
raise CommandError(f"{status} requires project_id and a 'branch' in the payload")
# Pin an approval to the reviewed commit so later pushes invalidate it. Prefer an
# explicit sha; else read HEAD of the target agent's working dir (or the project root).
approved_sha = p.get("sha")
if approved_sha is None and status == "approved":
with connection() as conn:
working_dir = None
if command.get("agent_name"):
agent = repo.get_agent_by_name(conn, command["project_id"], command["agent_name"])
working_dir = agent["working_dir"] if agent else None
if working_dir is None:
project = repo.get_project(conn, command["project_id"])
working_dir = project["root_dir"] if project else None
if working_dir:
approved_sha = gitops.head_sha(working_dir)
actor = command.get("requested_by") or "operator:web"
with connection() as conn:
approval = repo.record_approval(
conn,
project_id=command["project_id"],
branch=branch,
status=status,
pr_ref=p.get("pr"),
note=p.get("note"),
approved_sha=approved_sha,
actor=actor,
)
return {
"approval_id": approval["id"],
"branch": branch,
"status": status,
"approved_sha": approved_sha,
}
def _cmd_approve(command: dict) -> dict:
return _record_verdict(command, "approved")
def _cmd_reject(command: dict) -> dict:
return _record_verdict(command, "rejected")
def _cmd_forge_init(command: dict) -> dict:
project_id = command.get("project_id")
if not project_id:
raise CommandError("forge_init requires project_id")
with connection() as conn:
project = repo.get_project(conn, project_id)
if project is None:
raise CommandError(f"project '{project_id}' not registered")
root = project["root_dir"]
written = skills_gen.write_skills(root)
result = {"written": len(written)}
if not _payload(command).get("no_commit"):
rel = os.path.join(".claude", "skills")
ok_add, _ = gitops.add(root, [rel])
ok_commit, out = gitops.commit(root, "chore: add handler forge-workflow skills")
result["committed"] = bool(ok_add and ok_commit)
if not result["committed"]:
result["commit_note"] = out
return result
def _cmd_poll_ci(command: dict) -> dict:
return poller.sweep(project_id=command.get("project_id"))
_DISPATCH = {
"spawn": _cmd_spawn,
"kill": _cmd_kill,
"resume": _cmd_resume,
"approve": _cmd_approve,
"reject": _cmd_reject,
"forge_init": _cmd_forge_init,
"poll_ci": _cmd_poll_ci,
}
def execute_command(command: dict) -> dict:
"""Dispatch a claimed command to its handler and return a JSON-safe result.
Raises on any failure; the caller records that as a ``failed`` command. This is the
pure seam tests exercise directly (with the tmux/gitops/forge/spawn.resume mocks).
"""
handler = _DISPATCH.get(command["type"])
if handler is None: # pragma: no cover - CHECK constraint keeps types in the set
raise CommandError(f"unknown command type '{command['type']}'")
return handler(command)
def _run_one(command: dict) -> None:
"""Execute a claimed command and record done/failed — never raises."""
try:
result = execute_command(command)
with connection() as conn:
repo.finish_command(conn, command["id"], "done", result=result)
except Exception as exc: # noqa: BLE001 - one command must not kill the loop
with connection() as conn:
repo.finish_command(conn, command["id"], "failed", error=str(exc))
def drain(worker_id: str, limit: int | None = None) -> int:
"""Claim and run queued commands until the queue is empty (or ``limit`` reached).
Returns how many commands were processed. Each command is claimed in its own
transaction, executed, then finished in another — so the claim is committed (visible as
``running``) before the potentially slow control action runs.
"""
processed = 0
while limit is None or processed < limit:
with connection() as conn:
command = repo.claim_next_command(conn, worker_id)
if command is None:
break
_run_one(command)
processed += 1
return processed
def run(
worker_id: str | None = None,
poll_interval: float = 2.0,
ci_interval: float = 30.0,
iterations: int | None = None,
) -> None:
"""The control-container main loop: drain the command queue + sweep CI periodically.
``iterations`` bounds the loop for tests; production runs unbounded. Sleeps
``poll_interval`` only when a pass found no commands, so bursts drain promptly.
"""
worker_id = worker_id or f"worker-{os.getpid()}"
last_ci = 0.0
count = 0
while iterations is None or count < iterations:
did_work = drain(worker_id) > 0
now = time.monotonic()
if ci_interval > 0 and now - last_ci >= ci_interval:
try:
poller.sweep()
except Exception: # noqa: BLE001 - a CI sweep hiccup must not kill the worker
pass
last_ci = now
count += 1
if iterations is not None and count >= iterations:
break
if not did_work:
time.sleep(poll_interval)
+196 -3
View File
@@ -21,7 +21,16 @@ from typing import Any
from sqlalchemy import Connection, select
from .tables import agents, approvals, checkmarks, log_entries, projects, shared_context
from .tables import (
agents,
approvals,
checkmarks,
commands,
forge_hosts,
log_entries,
projects,
shared_context,
)
from .upsert import upsert_checkmark
@@ -237,12 +246,19 @@ def record_approval(
project_id: str,
branch: str,
status: str,
approved_by_agent_id: int,
approved_by_agent_id: int | None = None,
pr_ref: str | None = None,
note: str | None = None,
approved_sha: str | None = None,
actor: str | None = None,
) -> dict:
"""Insert an approval/rejection record for a branch (the senior agent's verdict)."""
"""Insert an approval/rejection record for a branch.
An agent verdict passes ``approved_by_agent_id``; an operator verdict from the
dashboard passes ``actor`` (e.g. ``operator:web``) and leaves the agent id null. The
deploy gate treats a null agent id as a genuinely different party than any pushing
agent, so operator approvals satisfy the "no self-approval" rule.
"""
result = conn.execute(
approvals.insert().values(
project_id=project_id,
@@ -251,6 +267,7 @@ def record_approval(
pr_ref=pr_ref,
status=status,
approved_by_agent_id=approved_by_agent_id,
actor=actor,
note=note,
created_at=_now(),
)
@@ -260,6 +277,17 @@ def record_approval(
return dict(row._mapping)
def list_approvals(
conn: Connection, project_id: str, branch: str | None = None, limit: int = 100
) -> list[dict]:
"""Approvals for a project (optionally one branch), newest first — the UI's view."""
stmt = select(approvals).where(approvals.c.project_id == project_id)
if branch is not None:
stmt = stmt.where(approvals.c.branch == branch)
rows = conn.execute(stmt.order_by(approvals.c.id.desc()).limit(limit)).all()
return [dict(r._mapping) for r in rows]
def update_ci_status(conn: Connection, log_entry_id: int, ci_status: str) -> bool:
"""Backfill a resolved CI verdict onto the log entry that recorded the push."""
result = conn.execute(
@@ -291,3 +319,168 @@ def set_shared_context(conn: Connection, key: str, value: str, agent_id: int | N
)
conn.execute(stmt)
return get_shared_context_key(conn, key)
# --------------------------------------------------- projects & agents (web management)
def update_project(conn: Connection, project_id: str, **fields: Any) -> dict | None:
"""Patch a project's editable columns (root_dir / git_remote / credential_ref).
Only known columns are applied; an empty patch is a no-op read. Returns the row.
"""
allowed = {"root_dir", "git_remote", "credential_ref"}
values = {k: v for k, v in fields.items() if k in allowed}
if values:
conn.execute(projects.update().where(projects.c.id == project_id).values(**values))
return get_project(conn, project_id)
def delete_project(conn: Connection, project_id: str) -> bool:
result = conn.execute(projects.delete().where(projects.c.id == project_id))
return result.rowcount > 0
def delete_agent(conn: Connection, project_id: str, name: str) -> bool:
result = conn.execute(
agents.delete().where(agents.c.project_id == project_id, agents.c.name == name)
)
return result.rowcount > 0
# ------------------------------------------------------------- command queue / audit log
def enqueue_command(
conn: Connection,
type: str,
*,
project_id: str | None = None,
agent_name: str | None = None,
payload: dict | None = None,
requested_by: str | None = None,
) -> dict:
"""Insert a ``queued`` control command for the worker to pick up. Returns the row."""
result = conn.execute(
commands.insert().values(
project_id=project_id,
agent_name=agent_name,
type=type,
payload=payload,
status="queued",
requested_by=requested_by,
created_at=_now(),
)
)
return get_command(conn, result.inserted_primary_key[0])
def get_command(conn: Connection, command_id: int) -> dict | None:
row = conn.execute(select(commands).where(commands.c.id == command_id)).first()
return _row_to_dict(row)
def list_commands(
conn: Connection, project_id: str | None = None, limit: int = 100, offset: int = 0
) -> list[dict]:
stmt = select(commands)
if project_id is not None:
stmt = stmt.where(commands.c.project_id == project_id)
rows = conn.execute(
stmt.order_by(commands.c.id.desc()).limit(limit).offset(offset)
).all()
return [dict(r._mapping) for r in rows]
def claim_next_command(conn: Connection, worker_id: str) -> dict | None:
"""Atomically claim the oldest queued command, flipping it to ``running``.
Postgres uses ``FOR UPDATE SKIP LOCKED`` so multiple workers never grab the same row;
on SQLite (single writer per transaction) the guarded ``WHERE status='queued'`` update
plus a rowcount check is enough. Returns the claimed row, or ``None`` when the queue is
empty or another worker won the race.
"""
sel = (
select(commands.c.id)
.where(commands.c.status == "queued")
.order_by(commands.c.id.asc())
.limit(1)
)
if conn.dialect.name == "postgresql":
sel = sel.with_for_update(skip_locked=True)
row = conn.execute(sel).first()
if row is None:
return None
command_id = row[0]
result = conn.execute(
commands.update()
.where(commands.c.id == command_id, commands.c.status == "queued")
.values(status="running", claimed_by=worker_id, claimed_at=_now())
)
if result.rowcount != 1:
return None # lost the race to another worker
return get_command(conn, command_id)
def finish_command(
conn: Connection,
command_id: int,
status: str,
result: dict | None = None,
error: str | None = None,
) -> None:
"""Mark a claimed command ``done`` or ``failed`` with its result/error."""
conn.execute(
commands.update()
.where(commands.c.id == command_id)
.values(status=status, result=result, error=error, finished_at=_now())
)
# ---------------------------------------------------------------- forge hosts (registry)
def list_hosts(conn: Connection) -> list[dict]:
rows = conn.execute(select(forge_hosts).order_by(forge_hosts.c.hostname)).all()
return [dict(r._mapping) for r in rows]
def get_host(conn: Connection, hostname: str) -> dict | None:
row = conn.execute(
select(forge_hosts).where(forge_hosts.c.hostname == hostname)
).first()
return _row_to_dict(row)
def create_host(
conn: Connection,
hostname: str,
forge_type: str,
token_env_var: str | None = None,
base_url: str | None = None,
) -> dict:
conn.execute(
forge_hosts.insert().values(
hostname=hostname,
forge_type=forge_type,
token_env_var=token_env_var,
base_url=base_url,
created_at=_now(),
)
)
return get_host(conn, hostname)
def update_host(conn: Connection, hostname: str, **fields: Any) -> dict | None:
allowed = {"forge_type", "token_env_var", "base_url"}
values = {k: v for k, v in fields.items() if k in allowed}
if values:
conn.execute(
forge_hosts.update().where(forge_hosts.c.hostname == hostname).values(**values)
)
return get_host(conn, hostname)
def delete_host(conn: Connection, hostname: str) -> bool:
result = conn.execute(forge_hosts.delete().where(forge_hosts.c.hostname == hostname))
return result.rowcount > 0
+53 -1
View File
@@ -31,6 +31,11 @@ GATE_STATUSES = ("pass", "fail", "unknown")
CI_STATUSES = ("not_applicable", "pending", "pass", "fail")
VISIBILITIES = ("project", "global")
APPROVAL_STATUSES = ("approved", "rejected")
# The control actions the API enqueues and the control-container worker executes.
COMMAND_TYPES = ("spawn", "kill", "resume", "approve", "reject", "forge_init", "poll_ci")
COMMAND_STATUSES = ("queued", "running", "done", "failed")
# Forge families a host can belong to (drives per-host token env conventions).
FORGE_TYPES = ("github", "gitlab", "gitea", "forgejo", "bitbucket")
def _in(column: str, values: tuple[str, ...]) -> str:
@@ -133,9 +138,56 @@ approvals = Table(
Column("approved_sha", String), # the HEAD the reviewer signed off on, when known
Column("pr_ref", String), # optional forge PR number/URL, for traceability
Column("status", String, nullable=False),
Column("approved_by_agent_id", BigInteger, ForeignKey("agents.id"), nullable=False),
# The reviewing agent, when an agent recorded the verdict. Nullable so an operator can
# approve/reject from the dashboard (no acting agent) — such rows set ``actor`` instead.
Column("approved_by_agent_id", BigInteger, ForeignKey("agents.id")),
# Human-readable actor label for a non-agent verdict, e.g. "operator:web". The deploy
# gate's "different agent than the pusher" check treats a null agent id as a genuine
# second party, so operator approvals satisfy it.
Column("actor", String),
Column("note", String),
Column("created_at", PortableTimestamp, nullable=False, server_default=func.now()),
CheckConstraint(_in("status", APPROVAL_STATUSES), name="ck_approvals_status"),
Index("ix_approvals_project_branch", "project_id", "branch"),
)
# The control-action queue + audit log (README §"web management"). The API writes a
# ``queued`` row; the worker in the control container claims it (status -> ``running``),
# dispatches to the matching control function, and writes ``done``/``failed`` back with a
# result/error. ``project_id`` is nullable because poll_ci can sweep every project at once.
commands = Table(
"commands",
metadata,
Column("id", PortableBigInt, primary_key=True, autoincrement=True),
Column("project_id", String, ForeignKey("projects.id")),
Column("agent_name", String), # target agent for spawn/kill/resume; null otherwise
Column("type", String, nullable=False),
Column("payload", PortableJSON), # type-specific args (role/worktree/task, branch/sha…)
Column("status", String, nullable=False, server_default="queued"),
Column("result", PortableJSON),
Column("error", String),
Column("requested_by", String), # actor label, e.g. "operator:web"
Column("claimed_by", String), # worker id that claimed the command
Column("created_at", PortableTimestamp, nullable=False, server_default=func.now()),
Column("claimed_at", PortableTimestamp),
Column("finished_at", PortableTimestamp),
CheckConstraint(_in("type", COMMAND_TYPES), name="ck_commands_type"),
CheckConstraint(_in("status", COMMAND_STATUSES), name="ck_commands_status"),
# The worker claims oldest-queued-first; this index serves that hot path.
Index("ix_commands_status_id", "status", "id"),
)
# Web-managed forge hosts (README §"web management"). Makes the host->token-env mapping
# that ``control.credentials`` used to hardcode into an editable registry, and lets
# operators register self-hosted forges without a code change. The built-in map in
# ``control.credentials`` remains the fallback when a host has no row here.
forge_hosts = Table(
"forge_hosts",
metadata,
Column("hostname", String, primary_key=True), # e.g. "github.com", "git.corp.internal"
Column("forge_type", String, nullable=False),
Column("token_env_var", String), # per-host env name to inject, e.g. "GITHUB_TOKEN"
Column("base_url", String), # HTTPS base for the credential-helper scope, when non-default
Column("created_at", PortableTimestamp, nullable=False, server_default=func.now()),
CheckConstraint(_in("forge_type", FORGE_TYPES), name="ck_forge_hosts_type"),
)
+9 -2
View File
@@ -159,7 +159,10 @@ def _approval_ok(
f"branch '{branch}' has no standing approval. A senior agent must approve it "
"before it can be merged or deployed."
)
if approval["approved_by_agent_id"] == ident.agent_id:
# No self-approval. A null approver id is an operator verdict from the dashboard — a
# genuinely different party than any pushing agent — so it never trips this check.
approver_id = approval.get("approved_by_agent_id")
if approver_id is not None and approver_id == ident.agent_id:
return False, (
f"branch '{branch}' was approved by this same agent. Review must come from a "
"different agent — no self-approval."
@@ -170,7 +173,11 @@ def _approval_ok(
f"the approval for '{branch}' was for commit {approved_sha[:12]}, but HEAD is "
f"now {current_sha[:12]}. The new commits must be re-reviewed."
)
return True, f"branch '{branch}' approved by agent id={approval['approved_by_agent_id']}"
if approver_id is not None:
by = f"agent id={approver_id}"
else:
by = approval.get("actor") or "operator"
return True, f"branch '{branch}' approved by {by}"
def handle_merge_deploy(conn: Connection, ident: Identity, hook_input: HookInput) -> dict:
@@ -0,0 +1,82 @@
"""web management: command queue, forge hosts, operator approvals
Revision ID: 0003_web_management
Revises: 0002_forge_approvals
Create Date: 2026-07-10
Adds the ``commands`` queue/audit table (the API enqueues control actions; the
control-container worker executes them) and the ``forge_hosts`` registry (web-managed
host->token-env mapping). Also relaxes ``approvals.approved_by_agent_id`` to nullable and
adds ``approvals.actor`` so an operator can approve/reject from the dashboard without an
acting agent. Hand-written like 0001/0002 so both dialects render correctly; the
nullability change goes through ``batch_alter_table`` so SQLite (no native DROP NOT NULL)
recreates the table while Postgres uses a plain ``ALTER COLUMN``.
"""
from __future__ import annotations
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
from handler.db.types import PortableBigInt, PortableJSON, PortableTimestamp
revision: str = "0003_web_management"
down_revision: str | None = "0002_forge_approvals"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
COMMAND_TYPES = "'spawn', 'kill', 'resume', 'approve', 'reject', 'forge_init', 'poll_ci'"
COMMAND_STATUSES = "'queued', 'running', 'done', 'failed'"
FORGE_TYPES = "'github', 'gitlab', 'gitea', 'forgejo', 'bitbucket'"
def upgrade() -> None:
# Operator approvals: drop NOT NULL on approved_by_agent_id + add the actor label.
with op.batch_alter_table("approvals", schema=None) as batch_op:
batch_op.alter_column(
"approved_by_agent_id", existing_type=sa.BigInteger(), nullable=True
)
batch_op.add_column(sa.Column("actor", sa.String()))
op.create_table(
"commands",
sa.Column("id", PortableBigInt, primary_key=True, autoincrement=True),
sa.Column("project_id", sa.String(), sa.ForeignKey("projects.id")),
sa.Column("agent_name", sa.String()),
sa.Column("type", sa.String(), nullable=False),
sa.Column("payload", PortableJSON),
sa.Column("status", sa.String(), nullable=False, server_default="queued"),
sa.Column("result", PortableJSON),
sa.Column("error", sa.String()),
sa.Column("requested_by", sa.String()),
sa.Column("claimed_by", sa.String()),
sa.Column("created_at", PortableTimestamp, nullable=False, server_default=sa.func.now()),
sa.Column("claimed_at", PortableTimestamp),
sa.Column("finished_at", PortableTimestamp),
sa.CheckConstraint(f"type IN ({COMMAND_TYPES})", name="ck_commands_type"),
sa.CheckConstraint(f"status IN ({COMMAND_STATUSES})", name="ck_commands_status"),
)
op.create_index("ix_commands_status_id", "commands", ["status", "id"])
op.create_table(
"forge_hosts",
sa.Column("hostname", sa.String(), primary_key=True),
sa.Column("forge_type", sa.String(), nullable=False),
sa.Column("token_env_var", sa.String()),
sa.Column("base_url", sa.String()),
sa.Column("created_at", PortableTimestamp, nullable=False, server_default=sa.func.now()),
sa.CheckConstraint(f"forge_type IN ({FORGE_TYPES})", name="ck_forge_hosts_type"),
)
def downgrade() -> None:
op.drop_table("forge_hosts")
op.drop_index("ix_commands_status_id", table_name="commands")
op.drop_table("commands")
with op.batch_alter_table("approvals", schema=None) as batch_op:
batch_op.drop_column("actor")
batch_op.alter_column(
"approved_by_agent_id", existing_type=sa.BigInteger(), nullable=False
)
+14 -18
View File
@@ -1,8 +1,9 @@
"""Answer + resume routes, including the mocked control seam."""
"""Answer + resume routes. Resume now enqueues a command for the control worker (the tmux
session lives in the control container), so we assert on the queued command, not an
in-process seam call."""
from __future__ import annotations
from handler.control import spawn
from handler.db import repository as repo
from handler.db.engine import get_engine
@@ -43,34 +44,29 @@ def test_answer_with_no_open_question_is_404(client, auth, env):
assert r.status_code == 404
def test_resume_calls_control_seam(client, auth, env, monkeypatch):
def test_resume_enqueues_command_with_the_answer(client, auth, env):
_seed_agent_with_question(env)
client.post(
"/projects/proj/agents/api/answer", json={"answer": "Postgres"}, headers=auth
)
calls = []
def fake_resume(agent, answer):
calls.append((agent["name"], answer))
return True, "delivered"
monkeypatch.setattr(spawn, "resume", fake_resume)
r = client.post("/projects/proj/agents/api/resume", json={}, headers=auth)
assert r.status_code == 200
assert r.json()["resumed"] is True
assert calls == [("api", "Postgres")]
assert r.status_code == 202
body = r.json()
assert body["type"] == "resume"
assert body["agent_name"] == "api"
assert body["status"] == "queued"
# The API resolves the stored answer and hands it to the worker via the payload.
assert body["payload"]["answer"] == "Postgres"
with get_engine().begin() as conn:
a = repo.get_agent_by_name(conn, "proj", "api")
assert a["status"] == "working"
commands = repo.list_commands(conn, project_id="proj")
assert [c["type"] for c in commands] == ["resume"]
def test_resume_without_answer_is_400(client, auth, env, monkeypatch):
def test_resume_without_answer_is_400(client, auth, env):
with get_engine().begin() as conn:
repo.create_project(conn, "proj", "/tmp/proj")
repo.create_agent(conn, "proj", "api", "/tmp/proj/api")
monkeypatch.setattr(spawn, "resume", lambda a, ans: (True, "x"))
r = client.post("/projects/proj/agents/api/resume", json={}, headers=auth)
assert r.status_code == 400
+155
View File
@@ -0,0 +1,155 @@
"""Web-management API surface: project/host CRUD, enqueue endpoints, and admin gating.
The conftest sets AUTH_TOKEN=test-token and SHARED_CONTEXT_WRITE_TOKEN=shared-token with no
ADMIN_TOKEN, so the effective admin token is the global test-token. The shared-token is a
valid-but-not-admin bearer: it passes require_auth (reads) but not require_admin (writes),
which is exactly what we use to prove the gate."""
from __future__ import annotations
import pytest
@pytest.fixture
def lowpriv(env):
"""A valid bearer that is NOT the admin token (the shared-context write token)."""
return {"Authorization": f"Bearer {env['shared_token']}"}
def _mk_project(client, auth, pid="proj"):
return client.post("/projects", json={"id": pid, "root_dir": "/tmp/proj"}, headers=auth)
# --- project CRUD ---------------------------------------------------------------------
def test_get_update_delete_project(client, auth):
_mk_project(client, auth)
assert client.get("/projects/proj", headers=auth).json()["id"] == "proj"
remote = "https://github.com/me/p.git"
r = client.patch("/projects/proj", json={"git_remote": remote}, headers=auth)
assert r.status_code == 200 and r.json()["git_remote"] == remote
assert client.delete("/projects/proj", headers=auth).status_code == 200
assert client.get("/projects/proj", headers=auth).status_code == 404
def test_credential_ref_cmd_scheme_rejected(client, auth):
r = client.post(
"/projects",
json={"id": "x", "root_dir": "/tmp/x", "credential_ref": "cmd:cat /etc/passwd"},
headers=auth,
)
assert r.status_code == 422
# env:/file:/db: are accepted.
ok = client.post(
"/projects",
json={"id": "y", "root_dir": "/tmp/y", "credential_ref": "env:TOK"},
headers=auth,
)
assert ok.status_code == 201
def test_patch_project_cmd_scheme_rejected(client, auth):
_mk_project(client, auth)
r = client.patch("/projects/proj", json={"credential_ref": "cmd:whoami"}, headers=auth)
assert r.status_code == 422
# --- admin gating ---------------------------------------------------------------------
def test_reads_allowed_but_writes_need_admin(client, auth, lowpriv):
_mk_project(client, auth)
# low-priv token can read...
assert client.get("/projects", headers=lowpriv).status_code == 200
assert client.get("/hosts", headers=lowpriv).status_code == 200
# ...but not perform admin actions.
patch = client.patch("/projects/proj", json={"root_dir": "/x"}, headers=lowpriv)
assert patch.status_code == 403
assert client.delete("/projects/proj", headers=lowpriv).status_code == 403
host = client.post("/hosts", json={"hostname": "h", "forge_type": "gitea"}, headers=lowpriv)
assert host.status_code == 403
spawn = client.post("/projects/proj/agents/spawn", json={"name": "j"}, headers=lowpriv)
assert spawn.status_code == 403
# --- enqueue endpoints ----------------------------------------------------------------
def test_spawn_enqueues_command(client, auth):
_mk_project(client, auth)
r = client.post(
"/projects/proj/agents/spawn",
json={"name": "junior", "role": "junior", "worktree": "feat/x", "task": "do it"},
headers=auth,
)
assert r.status_code == 202
body = r.json()
assert body["type"] == "spawn" and body["status"] == "queued"
assert body["agent_name"] == "junior"
assert body["payload"]["role"] == "junior" and body["payload"]["worktree"] == "feat/x"
# visible on the commands feed
assert any(c["id"] == body["id"] for c in client.get("/commands", headers=auth).json())
def test_kill_enqueues_command(client, auth):
_mk_project(client, auth)
client.post(
"/projects/proj/agents",
json={"name": "api", "working_dir": "/tmp/proj/api"},
headers=auth,
)
r = client.post("/projects/proj/agents/api/kill", headers=auth)
assert r.status_code == 202 and r.json()["type"] == "kill"
def test_approval_enqueues_correct_command_type(client, auth):
_mk_project(client, auth)
r = client.post(
"/projects/proj/approvals",
json={"branch": "feat/x", "status": "rejected", "note": "nit"},
headers=auth,
)
assert r.status_code == 202
# verdict 'rejected' maps to command type 'reject'
assert r.json()["type"] == "reject"
assert r.json()["payload"]["branch"] == "feat/x"
def test_forge_init_and_poll_ci_enqueue(client, auth):
_mk_project(client, auth)
assert client.post("/projects/proj/forge-init", headers=auth).json()["type"] == "forge_init"
assert client.post("/projects/proj/poll-ci", headers=auth).json()["type"] == "poll_ci"
assert client.post("/poll-ci", headers=auth).json()["project_id"] is None
def test_command_status_polling(client, auth):
_mk_project(client, auth)
cmd = client.post("/projects/proj/poll-ci", headers=auth).json()
got = client.get(f"/commands/{cmd['id']}", headers=auth)
assert got.status_code == 200 and got.json()["id"] == cmd["id"]
assert client.get("/commands/999999", headers=auth).status_code == 404
# --- hosts ----------------------------------------------------------------------------
def test_host_crud(client, auth):
r = client.post(
"/hosts",
json={"hostname": "git.corp", "forge_type": "gitea", "token_env_var": "GITEA_TOKEN"},
headers=auth,
)
assert r.status_code == 201
assert client.get("/hosts/git.corp", headers=auth).json()["token_env_var"] == "GITEA_TOKEN"
patch = client.patch("/hosts/git.corp", json={"base_url": "https://git.corp"}, headers=auth)
assert patch.status_code == 200
assert client.delete("/hosts/git.corp", headers=auth).status_code == 200
assert client.get("/hosts/git.corp", headers=auth).status_code == 404
def test_host_bad_forge_type_422(client, auth):
r = client.post("/hosts", json={"hostname": "h", "forge_type": "svn"}, headers=auth)
assert r.status_code == 422
+53
View File
@@ -0,0 +1,53 @@
"""Host-aware credential resolution: the forge_hosts registry overrides the built-in map
when a connection is supplied, and behaviour is unchanged when it isn't (regression)."""
from __future__ import annotations
from handler.control import credentials
from handler.db import repository as repo
from handler.db.engine import get_engine
def test_registry_host_overrides_builtin_env_var(env):
with get_engine().begin() as conn:
repo.create_host(conn, "github.com", "github", token_env_var="CORP_GH_TOKEN")
e = credentials.credential_env("tok", "https://github.com/me/repo.git", conn)
# Registry wins over the built-in GITHUB_TOKEN mapping.
assert e["CORP_GH_TOKEN"] == "tok"
assert e["FORGE_TOKEN"] == "tok"
assert "GITHUB_TOKEN" not in e
def test_registry_enables_self_hosted_host(env):
with get_engine().begin() as conn:
repo.create_host(conn, "git.corp.internal", "gitea", token_env_var="CORP_TOKEN")
e = credentials.credential_env("tok", "https://git.corp.internal/me/repo.git", conn)
assert e["CORP_TOKEN"] == "tok"
def test_fallback_to_builtin_when_no_row(env):
with get_engine().begin() as conn:
e = credentials.credential_env("tok", "https://github.com/me/repo.git", conn)
# No forge_hosts row -> built-in map still applies.
assert e["GITHUB_TOKEN"] == "tok"
def test_no_conn_behaviour_is_unchanged():
# The 2-arg form (no registry) must match the pre-existing built-in behaviour.
e = credentials.credential_env("tok", "https://github.com/me/repo.git")
assert e == {"FORGE_TOKEN": "tok", "GITHUB_TOKEN": "tok"}
def test_credential_config_uses_registry_base_url(env):
with get_engine().begin() as conn:
repo.create_host(conn, "git.corp", "gitea", base_url="https://git.corp:8443")
key, value = credentials.git_credential_config("https://git.corp/me/repo.git", conn)
assert key == "credential.https://git.corp:8443.helper"
assert "$FORGE_TOKEN" in value
def test_db_scheme_is_reserved_not_yet_resolvable():
import pytest
with pytest.raises(credentials.CredentialError, match="reserved"):
credentials.resolve("db:42")
+60
View File
@@ -0,0 +1,60 @@
"""End-to-end web management: the dashboard's HTTP calls -> command queue -> worker ->
real ``spawn.spawn`` -> tmux seam. Proves the full container-split flow works with only the
tmux/claude boundary faked, not the control layer itself."""
from __future__ import annotations
from handler.control import worker
from handler.db import repository as repo
from handler.db.engine import get_engine
def _spawnable_project(root):
root.mkdir(parents=True, exist_ok=True)
(root / ".mise.toml").write_text("[tasks.test]\nrun = 'pytest'\n")
with get_engine().begin() as conn:
repo.create_project(conn, "proj", str(root))
def test_spawn_via_api_then_worker_creates_agent_and_session(client, auth, env, fake_tmux):
_spawnable_project(env["tmp"] / "proj")
# 1. The dashboard enqueues a spawn (202 + a queued command).
r = client.post(
"/projects/proj/agents/spawn",
json={"name": "api", "task": "build the thing"},
headers=auth,
)
assert r.status_code == 202
command_id = r.json()["id"]
assert r.json()["status"] == "queued"
# No agent yet — the worker hasn't run.
assert client.get("/projects/proj/agents", headers=auth).json() == []
# 2. The control worker drains the queue (runs the real spawn.spawn).
assert worker.drain("test-worker") == 1
# 3. The command is done and the agent + tmux session now exist.
got = client.get(f"/commands/{command_id}", headers=auth).json()
assert got["status"] == "done"
assert got["result"]["name"] == "api"
agents = client.get("/projects/proj/agents", headers=auth).json()
assert [a["name"] for a in agents] == ["api"]
assert fake_tmux["calls"]["new_session"][0]["name"] == "proj__api"
def test_kill_via_api_then_worker(client, auth, env, fake_tmux):
_spawnable_project(env["tmp"] / "proj")
client.post("/projects/proj/agents/spawn", json={"name": "api"}, headers=auth)
worker.drain("w")
r = client.post("/projects/proj/agents/api/kill", headers=auth)
assert r.status_code == 202
worker.drain("w")
assert client.get(f"/commands/{r.json()['id']}", headers=auth).json()["status"] == "done"
assert "proj__api" in fake_tmux["calls"]["kill_session"]
with get_engine().begin() as conn:
assert repo.get_agent_by_name(conn, "proj", "api")["status"] == "done"
+85
View File
@@ -0,0 +1,85 @@
"""DAL for web management: project/agent mutation, the command queue, and hosts."""
from __future__ import annotations
from handler.db import repository as repo
def test_update_and_delete_project(conn):
repo.create_project(conn, "p", "/tmp/p", git_remote="https://github.com/me/p.git")
updated = repo.update_project(conn, "p", git_remote="https://gitea.corp/me/p.git",
credential_ref="env:TOK")
assert updated["git_remote"] == "https://gitea.corp/me/p.git"
assert updated["credential_ref"] == "env:TOK"
# An unknown field is ignored, not applied.
repo.update_project(conn, "p", nonsense="x")
assert repo.delete_project(conn, "p") is True
assert repo.get_project(conn, "p") is None
def test_delete_agent_row(conn):
repo.create_project(conn, "p", "/tmp/p")
repo.create_agent(conn, "p", "api", "/tmp/p/api")
assert repo.delete_agent(conn, "p", "api") is True
assert repo.get_agent_by_name(conn, "p", "api") is None
def test_enqueue_get_and_list_command(conn):
repo.create_project(conn, "p", "/tmp/p")
cmd = repo.enqueue_command(
conn, "spawn", project_id="p", agent_name="junior",
payload={"role": "junior"}, requested_by="operator:web",
)
assert cmd["status"] == "queued"
assert cmd["type"] == "spawn"
assert cmd["payload"] == {"role": "junior"}
assert repo.get_command(conn, cmd["id"])["agent_name"] == "junior"
assert [c["id"] for c in repo.list_commands(conn, project_id="p")] == [cmd["id"]]
def test_claim_is_atomic_and_fifo(conn):
repo.enqueue_command(conn, "poll_ci")
second = repo.enqueue_command(conn, "poll_ci")
first_claim = repo.claim_next_command(conn, "worker-1")
assert first_claim["status"] == "running"
assert first_claim["claimed_by"] == "worker-1"
# Oldest-first: the second claim gets the later row, never the same one twice.
second_claim = repo.claim_next_command(conn, "worker-2")
assert second_claim["id"] == second["id"]
assert second_claim["id"] != first_claim["id"]
# Queue drained -> None.
assert repo.claim_next_command(conn, "worker-3") is None
def test_finish_command_records_result(conn):
cmd = repo.enqueue_command(conn, "poll_ci")
repo.claim_next_command(conn, "w")
repo.finish_command(conn, cmd["id"], "done", result={"checked": 3})
done = repo.get_command(conn, cmd["id"])
assert done["status"] == "done"
assert done["result"] == {"checked": 3}
assert done["finished_at"] is not None
def test_hosts_crud(conn):
created = repo.create_host(conn, "git.corp", "gitea", token_env_var="GITEA_TOKEN")
assert created["forge_type"] == "gitea"
assert repo.get_host(conn, "git.corp")["token_env_var"] == "GITEA_TOKEN"
repo.update_host(conn, "git.corp", base_url="https://git.corp")
assert repo.get_host(conn, "git.corp")["base_url"] == "https://git.corp"
assert [h["hostname"] for h in repo.list_hosts(conn)] == ["git.corp"]
assert repo.delete_host(conn, "git.corp") is True
assert repo.get_host(conn, "git.corp") is None
def test_operator_approval_has_no_agent_and_lists(conn):
repo.create_project(conn, "p", "/tmp/p")
ap = repo.record_approval(conn, "p", "feat/x", "approved", actor="operator:web")
assert ap["approved_by_agent_id"] is None
assert ap["actor"] == "operator:web"
listed = repo.list_approvals(conn, "p")
assert [a["id"] for a in listed] == [ap["id"]]
assert repo.list_approvals(conn, "p", branch="other") == []
+126
View File
@@ -0,0 +1,126 @@
"""The control worker: command dispatch + the claim/finish plumbing.
Uses the same mock seams as the CLI tests (tmux/gitops/forge/spawn.resume) plus direct
monkeypatching of spawn.spawn/kill so we exercise the worker's routing, not the full spawn
machinery (already covered by test_control_spawn)."""
from __future__ import annotations
from handler.control import poller, spawn, worker
from handler.db import repository as repo
from handler.db.engine import get_engine
def _seed_project(agent=None):
with get_engine().begin() as conn:
repo.create_project(conn, "p", "/tmp/p")
if agent:
repo.create_agent(conn, "p", agent, f"/tmp/p/{agent}", status="paused_for_input")
def _enqueue(**kw):
with get_engine().begin() as conn:
return repo.enqueue_command(conn, **kw)
def _get(cmd_id):
with get_engine().begin() as conn:
return repo.get_command(conn, cmd_id)
def test_spawn_command_calls_spawn_and_records_result(env, monkeypatch):
_seed_project()
calls = {}
def fake_spawn(project_id, name, **kw):
calls.update(project_id=project_id, name=name, **kw)
return {"id": 42, "name": name, "working_dir": "/tmp/p/j", "forge_note": None}
monkeypatch.setattr(spawn, "spawn", fake_spawn)
cmd = _enqueue(
type="spawn", project_id="p", agent_name="j",
payload={"role": "junior", "worktree": "feat/x"},
)
assert worker.drain("w") == 1
done = _get(cmd["id"])
assert done["status"] == "done"
assert done["result"]["agent_id"] == 42
assert calls["project_id"] == "p" and calls["name"] == "j"
assert calls["role"] == "junior" and calls["worktree_branch"] == "feat/x"
def test_kill_command_calls_kill(env, monkeypatch):
_seed_project("api")
killed = {}
monkeypatch.setattr(spawn, "kill", lambda p, n: killed.update(project=p, name=n))
cmd = _enqueue(type="kill", project_id="p", agent_name="api")
worker.drain("w")
assert _get(cmd["id"])["status"] == "done"
assert killed == {"project": "p", "name": "api"}
def test_resume_command_feeds_answer_and_sets_working(env, monkeypatch):
_seed_project("api")
seen = {}
def fake_resume(agent, ans):
seen.update(name=agent["name"], ans=ans)
return True, "ok"
monkeypatch.setattr(spawn, "resume", fake_resume)
cmd = _enqueue(type="resume", project_id="p", agent_name="api", payload={"answer": "Postgres"})
worker.drain("w")
assert _get(cmd["id"])["status"] == "done"
assert seen == {"name": "api", "ans": "Postgres"}
with get_engine().begin() as conn:
assert repo.get_agent_by_name(conn, "p", "api")["status"] == "working"
def test_approve_command_records_operator_verdict_with_head_sha(env, fake_gitops):
_seed_project("senior")
cmd = _enqueue(
type="approve", project_id="p", agent_name="senior", payload={"branch": "feat/x"}
)
worker.drain("w")
assert _get(cmd["id"])["status"] == "done"
with get_engine().begin() as conn:
ap = repo.get_latest_approval(conn, "p", "feat/x")
assert ap["status"] == "approved"
assert ap["actor"] == "operator:web"
assert ap["approved_by_agent_id"] is None
assert ap["approved_sha"] == fake_gitops["sha"] # read from the agent's working dir
def test_poll_ci_command_returns_summary(env, monkeypatch):
_seed_project()
summary = {"checked": 0, "resolved": 0, "pending": 0}
monkeypatch.setattr(poller, "sweep", lambda project_id=None: summary)
cmd = _enqueue(type="poll_ci", project_id="p")
worker.drain("w")
done = _get(cmd["id"])
assert done["status"] == "done"
assert done["result"] == {"checked": 0, "resolved": 0, "pending": 0}
def test_bad_command_is_recorded_failed_not_raised(env):
# spawn with no agent name -> CommandError -> the worker records 'failed', keeps going.
_seed_project()
cmd = _enqueue(type="spawn", project_id="p")
assert worker.drain("w") == 1
failed = _get(cmd["id"])
assert failed["status"] == "failed"
assert "agent name" in failed["error"]
def test_drain_processes_multiple_then_stops(env, monkeypatch):
_seed_project()
monkeypatch.setattr(poller, "sweep", lambda project_id=None: {"checked": 0})
_enqueue(type="poll_ci", project_id="p")
_enqueue(type="poll_ci", project_id="p")
assert worker.drain("w") == 2
assert worker.drain("w") == 0 # queue now empty