Compare commits
12
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
da67ec9714 | ||
|
|
2e93ed904a | ||
|
|
b901fa3a5a | ||
|
|
b9778f1f69 | ||
|
|
41c4c4f003 | ||
|
|
d119c2d87c | ||
|
|
0a143c6b98 | ||
|
|
7ef76b739f | ||
|
|
a46878f9b8 | ||
|
|
f73c7946d0 | ||
|
|
ecd7ec40dd | ||
|
|
82bf1bd118 |
@@ -0,0 +1,77 @@
|
||||
"""Actions API for unified Agent Event Bus."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
try:
|
||||
from fastapi import APIRouter
|
||||
except Exception:
|
||||
APIRouter = None
|
||||
|
||||
from .agent_event_bus import AgentEventBus
|
||||
|
||||
bus = AgentEventBus()
|
||||
|
||||
|
||||
def create_agent_event(payload: dict[str, Any]) -> dict[str, Any]:
|
||||
return {"status": "ok", "event": bus.publish(**payload)}
|
||||
|
||||
|
||||
def list_agent_events(kind: str | None = None, status: str = "pending") -> dict[str, Any]:
|
||||
return {"status": "ok", "events": bus.list_events(kind, status)}
|
||||
|
||||
|
||||
def get_agent_event(event_id: str) -> dict[str, Any]:
|
||||
return {"status": "ok", "event": bus.get(event_id)}
|
||||
|
||||
|
||||
def transition_agent_event(event_id: str, status: str, actor: str = "system") -> dict[str, Any]:
|
||||
return {"status": "ok", "event": bus.transition(event_id, status, actor)}
|
||||
|
||||
|
||||
def agent_event_stats() -> dict[str, Any]:
|
||||
return {"status": "ok", "stats": bus.stats()}
|
||||
|
||||
|
||||
# Compatibility adapters
|
||||
|
||||
def list_gpt_inbox_messages():
|
||||
return list_agent_events(kind="gpt_inbox")
|
||||
|
||||
|
||||
def ack_gpt_inbox_message(event_id: str):
|
||||
return transition_agent_event(event_id, "completed", "custom-gpt")
|
||||
|
||||
|
||||
def build_router():
|
||||
if APIRouter is None:
|
||||
return None
|
||||
router = APIRouter(prefix="/agent-events", tags=["agent-events"])
|
||||
|
||||
@router.post("", operation_id="createAgentEvent")
|
||||
def create(payload: dict[str, Any]):
|
||||
return create_agent_event(payload)
|
||||
|
||||
@router.get("", operation_id="listAgentEvents")
|
||||
def listing(kind: str | None = None, status: str = "pending"):
|
||||
return list_agent_events(kind, status)
|
||||
|
||||
@router.get("/{event_id}", operation_id="getAgentEvent")
|
||||
def get(event_id: str):
|
||||
return get_agent_event(event_id)
|
||||
|
||||
@router.post("/{event_id}/transition", operation_id="transitionAgentEvent")
|
||||
def transition(event_id: str, payload: dict[str, Any]):
|
||||
return transition_agent_event(event_id, payload["status"], payload.get("actor", "system"))
|
||||
|
||||
@router.get("/stats/overview", operation_id="getAgentEventStats")
|
||||
def stats():
|
||||
return agent_event_stats()
|
||||
|
||||
return router
|
||||
|
||||
|
||||
router = build_router()
|
||||
|
||||
__all__ = ["router", "create_agent_event", "list_agent_events", "get_agent_event", "transition_agent_event"]
|
||||
@@ -0,0 +1,125 @@
|
||||
"""Unified Agent Event Bus.
|
||||
|
||||
Unifies commands, GPT messages, replies, approvals, audit and reports into one
|
||||
state machine. Compatibility layers can filter events by kind instead of keeping
|
||||
separate queues.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import asdict, dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
from hashlib import sha256
|
||||
from pathlib import Path
|
||||
import json
|
||||
from typing import Any
|
||||
|
||||
|
||||
def utc_now() -> str:
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
|
||||
@dataclass
|
||||
class AgentEvent:
|
||||
id: str
|
||||
kind: str
|
||||
source: str
|
||||
target: str
|
||||
summary: str
|
||||
payload: dict[str, Any]
|
||||
status: str = "pending"
|
||||
correlation_id: str | None = None
|
||||
created_at: str = field(default_factory=utc_now)
|
||||
updated_at: str = field(default_factory=utc_now)
|
||||
actor: str | None = None
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return asdict(self)
|
||||
|
||||
|
||||
class AgentEventBus:
|
||||
STATES = ("pending", "running", "completed", "failed")
|
||||
|
||||
def __init__(self, root: str = "agents/events"):
|
||||
self.root = Path(root)
|
||||
for state in self.STATES:
|
||||
(self.root / state).mkdir(parents=True, exist_ok=True)
|
||||
(self.root / "audit").mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def publish(
|
||||
self,
|
||||
kind: str,
|
||||
source: str,
|
||||
target: str,
|
||||
summary: str,
|
||||
payload: dict[str, Any],
|
||||
correlation_id: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
event_id = "evt_" + sha256(
|
||||
json.dumps(
|
||||
{"kind": kind, "source": source, "summary": summary, "payload": payload},
|
||||
sort_keys=True,
|
||||
ensure_ascii=False,
|
||||
).encode("utf-8")
|
||||
).hexdigest()[:16]
|
||||
event = AgentEvent(event_id, kind, source, target, summary, payload, correlation_id=correlation_id)
|
||||
self._write("pending", event.to_dict())
|
||||
self._audit("created", event.to_dict())
|
||||
return event.to_dict()
|
||||
|
||||
def list_events(self, kind: str | None = None, status: str = "pending", limit: int = 100) -> list[dict[str, Any]]:
|
||||
result = []
|
||||
for path in sorted((self.root / status).glob("*.json")):
|
||||
event = self._read(path)
|
||||
if kind and event.get("kind") != kind:
|
||||
continue
|
||||
result.append(event)
|
||||
return result[:limit]
|
||||
|
||||
def get(self, event_id: str) -> dict[str, Any] | None:
|
||||
for state in self.STATES:
|
||||
path = self.root / state / f"{event_id}.json"
|
||||
if path.exists():
|
||||
return self._read(path)
|
||||
return None
|
||||
|
||||
def transition(self, event_id: str, status: str, actor: str | None = None) -> dict[str, Any]:
|
||||
event = self.get(event_id)
|
||||
if not event:
|
||||
return {"status": "not_found", "id": event_id}
|
||||
if status not in self.STATES:
|
||||
return {"status": "invalid_state"}
|
||||
old = event.get("status", "pending")
|
||||
old_path = self.root / old / f"{event_id}.json"
|
||||
event["status"] = status
|
||||
event["actor"] = actor
|
||||
event["updated_at"] = utc_now()
|
||||
self._write(status, event)
|
||||
if old_path.exists() and old != status:
|
||||
old_path.unlink()
|
||||
self._audit("transition", event, {"from": old, "to": status})
|
||||
return event
|
||||
|
||||
def stats(self) -> dict[str, Any]:
|
||||
return {
|
||||
"events": {state: len(list((self.root / state).glob("*.json"))) for state in self.STATES},
|
||||
"audit_events": len(list((self.root / "audit").glob("*.json"))),
|
||||
"status": "healthy",
|
||||
}
|
||||
|
||||
def _write(self, state: str, data: dict[str, Any]):
|
||||
(self.root / state / f"{data['id']}.json").write_text(
|
||||
json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8"
|
||||
)
|
||||
|
||||
def _read(self, path: Path):
|
||||
return json.loads(path.read_text(encoding="utf-8"))
|
||||
|
||||
def _audit(self, action: str, event: dict[str, Any], extra: dict[str, Any] | None = None):
|
||||
data = {"action": action, "event": event, "extra": extra or {}, "time": utc_now()}
|
||||
(self.root / "audit" / f"{event['id']}-{action}.json").write_text(
|
||||
json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8"
|
||||
)
|
||||
|
||||
|
||||
__all__ = ["AgentEvent", "AgentEventBus"]
|
||||
@@ -0,0 +1,50 @@
|
||||
"""Admin dashboard data model for the unified Agent Event Bus."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import Counter
|
||||
from typing import Any
|
||||
|
||||
from .agent_event_bus import AgentEventBus
|
||||
|
||||
|
||||
class AgentEventDashboard:
|
||||
def __init__(self, bus: AgentEventBus | None = None):
|
||||
self.bus = bus or AgentEventBus()
|
||||
|
||||
def overview(self) -> dict[str, Any]:
|
||||
stats = self.bus.stats()
|
||||
pending = self.bus.list_events(status="pending")
|
||||
running = self.bus.list_events(status="running")
|
||||
completed = self.bus.list_events(status="completed")
|
||||
failed = self.bus.list_events(status="failed")
|
||||
all_visible = pending + running + completed + failed
|
||||
by_kind = Counter(event.get("kind", "unknown") for event in all_visible)
|
||||
by_target = Counter(event.get("target", "unknown") for event in all_visible)
|
||||
return {
|
||||
"module": "agent-event-bus",
|
||||
"title": "Unified Agent Event Bus",
|
||||
"route": "/admin/ai-evolution/events",
|
||||
"metrics": {
|
||||
**stats,
|
||||
"by_kind": dict(by_kind),
|
||||
"by_target": dict(by_target),
|
||||
"gpt_inbox_pending": by_kind.get("gpt_inbox", 0),
|
||||
"openclaw_pending": len([e for e in pending if e.get("target") == "openclaw"]),
|
||||
},
|
||||
"panels": [
|
||||
"event timeline",
|
||||
"pending commands",
|
||||
"GPT inbox",
|
||||
"agent replies",
|
||||
"failure reports",
|
||||
"audit trail",
|
||||
],
|
||||
"audit": {
|
||||
"enabled": True,
|
||||
"location": "agents/events/audit",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
__all__ = ["AgentEventDashboard"]
|
||||
@@ -0,0 +1,328 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
AIOS Autonomous Controller
|
||||
|
||||
Detect -> Diagnose -> Fix -> Test -> Patch -> Approve -> Commit -> Notify
|
||||
|
||||
Runs alongside the command watcher. Checks system health periodically.
|
||||
When an issue is found, creates a patch in agents/patches/pending/
|
||||
with a structured request for owner approval.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import hashlib
|
||||
import time
|
||||
import sys
|
||||
import logging
|
||||
import subprocess
|
||||
import requests
|
||||
from pathlib import Path
|
||||
from datetime import datetime, timezone, timedelta
|
||||
|
||||
WORKSPACE = Path(os.environ.get("AIOS_WORKSPACE", "/data/aios-gitops/workspace"))
|
||||
ACTIVE_DIR = WORKSPACE / "agents/commands/active/"
|
||||
RUNNING_DIR = WORKSPACE / "agents/commands/running/"
|
||||
COMPLETED_DIR = WORKSPACE / "agents/commands/completed/"
|
||||
FAILED_DIR = WORKSPACE / "agents/commands/failed/"
|
||||
PATCHES_PENDING_DIR = WORKSPACE / "agents/patches/pending/"
|
||||
PATCHES_APPLIED_DIR = WORKSPACE / "agents/patches/applied/"
|
||||
PATCHES_REJECTED_DIR = WORKSPACE / "agents/patches/rejected/"
|
||||
AUDIT_DIR = WORKSPACE / "audit/controller-runs/"
|
||||
STATE_FILE = WORKSPACE / "agents/state/controller-state.json"
|
||||
HEALTH_FILE = WORKSPACE / "agents/state/health.json"
|
||||
POLL_INTERVAL = int(os.environ.get("CONTROLLER_POLL_INTERVAL_SECONDS", "300"))
|
||||
GITEA_URL = os.environ.get("GITEA_URL", "http://127.0.0.1:3000")
|
||||
API_URL = os.environ.get("AIOS_API_URL", "http://127.0.0.1:18890")
|
||||
GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "")
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s [controller] %(levelname)s %(message)s",
|
||||
datefmt="%Y-%m-%dT%H:%M:%S%z",
|
||||
)
|
||||
log = logging.getLogger("autonomous-controller")
|
||||
|
||||
CONTROLLER_VERSION = "1.0.0"
|
||||
|
||||
|
||||
def load_state():
|
||||
if STATE_FILE.exists():
|
||||
try:
|
||||
return json.loads(STATE_FILE.read_text())
|
||||
except (json.JSONDecodeError, OSError):
|
||||
pass
|
||||
return {"controller_version": CONTROLLER_VERSION, "checks": {}, "patches": {}}
|
||||
|
||||
|
||||
def save_state(state):
|
||||
STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||||
STATE_FILE.write_text(json.dumps(state, indent=2, ensure_ascii=False))
|
||||
|
||||
|
||||
def write_audit_record(check_name, status, details):
|
||||
record = {
|
||||
"controller_run_id": f"controller-{int(time.time())}",
|
||||
"check": check_name,
|
||||
"status": status,
|
||||
"controller_version": CONTROLLER_VERSION,
|
||||
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
|
||||
"details": details,
|
||||
}
|
||||
AUDIT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
af = AUDIT_DIR / f"{record['controller_run_id']}.json"
|
||||
af.write_text(json.dumps(record, indent=2, ensure_ascii=False))
|
||||
return record
|
||||
|
||||
|
||||
def create_patch(check_name, diagnosis, proposed_fix, files_to_change):
|
||||
"""Create a structured patch request for owner approval."""
|
||||
patch_id = f"patch-{int(time.time())}"
|
||||
patch = {
|
||||
"patch_id": patch_id,
|
||||
"check": check_name,
|
||||
"diagnosis": diagnosis,
|
||||
"proposed_fix": proposed_fix,
|
||||
"files_to_change": files_to_change,
|
||||
"created_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
|
||||
"status": "pending",
|
||||
"approved": False,
|
||||
"applied": False,
|
||||
}
|
||||
PATCHES_PENDING_DIR.mkdir(parents=True, exist_ok=True)
|
||||
pf = PATCHES_PENDING_DIR / f"{patch_id}.json"
|
||||
pf.write_text(json.dumps(patch, indent=2, ensure_ascii=False))
|
||||
log.info(f"Created patch: {patch_id}")
|
||||
return patch
|
||||
|
||||
|
||||
def get_gitea_version():
|
||||
try:
|
||||
r = requests.get(f"{GITEA_URL}/api/v1/version", timeout=5)
|
||||
if r.status_code == 200:
|
||||
return {"healthy": True, "data": r.json()}
|
||||
return {"healthy": False, "error": f"HTTP {r.status_code}"}
|
||||
except Exception as e:
|
||||
return {"healthy": False, "error": str(e)}
|
||||
|
||||
|
||||
def get_gitea_repo_info(owner="aios", repo="aios-v23"):
|
||||
try:
|
||||
branches = []
|
||||
page = 1
|
||||
while True:
|
||||
r = requests.get(
|
||||
f"{GITEA_URL}/api/v1/repos/{owner}/{repo}/branches",
|
||||
params={"page": page, "limit": 50},
|
||||
headers={"Authorization": f"token {GITEA_TOKEN}"},
|
||||
timeout=10
|
||||
)
|
||||
if r.status_code != 200:
|
||||
break
|
||||
data = r.json()
|
||||
if not data:
|
||||
break
|
||||
branches.extend(data)
|
||||
page += 1
|
||||
tags = []
|
||||
page = 1
|
||||
while True:
|
||||
r = requests.get(
|
||||
f"{GITEA_URL}/api/v1/repos/{owner}/{repo}/tags",
|
||||
params={"page": page, "limit": 50},
|
||||
headers={"Authorization": f"token {GITEA_TOKEN}"},
|
||||
timeout=10
|
||||
)
|
||||
if r.status_code != 200:
|
||||
break
|
||||
data = r.json()
|
||||
if not data:
|
||||
break
|
||||
tags.extend(data)
|
||||
page += 1
|
||||
return {
|
||||
"healthy": True,
|
||||
"branches": len(branches),
|
||||
"tags": len(tags),
|
||||
"branch_list": [b["name"] for b in branches],
|
||||
"tag_list": [t["name"] for t in tags],
|
||||
}
|
||||
except Exception as e:
|
||||
return {"healthy": False, "error": str(e)}
|
||||
|
||||
|
||||
def get_api_baseline():
|
||||
try:
|
||||
r = requests.get(f"{API_URL}/admin/ai-evolution/api/data", timeout=5)
|
||||
if r.status_code == 200:
|
||||
data = r.json()
|
||||
return {"healthy": True, "baseline": data.get("baseline")}
|
||||
return {"healthy": False, "error": f"HTTP {r.status_code}"}
|
||||
except Exception as e:
|
||||
return {"healthy": False, "error": str(e)}
|
||||
|
||||
|
||||
def check_barerepo_branch_sync():
|
||||
bare_repo = Path("/data/aios-gitops/runtime/gitea-bin/data/repositories/aios/aios-v23.git")
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", "--git-dir", str(bare_repo), "branch", "-a"],
|
||||
capture_output=True, text=True, timeout=10
|
||||
)
|
||||
if result.returncode != 0:
|
||||
return {"healthy": False, "error": result.stderr[:200]}
|
||||
bare_branches = [
|
||||
l.strip().replace("remotes/origin/", "").replace("* ", "")
|
||||
for l in result.stdout.strip().split("\n") if l.strip()
|
||||
]
|
||||
bare_set = set(bare_branches)
|
||||
repo_info = get_gitea_repo_info()
|
||||
if not repo_info["healthy"]:
|
||||
return {"healthy": False, "error": repo_info.get("error")}
|
||||
gitea_set = set(repo_info["branch_list"])
|
||||
missing_in_gitea = bare_set - gitea_set
|
||||
extra_in_gitea = gitea_set - bare_set
|
||||
return {
|
||||
"healthy": len(missing_in_gitea) == 0,
|
||||
"bare_count": len(bare_set),
|
||||
"gitea_count": len(gitea_set),
|
||||
"missing_in_gitea": list(missing_in_gitea),
|
||||
"extra_in_gitea": list(extra_in_gitea),
|
||||
}
|
||||
except Exception as e:
|
||||
return {"healthy": False, "error": str(e)}
|
||||
|
||||
|
||||
def check_stuck_commands():
|
||||
stuck = []
|
||||
threshold_minutes = 30
|
||||
if RUNNING_DIR.exists():
|
||||
now = time.time()
|
||||
for cmd_file in RUNNING_DIR.iterdir():
|
||||
if cmd_file.name.endswith(".md"):
|
||||
mtime = cmd_file.stat().st_mtime
|
||||
age_minutes = (now - mtime) / 60
|
||||
if age_minutes > threshold_minutes:
|
||||
stuck.append({
|
||||
"name": cmd_file.name,
|
||||
"age_minutes": round(age_minutes, 1),
|
||||
"path": str(cmd_file),
|
||||
})
|
||||
return {"healthy": len(stuck) == 0, "stuck_count": len(stuck), "stuck_commands": stuck}
|
||||
|
||||
|
||||
|
||||
def run_all_checks():
|
||||
"""Run all health checks and return results."""
|
||||
state = load_state()
|
||||
results = {}
|
||||
patches_created = []
|
||||
|
||||
# 1. Gitea version
|
||||
log.info("Check: Gitea API health")
|
||||
results["gitea_version"] = get_gitea_version()
|
||||
write_audit_record("gitea_version", "ok" if results["gitea_version"]["healthy"] else "fail", results["gitea_version"])
|
||||
|
||||
# 2. API baseline
|
||||
log.info("Check: API baseline")
|
||||
results["api_baseline"] = get_api_baseline()
|
||||
write_audit_record("api_baseline", "ok" if results["api_baseline"]["healthy"] else "fail", results["api_baseline"])
|
||||
if results["api_baseline"]["healthy"]:
|
||||
log.info(f" Baseline: {results['api_baseline'].get('baseline')}")
|
||||
|
||||
# 3. Bare repo branch sync
|
||||
log.info("Check: bare repo / Gitea branch sync")
|
||||
results["branch_sync"] = check_barerepo_branch_sync()
|
||||
status = "ok" if results["branch_sync"]["healthy"] else "degraded"
|
||||
write_audit_record("branch_sync", status, results["branch_sync"])
|
||||
missing = results["branch_sync"].get("missing_in_gitea", [])
|
||||
if missing:
|
||||
log.warning(f" {len(missing)} branches missing in Gitea DB")
|
||||
for b in missing:
|
||||
log.warning(f" -> {b}")
|
||||
# Auto-create a patch if branches are out of sync
|
||||
patch = create_patch(
|
||||
"branch_sync",
|
||||
f"{len(missing)} branches present in bare repo but not in Gitea Web UI",
|
||||
"Run SQL INSERT to sync branches: INSERT INTO branch (repo_id, name, commit_id, ...) "
|
||||
"for each missing branch. Verify with bare repo SHA.",
|
||||
["runtime/gitea-bin/data/... (SQL via gitea CLI or direct DB)"]
|
||||
)
|
||||
patches_created.append(patch)
|
||||
|
||||
# 4. Stuck commands
|
||||
log.info("Check: stuck commands")
|
||||
results["stuck_commands"] = check_stuck_commands()
|
||||
status = "ok" if results["stuck_commands"]["healthy"] else "degraded"
|
||||
write_audit_record("stuck_commands", status, results["stuck_commands"])
|
||||
stuck = results["stuck_commands"].get("stuck_commands", [])
|
||||
if stuck:
|
||||
log.warning(f" {len(stuck)} commands stuck in running/")
|
||||
for c in stuck:
|
||||
log.warning(f" -> {c['name']} (age: {c['age_minutes']}m)")
|
||||
|
||||
# Update state
|
||||
state["checks"] = {
|
||||
"last_run": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
|
||||
"results": {k: {"healthy": v.get("healthy"), "error": v.get("error")} for k, v in results.items()},
|
||||
"patches_created": len(patches_created),
|
||||
}
|
||||
save_state(state)
|
||||
|
||||
# Write health summary
|
||||
all_healthy = all(v.get("healthy", False) for v in results.values())
|
||||
summary = {
|
||||
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
|
||||
"controller_version": CONTROLLER_VERSION,
|
||||
"all_healthy": all_healthy,
|
||||
"checks": {k: v.get("healthy") for k, v in results.items()},
|
||||
"patches_pending": len(patches_created),
|
||||
}
|
||||
HEALTH_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||||
HEALTH_FILE.write_text(json.dumps(summary, indent=2, ensure_ascii=False))
|
||||
|
||||
return results, patches_created
|
||||
|
||||
|
||||
def run_forever():
|
||||
log.info(f"Autonomous Controller v{CONTROLLER_VERSION} started")
|
||||
log.info(f"Poll interval: {POLL_INTERVAL}s")
|
||||
log.info(f"Workspace: {WORKSPACE}")
|
||||
log.info(f"Gitea: {GITEA_URL}, API: {API_URL}")
|
||||
|
||||
while True:
|
||||
try:
|
||||
log.info("--- Health check cycle ---")
|
||||
results, patches = run_all_checks()
|
||||
healthy_count = sum(1 for v in results.values() if v.get("healthy"))
|
||||
total = len(results)
|
||||
log.info(f"Results: {healthy_count}/{total} healthy")
|
||||
if patches:
|
||||
log.warning(f"Created {len(patches)} patch(es) for owner review")
|
||||
else:
|
||||
log.info("No issues detected requiring patches")
|
||||
except Exception as e:
|
||||
log.exception(f"Check cycle failed: {e}")
|
||||
|
||||
time.sleep(POLL_INTERVAL)
|
||||
|
||||
|
||||
def run_once():
|
||||
log.info("Running one-shot health check")
|
||||
results, patches = run_all_checks()
|
||||
for name, result in results.items():
|
||||
status = "OK" if result.get("healthy") else "FAIL"
|
||||
log.info(f" [{status}] {name}")
|
||||
if not result.get("healthy") and result.get("error"):
|
||||
log.info(f" error: {result['error'][:200]}")
|
||||
if patches:
|
||||
log.info(f"Patches created: {len(patches)}")
|
||||
else:
|
||||
log.info("No patches created")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) > 1 and sys.argv[1] == "--once":
|
||||
run_once()
|
||||
else:
|
||||
run_forever()
|
||||
@@ -0,0 +1,10 @@
|
||||
{
|
||||
"agent_run_id": "openclaw-20260708-003",
|
||||
"agent": "openclaw",
|
||||
"command": "CMD-20260708-enable-aios-autonomous-operation-and-notification",
|
||||
"start_time": "2026-07-08T15:00:00+08:00",
|
||||
"end_time": "2026-07-08T16:20:00+08:00",
|
||||
"final_result": "success",
|
||||
"interaction_count": 75,
|
||||
"summary": "Initialized autonomous operation mode: Git command watcher running, Web UI dashboard live, audit records created. Pending: WeChat notification system integration."
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
{"owner_approved": true, "version": "V1R4C0-test"}
|
||||
@@ -0,0 +1,35 @@
|
||||
agent_event_bus:
|
||||
enabled: true
|
||||
replacement_for:
|
||||
- agents/commands
|
||||
- agents/messages
|
||||
storage:
|
||||
root: agents/events
|
||||
states:
|
||||
- pending
|
||||
- running
|
||||
- completed
|
||||
- failed
|
||||
audit: agents/events/audit
|
||||
supported_kinds:
|
||||
- command
|
||||
- agent_status
|
||||
- gpt_inbox
|
||||
- gpt_reply
|
||||
- audit_event
|
||||
- approval_request
|
||||
- failure_report
|
||||
- daily_report
|
||||
api:
|
||||
operations:
|
||||
- createAgentEvent
|
||||
- listAgentEvents
|
||||
- getAgentEvent
|
||||
- transitionAgentEvent
|
||||
- getAgentEventStats
|
||||
dashboard:
|
||||
route: /admin/ai-evolution/events
|
||||
governance:
|
||||
secrets_in_git: false
|
||||
direct_main_push: false
|
||||
owner_gate_required: true
|
||||
@@ -0,0 +1,19 @@
|
||||
openapi: 3.0.0
|
||||
info:
|
||||
title: AIOS Unified Agent Event Bus API
|
||||
version: V1R3C1+
|
||||
paths:
|
||||
/agent-events:
|
||||
post:
|
||||
operationId: createAgentEvent
|
||||
get:
|
||||
operationId: listAgentEvents
|
||||
/agent-events/{event_id}:
|
||||
get:
|
||||
operationId: getAgentEvent
|
||||
/agent-events/{event_id}/transition:
|
||||
post:
|
||||
operationId: transitionAgentEvent
|
||||
/agent-events/stats/overview:
|
||||
get:
|
||||
operationId: getAgentEventStats
|
||||
@@ -0,0 +1,38 @@
|
||||
# CMD-20260709-enable-unified-agent-event-bus
|
||||
|
||||
## Goal
|
||||
|
||||
Replace separate command/message queues with the unified Agent Event Bus.
|
||||
|
||||
## Deploy
|
||||
|
||||
1. Deploy `active-runtime/app/agent_event_bus.py`.
|
||||
2. Enable Event API:
|
||||
- createAgentEvent
|
||||
- listAgentEvents
|
||||
- getAgentEvent
|
||||
- transitionAgentEvent
|
||||
- getAgentEventStats
|
||||
3. Switch Web UI admin evolution page to:
|
||||
- `/admin/ai-evolution/events`
|
||||
4. Enable audit path:
|
||||
- `agents/events/audit`
|
||||
5. Keep compatibility adapters for old GPT inbox calls.
|
||||
|
||||
## Validation signals
|
||||
|
||||
- AIOS_AGENT_EVENT_BUS_READY
|
||||
- AIOS_AGENT_EVENT_AUDIT_READY
|
||||
- AIOS_AGENT_EVENT_DASHBOARD_READY
|
||||
|
||||
## Migration
|
||||
|
||||
- Do not delete existing command/message data.
|
||||
- New events go to `agents/events`.
|
||||
- Legacy queues become compatibility views.
|
||||
|
||||
## Governance
|
||||
|
||||
- No secrets in Git.
|
||||
- No direct main push.
|
||||
- Owner gate remains enabled.
|
||||
@@ -0,0 +1,9 @@
|
||||
{
|
||||
"type": "branch",
|
||||
"branch": "patch/agent-event-bus-unified",
|
||||
"base": "main",
|
||||
"created_by": "chatgpt-gitops-bot",
|
||||
"created_at": "2026-07-08T08:36:38Z",
|
||||
"direct_main_push": false,
|
||||
"owner_approval_required": true
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
from active_runtime.app.agent_event_bus import AgentEventBus
|
||||
from active_runtime.app.agent_event_dashboard import AgentEventDashboard
|
||||
|
||||
|
||||
def test_unified_event_flow():
|
||||
bus = AgentEventBus(root="/tmp/aios-agent-events-test")
|
||||
event = bus.publish(
|
||||
kind="gpt_inbox",
|
||||
source="openclaw",
|
||||
target="custom-gpt",
|
||||
summary="execution report",
|
||||
payload={"status": "done"},
|
||||
)
|
||||
|
||||
assert event["status"] == "pending"
|
||||
assert len(bus.list_events(kind="gpt_inbox")) == 1
|
||||
|
||||
done = bus.transition(event["id"], "completed", "custom-gpt")
|
||||
assert done["status"] == "completed"
|
||||
|
||||
dashboard = AgentEventDashboard(bus).overview()
|
||||
assert dashboard["audit"]["enabled"] is True
|
||||
Reference in New Issue
Block a user