feat: 完成项目助手与监控演示并修复聊天及云部署配置
This commit is contained in:
NathanAlex7
@@ -0,0 +1,295 @@
|
||||
"""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
|
||||
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"
|
||||
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()
|
||||
threading.Thread(target=self.heartbeat, args=(run, done), daemon=True).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))
|
||||
event(db, run["project_id"], "run.failed", self.owner, {"error_type": type(exc).__name__}, run["id"])
|
||||
finally:
|
||||
done.set()
|
||||
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))
|
||||
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"])
|
||||
Reference in New Issue
Block a user