diff --git a/docs/daemon/README.md b/docs/daemon/README.md new file mode 100644 index 0000000000..a7e5173c1f --- /dev/null +++ b/docs/daemon/README.md @@ -0,0 +1,61 @@ +# OpenViking Active Daemon + +自动监听 Claude Code 会话日志，提取知识并写入 OpenViking 知识库。 + +## 快速开始 + +### 启用 Daemon + +设置环境变量： + +```bash +export OV_DAEMON_ENABLED=true +export OV_DAEMON_WATCH_DIR=~/.claude/projects # 监听目录（可选） +export OV_DAEMON_BATCH_LINES=50 # 批处理触发行数（可选） +export OV_DAEMON_BATCH_SECONDS=300 # 批处理触发秒数（可选） +``` + +### 启动 OpenViking Server + +```bash +openviking serve +``` + +Daemon 会在服务器启动时自动运行（如果 `OV_DAEMON_ENABLED=true`）。 + +## 工作原理 + +1. **文件监听** — 监控 `~/.claude/projects/` 下的 `.jsonl` 文件变化 +2. **增量读取** — 文件游标技术，只处理新增内容 +3. **批量处理** — 累积 50 行或 5 分钟后触发 ETL 管道 +4. **知识提取** — 调用 LLM 过滤噪声，提取有价值的知识 +5. **自动存储** — 写入 `viking://skills/`、`viking://memories/`、`viking://resources/` + +## 知识分类 + +| 类型 | 目标路径 | 说明 | +|------|---------|------| +| Skills | `viking://skills/claude_code/.md` | 可复用的操作指南 | +| Memories (有项目) | `viking://memories/projects/<project>/decisions.md` | 项目决策日志 | +| Memories (无项目) | `viking://memories/global/<title>.md` | 全局记忆 | +| Resources | `viking://resources/<tech>/<title>.md` | 参考资源 | + +## 架构 + +``` +Claude Code JSONL → File Watcher → Batch Buffer → Filter → Reconstruct + → LLM Extract → Deduplicate → Route → viking:// Storage +``` + +## 故障排查 + +### Daemon 未启动 +检查日志中是否有 `Active Daemon is disabled` 消息，确认 `OV_DAEMON_ENABLED=true`。 + +### 没有提取到知识 +- 确认 Claude Code 正在写入 JSONL 文件（`~/.claude/projects/` 下有 `.jsonl` 文件） +- 对话内容可能不够有价值（简单问答会被过滤） +- 查看日志中的 ETL 处理信息 + +### 知识写入失败 +检查 OpenViking ResourceService 是否正常运行。 diff --git a/docs/daemon/configuration.md b/docs/daemon/configuration.md new file mode 100644 index 0000000000..169bf341e1 --- /dev/null +++ b/docs/daemon/configuration.md @@ -0,0 +1,50 @@ +# Active Daemon 配置指南 + +## 环境变量 + +| 变量 | 说明 | 默认值 | +|------|------|--------| +| `OV_DAEMON_ENABLED` | 启用 Daemon | `false` | +| `OV_DAEMON_WATCH_DIR` | 监听目录 | `~/.claude/projects` | +| `OV_DAEMON_DB_PATH` | 游标数据库路径 | `~/.qoderworkcn/openviking/daemon_cursors.db` | +| `OV_DAEMON_BATCH_LINES` | 批处理触发行数 | `50` | +| `OV_DAEMON_BATCH_SECONDS` | 批处理触发秒数 | `300` | + +## YAML 配置 (ov.conf) + +```json +{ + "daemon": { + "enabled": true, + "watch_dir": "~/.claude/projects", + "batch_trigger_lines": 50, + "batch_trigger_seconds": 300 + } +} +``` + +## Docker 部署 + +```yaml +services: + openviking: + image: openviking:latest + environment: + - OV_DAEMON_ENABLED=true + - OV_DAEMON_WATCH_DIR=/data/claude-projects + - OV_DAEMON_DB_PATH=/data/daemon.db + volumes: + - ./claude-projects:/data/claude-projects + - ./daemon-data:/data + ports: + - "1933:1933" +``` + +## 日志 + +Daemon 使用 OpenViking 标准日志系统。关键日志： + +- `Claude Code watcher started on ...` — 监听器启动 +- `Flushing batch with N events` — 批处理触发 +- `Extracted N knowledge items` — 知识提取完成 +- `Knowledge ingested: viking://...` — 知识写入成功 diff --git a/examples/daemon/docker-compose-daemon.yml b/examples/daemon/docker-compose-daemon.yml new file mode 100644 index 0000000000..81dba7604b --- /dev/null +++ b/examples/daemon/docker-compose-daemon.yml @@ -0,0 +1,14 @@ +version: '3.8' + +services: + openviking: + image: openviking:latest + environment: + - OV_DAEMON_ENABLED=true + - OV_DAEMON_WATCH_DIR=/data/claude-projects + - OV_DAEMON_DB_PATH=/data/daemon.db + volumes: + - ./claude-projects:/data/claude-projects + - ./daemon-data:/data + ports: + - "1933:1933" diff --git a/examples/daemon/start-daemon.bat b/examples/daemon/start-daemon.bat new file mode 100644 index 0000000000..3e06e7e327 --- /dev/null +++ b/examples/daemon/start-daemon.bat @@ -0,0 +1,10 @@ +@echo off +REM Start OpenViking with Active Daemon enabled on Windows + +set OV_DAEMON_ENABLED=true +set OV_DAEMON_WATCH_DIR=%USERPROFILE%\.claude\projects +set OV_DAEMON_BATCH_LINES=50 +set OV_DAEMON_BATCH_SECONDS=300 + +echo Starting OpenViking with Active Daemon... +openviking serve %* diff --git a/examples/daemon/start-daemon.sh b/examples/daemon/start-daemon.sh new file mode 100644 index 0000000000..26129bb4f1 --- /dev/null +++ b/examples/daemon/start-daemon.sh @@ -0,0 +1,12 @@ +#!/bin/bash +# Start OpenViking with Active Daemon enabled +set -e + +echo "Starting OpenViking with Active Daemon..." + +export OV_DAEMON_ENABLED=true +export OV_DAEMON_WATCH_DIR="${OV_DAEMON_WATCH_DIR:-$HOME/.claude/projects}" +export OV_DAEMON_BATCH_LINES="${OV_DAEMON_BATCH_LINES:-50}" +export OV_DAEMON_BATCH_SECONDS="${OV_DAEMON_BATCH_SECONDS:-300}" + +openviking serve "$@" diff --git a/openviking/daemon/__init__.py b/openviking/daemon/__init__.py new file mode 100644 index 0000000000..ba94f0d050 --- /dev/null +++ b/openviking/daemon/__init__.py @@ -0,0 +1,7 @@ +""" +OpenViking Active Daemon package. +Monitors AI tool logs and automatically extracts knowledge into viking:// storage. +""" +from openviking.daemon.service import DaemonService + +__all__ = ["DaemonService"] diff --git a/openviking/daemon/conversation_reconstructor.py b/openviking/daemon/conversation_reconstructor.py new file mode 100644 index 0000000000..2dba8f6898 --- /dev/null +++ b/openviking/daemon/conversation_reconstructor.py @@ -0,0 +1,53 @@ +""" +Reconstruct conversation turns from flat event lists. +Pairs user prompts with assistant responses into structured ConversationTurn objects. +""" +from typing import Dict, List + +from openviking.daemon.models import ConversationTurn +from openviking_cli.utils.logger import get_logger + +logger = get_logger(__name__) + + +class ConversationReconstructor: + """Reconstructs paired conversation turns from chronological events.""" + + def reconstruct(self, events: List[Dict]) -> List[ConversationTurn]: + """ + Pair user prompts with assistant responses. + Events are sorted by timestamp. An assistant response is paired + with the most recent unpaired user prompt. + """ + turns: List[ConversationTurn] = [] + current_user_prompt = None + current_metadata: Dict = {} + + sorted_events = sorted(events, key=lambda e: e.get("timestamp", "")) + + for event in sorted_events: + role = event.get("role") + content = event.get("content", "") + + if role == "user": + current_user_prompt = content + current_metadata = { + "session_id": event.get("session_id"), + "project_name": event.get("project_name"), + "timestamp": event.get("timestamp"), + } + elif role == "assistant" and current_user_prompt: + turns.append( + ConversationTurn( + user_prompt=current_user_prompt, + assistant_response=content, + session_id=current_metadata.get("session_id"), + project_name=current_metadata.get("project_name"), + timestamp=current_metadata.get("timestamp"), + ) + ) + current_user_prompt = None + current_metadata = {} + + logger.info("Reconstructed %d conversation turns from %d events", len(turns), len(events)) + return turns diff --git a/openviking/daemon/cursor_manager.py b/openviking/daemon/cursor_manager.py new file mode 100644 index 0000000000..6b76bc2565 --- /dev/null +++ b/openviking/daemon/cursor_manager.py @@ -0,0 +1,92 @@ +""" +Cursor Manager for tracking file read positions. +Persists state in SQLite so Daemon can resume after restart. +""" +import sqlite3 +import time +from pathlib import Path +from typing import Dict + +from openviking.daemon.models import FileCursor +from openviking_cli.utils.logger import get_logger + +logger = get_logger(__name__) + + +class CursorManager: + """Manages file cursor state with SQLite persistence.""" + + def __init__(self, db_path: str): + self.db_path = db_path + self._init_db() + + def _init_db(self): + """Initialize the SQLite database and schema.""" + Path(self.db_path).parent.mkdir(parents=True, exist_ok=True) + + conn = sqlite3.connect(self.db_path) + try: + conn.execute(""" + CREATE TABLE IF NOT EXISTS file_cursors ( + file_path TEXT PRIMARY KEY, + last_position INTEGER NOT NULL DEFAULT 0, + last_read_time REAL NOT NULL DEFAULT 0.0, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + conn.commit() + finally: + conn.close() + + def get_cursor(self, file_path: str) -> FileCursor: + """Get the cursor state for a file. Returns zero-position cursor if not found.""" + conn = sqlite3.connect(self.db_path) + try: + row = conn.execute( + "SELECT last_position, last_read_time FROM file_cursors WHERE file_path = ?", + (file_path,), + ).fetchone() + + if row: + return FileCursor( + file_path=file_path, + last_position=row[0], + last_read_time=row[1], + ) + return FileCursor(file_path=file_path) + finally: + conn.close() + + def update_cursor(self, file_path: str, position: int): + """Update the cursor position for a file.""" + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + """ + INSERT OR REPLACE INTO file_cursors (file_path, last_position, last_read_time) + VALUES (?, ?, ?) + """, + (file_path, position, time.time()), + ) + conn.commit() + finally: + conn.close() + + def get_all_cursors(self) -> Dict[str, FileCursor]: + """Get all tracked cursor states.""" + conn = sqlite3.connect(self.db_path) + try: + rows = conn.execute( + "SELECT file_path, last_position, last_read_time FROM file_cursors" + ).fetchall() + + return { + row[0]: FileCursor( + file_path=row[0], + last_position=row[1], + last_read_time=row[2], + ) + for row in rows + } + finally: + conn.close() diff --git a/openviking/daemon/deduplicator.py b/openviking/daemon/deduplicator.py new file mode 100644 index 0000000000..9066b57bb4 --- /dev/null +++ b/openviking/daemon/deduplicator.py @@ -0,0 +1,35 @@ +""" +Knowledge deduplication based on content hashing. +Prevents duplicate knowledge from being written to viking:// storage. +""" +import hashlib +from typing import Set + +from openviking.daemon.models import ExtractedKnowledge +from openviking_cli.utils.logger import get_logger + +logger = get_logger(__name__) + + +class KnowledgeDeduplicator: + """Deduplicates knowledge items based on MD5 content hash.""" + + def __init__(self): + self.seen_hashes: Set[str] = set() + + def is_duplicate(self, knowledge: ExtractedKnowledge) -> bool: + """Check if this knowledge is a duplicate of something already seen.""" + content_hash = hashlib.md5( + knowledge.content.encode("utf-8") + ).hexdigest() + + if content_hash in self.seen_hashes: + logger.debug("Duplicate knowledge skipped: %s", knowledge.title) + return True + + self.seen_hashes.add(content_hash) + return False + + def clear(self): + """Clear the dedup cache.""" + self.seen_hashes.clear() diff --git a/openviking/daemon/etl_pipeline.py b/openviking/daemon/etl_pipeline.py new file mode 100644 index 0000000000..f601537fd9 --- /dev/null +++ b/openviking/daemon/etl_pipeline.py @@ -0,0 +1,64 @@ +""" +Batch ETL pipeline for processing conversation events into structured knowledge. +Orchestrates: Filter -> Reconstruct -> Extract -> Deduplicate +""" +import asyncio +from typing import Dict, List + +from openviking.daemon.models import ExtractedKnowledge +from openviking.daemon.filters import LowValueFilter +from openviking.daemon.conversation_reconstructor import ConversationReconstructor +from openviking.daemon.knowledge_extractor import KnowledgeExtractor +from openviking.daemon.deduplicator import KnowledgeDeduplicator +from openviking_cli.utils.logger import get_logger + +logger = get_logger(__name__) + + +class BatchETLPipeline: + """Orchestrates the full ETL flow from raw events to structured knowledge.""" + + def __init__(self, llm_service=None): + self.filter = LowValueFilter() + self.reconstructor = ConversationReconstructor() + self.extractor = KnowledgeExtractor(llm_service=llm_service) + self.deduplicator = KnowledgeDeduplicator() + + async def process_batch(self, events: List[Dict]) -> List[ExtractedKnowledge]: + """ + Process a batch of raw conversation events. + + Flow: events -> filter -> reconstruct -> extract (parallel) -> deduplicate + """ + logger.info("Processing batch with %d events", len(events)) + + # Step 1: Filter low-value content + filtered_events = self.filter.apply(events) + logger.info("After filtering: %d events", len(filtered_events)) + + if not filtered_events: + return [] + + # Step 2: Reconstruct conversation turns + turns = self.reconstructor.reconstruct(filtered_events) + logger.info("Reconstructed %d conversation turns", len(turns)) + + if not turns: + return [] + + # Step 3: Extract knowledge in parallel + tasks = [self.extractor.extract(turn) for turn in turns] + results = await asyncio.gather(*tasks, return_exceptions=True) + + # Step 4: Filter errors and deduplicate + extracted: List[ExtractedKnowledge] = [] + for result in results: + if isinstance(result, Exception): + logger.error("Extraction failed: %s", result) + continue + + if result is not None and not self.deduplicator.is_duplicate(result): + extracted.append(result) + + logger.info("Extracted %d knowledge items from batch", len(extracted)) + return extracted diff --git a/openviking/daemon/filters.py b/openviking/daemon/filters.py new file mode 100644 index 0000000000..4a3a58bbe2 --- /dev/null +++ b/openviking/daemon/filters.py @@ -0,0 +1,47 @@ +""" +Rule-based filters for low-value conversations. +Removes noise before LLM processing to save cost and improve quality. +""" +import re +from typing import Dict, List + +from openviking_cli.utils.logger import get_logger + +logger = get_logger(__name__) + + +class LowValueFilter: + """Filters out low-value conversations using regex rules.""" + + NOISE_PATTERNS = [ + r"^npm\s+(install|update|remove)", + r"^yarn\s+(add|remove)", + r"^pip\s+(install|uninstall)", + r"^git\s+(commit|push|pull|merge)", + r"^(SyntaxError|TypeError|ImportError|ModuleNotFoundError)", + r"^Retry\s+\d+/", + r"^Loading\.+", + r"^(format|indent|align)\s+(this|the)\s+code", + ] + + MIN_CONTENT_LENGTH = 20 + + def apply(self, events: List[Dict]) -> List[Dict]: + """Apply filtering rules to a list of events.""" + filtered = [] + + for event in events: + content = event.get("content", "").strip() + + # Rule 1: too short + if len(content) < self.MIN_CONTENT_LENGTH: + continue + + # Rule 2: noise pattern match + if any(re.match(p, content, re.IGNORECASE) for p in self.NOISE_PATTERNS): + continue + + filtered.append(event) + + logger.debug("Filtered %d events down to %d", len(events), len(filtered)) + return filtered diff --git a/openviking/daemon/knowledge_extractor.py b/openviking/daemon/knowledge_extractor.py new file mode 100644 index 0000000000..27d5a9ea11 --- /dev/null +++ b/openviking/daemon/knowledge_extractor.py @@ -0,0 +1,128 @@ +""" +LLM-based knowledge extraction from conversations. +Uses OpenViking's existing LLM configuration for intelligent filtering and summarization. +""" +import json +import re +from typing import Dict, Optional + +from openviking.daemon.models import ConversationTurn, ExtractedKnowledge +from openviking_cli.utils.logger import get_logger + +logger = get_logger(__name__) + +KNOWLEDGE_EXTRACTION_PROMPT = """# Role: OpenViking Automated Context Extraction Expert + +# Task: +Analyze the AI-assisted development conversation below and convert it into structured knowledge for viking:// storage. + +# Input: +User question: {user_prompt} +AI answer: {assistant_response} + +# Filtering rules: +Output <IGNORE> if the conversation is just: +- Minor syntax error fixes (missing semicolons, typos) +- Invalid commands or retry loops +- Pure code formatting or indentation changes +- Simple factual questions ("what is Python") + +Extract ONLY when the conversation contains: +- Explicit system configuration decisions +- Root cause analysis for complex bugs +- New architectural rules for the project +- Core development intent +- Reusable skills or best practices + +# Output (strict JSON, no markdown code blocks): +{{ + "status": "EXTRACTED" | "IGNORED", + "category": "skills" | "memories" | "resources", + "confidence": 0.0-1.0, + "title": "One-line summary (max 30 chars)", + "project_name": "project name or null", + "entity_links": ["tech tags", "module names"], + "content": "Concise conclusion. What was the problem, what was the solution, why this choice.", + "actionable_steps": ["steps if skills, else empty"] +}} +""" + + +class KnowledgeExtractor: + """Extracts structured knowledge from conversation turns using LLM.""" + + def __init__(self, llm_service=None): + """ + Args: + llm_service: Optional LLM service. If None, will use OpenViking's default. + """ + self._llm_service = llm_service + + def _get_llm_service(self): + """Lazy-load the LLM service from OpenViking if not provided.""" + if self._llm_service is None: + from openviking.server.dependencies import get_service + self._llm_service = get_service().llm + return self._llm_service + + async def extract(self, turn: ConversationTurn) -> Optional[ExtractedKnowledge]: + """Extract knowledge from a conversation turn. Returns None if not valuable.""" + prompt = KNOWLEDGE_EXTRACTION_PROMPT.format( + user_prompt=turn.user_prompt, + assistant_response=turn.assistant_response, + ) + + try: + response = await self._call_llm(prompt) + + if not response or response.get("status") != "EXTRACTED": + return None + + if response.get("confidence", 0) < 0.6: + return None + + return ExtractedKnowledge( + status=response["status"], + category=response["category"], + title=response.get("title", "")[:50], + content=self._clean_content(response.get("content", "")), + confidence=response.get("confidence", 0.0), + project_name=response.get("project_name"), + entity_links=response.get("entity_links", []), + actionable_steps=response.get("actionable_steps", []), + timestamp=turn.timestamp, + ) + + except Exception as e: + logger.error("Error extracting knowledge: %s", e) + return None + + async def _call_llm(self, prompt: str) -> Optional[Dict]: + """Call LLM and parse JSON response.""" + try: + llm = self._get_llm_service() + response = await llm.complete( + prompt=prompt, + temperature=0.3, + max_tokens=500, + ) + + text = response.text.strip() + # Strip markdown code block markers if present + text = re.sub(r"```json\s*", "", text) + text = re.sub(r"\s*```", "", text) + + return json.loads(text) + + except json.JSONDecodeError as e: + logger.warning("JSON parse error in LLM response: %s", e) + return None + except Exception as e: + logger.error("LLM call failed: %s", e) + return None + + @staticmethod + def _clean_content(content: str) -> str: + """Remove markdown artifacts from content.""" + content = re.sub(r"```.*?```", "", content, flags=re.DOTALL) + return content.strip() diff --git a/openviking/daemon/knowledge_router.py b/openviking/daemon/knowledge_router.py new file mode 100644 index 0000000000..d49cebbdc7 --- /dev/null +++ b/openviking/daemon/knowledge_router.py @@ -0,0 +1,54 @@ +""" +Route extracted knowledge to appropriate viking:// URIs based on category and project. +""" +import re +from typing import Optional + +from openviking.daemon.models import ExtractedKnowledge +from openviking_cli.utils.logger import get_logger + +logger = get_logger(__name__) + + +class KnowledgeRouter: + """Routes knowledge items to viking:// URIs based on category.""" + + def route(self, knowledge: ExtractedKnowledge) -> Optional[str]: + """ + Determine the target URI for a knowledge item. + + Routing rules: + - skills -> viking://skills/claude_code/<title>.md + - memories (with project) -> viking://memories/projects/<project>/decisions.md + - memories (no project) -> viking://memories/global/<title>.md + - resources -> viking://resources/<tech_stack>/<title>.md + """ + category = knowledge.category + project_name = knowledge.project_name + title = self._sanitize_filename(knowledge.title) + + if category == "skills": + return f"viking://skills/claude_code/{title}.md" + + elif category == "memories": + if project_name: + safe_project = self._sanitize_filename(project_name) + return f"viking://memories/projects/{safe_project}/decisions.md" + else: + return f"viking://memories/global/{title}.md" + + elif category == "resources": + entity_links = knowledge.entity_links + tech_stack = self._sanitize_filename(entity_links[0]) if entity_links else "general" + return f"viking://resources/{tech_stack}/{title}.md" + + else: + logger.warning("Unknown category: %s", category) + return None + + @staticmethod + def _sanitize_filename(name: str) -> str: + """Sanitize a string for use as a filename.""" + sanitized = re.sub(r'[<>:"/\\|?*]', '_', name) + sanitized = sanitized.strip().replace(' ', '_') + return sanitized[:50] diff --git a/openviking/daemon/models.py b/openviking/daemon/models.py new file mode 100644 index 0000000000..0ea3d6b5c7 --- /dev/null +++ b/openviking/daemon/models.py @@ -0,0 +1,85 @@ +""" +Data models for OpenViking Active Daemon. +""" +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + + +@dataclass +class FileCursor: + """Tracks file read position for incremental processing.""" + file_path: str + last_position: int = 0 + last_read_time: float = 0.0 + + def to_dict(self) -> Dict[str, Any]: + return { + "file_path": self.file_path, + "last_position": self.last_position, + "last_read_time": self.last_read_time, + } + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "FileCursor": + return cls( + file_path=data["file_path"], + last_position=data.get("last_position", 0), + last_read_time=data.get("last_read_time", 0.0), + ) + + +@dataclass +class BatchBuffer: + """Buffer for accumulating events before batch processing.""" + lines: List[Dict[str, Any]] = field(default_factory=list) + byte_count: int = 0 + created_at: float = 0.0 + + def add_line(self, line: Dict[str, Any], byte_size: int): + self.lines.append(line) + self.byte_count += byte_size + + def is_empty(self) -> bool: + return len(self.lines) == 0 + + def clear(self): + self.lines.clear() + self.byte_count = 0 + self.created_at = 0.0 + + +@dataclass +class ConversationTurn: + """A complete user-assistant conversation turn.""" + user_prompt: str + assistant_response: str + session_id: Optional[str] = None + project_name: Optional[str] = None + timestamp: Optional[str] = None + + +@dataclass +class ExtractedKnowledge: + """Structured knowledge extracted from a conversation.""" + status: str # "EXTRACTED" | "IGNORED" + category: str # "skills" | "memories" | "resources" + title: str + content: str + confidence: float = 0.0 + project_name: Optional[str] = None + entity_links: List[str] = field(default_factory=list) + actionable_steps: List[str] = field(default_factory=list) + timestamp: Optional[str] = None + + def to_dict(self) -> Dict[str, Any]: + return { + "status": self.status, + "category": self.category, + "title": self.title, + "content": self.content, + "confidence": self.confidence, + "project_name": self.project_name, + "entity_links": self.entity_links, + "actionable_steps": self.actionable_steps, + "timestamp": self.timestamp, + } diff --git a/openviking/daemon/service.py b/openviking/daemon/service.py new file mode 100644 index 0000000000..d8e14baef8 --- /dev/null +++ b/openviking/daemon/service.py @@ -0,0 +1,160 @@ +""" +OpenViking Active Daemon main service. +Orchestrates file watching, ETL processing, and knowledge storage. +""" +import asyncio +import os +from pathlib import Path +from typing import Optional + +from openviking.daemon.cursor_manager import CursorManager +from openviking.daemon.watchers.claude_code_watcher import ClaudeCodeWatcher +from openviking.daemon.etl_pipeline import BatchETLPipeline +from openviking.daemon.storage_adapter import VikingStorageAdapter +from openviking_cli.utils.logger import get_logger + +logger = get_logger(__name__) + + +class DaemonService: + """ + OpenViking Active Daemon main service. + Monitors Claude Code JSONL files and extracts knowledge into viking:// storage. + """ + + def __init__( + self, + resource_service, + watch_dir: Optional[str] = None, + db_path: Optional[str] = None, + batch_trigger_lines: int = 50, + batch_trigger_seconds: int = 300, + ): + self.resource_service = resource_service + + # Default paths + home = Path.home() + self.watch_dir = watch_dir or str(home / ".claude" / "projects") + self.db_path = db_path or str( + home / ".qoderworkcn" / "openviking" / "daemon_cursors.db" + ) + + self.batch_trigger_lines = batch_trigger_lines + self.batch_trigger_seconds = batch_trigger_seconds + + # Components (initialized in start()) + self.cursor_manager: Optional[CursorManager] = None + self.watcher: Optional[ClaudeCodeWatcher] = None + self.etl_pipeline: Optional[BatchETLPipeline] = None + self.storage_adapter: Optional[VikingStorageAdapter] = None + + self._running = False + self._etl_task: Optional[asyncio.Task] = None + self._batch_queue: asyncio.Queue = asyncio.Queue() + + async def start(self): + """Start the Daemon service.""" + logger.info("Starting OpenViking Active Daemon...") + + # Initialize components + self.cursor_manager = CursorManager(self.db_path) + self.etl_pipeline = BatchETLPipeline() + self.storage_adapter = VikingStorageAdapter(self.resource_service) + + # Ensure watch directory exists + Path(self.watch_dir).mkdir(parents=True, exist_ok=True) + + # Start the ETL processing loop as a background task + self._etl_task = asyncio.create_task(self._etl_loop()) + + # Start the file watcher + # The watcher runs in a separate thread (watchdog), so we pass + # a sync callback that puts events onto the async queue + self.watcher = ClaudeCodeWatcher( + watch_dir=self.watch_dir, + cursor_manager=self.cursor_manager, + batch_callback=self._enqueue_batch, + batch_trigger_lines=self.batch_trigger_lines, + batch_trigger_seconds=self.batch_trigger_seconds, + ) + self.watcher.start() + + self._running = True + logger.info("Daemon started, watching: %s", self.watch_dir) + + async def stop(self): + """Stop the Daemon service.""" + logger.info("Stopping OpenViking Active Daemon...") + + self._running = False + + if self.watcher: + self.watcher.stop() + + # Signal the ETL loop to stop + if self._etl_task: + await self._batch_queue.put(None) # sentinel + try: + await asyncio.wait_for(self._etl_task, timeout=10) + except asyncio.TimeoutError: + self._etl_task.cancel() + + logger.info("Daemon stopped") + + def _enqueue_batch(self, events): + """Sync callback from watcher thread - puts events onto async queue.""" + try: + self._batch_queue.put_nowait(events) + except Exception as e: + logger.error("Failed to enqueue batch: %s", e) + + async def _etl_loop(self): + """Background loop that processes batches from the queue.""" + logger.info("ETL processing loop started") + + while self._running: + try: + events = await asyncio.wait_for( + self._batch_queue.get(), timeout=5.0 + ) + except asyncio.TimeoutError: + continue + + if events is None: + break # shutdown sentinel + + try: + extracted = await self.etl_pipeline.process_batch(events) + if not extracted: + logger.info("No knowledge extracted from batch") + continue + + # Write to OpenViking + for knowledge in extracted: + try: + from openviking.server.identity import RequestContext + ctx = RequestContext(user_id="daemon", session_id="daemon-session") + success = await self.storage_adapter.write_knowledge( + knowledge, ctx + ) + if success: + logger.info("Successfully wrote: %s", knowledge.title) + else: + logger.warning("Failed to write: %s", knowledge.title) + except Exception as e: + logger.error("Error writing knowledge: %s", e) + + except Exception as e: + logger.error("Error in ETL processing: %s", e, exc_info=True) + + logger.info("ETL processing loop stopped") + + async def flush(self): + """Force flush any buffered events.""" + if self.watcher: + self.watcher.flush() + logger.info("Manual flush triggered") + + @property + def is_running(self) -> bool: + return self._running diff --git a/openviking/daemon/storage_adapter.py b/openviking/daemon/storage_adapter.py new file mode 100644 index 0000000000..41d5924bbb --- /dev/null +++ b/openviking/daemon/storage_adapter.py @@ -0,0 +1,152 @@ +""" +Adapter for writing extracted knowledge to OpenViking via ResourceService. +Generates temporary Markdown files and ingests them through the standard resource pipeline. +""" +import os +import tempfile +from datetime import datetime +from typing import Optional + +from openviking.daemon.models import ExtractedKnowledge +from openviking.daemon.knowledge_router import KnowledgeRouter +from openviking_cli.utils.logger import get_logger + +logger = get_logger(__name__) + + +class VikingStorageAdapter: + """Writes structured knowledge to OpenViking via ResourceService.add_resource().""" + + def __init__(self, resource_service): + self.resource_service = resource_service + self.router = KnowledgeRouter() + + async def write_knowledge( + self, + knowledge: ExtractedKnowledge, + ctx, + ) -> bool: + """ + Write a knowledge item to viking:// storage. + + Args: + knowledge: The extracted knowledge to write + ctx: OpenViking RequestContext + + Returns: + True if successful, False otherwise + """ + uri = self.router.route(knowledge) + if not uri: + logger.warning("Cannot route knowledge: %s", knowledge.title) + return False + + temp_file = self._generate_temp_markdown(knowledge) + + try: + await self.resource_service.add_resource( + path=temp_file, + ctx=ctx, + to=uri, + build_index=True, + summarize=False, + ) + logger.info("Knowledge ingested: %s", uri) + return True + + except Exception as e: + logger.error("Failed to ingest knowledge to %s: %s", uri, e) + return False + + finally: + if os.path.exists(temp_file): + os.unlink(temp_file) + + def _generate_temp_markdown(self, knowledge: ExtractedKnowledge) -> str: + """Generate a temporary Markdown file for the knowledge item.""" + fd, path = tempfile.mkstemp(suffix=".md") + content = self._format_content(knowledge) + + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(content) + + return path + + def _format_content(self, knowledge: ExtractedKnowledge) -> str: + """Format knowledge content as Markdown based on category.""" + timestamp = datetime.now().isoformat() + + if knowledge.category == "skills": + return self._format_skill(knowledge, timestamp) + elif knowledge.category == "memories": + return self._format_memory(knowledge, timestamp) + else: + return self._format_resource(knowledge, timestamp) + + @staticmethod + def _format_skill(knowledge: ExtractedKnowledge, timestamp: str) -> str: + """Format a skill knowledge item.""" + steps = "" + if knowledge.actionable_steps: + steps = "\n## Steps\n\n" + for i, step in enumerate(knowledge.actionable_steps, 1): + steps += f"{i}. {step}\n" + else: + steps = "\n## Steps\n\n(No specific steps)\n" + + return f"""# {knowledge.title} + +**Extracted**: {timestamp} +**Source**: Claude Code session +**Confidence**: {knowledge.confidence} + +## Content + +{knowledge.content} +{steps} +--- +*Auto-generated by OpenViking Active Daemon* +""" + + @staticmethod + def _format_memory(knowledge: ExtractedKnowledge, timestamp: str) -> str: + """Format a memory knowledge item (append-friendly).""" + ts = "" + if knowledge.timestamp: + try: + ts = datetime.fromisoformat(knowledge.timestamp).strftime("%Y-%m-%d %H:%M") + except ValueError: + ts = timestamp + else: + ts = timestamp + + tags = ", ".join(knowledge.entity_links) if knowledge.entity_links else "none" + + return f"""## [{ts}] {knowledge.title} + +{knowledge.content} + +**Tags**: {tags} + +--- + +""" + + @staticmethod + def _format_resource(knowledge: ExtractedKnowledge, timestamp: str) -> str: + """Format a resource knowledge item.""" + tech = knowledge.entity_links[0] if knowledge.entity_links else "general" + + return f"""# {knowledge.title} + +**Category**: {tech} +**Extracted**: {timestamp} +**Source**: Claude Code session + +## Content + +{knowledge.content} + +--- +*Auto-generated by OpenViking Active Daemon* +""" diff --git a/openviking/daemon/watchers/__init__.py b/openviking/daemon/watchers/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/openviking/daemon/watchers/claude_code_watcher.py b/openviking/daemon/watchers/claude_code_watcher.py new file mode 100644 index 0000000000..91eca08159 --- /dev/null +++ b/openviking/daemon/watchers/claude_code_watcher.py @@ -0,0 +1,165 @@ +""" +Claude Code JSONL file watcher using watchdog for incremental monitoring. +""" +import json +import os +import time +from typing import Callable, Dict, List, Optional + +from watchdog.observers import Observer +from watchdog.events import FileSystemEventHandler + +from openviking.daemon.models import BatchBuffer +from openviking.daemon.cursor_manager import CursorManager +from openviking_cli.utils.logger import get_logger + +logger = get_logger(__name__) + + +class ClaudeCodeLogHandler(FileSystemEventHandler): + """Handles file system events for Claude Code JSONL files.""" + + def __init__( + self, + cursor_manager: CursorManager, + batch_callback: Callable[[List[Dict]], None], + batch_trigger_lines: int = 50, + batch_trigger_seconds: int = 300, + ): + super().__init__() + self.cursor_manager = cursor_manager + self.batch_callback = batch_callback + self.batch_trigger_lines = batch_trigger_lines + self.batch_trigger_seconds = batch_trigger_seconds + self.buffer = BatchBuffer() + self.last_batch_time = time.time() + + def on_modified(self, event): + """Handle file modification events.""" + if event.is_directory or not event.src_path.endswith(".jsonl"): + return + try: + self._process_file(event.src_path) + except Exception as e: + logger.error("Error processing %s: %s", event.src_path, e) + + def _process_file(self, file_path: str): + """Incrementally read new lines from a file and add to buffer.""" + cursor = self.cursor_manager.get_cursor(file_path) + current_size = os.path.getsize(file_path) + + if current_size <= cursor.last_position: + return + + with open(file_path, "r", encoding="utf-8") as f: + f.seek(cursor.last_position) + new_lines = f.readlines() + new_position = f.tell() + + parsed_events = [] + for line in new_lines: + event = self._parse_line(line) + if event: + parsed_events.append(event) + + filtered_events = self._filter_events(parsed_events) + + for event in filtered_events: + self.buffer.add_line(event, len(json.dumps(event))) + + self.cursor_manager.update_cursor(file_path, new_position) + self._check_batch_trigger() + + @staticmethod + def _parse_line(line: str) -> Optional[Dict]: + """Parse a single JSONL line. Returns None on failure.""" + try: + return json.loads(line.strip()) + except (json.JSONDecodeError, AttributeError): + return None + + @staticmethod + def _filter_events(events: List[Dict]) -> List[Dict]: + """Filter to only user/assistant message events.""" + return [ + e for e in events + if e.get("role") in ("user", "assistant") and e.get("type") == "message" + ] + + def _check_batch_trigger(self): + """Check if batch processing should be triggered.""" + should_trigger = False + + if time.time() - self.last_batch_time >= self.batch_trigger_seconds: + should_trigger = True + + if len(self.buffer.lines) >= self.batch_trigger_lines: + should_trigger = True + + if should_trigger and not self.buffer.is_empty(): + self._flush_buffer() + + def _flush_buffer(self): + """Send buffered events to the batch callback and reset buffer.""" + if self.buffer.is_empty(): + return + + logger.info("Flushing batch with %d events", len(self.buffer.lines)) + + try: + self.batch_callback(self.buffer.lines.copy()) + except Exception as e: + logger.error("Error in batch callback: %s", e) + + self.buffer.clear() + self.last_batch_time = time.time() + + def force_flush(self): + """Force flush the buffer regardless of trigger conditions.""" + self._flush_buffer() + + +class ClaudeCodeWatcher: + """Monitors Claude Code JSONL log files for new conversation data.""" + + def __init__( + self, + watch_dir: str, + cursor_manager: CursorManager, + batch_callback: Callable[[List[Dict]], None], + batch_trigger_lines: int = 50, + batch_trigger_seconds: int = 300, + ): + self.watch_dir = watch_dir + self.cursor_manager = cursor_manager + self.batch_callback = batch_callback + self.batch_trigger_lines = batch_trigger_lines + self.batch_trigger_seconds = batch_trigger_seconds + self.observer: Optional[Observer] = None + self.handler: Optional[ClaudeCodeLogHandler] = None + + def start(self): + """Start watching for file changes.""" + self.handler = ClaudeCodeLogHandler( + cursor_manager=self.cursor_manager, + batch_callback=self.batch_callback, + batch_trigger_lines=self.batch_trigger_lines, + batch_trigger_seconds=self.batch_trigger_seconds, + ) + + self.observer = Observer() + self.observer.schedule(self.handler, self.watch_dir, recursive=True) + self.observer.start() + logger.info("Claude Code watcher started on %s", self.watch_dir) + + def stop(self): + """Stop watching for file changes.""" + if self.observer: + self.observer.stop() + self.observer.join() + logger.info("Claude Code watcher stopped") + + def flush(self): + """Force flush any buffered events.""" + if self.handler: + self.handler.force_flush() diff --git a/openviking/server/app.py b/openviking/server/app.py index bdea7a5803..e4e5e2b7df 100644 --- a/openviking/server/app.py +++ b/openviking/server/app.py @@ -33,6 +33,7 @@ bot_router, console_router, content_router, + daemon_router, debug_router, filesystem_router, metrics_router, @@ -275,6 +276,32 @@ async def _oauth_gc_loop(store) -> None: # noqa: ANN001 task_tracker = get_task_tracker() task_tracker.start_cleanup_loop() + # Start Active Daemon if enabled + daemon_service = None + if config.daemon.enabled: + try: + from openviking.daemon.service import DaemonService + from openviking.server.config import DaemonConfig + + # Check env var override + daemon_config = config.daemon + if not daemon_config.enabled: + daemon_config = DaemonConfig.from_env() + + if daemon_config.enabled: + resource_service = service.resources + daemon_service = DaemonService( + resource_service=resource_service, + watch_dir=daemon_config.watch_dir, + db_path=daemon_config.db_path, + batch_trigger_lines=daemon_config.batch_trigger_lines, + batch_trigger_seconds=daemon_config.batch_trigger_seconds, + ) + await daemon_service.start() + logger.info("Active Daemon started, watching: %s", daemon_service.watch_dir) + except Exception as e: + logger.warning("Failed to start Active Daemon: %s", e) + # Initialize tracing and OTLP log export from server.observability. from openviking.telemetry import tracer_module @@ -289,6 +316,14 @@ async def _oauth_gc_loop(store) -> None: # noqa: ANN001 await _initialize_runtime_state(app, service, config) yield + # Stop Active Daemon on shutdown + if daemon_service is not None: + try: + await daemon_service.stop() + logger.info("Active Daemon stopped") + except Exception as e: + logger.warning("Failed to stop Active Daemon: %s", e) + # Cleanup from openviking.metrics.global_api import shutdown_metrics_async from openviking.observability.usage_audit import shutdown_usage_audit @@ -537,6 +572,7 @@ async def general_error_handler(request: Request, exc: Exception): app.include_router(watches_router) app.include_router(webdav_router) app.include_router(bot_router, prefix="/bot/v1") + app.include_router(daemon_router) # OAuth 2.1: when enabled, mount the official MCP SDK auth routes # (DCR / authorize / token / metadata) plus our authorize page + consent / diff --git a/openviking/server/bootstrap.py b/openviking/server/bootstrap.py index 7fa0228e25..36a61a4256 100644 --- a/openviking/server/bootstrap.py +++ b/openviking/server/bootstrap.py @@ -17,7 +17,7 @@ import uvicorn from openviking.server.app import create_app -from openviking.server.config import load_server_config +from openviking.server.config import DaemonConfig, load_server_config from openviking_cli.utils.config import OPENVIKING_CONFIG_ENV from openviking_cli.utils.config.config_loader import resolve_config_path from openviking_cli.utils.config.consts import ( @@ -169,6 +169,12 @@ def main(): dest="with_bot", help="Enable Bot API proxy to Vikingbot (requires Vikingbot running)", ) + parser.add_argument( + "--with-daemon", + action="store_true", + dest="with_daemon", + help="Enable Active Daemon for automatic knowledge extraction from Claude Code logs", + ) parser.add_argument( "--bot-port", type=int, @@ -243,6 +249,8 @@ def main(): config.workers = args.workers if args.with_bot: config.with_bot = True + if args.with_daemon: + config.daemon.enabled = True # Configure logging for Uvicorn configure_uvicorn_logging() diff --git a/openviking/server/config.py b/openviking/server/config.py index 88943f8f4a..05ceeec4f6 100644 --- a/openviking/server/config.py +++ b/openviking/server/config.py @@ -156,6 +156,30 @@ class ToolOutputExternalizationConfig(BaseModel): model_config = {"extra": "forbid"} +class DaemonConfig(BaseModel): + """Configuration for OpenViking Active Daemon.""" + + enabled: bool = False + watch_dir: Optional[str] = None + db_path: Optional[str] = None + batch_trigger_lines: int = Field(50, gt=0) + batch_trigger_seconds: int = Field(300, gt=0) + + model_config = {"extra": "forbid"} + + @classmethod + def from_env(cls) -> "DaemonConfig": + """Load configuration from OV_DAEMON_* environment variables.""" + import os + return cls( + enabled=os.getenv("OV_DAEMON_ENABLED", "false").lower() == "true", + watch_dir=os.getenv("OV_DAEMON_WATCH_DIR"), + db_path=os.getenv("OV_DAEMON_DB_PATH"), + batch_trigger_lines=int(os.getenv("OV_DAEMON_BATCH_LINES", "50")), + batch_trigger_seconds=int(os.getenv("OV_DAEMON_BATCH_SECONDS", "300")), + ) + + class ServerConfig(BaseModel): host: str = "127.0.0.1" port: int = 1933 @@ -180,6 +204,7 @@ class ServerConfig(BaseModel): tool_output_externalization: ToolOutputExternalizationConfig = Field( default_factory=ToolOutputExternalizationConfig ) + daemon: DaemonConfig = Field(default_factory=DaemonConfig) model_config = {"extra": "forbid"} diff --git a/openviking/server/routers/__init__.py b/openviking/server/routers/__init__.py index 1d35086880..83d50291cf 100644 --- a/openviking/server/routers/__init__.py +++ b/openviking/server/routers/__init__.py @@ -6,6 +6,7 @@ from openviking.server.routers.bot import router as bot_router from openviking.server.routers.console import router as console_router from openviking.server.routers.content import router as content_router +from openviking.server.routers.daemon import router as daemon_router from openviking.server.routers.debug import router as debug_router from openviking.server.routers.filesystem import router as filesystem_router from openviking.server.routers.metrics import router as metrics_router @@ -26,6 +27,7 @@ __all__ = [ "admin_router", "bot_router", + "daemon_router", "system_router", "resources_router", "filesystem_router", diff --git a/openviking/server/routers/daemon.py b/openviking/server/routers/daemon.py new file mode 100644 index 0000000000..86849b1ddd --- /dev/null +++ b/openviking/server/routers/daemon.py @@ -0,0 +1,43 @@ +# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd. +# SPDX-License-Identifier: AGPL-3.0 +"""Daemon status API endpoints.""" +from typing import Any, Dict + +from fastapi import APIRouter + +router = APIRouter(prefix="/api/v1/daemon", tags=["daemon"]) + + +@router.get("/status") +async def get_daemon_status() -> Dict[str, Any]: + """ + Get the current status of the Active Daemon. + + Returns: + { + "enabled": bool, + "running": bool, + "watch_dir": str | null, + "db_path": str | null, + "batch_trigger_lines": int, + "batch_trigger_seconds": int, + "cursor_count": int, + "last_flush_time": str | null + } + """ + # This is a placeholder — actual implementation needs access to DaemonService instance + # For now, return static config info + from openviking.server.config import DaemonConfig + + daemon_config = DaemonConfig.from_env() + + return { + "enabled": daemon_config.enabled, + "running": False, # Would need to track actual state + "watch_dir": daemon_config.watch_dir, + "db_path": daemon_config.db_path, + "batch_trigger_lines": daemon_config.batch_trigger_lines, + "batch_trigger_seconds": daemon_config.batch_trigger_seconds, + "cursor_count": 0, + "last_flush_time": None, + } diff --git a/openviking/session/memory/memory_type_registry.py b/openviking/session/memory/memory_type_registry.py index a07c20d179..b8ffb52fea 100644 --- a/openviking/session/memory/memory_type_registry.py +++ b/openviking/session/memory/memory_type_registry.py @@ -301,13 +301,20 @@ async def initialize_memory_files( pass # Add MEMORY_FIELDS comment with field metadata + from datetime import datetime, timezone + from openviking.session.memory.dataclass import MemoryFile from openviking.session.memory.utils.memory_file_utils import MemoryFileUtils + now = datetime.now(timezone.utc) + extra = dict(fields_with_init) + extra["created_at"] = now + extra["updated_at"] = now + mf = MemoryFile( uri=file_uri, memory_type=schema.memory_type, - extra_fields=fields_with_init, + extra_fields=extra, ) full_content = MemoryFileUtils.write( mf, diff --git a/openviking/session/memory/memory_updater.py b/openviking/session/memory/memory_updater.py index b7c0eac2c8..49cc190b5a 100644 --- a/openviking/session/memory/memory_updater.py +++ b/openviking/session/memory/memory_updater.py @@ -11,6 +11,7 @@ import re from dataclasses import dataclass +from datetime import datetime, timezone from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple if TYPE_CHECKING: @@ -695,6 +696,16 @@ async def _apply_upsert( if key not in schema_field_names and key not in metadata and val is not None: metadata[key] = val + # Inject system-managed timestamps into MEMORY_FIELDS. + # created_at: inherited from existing file for updates; set to now for new files. + # updated_at: always refreshed to the current time. + now = datetime.now(timezone.utc) + if old_content and old_content.extra_fields.get("created_at"): + metadata["created_at"] = old_content.extra_fields["created_at"] + else: + metadata["created_at"] = now + metadata["updated_at"] = now + # Handle links/backlinks fields: merge with existing incoming_links_by_uri = getattr(resolved_op, "_incoming_links_by_uri", {}) incoming_backlinks_by_uri = getattr(resolved_op, "_incoming_backlinks_by_uri", {}) diff --git a/tests/daemon/__init__.py b/tests/daemon/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/daemon/test_claude_code_watcher.py b/tests/daemon/test_claude_code_watcher.py new file mode 100644 index 0000000000..880a5d2634 --- /dev/null +++ b/tests/daemon/test_claude_code_watcher.py @@ -0,0 +1,88 @@ +"""Tests for ClaudeCodeWatcher parsing and filtering logic.""" +from openviking.daemon.watchers.claude_code_watcher import ClaudeCodeLogHandler + + +class FakeCursorManager: + """Minimal stub for testing.""" + def __init__(self): + self.cursors = {} + self.updates = [] + + def get_cursor(self, file_path): + from openviking.daemon.models import FileCursor + return self.cursors.get(file_path, FileCursor(file_path=file_path)) + + def update_cursor(self, file_path, position): + self.updates.append((file_path, position)) + + +def _make_handler(batch_trigger_lines=50, batch_trigger_seconds=300): + batches = [] + cursor_mgr = FakeCursorManager() + handler = ClaudeCodeLogHandler( + cursor_manager=cursor_mgr, + batch_callback=lambda lines: batches.append(lines), + batch_trigger_lines=batch_trigger_lines, + batch_trigger_seconds=batch_trigger_seconds, + ) + return handler, batches + + +def test_parse_valid_jsonl_line(): + handler, _ = _make_handler() + line = '{"timestamp": "2026-06-15T10:30:00Z", "role": "user", "content": "Hello", "type": "message"}' + event = handler._parse_line(line) + assert event is not None + assert event["role"] == "user" + assert event["content"] == "Hello" + + +def test_parse_invalid_line(): + handler, _ = _make_handler() + assert handler._parse_line("not valid json") is None + assert handler._parse_line("") is None + + +def test_filter_keeps_user_and_assistant_messages(): + handler, _ = _make_handler() + events = [ + {"role": "user", "type": "message", "content": "User question"}, + {"role": "assistant", "type": "message", "content": "AI answer"}, + {"role": "assistant", "type": "tool_call", "content": "Tool call"}, + {"role": "system", "type": "message", "content": "System msg"}, + ] + filtered = handler._filter_events(events) + assert len(filtered) == 2 + assert all(e["type"] == "message" for e in filtered) + + +def test_filter_excludes_tool_calls(): + handler, _ = _make_handler() + events = [ + {"role": "assistant", "type": "tool_call", "content": "call"}, + {"role": "assistant", "type": "tool_result", "content": "result"}, + ] + filtered = handler._filter_events(events) + assert len(filtered) == 0 + + +def test_buffer_add_and_flush(): + handler, batches = _make_handler(batch_trigger_lines=2) + handler.buffer.add_line({"role": "user", "content": "a"}, 10) + handler.buffer.add_line({"role": "assistant", "content": "b"}, 10) + handler._check_batch_trigger() + assert len(batches) == 1 + assert len(batches[0]) == 2 + + +def test_force_flush_empty_buffer(): + handler, batches = _make_handler() + handler.force_flush() + assert len(batches) == 0 + + +def test_force_flush_with_data(): + handler, batches = _make_handler() + handler.buffer.add_line({"role": "user", "content": "test"}, 10) + handler.force_flush() + assert len(batches) == 1 diff --git a/tests/daemon/test_conversation_reconstructor.py b/tests/daemon/test_conversation_reconstructor.py new file mode 100644 index 0000000000..b8c588855d --- /dev/null +++ b/tests/daemon/test_conversation_reconstructor.py @@ -0,0 +1,58 @@ +"""Tests for ConversationReconstructor.""" +from openviking.daemon.conversation_reconstructor import ConversationReconstructor + + +def test_reconstruct_simple_conversation(): + r = ConversationReconstructor() + events = [ + {"role": "user", "content": "How to configure PostgreSQL?", "timestamp": "2026-06-15T10:00:00Z"}, + {"role": "assistant", "content": "Edit postgresql.conf", "timestamp": "2026-06-15T10:00:01Z"}, + ] + turns = r.reconstruct(events) + assert len(turns) == 1 + assert turns[0].user_prompt == "How to configure PostgreSQL?" + assert turns[0].assistant_response == "Edit postgresql.conf" + + +def test_skip_orphaned_assistant(): + r = ConversationReconstructor() + events = [ + {"role": "assistant", "content": "Orphan answer", "timestamp": "2026-06-15T10:00:00Z"}, + ] + turns = r.reconstruct(events) + assert len(turns) == 0 + + +def test_multiple_turns(): + r = ConversationReconstructor() + events = [ + {"role": "user", "content": "Q1", "timestamp": "2026-06-15T10:00:00Z"}, + {"role": "assistant", "content": "A1", "timestamp": "2026-06-15T10:00:01Z"}, + {"role": "user", "content": "Q2", "timestamp": "2026-06-15T10:00:02Z"}, + {"role": "assistant", "content": "A2", "timestamp": "2026-06-15T10:00:03Z"}, + ] + turns = r.reconstruct(events) + assert len(turns) == 2 + assert turns[0].user_prompt == "Q1" + assert turns[1].user_prompt == "Q2" + + +def test_unpaired_user_prompt(): + r = ConversationReconstructor() + events = [ + {"role": "user", "content": "No answer", "timestamp": "2026-06-15T10:00:00Z"}, + ] + turns = r.reconstruct(events) + assert len(turns) == 0 + + +def test_preserves_metadata(): + r = ConversationReconstructor() + events = [ + {"role": "user", "content": "Q", "timestamp": "2026-06-15T10:00:00Z", "session_id": "s1", "project_name": "proj"}, + {"role": "assistant", "content": "A", "timestamp": "2026-06-15T10:00:01Z"}, + ] + turns = r.reconstruct(events) + assert turns[0].session_id == "s1" + assert turns[0].project_name == "proj" + assert turns[0].timestamp == "2026-06-15T10:00:00Z" diff --git a/tests/daemon/test_cursor_manager.py b/tests/daemon/test_cursor_manager.py new file mode 100644 index 0000000000..375df56d09 --- /dev/null +++ b/tests/daemon/test_cursor_manager.py @@ -0,0 +1,61 @@ +"""Tests for CursorManager.""" +import os +import tempfile + +import pytest + +from openviking.daemon.cursor_manager import CursorManager + + +@pytest.fixture +def temp_db(): + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + yield path + if os.path.exists(path): + os.unlink(path) + + +def test_default_cursor(temp_db): + manager = CursorManager(temp_db) + cursor = manager.get_cursor("/path/to/file.jsonl") + assert cursor.last_position == 0 + assert cursor.last_read_time == 0.0 + + +def test_save_and_load_cursor(temp_db): + manager = CursorManager(temp_db) + manager.update_cursor("/path/to/file.jsonl", 1024) + + cursor = manager.get_cursor("/path/to/file.jsonl") + assert cursor.last_position == 1024 + assert cursor.last_read_time > 0 + + +def test_persist_across_instances(temp_db): + manager1 = CursorManager(temp_db) + manager1.update_cursor("/path/to/file.jsonl", 2048) + + manager2 = CursorManager(temp_db) + cursor = manager2.get_cursor("/path/to/file.jsonl") + assert cursor.last_position == 2048 + + +def test_get_all_cursors(temp_db): + manager = CursorManager(temp_db) + manager.update_cursor("/path/file1.jsonl", 100) + manager.update_cursor("/path/file2.jsonl", 200) + + cursors = manager.get_all_cursors() + assert len(cursors) == 2 + assert cursors["/path/file1.jsonl"].last_position == 100 + assert cursors["/path/file2.jsonl"].last_position == 200 + + +def test_update_existing_cursor(temp_db): + manager = CursorManager(temp_db) + manager.update_cursor("/path/file.jsonl", 100) + manager.update_cursor("/path/file.jsonl", 500) + + cursor = manager.get_cursor("/path/file.jsonl") + assert cursor.last_position == 500 diff --git a/tests/daemon/test_deduplicator.py b/tests/daemon/test_deduplicator.py new file mode 100644 index 0000000000..569c14a3fb --- /dev/null +++ b/tests/daemon/test_deduplicator.py @@ -0,0 +1,42 @@ +"""Tests for KnowledgeDeduplicator.""" +from openviking.daemon.deduplicator import KnowledgeDeduplicator +from openviking.daemon.models import ExtractedKnowledge + + +def _make_knowledge(title: str, content: str) -> ExtractedKnowledge: + return ExtractedKnowledge( + status="EXTRACTED", + category="memories", + title=title, + content=content, + ) + + +def test_first_occurrence_not_duplicate(): + dedup = KnowledgeDeduplicator() + k = _make_knowledge("Test", "Some unique content") + assert not dedup.is_duplicate(k) + + +def test_same_content_is_duplicate(): + dedup = KnowledgeDeduplicator() + k1 = _make_knowledge("Title A", "Same content") + k2 = _make_knowledge("Title B", "Same content") + assert not dedup.is_duplicate(k1) + assert dedup.is_duplicate(k2) + + +def test_different_content_not_duplicate(): + dedup = KnowledgeDeduplicator() + k1 = _make_knowledge("A", "Content 1") + k2 = _make_knowledge("B", "Content 2") + assert not dedup.is_duplicate(k1) + assert not dedup.is_duplicate(k2) + + +def test_clear_resets_cache(): + dedup = KnowledgeDeduplicator() + k = _make_knowledge("Test", "Content") + dedup.is_duplicate(k) + dedup.clear() + assert not dedup.is_duplicate(k) diff --git a/tests/daemon/test_filters.py b/tests/daemon/test_filters.py new file mode 100644 index 0000000000..52d0c07d50 --- /dev/null +++ b/tests/daemon/test_filters.py @@ -0,0 +1,53 @@ +"""Tests for LowValueFilter.""" +from openviking.daemon.filters import LowValueFilter + + +def test_filter_short_content(): + f = LowValueFilter() + events = [ + {"content": "Short"}, + {"content": "This is a longer meaningful conversation about architecture"}, + ] + filtered = f.apply(events) + assert len(filtered) == 1 + + +def test_filter_noise_patterns(): + f = LowValueFilter() + events = [ + {"content": "npm install lodash --save"}, + {"content": "git commit -m 'fix bug'"}, + {"content": "Let's discuss the architecture design pattern for the new module"}, + ] + filtered = f.apply(events) + assert len(filtered) == 1 + assert "architecture" in filtered[0]["content"] + + +def test_filter_pip_install(): + f = LowValueFilter() + events = [ + {"content": "pip install requests library for HTTP calls"}, + ] + filtered = f.apply(events) + assert len(filtered) == 0 + + +def test_preserves_valid_content(): + f = LowValueFilter() + events = [ + {"content": "We decided to use PostgreSQL instead of MySQL for better JSON support"}, + {"content": "The memory leak was caused by unclosed database connections"}, + ] + filtered = f.apply(events) + assert len(filtered) == 2 + + +def test_empty_content_filtered(): + f = LowValueFilter() + events = [ + {"content": ""}, + {"content": " "}, + ] + filtered = f.apply(events) + assert len(filtered) == 0 diff --git a/tests/daemon/test_integration.py b/tests/daemon/test_integration.py new file mode 100644 index 0000000000..d03f07e3e4 --- /dev/null +++ b/tests/daemon/test_integration.py @@ -0,0 +1,189 @@ +"""Integration tests for OpenViking Active Daemon.""" +import json +import os +import tempfile +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from openviking.daemon.cursor_manager import CursorManager +from openviking.daemon.deduplicator import KnowledgeDeduplicator +from openviking.daemon.etl_pipeline import BatchETLPipeline +from openviking.daemon.filters import LowValueFilter +from openviking.daemon.conversation_reconstructor import ConversationReconstructor +from openviking.daemon.knowledge_router import KnowledgeRouter +from openviking.daemon.storage_adapter import VikingStorageAdapter +from openviking.daemon.models import ExtractedKnowledge + + +@pytest.fixture +def temp_watch_dir(): + with tempfile.TemporaryDirectory() as tmpdir: + yield Path(tmpdir) + + +@pytest.fixture +def temp_db(): + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + yield path + if os.path.exists(path): + os.unlink(path) + + +def test_jsonl_file_created_and_readable(temp_watch_dir): + """Verify we can create and read JSONL files in the watch directory.""" + jsonl_file = temp_watch_dir / "test_session.jsonl" + + events = [ + { + "timestamp": "2026-06-15T10:00:00Z", + "role": "user", + "content": "How to configure PostgreSQL for high availability?", + "type": "message", + }, + { + "timestamp": "2026-06-15T10:00:01Z", + "role": "assistant", + "content": "Edit postgresql.conf, set max_connections=100 and enable WAL archiving", + "type": "message", + }, + ] + + with open(jsonl_file, "w", encoding="utf-8") as f: + for event in events: + f.write(json.dumps(event) + "\n") + + assert jsonl_file.exists() + lines = jsonl_file.read_text(encoding="utf-8").strip().split("\n") + assert len(lines) == 2 + + parsed = [json.loads(line) for line in lines] + assert parsed[0]["role"] == "user" + assert parsed[1]["role"] == "assistant" + + +def test_cursor_tracks_jsonl_progress(temp_watch_dir, temp_db): + """Verify CursorManager correctly tracks incremental reads.""" + jsonl_file = temp_watch_dir / "session.jsonl" + + # Write first batch + with open(jsonl_file, "w", encoding="utf-8") as f: + f.write(json.dumps({"role": "user", "content": "First question", "type": "message"}) + "\n") + + cursor_mgr = CursorManager(temp_db) + cursor = cursor_mgr.get_cursor(str(jsonl_file)) + assert cursor.last_position == 0 + + # Read the file + with open(jsonl_file, "r", encoding="utf-8") as f: + f.seek(cursor.last_position) + lines = f.readlines() + new_position = f.tell() + + assert len(lines) == 1 + assert new_position > 0 + + cursor_mgr.update_cursor(str(jsonl_file), new_position) + + # Verify cursor persisted + cursor2 = cursor_mgr.get_cursor(str(jsonl_file)) + assert cursor2.last_position == new_position + + # Append more data + with open(jsonl_file, "a", encoding="utf-8") as f: + f.write(json.dumps({"role": "assistant", "content": "Answer", "type": "message"}) + "\n") + + # Read incrementally + with open(jsonl_file, "r", encoding="utf-8") as f: + f.seek(cursor2.last_position) + new_lines = f.readlines() + + assert len(new_lines) == 1 + assert json.loads(new_lines[0])["role"] == "assistant" + + +def test_filter_and_reconstruct_pipeline(): + """Verify the filter -> reconstruct pipeline works end-to-end.""" + events = [ + {"role": "user", "content": "Hi", "type": "message", "timestamp": "2026-06-15T10:00:00Z"}, # Too short + {"role": "user", "content": "npm install lodash --save-dev", "type": "message", "timestamp": "2026-06-15T10:00:01Z"}, # Noise + {"role": "user", "content": "How should we handle database migrations in production?", "type": "message", "timestamp": "2026-06-15T10:00:02Z"}, + {"role": "assistant", "content": "Use Alembic with versioned migration scripts and run them in a CI pipeline before deployment", "type": "message", "timestamp": "2026-06-15T10:00:03Z"}, + ] + + # Step 1: Filter + filt = LowValueFilter() + filtered = filt.apply(events) + assert len(filtered) == 2 # "Hi" and "npm install" removed + + # Step 2: Reconstruct + reconstructor = ConversationReconstructor() + turns = reconstructor.reconstruct(filtered) + assert len(turns) == 1 + assert "database migrations" in turns[0].user_prompt + assert "Alembic" in turns[0].assistant_response + + +def test_knowledge_router_all_categories(): + """Verify router handles all categories correctly.""" + router = KnowledgeRouter() + + skill = ExtractedKnowledge(status="EXTRACTED", category="skills", title="PG Config", content="...") + assert "skills/claude_code" in router.route(skill) + + mem_proj = ExtractedKnowledge(status="EXTRACTED", category="memories", title="Arch", content="...", project_name="myapp") + assert "memories/projects/myapp" in router.route(mem_proj) + + mem_global = ExtractedKnowledge(status="EXTRACTED", category="memories", title="General", content="...") + assert "memories/global" in router.route(mem_global) + + resource = ExtractedKnowledge(status="EXTRACTED", category="resources", title="Redis", content="...", entity_links=["Redis"]) + assert "resources/Redis" in router.route(resource) + + +def test_deduplicator_prevents_duplicates(): + """Verify deduplicator blocks duplicate knowledge.""" + dedup = KnowledgeDeduplicator() + + k1 = ExtractedKnowledge(status="EXTRACTED", category="memories", title="A", content="Same content here") + k2 = ExtractedKnowledge(status="EXTRACTED", category="memories", title="B", content="Same content here") + k3 = ExtractedKnowledge(status="EXTRACTED", category="memories", title="C", content="Different content") + + assert not dedup.is_duplicate(k1) + assert dedup.is_duplicate(k2) + assert not dedup.is_duplicate(k3) + + +def test_storage_adapter_formats_content(): + """Verify storage adapter generates correct Markdown for each category.""" + mock_service = MagicMock() + adapter = VikingStorageAdapter(mock_service) + + skill = ExtractedKnowledge( + status="EXTRACTED", category="skills", title="Test Skill", + content="Do this thing", confidence=0.9, + actionable_steps=["Step 1", "Step 2"], + ) + content = adapter._format_content(skill) + assert "Test Skill" in content + assert "Step 1" in content + assert "Step 2" in content + + memory = ExtractedKnowledge( + status="EXTRACTED", category="memories", title="Decision", + content="We chose X", entity_links=["tag1"], + timestamp="2026-06-15T10:00:00Z", + ) + content = adapter._format_content(memory) + assert "Decision" in content + assert "tag1" in content + + resource = ExtractedKnowledge( + status="EXTRACTED", category="resources", title="Guide", + content="Reference material", entity_links=["Docker"], + ) + content = adapter._format_content(resource) + assert "Guide" in content + assert "Docker" in content diff --git a/tests/daemon/test_knowledge_extractor.py b/tests/daemon/test_knowledge_extractor.py new file mode 100644 index 0000000000..56391929d2 --- /dev/null +++ b/tests/daemon/test_knowledge_extractor.py @@ -0,0 +1,123 @@ +"""Tests for KnowledgeExtractor.""" +import json +from dataclasses import dataclass +from typing import Optional + +import pytest + +from openviking.daemon.knowledge_extractor import KnowledgeExtractor +from openviking.daemon.models import ConversationTurn, ExtractedKnowledge + + +@dataclass +class MockLLMResponse: + text: str + + +class MockLLMService: + """Mock LLM service that returns canned responses.""" + + def __init__(self, response_text: str): + self.response_text = response_text + + async def complete(self, **kwargs): + return MockLLMResponse(text=self.response_text) + + +def _make_turn(user="How to configure PostgreSQL?", assistant="Edit postgresql.conf"): + return ConversationTurn( + user_prompt=user, + assistant_response=assistant, + timestamp="2026-06-15T10:00:00Z", + ) + + +@pytest.mark.asyncio +async def test_extract_valid_knowledge(): + llm_response = json.dumps({ + "status": "EXTRACTED", + "category": "skills", + "confidence": 0.9, + "title": "PostgreSQL Config", + "content": "Configure PostgreSQL by editing postgresql.conf", + "project_name": "my-project", + "entity_links": ["PostgreSQL"], + "actionable_steps": ["Edit postgresql.conf"], + }) + extractor = KnowledgeExtractor(llm_service=MockLLMService(llm_response)) + result = await extractor.extract(_make_turn()) + + assert result is not None + assert result.status == "EXTRACTED" + assert result.category == "skills" + assert result.confidence == 0.9 + + +@pytest.mark.asyncio +async def test_extract_ignored_status(): + llm_response = json.dumps({ + "status": "IGNORED", + "category": "memories", + "confidence": 0.3, + "title": "Typo fix", + "content": "Fixed typo", + }) + extractor = KnowledgeExtractor(llm_service=MockLLMService(llm_response)) + result = await extractor.extract(_make_turn()) + + assert result is None + + +@pytest.mark.asyncio +async def test_extract_low_confidence(): + llm_response = json.dumps({ + "status": "EXTRACTED", + "category": "memories", + "confidence": 0.4, + "title": "Low confidence", + "content": "Some content", + }) + extractor = KnowledgeExtractor(llm_service=MockLLMService(llm_response)) + result = await extractor.extract(_make_turn()) + + assert result is None + + +@pytest.mark.asyncio +async def test_extract_invalid_json(): + extractor = KnowledgeExtractor(llm_service=MockLLMService("not valid json")) + result = await extractor.extract(_make_turn()) + + assert result is None + + +@pytest.mark.asyncio +async def test_extract_cleans_markdown(): + llm_response = json.dumps({ + "status": "EXTRACTED", + "category": "memories", + "confidence": 0.8, + "title": "Test", + "content": "```python\nsome code\n```\nActual content here", + }) + extractor = KnowledgeExtractor(llm_service=MockLLMService(llm_response)) + result = await extractor.extract(_make_turn()) + + assert result is not None + assert "```" not in result.content + + +@pytest.mark.asyncio +async def test_title_truncated(): + llm_response = json.dumps({ + "status": "EXTRACTED", + "category": "memories", + "confidence": 0.8, + "title": "A" * 100, + "content": "Content", + }) + extractor = KnowledgeExtractor(llm_service=MockLLMService(llm_response)) + result = await extractor.extract(_make_turn()) + + assert result is not None + assert len(result.title) <= 50 diff --git a/tests/daemon/test_knowledge_router.py b/tests/daemon/test_knowledge_router.py new file mode 100644 index 0000000000..f60382ea00 --- /dev/null +++ b/tests/daemon/test_knowledge_router.py @@ -0,0 +1,65 @@ +"""Tests for KnowledgeRouter.""" +from openviking.daemon.knowledge_router import KnowledgeRouter +from openviking.daemon.models import ExtractedKnowledge + + +def _make(**kwargs) -> ExtractedKnowledge: + defaults = { + "status": "EXTRACTED", + "category": "memories", + "title": "Test Title", + "content": "Some content", + } + defaults.update(kwargs) + return ExtractedKnowledge(**defaults) + + +def test_route_skills(): + router = KnowledgeRouter() + k = _make(category="skills", title="PostgreSQL Config") + uri = router.route(k) + assert uri.startswith("viking://skills/claude_code/") + assert uri.endswith(".md") + + +def test_route_memories_with_project(): + router = KnowledgeRouter() + k = _make(category="memories", title="Arch Decision", project_name="my-project") + uri = router.route(k) + assert "my-project" in uri + assert uri.endswith("decisions.md") + + +def test_route_memories_global(): + router = KnowledgeRouter() + k = _make(category="memories", title="Global Memory") + uri = router.route(k) + assert uri.startswith("viking://memories/global/") + + +def test_route_resources(): + router = KnowledgeRouter() + k = _make(category="resources", title="Redis Guide", entity_links=["Redis"]) + uri = router.route(k) + assert "Redis" in uri + assert uri.endswith(".md") + + +def test_route_resources_no_tags(): + router = KnowledgeRouter() + k = _make(category="resources", title="General", entity_links=[]) + uri = router.route(k) + assert "general" in uri + + +def test_route_unknown_category(): + router = KnowledgeRouter() + k = _make(category="unknown") + uri = router.route(k) + assert uri is None + + +def test_sanitize_filename(): + router = KnowledgeRouter() + assert router._sanitize_filename('file<>:name') == 'file___name' + assert router._sanitize_filename("a" * 100) == "a" * 50 diff --git a/tests/session/memory/test_memory_updater.py b/tests/session/memory/test_memory_updater.py index 82161d7667..e0a8e5c245 100644 --- a/tests/session/memory/test_memory_updater.py +++ b/tests/session/memory/test_memory_updater.py @@ -847,3 +847,95 @@ async def mock_write_file(uri, content, **kwargs): assert "ALPHA" in parsed["content"] assert "BETA" in parsed["content"] assert "gamma" in parsed["content"] + + +class TestUpsertTimestamps: + """Regression tests: _apply_upsert must inject created_at and updated_at + into MEMORY_FIELDS for both new and updated files.""" + + def _make_updater(self, memory_type="notes"): + schema = MemoryTypeSchema( + memory_type=memory_type, + description="notes", + fields=[ + MemoryField( + name="content", field_type=FieldType.STRING, merge_op=MergeOp.PATCH + ), + ], + ) + registry = MemoryTypeRegistry() + registry.register(schema) + return MemoryUpdater(registry=registry) + + def _make_fs(self): + store: dict[str, str] = {} + mock_fs = MagicMock() + + async def mock_read_file(uri, **kwargs): + return store.get(uri) + + async def mock_write_file(uri, content, **kwargs): + store[uri] = content + + mock_fs.read_file = mock_read_file + mock_fs.write_file = mock_write_file + return store, mock_fs + + @pytest.mark.asyncio + async def test_upsert_injects_timestamps_on_new_file(self): + """A brand-new file (no old_content) must get created_at and updated_at.""" + uri = "viking://user/test/memories/notes/new.md" + updater = self._make_updater() + store, mock_fs = self._make_fs() + updater._get_viking_fs = MagicMock(return_value=mock_fs) + + op = ResolvedOperation( + old_memory_file_content=None, + memory_fields={"content": "Hello world"}, + memory_type="notes", + uris=[uri], + ) + await updater._apply_upsert(op, MagicMock()) + + parsed = parse_memory_file_with_fields(store[uri]) + assert "created_at" in parsed, "created_at must be present in MEMORY_FIELDS" + assert "updated_at" in parsed, "updated_at must be present in MEMORY_FIELDS" + assert parsed["created_at"] == parsed["updated_at"] + + @pytest.mark.asyncio + async def test_upsert_preserves_created_at_updates_updated_at(self): + """Updating an existing file must keep created_at and refresh updated_at.""" + from datetime import datetime, timezone + + uri = "viking://user/test/memories/notes/existing.md" + updater = self._make_updater() + store, mock_fs = self._make_fs() + updater._get_viking_fs = MagicMock(return_value=mock_fs) + + old_created = "2026-01-01T00:00:00+00:00" + old_updated = "2026-01-01T00:00:00+00:00" + + # Seed the store with an existing file that has timestamps + existing = MemoryFile( + uri=uri, + content="Old content", + extra_fields={"created_at": old_created, "updated_at": old_updated}, + ) + existing_content = MemoryFileUtils.write(existing) + store[uri] = existing_content + + patch = StrPatch(blocks=[SearchReplaceBlock(search="Old", replace="New")]) + op = ResolvedOperation( + old_memory_file_content=existing, + memory_fields={"content": patch}, + memory_type="notes", + uris=[uri], + ) + await updater._apply_upsert(op, MagicMock()) + + parsed = parse_memory_file_with_fields(store[uri]) + assert parsed["created_at"] == old_created, "created_at must be preserved from original" + assert parsed["updated_at"] != old_updated, "updated_at must be refreshed" + # updated_at should be a recent UTC time + updated_dt = datetime.fromisoformat(parsed["updated_at"]) + assert updated_dt.tzinfo is not None, "updated_at must be timezone-aware" diff --git a/web-studio/src/i18n/locales/en.ts b/web-studio/src/i18n/locales/en.ts index 7326f436a8..c10333bb15 100644 --- a/web-studio/src/i18n/locales/en.ts +++ b/web-studio/src/i18n/locales/en.ts @@ -301,6 +301,18 @@ const en = { usageDisabled: 'Usage/Audit is not initialized, so live usage stats are unavailable.', }, + daemon: { + status: 'Active Daemon', + enabled: 'Enabled', + disabled: 'Disabled', + running: 'Running', + stopped: 'Stopped', + watchDir: 'Watch Directory', + batchLines: 'Batch Lines', + batchSeconds: 'Batch Seconds', + cursorCount: 'Tracked Files', + lastFlush: 'Last Flush', + }, operations: { page: { placeholder: 'Operations dashboard is under construction.', diff --git a/web-studio/src/i18n/locales/zh-CN.ts b/web-studio/src/i18n/locales/zh-CN.ts index 031b077cb0..a5cae301d0 100644 --- a/web-studio/src/i18n/locales/zh-CN.ts +++ b/web-studio/src/i18n/locales/zh-CN.ts @@ -298,6 +298,18 @@ const zhCN = { }, usageDisabled: 'Usage/Audit 未初始化，暂无实时统计。', }, + daemon: { + status: '主动守护进程', + enabled: '已启用', + disabled: '已禁用', + running: '运行中', + stopped: '已停止', + watchDir: '监听目录', + batchLines: '批处理行数', + batchSeconds: '批处理间隔(秒)', + cursorCount: '追踪文件数', + lastFlush: '上次刷新', + }, operations: { page: { placeholder: '运维面板能力尚未接入。', diff --git a/web-studio/src/routes/home/-components/daemon-status-card.tsx b/web-studio/src/routes/home/-components/daemon-status-card.tsx new file mode 100644 index 0000000000..5bf2a2d93c --- /dev/null +++ b/web-studio/src/routes/home/-components/daemon-status-card.tsx @@ -0,0 +1,148 @@ +import { useQuery } from '@tanstack/react-query'; +import { Activity, Clock, Database, FileText, ToggleLeft, ToggleRight } from 'lucide-react'; +import { useTranslation } from 'react-i18next'; +import { Card, CardContent, CardHeader, CardTitle } from '@/components/ui/card'; +import { Badge } from '@/components/ui/badge'; +import { Skeleton } from '@/components/ui/skeleton'; +import { ovClient } from '@/lib/ov-client/client'; + +interface DaemonStatus { + enabled: boolean; + running: boolean; + watch_dir: string | null; + db_path: string | null; + batch_trigger_lines: number; + batch_trigger_seconds: number; + cursor_count: number; + last_flush_time: string | null; +} + +async function fetchDaemonStatus(): Promise<DaemonStatus> { + const response = await ovClient.instance.get('/api/v1/daemon/status'); + return response.data as DaemonStatus; +} + +export function DaemonStatusCard() { + const { t } = useTranslation(); + + const { data, isLoading, error } = useQuery({ + queryKey: ['daemon-status'], + queryFn: fetchDaemonStatus, + refetchInterval: 30000, // Refresh every 30 seconds + }); + + if (isLoading) { + return ( + <Card> + <CardHeader> + <CardTitle className="flex items-center gap-2"> + <Activity className="h-4 w-4" /> + <Skeleton className="h-5 w-32" /> + </CardTitle> + </CardHeader> + <CardContent> + <Skeleton className="h-4 w-full mb-2" /> + <Skeleton className="h-4 w-3/4" /> + </CardContent> + </Card> + ); + } + + if (error || !data) { + return ( + <Card> + <CardHeader> + <CardTitle className="flex items-center gap-2 text-muted-foreground"> + <Activity className="h-4 w-4" /> + {t('daemon.status')} + </CardTitle> + </CardHeader> + <CardContent> + <p className="text-sm text-muted-foreground">Failed to load daemon status</p> + </CardContent> + </Card> + ); + } + + const statusColor = data.enabled && data.running + ? 'bg-green-500' + : data.enabled + ? 'bg-yellow-500' + : 'bg-gray-400'; + + const statusLabel = data.enabled && data.running + ? t('daemon.running') + : data.enabled + ? t('daemon.stopped') + : t('daemon.disabled'); + + return ( + <Card> + <CardHeader> + <CardTitle className="flex items-center justify-between"> + <div className="flex items-center gap-2"> + <Activity className="h-4 w-4" /> + {t('daemon.status')} + </div> + <Badge variant={data.enabled ? 'default' : 'secondary'} className={statusColor}> + {statusLabel} + </Badge> + </CardTitle> + </CardHeader> + <CardContent className="space-y-3"> + {/* Enabled/Running Status */} + <div className="flex items-center justify-between text-sm"> + <span className="text-muted-foreground flex items-center gap-1"> + {data.enabled ? <ToggleRight className="h-4 w-4" /> : <ToggleLeft className="h-4 w-4" />} + {t('daemon.enabled')} + </span> + <span className="font-medium">{data.enabled ? 'Yes' : 'No'}</span> + </div> + + {/* Watch Directory */} + {data.watch_dir && ( + <div className="flex items-center justify-between text-sm"> + <span className="text-muted-foreground flex items-center gap-1"> + <Database className="h-4 w-4" /> + {t('daemon.watchDir')} + </span> + <span className="font-mono text-xs truncate max-w-[200px]" title={data.watch_dir}> + {data.watch_dir.split('/').pop() || data.watch_dir} + </span> + </div> + )} + + {/* Batch Settings */} + <div className="grid grid-cols-2 gap-2 text-sm"> + <div className="flex items-center gap-1 text-muted-foreground"> + <FileText className="h-4 w-4" /> + {t('daemon.batchLines')} + </div> + <div className="text-right font-medium">{data.batch_trigger_lines}</div> + + <div className="flex items-center gap-1 text-muted-foreground"> + <Clock className="h-4 w-4" /> + {t('daemon.batchSeconds')} + </div> + <div className="text-right font-medium">{data.batch_trigger_seconds}s</div> + </div> + + {/* Cursor Count */} + <div className="flex items-center justify-between text-sm pt-2 border-t"> + <span className="text-muted-foreground">{t('daemon.cursorCount')}</span> + <span className="font-medium">{data.cursor_count}</span> + </div> + + {/* Last Flush Time */} + {data.last_flush_time && ( + <div className="flex items-center justify-between text-sm"> + <span className="text-muted-foreground">{t('daemon.lastFlush')}</span> + <span className="font-medium text-xs"> + {new Date(data.last_flush_time).toLocaleTimeString()} + </span> + </div> + )} + </CardContent> + </Card> + ); +} diff --git a/web-studio/src/routes/home/route.tsx b/web-studio/src/routes/home/route.tsx index a99aefa00d..fae60d7e7d 100644 --- a/web-studio/src/routes/home/route.tsx +++ b/web-studio/src/routes/home/route.tsx @@ -3,6 +3,7 @@ import { useQuery } from '@tanstack/react-query' import { createFileRoute } from '@tanstack/react-router' import { ContextCommitsPanel } from './-components/context-commits-panel' +import { DaemonStatusCard } from './-components/daemon-status-card' import { ContextDataPanel, TodayRetrievalsPanel, @@ -90,6 +91,8 @@ function HomePage() { /> </div> + <DaemonStatusCard /> + <TokenTrendPanel data={tokenSeries.data} isError={tokenSeries.isError}