nearxos
6185b9b85a
Self-hosted, Docker-based agentic troubleshooting platform: FastAPI backend + LangGraph agent, Next.js UI, tiered LLM routing (local Ollama -> Gemini -> DeepSeek -> OpenRouter), MCP server manager, encrypted device credentials, RBAC, audit log, project-memory + Obsidian integrations, and editable troubleshooting decision rules tuned for the GeneseasX vessel stack. Co-authored-by: Cursor <cursoragent@cursor.com>
97 lines
2.9 KiB
Python
97 lines
2.9 KiB
Python
"""Task event bus: persist events to DB and publish to Redis for live streaming."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
import redis.asyncio as aioredis
|
|
|
|
from app.config import settings
|
|
from app.database import SessionLocal
|
|
from app.models.task import TaskEvent
|
|
|
|
_redis: aioredis.Redis | None = None
|
|
|
|
# Redis keys
|
|
TASK_QUEUE = "agentic:task_queue"
|
|
|
|
# High-frequency live-only event kinds: streamed to the UI but not written to
|
|
# Postgres, to avoid a separate transaction per progress/tool-start tick.
|
|
EPHEMERAL_KINDS = frozenset({"progress", "tool_start"})
|
|
|
|
|
|
def get_redis() -> aioredis.Redis:
|
|
global _redis
|
|
if _redis is None:
|
|
_redis = aioredis.from_url(
|
|
settings.redis_url, decode_responses=True, max_connections=32
|
|
)
|
|
return _redis
|
|
|
|
|
|
def _channel(task_id: int) -> str:
|
|
return f"agentic:task:{task_id}:events"
|
|
|
|
|
|
async def dedupe_task_queue(task_id: int) -> int:
|
|
"""Remove stale duplicate entries for a task id from the Redis queue."""
|
|
redis = get_redis()
|
|
needle = str(task_id)
|
|
items = await redis.lrange(TASK_QUEUE, 0, -1)
|
|
removed = sum(1 for x in items if x == needle)
|
|
if removed:
|
|
await redis.delete(TASK_QUEUE)
|
|
for item in items:
|
|
if item != needle:
|
|
await redis.rpush(TASK_QUEUE, item)
|
|
return removed
|
|
|
|
|
|
async def enqueue_task(task_id: int) -> None:
|
|
await dedupe_task_queue(task_id)
|
|
await get_redis().rpush(TASK_QUEUE, str(task_id))
|
|
|
|
|
|
async def publish_event(
|
|
task_id: int,
|
|
kind: str,
|
|
message: str | None = None,
|
|
payload: dict[str, Any] | None = None,
|
|
persist: bool | None = None,
|
|
) -> None:
|
|
"""Persist an event and publish it to the task's Redis channel.
|
|
|
|
When ``persist`` is None, high-frequency ephemeral kinds (progress,
|
|
tool_start) are streamed live but not written to the DB, while all other
|
|
kinds are persisted. Pass an explicit bool to override.
|
|
"""
|
|
if persist is None:
|
|
persist = kind not in EPHEMERAL_KINDS
|
|
if persist:
|
|
async with SessionLocal() as db:
|
|
db.add(TaskEvent(task_id=task_id, kind=kind, message=message, payload=payload))
|
|
await db.commit()
|
|
|
|
event = {
|
|
"task_id": task_id,
|
|
"kind": kind,
|
|
"message": message,
|
|
"payload": payload,
|
|
"ts": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
await get_redis().publish(_channel(task_id), json.dumps(event))
|
|
|
|
|
|
async def subscribe_events(task_id: int):
|
|
"""Async generator yielding decoded events for a task."""
|
|
pubsub = get_redis().pubsub()
|
|
await pubsub.subscribe(_channel(task_id))
|
|
try:
|
|
async for message in pubsub.listen():
|
|
if message["type"] == "message":
|
|
yield json.loads(message["data"])
|
|
finally:
|
|
await pubsub.unsubscribe(_channel(task_id))
|
|
await pubsub.close()
|