mirror of
https://github.com/blw1138/cross-py-builder.git
synced 2026-09-07 21:41:09 -05:00
Brett Williams
18d8d2e398
Replace the CROSS_PY_WORKERS env list with a DB-backed worker store so agents can be added/removed (and their online status verified) from the browser. - db.py: add workers table + list_workers/add_worker/remove_worker. - workers.py: configured_workers() reads the DB instead of settings. - app.py: POST /api/workers (add, probes reachability) and DELETE /api/workers/<port>/<host>; startup log uses DB worker count. - UI: Workers card with add form, per-worker live status + OS/CPU, remove. - settings.py: drop CROSS_PY_WORKERS/_get_json_list; DB is single source. - README/DESIGN updated to describe UI-managed workers.
146 lines
3.8 KiB
Python
146 lines
3.8 KiB
Python
import json
|
|
import sqlite3
|
|
import threading
|
|
|
|
import ctrl.settings as settings
|
|
|
|
|
|
_local = threading.local()
|
|
|
|
|
|
def _conn():
|
|
conn = getattr(_local, "conn", None)
|
|
if conn is None:
|
|
conn = sqlite3.connect(settings.DB_PATH, timeout=30)
|
|
conn.row_factory = sqlite3.Row
|
|
_local.conn = conn
|
|
return conn
|
|
|
|
|
|
def init_db():
|
|
settings.ensure_dirs()
|
|
conn = _conn()
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS jobs (
|
|
id TEXT PRIMARY KEY,
|
|
status TEXT NOT NULL,
|
|
source_type TEXT NOT NULL,
|
|
source TEXT,
|
|
os_req TEXT,
|
|
cpu_req TEXT,
|
|
worker_host TEXT,
|
|
worker_job_id TEXT,
|
|
worker_url TEXT,
|
|
created_at TEXT NOT NULL,
|
|
started_at TEXT,
|
|
finished_at TEXT,
|
|
log TEXT DEFAULT '',
|
|
error TEXT,
|
|
artifacts TEXT DEFAULT '[]'
|
|
)
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS workers (
|
|
host TEXT NOT NULL,
|
|
port INTEGER NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
PRIMARY KEY (host, port)
|
|
)
|
|
"""
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
# ---- workers -------------------------------------------------------------
|
|
|
|
def list_workers():
|
|
rows = _conn().execute("SELECT host, port FROM workers ORDER BY host, port").fetchall()
|
|
return [{"host": r["host"], "port": r["port"]} for r in rows]
|
|
|
|
|
|
def add_worker(host, port):
|
|
import datetime
|
|
conn = _conn()
|
|
conn.execute(
|
|
"INSERT OR IGNORE INTO workers (host, port, created_at) VALUES (?, ?, ?)",
|
|
(host, int(port), datetime.datetime.now().isoformat()),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def remove_worker(host, port):
|
|
conn = _conn()
|
|
conn.execute("DELETE FROM workers WHERE host = ? AND port = ?", (host, int(port)))
|
|
conn.commit()
|
|
|
|
|
|
def create_job(source_type, source, os_req, cpu_req):
|
|
import uuid
|
|
import datetime
|
|
job_id = "JOB-" + uuid.uuid4().hex[:8]
|
|
conn = _conn()
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO jobs (id, status, source_type, source, os_req, cpu_req, created_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(job_id, "queued", source_type, source, os_req, cpu_req,
|
|
datetime.datetime.now().isoformat()),
|
|
)
|
|
conn.commit()
|
|
return job_id
|
|
|
|
|
|
def _row_to_dict(row):
|
|
d = dict(row)
|
|
try:
|
|
d["artifacts"] = json.loads(d.get("artifacts") or "[]")
|
|
except (TypeError, ValueError):
|
|
d["artifacts"] = []
|
|
return d
|
|
|
|
|
|
def get_job(job_id):
|
|
row = _conn().execute("SELECT * FROM jobs WHERE id = ?", (job_id,)).fetchone()
|
|
return _row_to_dict(row) if row else None
|
|
|
|
|
|
def list_jobs(limit=100):
|
|
rows = _conn().execute(
|
|
"SELECT * FROM jobs ORDER BY created_at DESC LIMIT ?", (limit,)
|
|
).fetchall()
|
|
return [_row_to_dict(r) for r in rows]
|
|
|
|
|
|
def update_job(job_id, **fields):
|
|
if not fields:
|
|
return
|
|
cols = ", ".join(f"{k} = ?" for k in fields)
|
|
values = []
|
|
for v in fields.values():
|
|
if isinstance(v, (list, dict)):
|
|
v = json.dumps(v)
|
|
values.append(v)
|
|
conn = _conn()
|
|
conn.execute(f"UPDATE jobs SET {cols} WHERE id = ?", (*values, job_id))
|
|
conn.commit()
|
|
|
|
|
|
def append_log(job_id, text):
|
|
conn = _conn()
|
|
conn.execute("UPDATE jobs SET log = log || ? WHERE id = ?", (text, job_id))
|
|
conn.commit()
|
|
|
|
|
|
def set_queued_jobs_for_worker(worker_url):
|
|
"""Reschedule any jobs currently stuck on a worker back to queued."""
|
|
conn = _conn()
|
|
conn.execute(
|
|
"UPDATE jobs SET status = 'queued', worker_url = NULL WHERE worker_url = ? AND status IN ('dispatching', 'building')",
|
|
(worker_url,),
|
|
)
|
|
conn.commit()
|