302 lines
18 KiB
Python
302 lines
18 KiB
Python
"""SQLite durable worker with leases, fencing, checkpointed steps and cancellation."""
|
|
import json
|
|
import os
|
|
import queue
|
|
import threading
|
|
import time
|
|
from datetime import datetime, timezone, timedelta
|
|
from copy import deepcopy
|
|
|
|
from services.assistant_store import get, put, listing, event, uid, now, digest, finish_time_event
|
|
from services.assistant_data import read_cards
|
|
from services.skill_executor import execute_skill, execute_rules
|
|
from services.assistant_experiences import apply_experiences
|
|
from services import assistant_model
|
|
from contracts.project_assistant import SkillResult, ProjectSnapshot
|
|
|
|
LEASE_SECONDS = 30
|
|
MODEL_SLOTS = threading.BoundedSemaphore(2)
|
|
|
|
|
|
def authorized(identity, project_id):
|
|
configured = os.getenv("XINGBANG_AUTH_USERS", "")
|
|
if identity.get("preview"):
|
|
return not configured and os.getenv("XINGBANG_REQUIRE_AUTH") != "1"
|
|
if identity.get("demo"):
|
|
return os.getenv("XINGBANG_DEMO_ROLE_LOGIN") == "1" and identity.get("role") in ("manager", "reviewer", "admin") and project_id in identity.get("projects", [])
|
|
try:
|
|
account = next((u for u in json.loads(configured or "[]") if u.get("id") == identity["id"]), None)
|
|
except ValueError:
|
|
return False
|
|
return bool(account and account.get("role") in ("manager", "reviewer", "admin") and ("*" in account.get("projects", []) or project_id in account.get("projects", [])) and set(identity.get("teams", [])).issubset(account.get("teams", [])))
|
|
|
|
|
|
def reconcile(db, project_id, run_id, checks):
|
|
prior = {f["code"]: f for f in listing(db, "finding", project_id)}
|
|
changes = {"resolved": [], "reopened": [], "retained": [], "new": []}
|
|
for check in checks:
|
|
old = prior.get(check["code"])
|
|
if not old and check["result"] in ("pass", "not_applicable"):
|
|
continue
|
|
finding = deepcopy(old) if old else {"id": uid("finding"), "project_id": project_id, "first_seen_run_id": run_id, "history": [], "status": "待补件", "human_status": None}
|
|
previous_status = finding["status"]
|
|
basis_changed = bool(old and old.get("basis_ids") != check.get("basis_ids"))
|
|
status = previous_status
|
|
if check["result"] == "pass" and check.get("evidence_ids"):
|
|
status = "待复核" if basis_changed else "已解决"
|
|
elif check["result"] in ("issue", "unknown", "error"):
|
|
if previous_status == "已解决" or basis_changed:
|
|
status = "待复核"
|
|
elif previous_status != "不适用":
|
|
status = "待补件" if check["result"] == "unknown" else previous_status
|
|
finding.update(deepcopy(check), last_seen_run_id=run_id, status=status, machine_result=check["result"], stale=False)
|
|
finding["history"].append({"time": now(), "run_id": run_id, "result": check["result"], "from": previous_status, "to": status, "basis_ids": check.get("basis_ids", []), "evidence_ids": check.get("evidence_ids", []), "type": "machine"})
|
|
put(db, "finding", finding)
|
|
key = "new" if not old else "resolved" if status == "已解决" and previous_status != status else "reopened" if previous_status == "已解决" and status != previous_status else "retained"
|
|
changes[key].append(finding["id"])
|
|
# Unexecuted checks remain unresolved/stale, never disappear into an implied pass.
|
|
executed = {c["code"] for c in checks}
|
|
for code, old in prior.items():
|
|
if code not in executed:
|
|
old["stale"] = True
|
|
put(db, "finding", old)
|
|
changes["retained"].append(old["id"])
|
|
return listing(db, "finding", project_id), changes
|
|
|
|
|
|
class DurableWorker:
|
|
def __init__(self, service):
|
|
self.service = service
|
|
self.owner = uid("worker")
|
|
self.stopped = threading.Event()
|
|
self.thread = None
|
|
|
|
def start(self):
|
|
self.thread = threading.Thread(target=self.loop, name="xingbang-durable-worker", daemon=True)
|
|
self.thread.start()
|
|
return self
|
|
|
|
def stop(self):
|
|
self.stopped.set()
|
|
if self.thread:
|
|
self.thread.join(timeout=2)
|
|
|
|
def loop(self):
|
|
while not self.stopped.is_set():
|
|
try:
|
|
worked = self.process_next()
|
|
except Exception:
|
|
worked = False
|
|
if not worked:
|
|
self.stopped.wait(.5)
|
|
|
|
def claim(self):
|
|
with self.service.connect() as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
row = db.execute("SELECT id FROM assistant_jobs WHERE state='queued' OR (state='running' AND lease_until<?) ORDER BY rowid LIMIT 1", (time.time(),)).fetchone()
|
|
if not row:
|
|
return None
|
|
run = get(db, "run", row[0])
|
|
attempts = db.execute("SELECT attempts FROM assistant_jobs WHERE id=?", (row[0],)).fetchone()[0]
|
|
if attempts >= 3:
|
|
run.update(status="failed", error="工作进程恢复次数超过上限，请创建新运行", ended_at=now())
|
|
put(db, "run", run)
|
|
db.execute("UPDATE assistant_jobs SET state='failed' WHERE id=?", (row[0],))
|
|
return None
|
|
db.execute("UPDATE assistant_jobs SET state='running',owner=?,lease_until=?,attempts=attempts+1 WHERE id=?", (self.owner, time.time() + LEASE_SECONDS, run["id"]))
|
|
run.update(status="running", started_at=run.get("started_at", now()), attempt=attempts + 1)
|
|
put(db, "run", run)
|
|
event(db, run["project_id"], "run.claimed", self.owner, {"attempt": attempts + 1}, run["id"])
|
|
return run
|
|
|
|
def valid(self, db, run_id):
|
|
row = db.execute("SELECT state,owner,lease_until FROM assistant_jobs WHERE id=?", (run_id,)).fetchone()
|
|
return bool(row and row[0] == "running" and row[1] == self.owner and row[2] > time.time() and not self.stopped.is_set())
|
|
|
|
def alive(self, run):
|
|
with self.service.connect() as db:
|
|
return self.valid(db, run["id"]) and authorized(run["identity"], run["project_id"])
|
|
|
|
def heartbeat(self, run, done):
|
|
while not done.wait(5) and not self.stopped.is_set():
|
|
with self.service.connect() as db:
|
|
db.execute("UPDATE assistant_jobs SET lease_until=? WHERE id=? AND owner=? AND state='running'", (time.time() + LEASE_SECONDS, run["id"], self.owner))
|
|
|
|
def save_step(self, run, step):
|
|
if step["status"] != "running":
|
|
step = SkillResult.model_validate(step).model_dump()
|
|
with self.service.connect() as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
if not self.valid(db, run["id"]) or not authorized(run["identity"], run["project_id"]):
|
|
raise InterruptedError("运行已取消或工作租约失效")
|
|
current = get(db, "run", run["id"])
|
|
current["steps"] = [s for s in current["steps"] if s["step_id"] != step["step_id"]] + [step]
|
|
run["steps"] = current["steps"]
|
|
put(db, "run", current)
|
|
event(db, run["project_id"], "step." + step["status"], self.owner, {"step_id": step["step_id"], "name": step["name"], "attempt": step["attempt"]}, run["id"])
|
|
|
|
def bounded_model(self, function, timeout, run):
|
|
if not MODEL_SLOTS.acquire(blocking=False):
|
|
raise TimeoutError("模型并发预算已占用")
|
|
result = queue.Queue(maxsize=1)
|
|
def call():
|
|
try:
|
|
result.put((True, function()))
|
|
except Exception as exc:
|
|
result.put((False, exc))
|
|
finally:
|
|
MODEL_SLOTS.release()
|
|
threading.Thread(target=call, daemon=True, name="xingbang-scoped-model").start()
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
if not self.alive(run):
|
|
raise InterruptedError("运行取消或权限失效")
|
|
try:
|
|
ok, value = result.get(timeout=min(.5, max(.01, deadline - time.monotonic())))
|
|
if not ok:
|
|
raise value
|
|
return value
|
|
except queue.Empty:
|
|
continue
|
|
raise TimeoutError("模型步骤超过时间预算")
|
|
|
|
def process_next(self):
|
|
run = self.claim()
|
|
if not run:
|
|
return False
|
|
done = threading.Event()
|
|
heartbeat_thread = threading.Thread(target=self.heartbeat, args=(run, done), daemon=True)
|
|
heartbeat_thread.start()
|
|
try:
|
|
if not authorized(run["identity"], run["project_id"]):
|
|
raise PermissionError("项目权限已撤销")
|
|
self.execute(run)
|
|
except Exception as exc:
|
|
with self.service.connect() as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
if self.valid(db, run["id"]):
|
|
current = get(db, "run", run["id"])
|
|
current.update(status="failed", ended_at=now(), error="运行未完成：" + type(exc).__name__)
|
|
put(db, "run", current)
|
|
db.execute("UPDATE assistant_jobs SET state='failed' WHERE id=? AND owner=?", (run["id"], self.owner))
|
|
finish_time_event(db, "run", run["id"])
|
|
event(db, run["project_id"], "run.failed", self.owner, {"error_type": type(exc).__name__}, run["id"])
|
|
finally:
|
|
done.set()
|
|
heartbeat_thread.join(timeout=2)
|
|
return True
|
|
|
|
def execute(self, run):
|
|
components = run["components"]
|
|
with self.service.connect() as db:
|
|
original = get(db, "input_snapshot", run["input_snapshot_id"])
|
|
if run.get("snapshot_id"):
|
|
snapshot = get(db, "snapshot", run["snapshot_id"])
|
|
else:
|
|
snapshot = None
|
|
if snapshot is None:
|
|
data_project = dict(original, id=run["project_id"])
|
|
values, sources, datasets, errors = read_cards(data_project, components)
|
|
snapshot = dict(original, id=uid("snapshot"), values=values, sources=sources, datasets=datasets, data_errors=errors)
|
|
snapshot = ProjectSnapshot.model_validate(snapshot).model_dump()
|
|
snapshot["content_hash"] = digest(snapshot)
|
|
with self.service.connect() as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
if not self.valid(db, run["id"]) or not authorized(run["identity"], run["project_id"]):
|
|
raise InterruptedError()
|
|
put(db, "snapshot", snapshot, immutable=True)
|
|
run["snapshot_id"] = snapshot["id"]
|
|
current = get(db, "run", run["id"])
|
|
current["snapshot_id"] = snapshot["id"]
|
|
# Freeze applicable experience versions once; recovery never reads a newer release.
|
|
current["experience_versions"] = [get(db, "experience_version", r["version_id"]) for r in listing(db, "experience_release")] if run.get("feature_flags", {}).get("experiences", True) else []
|
|
put(db, "run", current)
|
|
run.update(current)
|
|
values, sources, all_checks = deepcopy(snapshot["values"]), deepcopy(snapshot["sources"]), []
|
|
budget = components["plan"].get("budget", {"timeout_seconds": 120, "max_tool_calls": 12})
|
|
deadline = time.monotonic() + budget["timeout_seconds"]
|
|
experience_checks, applications, guidance = apply_experiences(run.get("experience_versions", []), snapshot, run["identity"], run["id"])
|
|
steps = [s for s in components["skills"] if s["engine"] != "model"] + [dict(r, engine="rules") for r in components["rules"]]
|
|
steps += [dict(id="experience.applications", name="已发布经验核查", version="1.0.0", engine="experience", content_hash=digest(guidance))]
|
|
steps += [s for s in components["skills"] if s["engine"] == "model"]
|
|
required_fields = {f for s in components["skills"] if s["engine"] == "required" for f in s.get("inputs", [])}
|
|
for skill in steps:
|
|
if not self.alive(run):
|
|
raise PermissionError("运行取消或项目权限失效")
|
|
key = skill["id"]
|
|
cached = next((s for s in run["steps"] if s["step_id"] == key and s["status"] in ("completed", "partial")), None)
|
|
if cached:
|
|
all_checks.extend(cached["findings"])
|
|
values.update(cached["outputs"])
|
|
sources.update(cached.get("output_sources", {}))
|
|
continue
|
|
step = {"run_id": run["id"], "step_id": key, "name": skill["name"], "skill": {"id": key, "version": skill["version"]},
|
|
"status": "running", "input_snapshot_id": snapshot["id"], "input_hash": digest({"values": values, "component": skill.get("content_hash")}),
|
|
"started_at": now(), "attempt": run["attempt"], "outputs": {}, "findings": [], "errors": [], "missing_inputs": [], "output_sources": {}}
|
|
self.save_step(run, step)
|
|
try:
|
|
if time.monotonic() >= deadline:
|
|
raise TimeoutError("运行时间预算已耗尽")
|
|
engine = skill["engine"]
|
|
if engine == "model":
|
|
payload, trace = self.bounded_model(lambda: assistant_model.invoke(skill, dict(snapshot, values=values, sources=sources), all_checks, components["rules"], run["requirement"], guidance,
|
|
run["model_config"], budget, cancelled=lambda: not self.alive(run) or time.monotonic() >= deadline, feedback=run.get("feedback_context")), max(.1, deadline - time.monotonic()), run)
|
|
step["outputs"] = {"requirement_response": payload["requirement_response"]}
|
|
if payload.get("experience_proposal"):
|
|
step["outputs"]["experience_proposal"] = payload["experience_proposal"]
|
|
step["findings"] = [dict(c, code=key + "." + c["code"], basis_ids=[key + "@" + skill["version"]], severity="attention") for c in payload["findings"]]
|
|
step["trace"] = trace
|
|
elif engine == "experience":
|
|
step["findings"] = experience_checks
|
|
elif engine == "rules":
|
|
step["findings"] = execute_rules(skill, values, sources)
|
|
else:
|
|
step["outputs"], step["findings"] = execute_skill(skill, values, sources)
|
|
if engine == "quality":
|
|
step["findings"] = [c for c in step["findings"] if not set(c["fields"]).issubset(required_fields)]
|
|
if engine == "ratio":
|
|
step["output_sources"] = {skill["output"]: sources.get(skill["numerator"], []) + sources.get(skill["denominator"], [])}
|
|
step["missing_inputs"] = [f for f in skill.get("inputs", []) if f != "*" and values.get(f) is None]
|
|
step["status"] = "partial" if any(c["result"] in ("unknown", "error") for c in step["findings"]) else "completed"
|
|
except Exception as exc:
|
|
step.update(status="failed", errors=[{"type": type(exc).__name__, "message": str(exc)[:400] if isinstance(exc, (ValueError, TimeoutError)) else "步骤执行失败，请检查内网模型或数据服务"}])
|
|
step["ended_at"] = now()
|
|
self.save_step(run, step)
|
|
values.update(step["outputs"])
|
|
sources.update(step["output_sources"])
|
|
all_checks.extend(step["findings"])
|
|
with self.service.connect() as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
if not self.valid(db, run["id"]) or not authorized(run["identity"], run["project_id"]):
|
|
raise PermissionError("发布前权限或租约已失效")
|
|
project = get(db, "project", run["project_id"])
|
|
current = get(db, "run", run["id"])
|
|
current["superseded"] = project["data_revision"] != snapshot["data_revision"]
|
|
if run.get("feedback_context") and not current["superseded"]:
|
|
proposal = next((s["outputs"].get("experience_proposal") for s in current["steps"] if s["status"] in ("completed", "partial") and s["outputs"].get("experience_proposal")), None)
|
|
if proposal:
|
|
candidate = self.service.save_experience(run["project_id"], run["actor_id"], {**proposal, "feedback_id": run["feedback_context"]["id"], "scope": "project", "expires_at": (datetime.now(timezone.utc) + timedelta(days=90)).isoformat()}, db=db)
|
|
current["proposed_experience_id"] = candidate["id"]
|
|
if not current["superseded"]:
|
|
findings, changes = reconcile(db, run["project_id"], run["id"], all_checks)
|
|
else:
|
|
findings, changes = [], {"reason": "运行期间项目已更新，未覆盖当前问题状态"}
|
|
cited = {key for check in all_checks for key in check.get("evidence_ids", [])}
|
|
document_refs = [{"id": chunk["evidence_id"], "project_id": run["project_id"], "snapshot_id": snapshot["id"], "document_id": document["id"],
|
|
"source": document["name"], "locator": chunk["locator"], "field": "document.text", "value": chunk["text"][:500], "version": document["version"]}
|
|
for document in snapshot["documents"] for chunk in document["chunks"] if chunk.get("evidence_id") in cited]
|
|
current.update(status="partial" if snapshot.get("data_errors") or current["superseded"] or any(s["status"] != "completed" for s in current["steps"]) else "completed",
|
|
ended_at=now(), findings=findings, checks=all_checks, change_summary=changes, data_errors=snapshot.get("data_errors", []),
|
|
evidence=[dict(r, snapshot_id=snapshot["id"]) for refs in snapshot["sources"].values() for r in refs] + document_refs,
|
|
summary={"coverage": values.get("metrics.coverage"), "open": sum(f["status"] not in ("已解决", "不适用") for f in findings), "resolved": sum(f["status"] == "已解决" for f in findings)})
|
|
put(db, "run", current)
|
|
for application in applications:
|
|
put(db, "experience_application", application, immutable=True)
|
|
db.execute("UPDATE assistant_jobs SET state=? WHERE id=? AND owner=?", (current["status"], run["id"], self.owner))
|
|
finish_time_event(db, "run", run["id"])
|
|
self.service.mark_stale(db, run["project_id"], ["*"])
|
|
# A new result invalidates existing drafts, not the findings just reconciled.
|
|
for finding in findings:
|
|
put(db, "finding", finding)
|
|
event(db, run["project_id"], "run." + current["status"], self.owner, {"summary": current["summary"], "superseded": current["superseded"]}, run["id"])
|