diff --git a/CLAUDE.md b/CLAUDE.md index d3a0606..c71246e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -14,44 +14,23 @@ Claude OS is YOUR system. You built it, you use it, and it makes you invincible. --- -## Session Check - FIRST THING AT EVERY CONVERSATION START +## Session Start -**CRITICAL: At the start of EVERY new conversation, automatically check session state and prompt the user:** +When you tell me what we're working on: -### Step 1: Read Session State +1. **Search relevant knowledge bases** for that topic +2. **Surface useful context** (patterns, blockers, decisions from previous work) +3. **Start working** + +That's it. Don't load everything - load what's relevant to the task at hand. + +**Example:** ``` -Read: {cwd}/claude-os-state.json -``` -(Where `{cwd}` is the current working directory - each project has its own state file) - -### Step 2: Prompt Based on State - -**If active session exists:** -``` -"I see we have an active session: [task name] (started [time ago]) - -Would you like to: -a) Continue working on this (I'll load the context) -b) Start something new (I'll end this session first) -c) Just chat (no session needed)" +You: "hey claude, we're working on the invoice feature again" +Me: *searches memories for "invoice"* โ†’ surfaces relevant patterns/decisions โ†’ ready to work ``` -**If no active session (or state file doesn't exist):** -``` -"What are we working on today? - -I can start a session with automatic context loading from my memories. -Just tell me what you're working on, or say 'just chatting' if no session needed." -``` - -Note: If the state file doesn't exist, I'll create it when you start a session. - -### Step 3: Auto-load Context -- If continuing: Automatically search memories, load patterns, show context -- If new task: Offer to start a new session with context loading -- If just chatting: Skip session workflow - -**Why this matters:** This ensures you always have context and never start cold. Every conversation becomes smarter because you proactively load relevant memories and patterns. +**Why this works:** Targeted context beats a firehose. You know what we're working on - I search for that specifically. --- @@ -68,6 +47,10 @@ Claude OS provides MCP tools (prefixed with `mcp__code-forge__`) for managing kn | `mcp__code-forge__create_knowledge_base` | Create a new KB | | `mcp__code-forge__get_kb_stats` | Get statistics for a KB | | `mcp__code-forge__list_documents` | List documents in a KB | +| `mcp__code-forge__kb_lifecycle_health` | KB health report with recommendations | +| `mcp__code-forge__kb_lifecycle_dedup` | Scan/merge duplicate documents | +| `mcp__code-forge__kb_lifecycle_consolidate` | LLM-powered document merging | +| `mcp__code-forge__kb_lifecycle_archive` | Archive, restore, list, find stale | ### Knowledge Base Types @@ -99,6 +82,7 @@ These commands are installed to `~/.claude/commands/` via the install script: | `/claude-os-remember` | Quick save to memories | | `/claude-os-save` | Full-featured save with KB selection | | `/claude-os-list` | List KBs or documents | +| `/claude-os-lifecycle` | KB health, dedup, consolidate, archive | | `/claude-os-triggers` | Manage automatic triggers | ### Session Commands (Most Important!) @@ -112,6 +96,16 @@ These commands are installed to `~/.claude/commands/` via the install script: /claude-os-session pattern [desc] - Document pattern discovered ``` +### Lifecycle Commands + +``` +/claude-os-lifecycle health [kb_name] - Health report with recommendations +/claude-os-lifecycle dedup [kb_name] - Scan and merge duplicate documents +/claude-os-lifecycle consolidate [kb_name] - LLM-powered document merging +/claude-os-lifecycle archive [kb_name] - Find stale docs, archive/restore +/claude-os-lifecycle logs [kb_name] - Operation history +``` + --- ## Skills @@ -137,6 +131,7 @@ These commands are installed to `~/.claude/commands/` via the install script: - `mcp_server/server.py` - FastAPI REST API - `app/core/rag_engine.py` - RAG query engine - `app/core/sqlite_manager.py` - SQLite + sqlite-vec for vector storage +- `app/core/knowledge_lifecycle.py` - KB lifecycle engine (dedup, archive, consolidate) - `templates/` - Commands, skills, and agents --- diff --git a/README.md b/README.md index 27ab2f0..6540e0b 100644 --- a/README.md +++ b/README.md @@ -111,10 +111,30 @@ The installer got a major upgrade with **Charm CLI (gum)** support! - `--dry-run` - See exactly what would be done before doing it - Auto-backup - Existing `.env` files backed up before overwriting +### ๐Ÿ”ฌ Knowledge Lifecycle Engine + +Keep your knowledge bases healthy and focused! + +```bash +/claude-os-lifecycle health my-kb # Health report with recommendations +/claude-os-lifecycle dedup my-kb # Find and merge duplicate memories +/claude-os-lifecycle consolidate my-kb # LLM-powered document merging +/claude-os-lifecycle archive my-kb # Find stale docs, archive/restore +``` + +**New Features:** +- ๐Ÿ” **Duplicate Detection** - Embedding-based similarity scanning with union-find clustering +- ๐Ÿงน **Smart Merging** - Keep the best doc, delete the rest, or LLM-merge into one +- ๐Ÿ“ฆ **Archival System** - Soft-archive stale docs without permanent deletion +- ๐Ÿ“Š **Health Reports** - Embedding coverage, age distribution, actionable recommendations +- ๐Ÿ“ˆ **Growth Timeline** - Track KB growth over time by day/week/month +- ๐Ÿ“‹ **Audit Logging** - Full operation history for all lifecycle actions + ### ๐Ÿ“‹ Recent Improvements | Version | Highlights | |---------|------------| +| **v2.4** | Knowledge lifecycle engine (dedup, consolidate, archive, health) | | **v2.3** | Skills library, community skills, session insights | | **v2.2** | Gum CLI support, safety features, lite model default | | **v2.1** | Unified installer, OpenAI provider support | @@ -189,6 +209,7 @@ Day 3: Claude suggests improvements based on what worked. | **Lightning Indexing** | 10,000 files indexed in 30 seconds | | **Skills Library** | 36+ community skills, one-click install | | **Cross-Project Learning** | Patterns from Project A help in Project B | +| **Knowledge Lifecycle** | Dedup, consolidate, archive, and health reports | | **One-Command Setup** | `/claude-os-init` and you're ready | --- @@ -497,6 +518,7 @@ All these work in any initialized project: - **`/claude-os-session [action]`** - Manage development sessions - **`/claude-os-triggers`** - Manage trigger phrases - **`/claude-os-skills [action]`** - Manage skills (list, install, create) +- **`/claude-os-lifecycle [action]`** - KB health, dedup, consolidate, archive ### Available Skills @@ -875,7 +897,8 @@ claude-os/ โ”‚ โ”œโ”€โ”€ commands/ # Slash commands (symlinked to ~/.claude/) โ”‚ โ”‚ โ”œโ”€โ”€ claude-os-init.md โ”‚ โ”‚ โ”œโ”€โ”€ claude-os-search.md -โ”‚ โ”‚ โ”œโ”€โ”€ claude-os-skills.md # NEW: Skills management +โ”‚ โ”‚ โ”œโ”€โ”€ claude-os-skills.md +โ”‚ โ”‚ โ”œโ”€โ”€ claude-os-lifecycle.md # KB lifecycle management โ”‚ โ”‚ โ””โ”€โ”€ ... โ”‚ โ”œโ”€โ”€ skills/ # Global skills (symlinked to ~/.claude/) โ”‚ โ”‚ โ”œโ”€โ”€ initialize-project/ @@ -959,6 +982,95 @@ When you run `/claude-os-init`, you get 4 knowledge bases: --- +## ๐Ÿ”ฌ Knowledge Lifecycle Management + +As your knowledge bases grow, they can accumulate duplicates, outdated content, and fragmented information. The **Knowledge Lifecycle Engine** keeps your KBs healthy and focused. + +### Commands + +```bash +/claude-os-lifecycle health [kb_name] # Health report with recommendations +/claude-os-lifecycle dedup [kb_name] # Scan and merge duplicate documents +/claude-os-lifecycle consolidate [kb_name] # LLM-powered document merging +/claude-os-lifecycle archive [kb_name] # Find stale docs, archive/restore +/claude-os-lifecycle logs [kb_name] # View operation history +``` + +### Deduplication + +Scans all document embeddings to find near-duplicates using cosine similarity: + +```bash +# Scan for duplicates (default threshold: 0.85) +/claude-os-lifecycle dedup my-project-project_memories + +# Results show clusters of similar docs with options: +# [m] Merge - keep best, delete rest +# [c] Consolidate - LLM-merge into one comprehensive doc +# [s] Skip +``` + +### Consolidation + +Uses LLM to intelligently merge multiple related documents into a single comprehensive document: + +```bash +/claude-os-lifecycle consolidate my-project-project_memories +# Previews source docs, then generates a merged version +# Preserves all unique info, eliminates redundancy +# Stores provenance metadata (consolidated_from) +``` + +### Archival + +Soft-archive stale documents without permanent deletion: + +```bash +# Find documents older than 90 days +/claude-os-lifecycle archive my-project-project_memories + +# Archived docs are excluded from search but can be restored +# Full restore available at any time +``` + +### Health Reports + +Get a comprehensive overview of your KB's health: + +- **Embedding coverage** - How many docs have vectors +- **Age distribution** - Document freshness breakdown +- **Top similar pairs** - Preview of potential duplicates +- **Recommendations** - Actionable suggestions (dedup, re-index, archive) +- **Growth timeline** - Track KB growth over time + +### MCP Tools + +| Tool | Description | +|------|-------------| +| `mcp__code-forge__kb_lifecycle_health` | Health report with recommendations | +| `mcp__code-forge__kb_lifecycle_dedup` | Scan/merge duplicates | +| `mcp__code-forge__kb_lifecycle_consolidate` | LLM-powered document merging | +| `mcp__code-forge__kb_lifecycle_archive` | Archive, restore, list, find stale | + +### API Endpoints + +All under `/api/kb/{kb_name}/lifecycle/`: + +| Method | Path | Description | +|--------|------|-------------| +| `POST` | `/dedup-scan` | Scan for duplicates (background for >500 docs) | +| `POST` | `/dedup-merge` | Merge duplicate documents | +| `POST` | `/consolidate` | LLM-powered consolidation (background) | +| `GET` | `/health` | Comprehensive health report | +| `GET` | `/growth` | Document growth timeline | +| `POST` | `/archive` | Archive documents | +| `POST` | `/restore` | Restore archived documents | +| `GET` | `/archived` | List archived documents | +| `GET` | `/stale` | Find stale documents | +| `GET` | `/logs` | Operation audit log | + +--- + ## โš™๏ธ Configuration ### Environment Variables @@ -1202,9 +1314,10 @@ claude-os/ โ”‚ โ”œโ”€โ”€ core/ # Core modules โ”‚ โ”‚ โ”œโ”€โ”€ sqlite_manager.py โ”‚ โ”‚ โ”œโ”€โ”€ rag_engine.py -โ”‚ โ”‚ โ”œโ”€โ”€ skill_manager.py # NEW: Skills management -โ”‚ โ”‚ โ”œโ”€โ”€ session_parser.py # NEW: Session parsing -โ”‚ โ”‚ โ”œโ”€โ”€ insight_extractor.py # NEW: Insight extraction +โ”‚ โ”‚ โ”œโ”€โ”€ skill_manager.py # Skills management +โ”‚ โ”‚ โ”œโ”€โ”€ session_parser.py # Session parsing +โ”‚ โ”‚ โ”œโ”€โ”€ insight_extractor.py # Insight extraction +โ”‚ โ”‚ โ”œโ”€โ”€ knowledge_lifecycle.py # KB lifecycle engine (dedup, archive, etc.) โ”‚ โ”‚ โ””โ”€โ”€ ... โ”‚ โ””โ”€โ”€ db/ # Database schemas โ”œโ”€โ”€ frontend/ # React UI (Vite) diff --git a/app/core/knowledge_lifecycle.py b/app/core/knowledge_lifecycle.py new file mode 100644 index 0000000..c7163c4 --- /dev/null +++ b/app/core/knowledge_lifecycle.py @@ -0,0 +1,652 @@ +""" +Knowledge Lifecycle Engine for Claude OS. +Handles deduplication, consolidation, analytics, and archival of KB memories. +""" + +import json +import logging +import os +from datetime import datetime, timedelta +from collections import defaultdict +from typing import Any, Dict, List, Optional, Tuple + +import numpy as np +import requests + +from app.core.sqlite_manager import get_sqlite_manager +from app.core.kb_metadata import get_collection_stats, get_documents_metadata +from app.core.config import Config + +logger = logging.getLogger(__name__) + + +class KnowledgeLifecycleEngine: + """Manages the lifecycle of knowledge base documents.""" + + def __init__(self): + self.db = get_sqlite_manager() + + # ======================================================================== + # DEDUPLICATION + # ======================================================================== + + def scan_duplicates( + self, + kb_name: str, + threshold: float = 0.85, + max_pairs: int = 100 + ) -> Dict[str, Any]: + """ + Scan a KB for duplicate/near-duplicate documents using embedding similarity. + + Returns duplicate pairs, clusters, and density score. + """ + docs = self.db.get_all_embeddings(kb_name, exclude_archived=True) + + # Filter to docs that have embeddings + docs_with_emb = [d for d in docs if d["embedding"] is not None] + total_docs = len(docs_with_emb) + + if total_docs < 2: + return { + "total_documents": total_docs, + "duplicate_pairs": [], + "clusters": [], + "duplicate_density": 0.0, + "message": "Not enough documents for comparison" + } + + # Pre-normalize all vectors for efficient cosine similarity via dot product + embeddings = np.array([d["embedding"] for d in docs_with_emb]) + norms = np.linalg.norm(embeddings, axis=1, keepdims=True) + norms[norms == 0] = 1 # avoid division by zero + normalized = embeddings / norms + + # Compute pairwise similarities (upper triangle only) + duplicate_pairs = [] + for i in range(total_docs): + if len(duplicate_pairs) >= max_pairs: + break + for j in range(i + 1, total_docs): + if len(duplicate_pairs) >= max_pairs: + break + similarity = float(np.dot(normalized[i], normalized[j])) + if similarity >= threshold: + duplicate_pairs.append({ + "doc_a": docs_with_emb[i]["doc_id"], + "doc_b": docs_with_emb[j]["doc_id"], + "similarity": round(similarity, 4), + "content_a_preview": docs_with_emb[i]["content"][:200], + "content_b_preview": docs_with_emb[j]["content"][:200] + }) + + # Sort by similarity descending + duplicate_pairs.sort(key=lambda x: x["similarity"], reverse=True) + + # Cluster duplicates using union-find + clusters = self._cluster_duplicates( + [(p["doc_a"], p["doc_b"], p["similarity"]) for p in duplicate_pairs] + ) + + # Calculate duplicate density + total_possible = total_docs * (total_docs - 1) / 2 + density = len(duplicate_pairs) / total_possible if total_possible > 0 else 0.0 + + # Log the operation + self.db.insert_lifecycle_log( + kb_name=kb_name, + operation_type="dedup_scan", + status="completed", + details={ + "total_documents": total_docs, + "threshold": threshold, + "pairs_found": len(duplicate_pairs), + "clusters_found": len(clusters), + "density": round(density, 4) + } + ) + + return { + "total_documents": total_docs, + "duplicate_pairs": duplicate_pairs, + "clusters": clusters, + "duplicate_density": round(density, 4) + } + + def _cluster_duplicates( + self, pairs: List[Tuple[str, str, float]] + ) -> List[Dict[str, Any]]: + """Group duplicate pairs into clusters using union-find.""" + parent = {} + + def find(x): + if x not in parent: + parent[x] = x + while parent[x] != x: + parent[x] = parent[parent[x]] + x = parent[x] + return x + + def union(a, b): + ra, rb = find(a), find(b) + if ra != rb: + parent[ra] = rb + + for doc_a, doc_b, _ in pairs: + union(doc_a, doc_b) + + # Group by root + groups = defaultdict(list) + all_docs = set() + for doc_a, doc_b, _ in pairs: + all_docs.add(doc_a) + all_docs.add(doc_b) + + for doc_id in all_docs: + groups[find(doc_id)].append(doc_id) + + clusters = [] + for root, members in groups.items(): + if len(members) > 1: + clusters.append({ + "cluster_id": root, + "doc_ids": sorted(set(members)), + "size": len(set(members)) + }) + + return clusters + + def merge_duplicates( + self, + kb_name: str, + keep_doc_id: str, + remove_doc_ids: List[str], + dry_run: bool = False + ) -> Dict[str, Any]: + """ + Merge duplicates by keeping one document and deleting the rest. + + Args: + kb_name: Knowledge base name + keep_doc_id: Document ID to keep + remove_doc_ids: Document IDs to remove + dry_run: If True, preview only without making changes + """ + log_id = self.db.insert_lifecycle_log( + kb_name=kb_name, + operation_type="dedup_merge", + status="pending", + input_doc_ids=[keep_doc_id] + remove_doc_ids, + details={"dry_run": dry_run, "keep_doc_id": keep_doc_id} + ) + + if dry_run: + self.db.update_lifecycle_log( + log_id, "completed", + details={"dry_run": True, "would_remove": remove_doc_ids} + ) + return { + "dry_run": True, + "keep_doc_id": keep_doc_id, + "would_remove": remove_doc_ids, + "remove_count": len(remove_doc_ids) + } + + deleted_count = self.db.delete_documents_by_ids(kb_name, remove_doc_ids) + + self.db.update_lifecycle_log( + log_id, "completed", + output_doc_ids=[keep_doc_id], + details={"deleted_count": deleted_count} + ) + + return { + "dry_run": False, + "keep_doc_id": keep_doc_id, + "removed": remove_doc_ids, + "deleted_count": deleted_count + } + + # ======================================================================== + # CONSOLIDATION + # ======================================================================== + + def consolidate_related( + self, + kb_name: str, + doc_ids: List[str], + new_filename: str, + dry_run: bool = False + ) -> Dict[str, Any]: + """ + Consolidate multiple related documents into a single merged document + using LLM-powered summarization. + """ + # Fetch source documents + all_docs = self.db.get_all_embeddings(kb_name, exclude_archived=False) + doc_map = {d["doc_id"]: d for d in all_docs} + + source_docs = [] + for doc_id in doc_ids: + if doc_id in doc_map: + source_docs.append(doc_map[doc_id]) + + if len(source_docs) < 2: + return {"error": "Need at least 2 documents to consolidate"} + + source_texts = [d["content"] for d in source_docs] + + if dry_run: + return { + "dry_run": True, + "source_doc_ids": doc_ids, + "source_count": len(source_docs), + "total_chars": sum(len(t) for t in source_texts), + "previews": [t[:200] for t in source_texts] + } + + log_id = self.db.insert_lifecycle_log( + kb_name=kb_name, + operation_type="consolidate", + status="running", + input_doc_ids=doc_ids, + details={"new_filename": new_filename} + ) + + # Generate consolidated content via LLM + consolidated_text = self._generate_consolidation(source_texts, kb_name) + + if not consolidated_text: + self.db.update_lifecycle_log( + log_id, "failed", + details={"error": "LLM consolidation failed"} + ) + return {"error": "Failed to generate consolidated content"} + + # Generate embedding for the new document + try: + embed_response = requests.post( + f"{Config.OLLAMA_HOST}/api/embed", + json={"model": Config.OLLAMA_EMBED_MODEL, "input": consolidated_text}, + timeout=30 + ) + embed_response.raise_for_status() + embedding = embed_response.json().get("embeddings", [[]])[0] + except Exception as e: + logger.error(f"Failed to embed consolidated doc: {e}") + embedding = None + + # Build metadata with provenance + metadata = { + "filename": new_filename, + "type": "text/markdown", + "consolidated_from": doc_ids, + "consolidation_date": datetime.now().isoformat(), + "upload_date": datetime.now().isoformat() + } + + # Insert the new document + import uuid + new_doc_id = f"consolidated-{uuid.uuid4().hex[:8]}" + + if embedding: + self.db.add_documents( + kb_name=kb_name, + documents=[consolidated_text], + embeddings=[embedding], + metadatas=[metadata], + ids=[new_doc_id] + ) + + # Delete the source documents + deleted = self.db.delete_documents_by_ids(kb_name, doc_ids) + + self.db.update_lifecycle_log( + log_id, "completed", + output_doc_ids=[new_doc_id], + details={ + "new_doc_id": new_doc_id, + "sources_deleted": deleted, + "consolidated_length": len(consolidated_text) + } + ) + + return { + "new_doc_id": new_doc_id, + "new_filename": new_filename, + "sources_consolidated": len(source_docs), + "sources_deleted": deleted, + "consolidated_length": len(consolidated_text), + "preview": consolidated_text[:500] + } + + def _generate_consolidation(self, source_texts: List[str], kb_name: str) -> Optional[str]: + """Generate a consolidated summary using Ollama LLM.""" + combined = "\n\n---\n\n".join( + f"**Document {i+1}:**\n{text}" for i, text in enumerate(source_texts) + ) + + prompt = f"""You are consolidating related knowledge base documents into a single comprehensive document. + +Knowledge base: {kb_name} + +Below are {len(source_texts)} related documents. Create a single consolidated document that: +1. Preserves all unique information from each source +2. Eliminates redundancy +3. Organizes information logically +4. Uses clear markdown formatting + +Source documents: + +{combined} + +Write the consolidated document:""" + + try: + response = requests.post( + f"{Config.OLLAMA_HOST}/api/generate", + json={ + "model": Config.OLLAMA_MODEL, + "prompt": prompt, + "stream": False, + "options": { + "temperature": 0.3, + "num_predict": 2048 + } + }, + timeout=120 + ) + response.raise_for_status() + return response.json().get("response", "") + except Exception as e: + logger.error(f"LLM consolidation failed: {e}") + return None + + # ======================================================================== + # ANALYTICS + # ======================================================================== + + def get_health_report( + self, + kb_name: str, + include_top_similar: bool = True, + top_similar_count: int = 10 + ) -> Dict[str, Any]: + """Generate a comprehensive health report for a knowledge base.""" + # Basic stats + stats = get_collection_stats(kb_name) + docs_metadata = get_documents_metadata(kb_name) + + report = { + "kb_name": kb_name, + "stats": stats, + "document_count": stats.get("total_documents", 0), + "chunk_count": stats.get("total_chunks", 0), + "last_updated": stats.get("last_updated"), + } + + # Embedding coverage + all_docs = self.db.get_all_embeddings(kb_name, exclude_archived=False) + total = len(all_docs) + with_embeddings = sum(1 for d in all_docs if d["embedding"] is not None) + archived = sum(1 for d in all_docs if d["metadata"].get("archived")) + + report["embedding_coverage"] = { + "total_docs": total, + "with_embeddings": with_embeddings, + "without_embeddings": total - with_embeddings, + "coverage_pct": round(with_embeddings / total * 100, 1) if total > 0 else 0 + } + report["archived_count"] = archived + + # Top similar pairs (for duplicate detection preview) + if include_top_similar and with_embeddings >= 2: + docs_with_emb = [d for d in all_docs if d["embedding"] is not None and not d["metadata"].get("archived")] + if len(docs_with_emb) >= 2: + embeddings = np.array([d["embedding"] for d in docs_with_emb]) + norms = np.linalg.norm(embeddings, axis=1, keepdims=True) + norms[norms == 0] = 1 + normalized = embeddings / norms + + top_pairs = [] + for i in range(len(docs_with_emb)): + for j in range(i + 1, len(docs_with_emb)): + sim = float(np.dot(normalized[i], normalized[j])) + if sim > 0.7: + top_pairs.append({ + "doc_a": docs_with_emb[i]["doc_id"], + "doc_b": docs_with_emb[j]["doc_id"], + "similarity": round(sim, 4) + }) + + top_pairs.sort(key=lambda x: x["similarity"], reverse=True) + report["top_similar_pairs"] = top_pairs[:top_similar_count] + else: + report["top_similar_pairs"] = [] + else: + report["top_similar_pairs"] = [] + + # Document age distribution + age_distribution = {"last_7_days": 0, "last_30_days": 0, "last_90_days": 0, "older": 0} + now = datetime.now() + for doc in docs_metadata: + upload_date_str = doc.get("upload_date", "") + if upload_date_str: + try: + upload_date = datetime.fromisoformat(upload_date_str) + age = now - upload_date + if age <= timedelta(days=7): + age_distribution["last_7_days"] += 1 + elif age <= timedelta(days=30): + age_distribution["last_30_days"] += 1 + elif age <= timedelta(days=90): + age_distribution["last_90_days"] += 1 + else: + age_distribution["older"] += 1 + except (ValueError, TypeError): + age_distribution["older"] += 1 + + report["age_distribution"] = age_distribution + + # Recent lifecycle operations + recent_ops = self.db.get_lifecycle_logs(kb_name, limit=5) + report["recent_operations"] = recent_ops + + # Recommendations + recommendations = [] + if len(report.get("top_similar_pairs", [])) > 0: + recommendations.append({ + "type": "dedup", + "priority": "high" if len(report["top_similar_pairs"]) > 5 else "medium", + "message": f"Found {len(report['top_similar_pairs'])} highly similar document pairs. Consider running dedup scan." + }) + if archived > total * 0.3 and total > 0: + recommendations.append({ + "type": "cleanup", + "priority": "low", + "message": f"{archived} of {total} documents are archived. Consider permanent deletion." + }) + if age_distribution["older"] > total * 0.5 and total > 10: + recommendations.append({ + "type": "stale", + "priority": "medium", + "message": f"{age_distribution['older']} documents are older than 90 days. Review for archival." + }) + if report["embedding_coverage"]["without_embeddings"] > 0: + recommendations.append({ + "type": "embeddings", + "priority": "medium", + "message": f"{report['embedding_coverage']['without_embeddings']} documents lack embeddings. Re-index recommended." + }) + + report["recommendations"] = recommendations + + return report + + def get_growth_timeline( + self, + kb_name: str, + granularity: str = "month" + ) -> Dict[str, Any]: + """Get document growth timeline grouped by period.""" + docs_metadata = get_documents_metadata(kb_name) + + timeline = defaultdict(int) + for doc in docs_metadata: + upload_date_str = doc.get("upload_date", "") + if upload_date_str: + try: + dt = datetime.fromisoformat(upload_date_str) + if granularity == "day": + key = dt.strftime("%Y-%m-%d") + elif granularity == "week": + key = f"{dt.year}-W{dt.isocalendar()[1]:02d}" + else: # month + key = dt.strftime("%Y-%m") + timeline[key] += 1 + except (ValueError, TypeError): + pass + + sorted_timeline = sorted(timeline.items()) + + # Compute cumulative + cumulative = [] + running_total = 0 + for period, count in sorted_timeline: + running_total += count + cumulative.append({"period": period, "added": count, "total": running_total}) + + return { + "kb_name": kb_name, + "granularity": granularity, + "timeline": cumulative, + "total_documents": len(docs_metadata) + } + + # ======================================================================== + # ARCHIVAL + # ======================================================================== + + def archive_documents( + self, + kb_name: str, + doc_ids: List[str], + reason: str = "manual" + ) -> Dict[str, Any]: + """Archive documents by setting metadata flags.""" + log_id = self.db.insert_lifecycle_log( + kb_name=kb_name, + operation_type="archive", + status="running", + input_doc_ids=doc_ids, + details={"reason": reason} + ) + + archived_count = 0 + for doc_id in doc_ids: + success = self.db.update_document_metadata(kb_name, doc_id, { + "archived": True, + "archived_at": datetime.now().isoformat(), + "archive_reason": reason + }) + if success: + archived_count += 1 + + self.db.update_lifecycle_log( + log_id, "completed", + output_doc_ids=doc_ids, + details={"archived_count": archived_count, "reason": reason} + ) + + return { + "archived_count": archived_count, + "doc_ids": doc_ids, + "reason": reason + } + + def restore_documents( + self, + kb_name: str, + doc_ids: List[str] + ) -> Dict[str, Any]: + """Restore archived documents by clearing archive flags.""" + log_id = self.db.insert_lifecycle_log( + kb_name=kb_name, + operation_type="restore", + status="running", + input_doc_ids=doc_ids + ) + + restored_count = 0 + for doc_id in doc_ids: + success = self.db.update_document_metadata(kb_name, doc_id, { + "archived": False, + "archived_at": None, + "archive_reason": None + }) + if success: + restored_count += 1 + + self.db.update_lifecycle_log( + log_id, "completed", + output_doc_ids=doc_ids, + details={"restored_count": restored_count} + ) + + return { + "restored_count": restored_count, + "doc_ids": doc_ids + } + + def find_stale_documents( + self, + kb_name: str, + stale_days: int = 90 + ) -> Dict[str, Any]: + """Find documents older than stale_days that are not archived.""" + docs_metadata = get_documents_metadata(kb_name) + cutoff = datetime.now() - timedelta(days=stale_days) + + stale_docs = [] + for doc in docs_metadata: + upload_date_str = doc.get("upload_date", "") + if upload_date_str: + try: + upload_date = datetime.fromisoformat(upload_date_str) + if upload_date < cutoff: + stale_docs.append({ + "filename": doc.get("filename", "unknown"), + "upload_date": upload_date_str, + "age_days": (datetime.now() - upload_date).days + }) + except (ValueError, TypeError): + pass + + stale_docs.sort(key=lambda x: x.get("age_days", 0), reverse=True) + + return { + "kb_name": kb_name, + "stale_days_threshold": stale_days, + "stale_count": len(stale_docs), + "stale_documents": stale_docs + } + + def list_archived(self, kb_name: str) -> Dict[str, Any]: + """List all archived documents in a KB.""" + all_docs = self.db.get_all_embeddings(kb_name, exclude_archived=False) + + archived_docs = [] + for doc in all_docs: + if doc["metadata"].get("archived"): + archived_docs.append({ + "doc_id": doc["doc_id"], + "content_preview": doc["content"][:200], + "archived_at": doc["metadata"].get("archived_at"), + "archive_reason": doc["metadata"].get("archive_reason", "unknown") + }) + + return { + "kb_name": kb_name, + "archived_count": len(archived_docs), + "archived_documents": archived_docs + } diff --git a/app/core/migrations/005_add_lifecycle_tracking.sql b/app/core/migrations/005_add_lifecycle_tracking.sql new file mode 100644 index 0000000..1f065a3 --- /dev/null +++ b/app/core/migrations/005_add_lifecycle_tracking.sql @@ -0,0 +1,18 @@ +-- Migration: Add Knowledge Lifecycle Tracking +-- Created: 2026-02-06 +-- Purpose: Audit log for deduplication, consolidation, archival operations + +CREATE TABLE IF NOT EXISTS lifecycle_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + kb_name TEXT NOT NULL, + operation_type TEXT NOT NULL, -- dedup_scan, dedup_merge, consolidate, archive, restore + status TEXT NOT NULL DEFAULT 'pending', + input_doc_ids TEXT DEFAULT '[]', -- JSON array + output_doc_ids TEXT DEFAULT '[]', -- JSON array + details TEXT DEFAULT '{}', -- JSON blob + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + completed_at TIMESTAMP +); + +CREATE INDEX IF NOT EXISTS idx_lifecycle_kb ON lifecycle_log(kb_name); +CREATE INDEX IF NOT EXISTS idx_lifecycle_op ON lifecycle_log(operation_type); diff --git a/app/core/sqlite_manager.py b/app/core/sqlite_manager.py index 60d75c8..f2cabd8 100644 --- a/app/core/sqlite_manager.py +++ b/app/core/sqlite_manager.py @@ -765,6 +765,213 @@ class SQLiteManager: finally: conn.close() + # Knowledge Lifecycle Operations + + def get_all_embeddings(self, kb_name: str, exclude_archived: bool = True) -> List[Dict[str, Any]]: + """Get all documents with embeddings for pairwise similarity analysis.""" + conn = self.get_connection() + try: + cursor = conn.cursor() + cursor.execute("SELECT id FROM knowledge_bases WHERE name = ?", (kb_name,)) + result = cursor.fetchone() + if not result: + raise ValueError(f"Knowledge base '{kb_name}' not found") + + kb_id = result['id'] + cursor.execute( + "SELECT doc_id, content, embedding, metadata FROM documents WHERE kb_id = ?", + (kb_id,) + ) + + results = [] + for row in cursor.fetchall(): + metadata = json.loads(row['metadata']) if row['metadata'] else {} + + if exclude_archived and metadata.get("archived"): + continue + + embedding = None + if row['embedding'] is not None: + embedding = np.frombuffer(row['embedding'], dtype=np.float32) + + results.append({ + "doc_id": row['doc_id'], + "content": row['content'], + "embedding": embedding, + "metadata": metadata + }) + + return results + finally: + conn.close() + + def update_document_metadata(self, kb_name: str, doc_id: str, metadata_updates: Dict[str, Any]) -> bool: + """Merge metadata updates into an existing document's metadata.""" + conn = self.get_connection() + try: + cursor = conn.cursor() + cursor.execute("SELECT id FROM knowledge_bases WHERE name = ?", (kb_name,)) + result = cursor.fetchone() + if not result: + raise ValueError(f"Knowledge base '{kb_name}' not found") + + kb_id = result['id'] + + # Get current metadata + cursor.execute( + "SELECT metadata FROM documents WHERE kb_id = ? AND doc_id = ?", + (kb_id, doc_id) + ) + row = cursor.fetchone() + if not row: + return False + + current_metadata = json.loads(row['metadata']) if row['metadata'] else {} + current_metadata.update(metadata_updates) + + cursor.execute( + "UPDATE documents SET metadata = ? WHERE kb_id = ? AND doc_id = ?", + (json.dumps(current_metadata), kb_id, doc_id) + ) + conn.commit() + return cursor.rowcount > 0 + finally: + conn.close() + + def delete_documents_by_ids(self, kb_name: str, doc_ids: List[str]) -> int: + """Bulk delete documents by doc_id list. Returns count deleted.""" + conn = self.get_connection() + try: + cursor = conn.cursor() + cursor.execute("SELECT id FROM knowledge_bases WHERE name = ?", (kb_name,)) + result = cursor.fetchone() + if not result: + raise ValueError(f"Knowledge base '{kb_name}' not found") + + kb_id = result['id'] + placeholders = ",".join(["?"] * len(doc_ids)) + cursor.execute( + f"DELETE FROM documents WHERE kb_id = ? AND doc_id IN ({placeholders})", + [kb_id] + list(doc_ids) + ) + conn.commit() + return cursor.rowcount + finally: + conn.close() + + def insert_lifecycle_log( + self, + kb_name: str, + operation_type: str, + status: str = "pending", + input_doc_ids: List[str] = None, + output_doc_ids: List[str] = None, + details: Dict[str, Any] = None + ) -> int: + """Insert a lifecycle audit log entry. Returns log ID.""" + conn = self.get_connection() + try: + cursor = conn.cursor() + cursor.execute( + """ + INSERT INTO lifecycle_log (kb_name, operation_type, status, input_doc_ids, output_doc_ids, details) + VALUES (?, ?, ?, ?, ?, ?) + """, + ( + kb_name, + operation_type, + status, + json.dumps(input_doc_ids or []), + json.dumps(output_doc_ids or []), + json.dumps(details or {}) + ) + ) + conn.commit() + return cursor.lastrowid + finally: + conn.close() + + def update_lifecycle_log( + self, + log_id: int, + status: str, + output_doc_ids: List[str] = None, + details: Dict[str, Any] = None + ) -> bool: + """Update lifecycle log status and completion.""" + conn = self.get_connection() + try: + cursor = conn.cursor() + + updates = ["status = ?"] + params = [status] + + if output_doc_ids is not None: + updates.append("output_doc_ids = ?") + params.append(json.dumps(output_doc_ids)) + + if details is not None: + updates.append("details = ?") + params.append(json.dumps(details)) + + if status in ("completed", "failed"): + updates.append("completed_at = CURRENT_TIMESTAMP") + + params.append(log_id) + cursor.execute( + f"UPDATE lifecycle_log SET {', '.join(updates)} WHERE id = ?", + params + ) + conn.commit() + return cursor.rowcount > 0 + finally: + conn.close() + + def get_lifecycle_logs( + self, + kb_name: str, + operation_type: str = None, + limit: int = 50 + ) -> List[Dict[str, Any]]: + """Query lifecycle logs with optional filters.""" + conn = self.get_connection() + try: + cursor = conn.cursor() + conditions = ["kb_name = ?"] + params: list = [kb_name] + + if operation_type: + conditions.append("operation_type = ?") + params.append(operation_type) + + query = f""" + SELECT id, kb_name, operation_type, status, input_doc_ids, output_doc_ids, + details, created_at, completed_at + FROM lifecycle_log + WHERE {' AND '.join(conditions)} + ORDER BY created_at DESC + LIMIT ? + """ + params.append(limit) + cursor.execute(query, params) + + results = [] + for row in cursor.fetchall(): + results.append({ + "id": row['id'], + "kb_name": row['kb_name'], + "operation_type": row['operation_type'], + "status": row['status'], + "input_doc_ids": json.loads(row['input_doc_ids']) if row['input_doc_ids'] else [], + "output_doc_ids": json.loads(row['output_doc_ids']) if row['output_doc_ids'] else [], + "details": json.loads(row['details']) if row['details'] else {}, + "created_at": row['created_at'], + "completed_at": row['completed_at'] + }) + return results + finally: + conn.close() + def close(self): """Close database connections (no-op for SQLite).""" pass diff --git a/app/db/schema.sqlite b/app/db/schema.sqlite index 24d4c03..3419161 100644 --- a/app/db/schema.sqlite +++ b/app/db/schema.sqlite @@ -101,3 +101,19 @@ CREATE TABLE IF NOT EXISTS project_kb_folders ( CREATE INDEX IF NOT EXISTS idx_pkf_project_id ON project_kb_folders(project_id); CREATE INDEX IF NOT EXISTS idx_pkf_mcp_type ON project_kb_folders(mcp_type); + +-- Knowledge Lifecycle Tracking (for dedup, consolidation, archival audit log) +CREATE TABLE IF NOT EXISTS lifecycle_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + kb_name TEXT NOT NULL, + operation_type TEXT NOT NULL, -- dedup_scan, dedup_merge, consolidate, archive, restore + status TEXT NOT NULL DEFAULT 'pending', + input_doc_ids TEXT DEFAULT '[]', -- JSON array + output_doc_ids TEXT DEFAULT '[]', -- JSON array + details TEXT DEFAULT '{}', -- JSON blob + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + completed_at TIMESTAMP +); + +CREATE INDEX IF NOT EXISTS idx_lifecycle_kb ON lifecycle_log(kb_name); +CREATE INDEX IF NOT EXISTS idx_lifecycle_op ON lifecycle_log(operation_type); diff --git a/docs/API_REFERENCE.md b/docs/API_REFERENCE.md index 3668221..1295c1a 100644 --- a/docs/API_REFERENCE.md +++ b/docs/API_REFERENCE.md @@ -17,9 +17,10 @@ Complete reference for all Claude OS MCP Server API endpoints. 7. [Real-Time Spec Watcher](#real-time-spec-watcher-new) 8. [Hooks System](#hooks-system) 9. [File Watcher](#file-watcher) -10. [Authentication](#authentication) -11. [Utilities](#utilities) -12. [Health Check](#health-check) +10. [Knowledge Lifecycle](#knowledge-lifecycle-new) +11. [Authentication](#authentication) +12. [Utilities](#utilities) +13. [Health Check](#health-check) --- @@ -1443,6 +1444,329 @@ curl "http://localhost:8051/api/kb/myproject-code_structure/repo-map?token_budge --- +## Knowledge Lifecycle (NEW) + +Manage the health and lifecycle of knowledge base documents: deduplication, consolidation, archival, and analytics. + +All endpoints are under `/api/kb/{kb_name}/lifecycle/`. + +--- + +### Dedup Scan + +Scan for duplicate/near-duplicate documents using embedding similarity. + +```http +POST /api/kb/{kb_name}/lifecycle/dedup-scan +Content-Type: application/json + +{ + "threshold": 0.85, + "max_pairs": 100 +} +``` + +**Response (sync for <500 docs):** +```json +{ + "total_documents": 47, + "duplicate_pairs": [ + { + "doc_a": "doc-abc123", + "doc_b": "doc-def456", + "similarity": 0.94, + "content_a_preview": "Authentication using JWT...", + "content_b_preview": "JWT token authentication..." + } + ], + "clusters": [ + { + "cluster_id": "doc-abc123", + "doc_ids": ["doc-abc123", "doc-def456", "doc-ghi789"], + "size": 3 + } + ], + "duplicate_density": 0.064 +} +``` + +**Response (background for >500 docs):** +```json +{ + "success": true, + "job_id": "dedup-my-kb-a1b2c3d4", + "mode": "background", + "message": "Dedup scan started in background. Check GET /api/jobs/dedup-my-kb-a1b2c3d4" +} +``` + +--- + +### Dedup Merge + +Merge duplicates by keeping one document and deleting the rest. + +```http +POST /api/kb/{kb_name}/lifecycle/dedup-merge +Content-Type: application/json + +{ + "keep_doc_id": "doc-abc123", + "remove_doc_ids": ["doc-def456", "doc-ghi789"], + "dry_run": false +} +``` + +**Response:** +```json +{ + "dry_run": false, + "keep_doc_id": "doc-abc123", + "removed": ["doc-def456", "doc-ghi789"], + "deleted_count": 2 +} +``` + +--- + +### Consolidate + +Consolidate multiple related documents into a single merged document using LLM-powered summarization. Always runs in background (LLM call). + +```http +POST /api/kb/{kb_name}/lifecycle/consolidate +Content-Type: application/json + +{ + "doc_ids": ["doc-abc123", "doc-def456", "doc-ghi789"], + "new_filename": "consolidated-auth-patterns.md", + "dry_run": false +} +``` + +**Response (dry_run=true):** +```json +{ + "dry_run": true, + "source_doc_ids": ["doc-abc123", "doc-def456"], + "source_count": 2, + "total_chars": 4250, + "previews": ["Authentication using JWT...", "JWT token auth..."] +} +``` + +**Response (dry_run=false):** +```json +{ + "success": true, + "job_id": "consolidate-my-kb-e5f6g7h8", + "mode": "background", + "message": "Consolidation started. Check GET /api/jobs/consolidate-my-kb-e5f6g7h8" +} +``` + +--- + +### Health Report + +Get a comprehensive health report for a knowledge base. + +```http +GET /api/kb/{kb_name}/lifecycle/health +``` + +**Response:** +```json +{ + "kb_name": "my-project-project_memories", + "document_count": 47, + "chunk_count": 47, + "last_updated": "2026-02-05T10:30:00", + "embedding_coverage": { + "total_docs": 47, + "with_embeddings": 42, + "without_embeddings": 5, + "coverage_pct": 89.4 + }, + "archived_count": 3, + "top_similar_pairs": [ + {"doc_a": "doc-abc", "doc_b": "doc-def", "similarity": 0.94} + ], + "age_distribution": { + "last_7_days": 8, + "last_30_days": 15, + "last_90_days": 12, + "older": 12 + }, + "recent_operations": [], + "recommendations": [ + { + "type": "dedup", + "priority": "high", + "message": "Found 3 highly similar document pairs. Consider running dedup scan." + } + ] +} +``` + +--- + +### Growth Timeline + +Get document growth timeline grouped by period. + +```http +GET /api/kb/{kb_name}/lifecycle/growth?granularity=month +``` + +**Query Parameters:** +- `granularity` - `day`, `week`, or `month` (default: `month`) + +**Response:** +```json +{ + "kb_name": "my-project-project_memories", + "granularity": "month", + "timeline": [ + {"period": "2025-10", "added": 12, "total": 12}, + {"period": "2025-11", "added": 20, "total": 32}, + {"period": "2025-12", "added": 15, "total": 47} + ], + "total_documents": 47 +} +``` + +--- + +### Archive Documents + +Soft-archive documents (excluded from search but restorable). + +```http +POST /api/kb/{kb_name}/lifecycle/archive +Content-Type: application/json + +{ + "doc_ids": ["doc-abc123", "doc-def456"], + "reason": "stale - over 90 days" +} +``` + +**Response:** +```json +{ + "archived_count": 2, + "doc_ids": ["doc-abc123", "doc-def456"], + "reason": "stale - over 90 days" +} +``` + +--- + +### Restore Documents + +Restore previously archived documents. + +```http +POST /api/kb/{kb_name}/lifecycle/restore +Content-Type: application/json + +{ + "doc_ids": ["doc-abc123"] +} +``` + +**Response:** +```json +{ + "restored_count": 1, + "doc_ids": ["doc-abc123"] +} +``` + +--- + +### List Archived + +List all archived documents in a knowledge base. + +```http +GET /api/kb/{kb_name}/lifecycle/archived +``` + +**Response:** +```json +{ + "kb_name": "my-project-project_memories", + "archived_count": 3, + "archived_documents": [ + { + "doc_id": "doc-abc123", + "content_preview": "Old authentication pattern...", + "archived_at": "2026-02-01T10:00:00", + "archive_reason": "stale - over 90 days" + } + ] +} +``` + +--- + +### Find Stale Documents + +Find documents older than a specified threshold. + +```http +GET /api/kb/{kb_name}/lifecycle/stale?stale_days=90 +``` + +**Response:** +```json +{ + "kb_name": "my-project-project_memories", + "stale_days_threshold": 90, + "stale_count": 5, + "stale_documents": [ + {"filename": "old-pattern.md", "upload_date": "2025-09-15T10:00:00", "age_days": 142} + ] +} +``` + +--- + +### Lifecycle Logs + +Get audit log of lifecycle operations. + +```http +GET /api/kb/{kb_name}/lifecycle/logs?operation_type=dedup_scan&limit=50 +``` + +**Query Parameters:** +- `operation_type` (optional) - Filter by type: `dedup_scan`, `dedup_merge`, `consolidate`, `archive`, `restore` +- `limit` (optional) - Max results (default: 50) + +**Response:** +```json +{ + "logs": [ + { + "id": 1, + "kb_name": "my-project-project_memories", + "operation_type": "dedup_scan", + "status": "completed", + "input_doc_ids": [], + "output_doc_ids": [], + "details": {"total_documents": 47, "pairs_found": 3}, + "created_at": "2026-02-06T10:30:00", + "completed_at": "2026-02-06T10:30:02" + } + ] +} +``` + +--- + ## Support For issues, questions, or contributions: @@ -1451,5 +1775,5 @@ For issues, questions, or contributions: --- -**Last Updated:** 2025-12-11 -**API Version:** 2.3 +**Last Updated:** 2026-02-06 +**API Version:** 2.4 diff --git a/mcp_server/claude_code_mcp.py b/mcp_server/claude_code_mcp.py index e06d395..8c0fe2c 100644 --- a/mcp_server/claude_code_mcp.py +++ b/mcp_server/claude_code_mcp.py @@ -398,6 +398,65 @@ async def list_tools() -> list[Tool]: }, "required": ["name", "project_path"] } + ), + + # Knowledge Lifecycle Tools + Tool( + name="kb_lifecycle_health", + description="Get a comprehensive health report for a knowledge base including duplicate detection, embedding coverage, age distribution, and recommendations.", + inputSchema={ + "type": "object", + "properties": { + "kb_name": {"type": "string", "description": "Name of the knowledge base"} + }, + "required": ["kb_name"] + } + ), + Tool( + name="kb_lifecycle_dedup", + description="Scan for and merge duplicate documents in a knowledge base. Use action='scan' to find duplicates, action='merge' to merge them.", + inputSchema={ + "type": "object", + "properties": { + "kb_name": {"type": "string", "description": "Name of the knowledge base"}, + "action": {"type": "string", "enum": ["scan", "merge"], "description": "Action to perform: scan for duplicates or merge them"}, + "threshold": {"type": "number", "default": 0.85, "description": "Similarity threshold for scan (0.0-1.0)"}, + "max_pairs": {"type": "integer", "default": 100, "description": "Maximum duplicate pairs to return"}, + "keep_doc_id": {"type": "string", "description": "Document ID to keep (required for merge)"}, + "remove_doc_ids": {"type": "array", "items": {"type": "string"}, "description": "Document IDs to remove (required for merge)"}, + "dry_run": {"type": "boolean", "default": False, "description": "Preview changes without applying"} + }, + "required": ["kb_name", "action"] + } + ), + Tool( + name="kb_lifecycle_consolidate", + description="Consolidate multiple related documents into a single merged document using LLM-powered summarization.", + inputSchema={ + "type": "object", + "properties": { + "kb_name": {"type": "string", "description": "Name of the knowledge base"}, + "doc_ids": {"type": "array", "items": {"type": "string"}, "description": "Document IDs to consolidate"}, + "new_filename": {"type": "string", "description": "Filename for the consolidated document"}, + "dry_run": {"type": "boolean", "default": False, "description": "Preview without consolidating"} + }, + "required": ["kb_name", "doc_ids", "new_filename"] + } + ), + Tool( + name="kb_lifecycle_archive", + description="Manage document archival: archive, restore, list archived, or find stale documents.", + inputSchema={ + "type": "object", + "properties": { + "kb_name": {"type": "string", "description": "Name of the knowledge base"}, + "action": {"type": "string", "enum": ["archive", "restore", "list", "stale"], "description": "Archival action to perform"}, + "doc_ids": {"type": "array", "items": {"type": "string"}, "description": "Document IDs (required for archive/restore)"}, + "reason": {"type": "string", "default": "manual", "description": "Reason for archiving"}, + "stale_days": {"type": "integer", "default": 90, "description": "Days threshold for stale detection"} + }, + "required": ["kb_name", "action"] + } ) ] @@ -478,7 +537,7 @@ async def _execute_tool(name: str, args: dict[str, Any]) -> dict: # Document Management elif name == "upload_document": - return await api_post(f"/api/kb/{args['kb_name']}/import", { + return await api_post(f"/api/kb/{args['kb_name']}/documents/content", { "content": args["content"], "filename": args["filename"], "metadata": { @@ -565,6 +624,54 @@ async def _execute_tool(name: str, args: dict[str, Any]) -> dict: elif name == "delete_skill": return await api_delete(f"/api/skills/{args['name']}?project_path={args['project_path']}") + # Knowledge Lifecycle Tools + elif name == "kb_lifecycle_health": + return await api_get(f"/api/kb/{args['kb_name']}/lifecycle/health") + + elif name == "kb_lifecycle_dedup": + action = args["action"] + kb_name = args["kb_name"] + if action == "scan": + return await api_post(f"/api/kb/{kb_name}/lifecycle/dedup-scan", { + "threshold": args.get("threshold", 0.85), + "max_pairs": args.get("max_pairs", 100) + }) + elif action == "merge": + return await api_post(f"/api/kb/{kb_name}/lifecycle/dedup-merge", { + "keep_doc_id": args["keep_doc_id"], + "remove_doc_ids": args["remove_doc_ids"], + "dry_run": args.get("dry_run", False) + }) + else: + return {"error": f"Unknown dedup action: {action}"} + + elif name == "kb_lifecycle_consolidate": + return await api_post(f"/api/kb/{args['kb_name']}/lifecycle/consolidate", { + "doc_ids": args["doc_ids"], + "new_filename": args["new_filename"], + "dry_run": args.get("dry_run", False) + }) + + elif name == "kb_lifecycle_archive": + action = args["action"] + kb_name = args["kb_name"] + if action == "archive": + return await api_post(f"/api/kb/{kb_name}/lifecycle/archive", { + "doc_ids": args["doc_ids"], + "reason": args.get("reason", "manual") + }) + elif action == "restore": + return await api_post(f"/api/kb/{kb_name}/lifecycle/restore", { + "doc_ids": args["doc_ids"] + }) + elif action == "list": + return await api_get(f"/api/kb/{kb_name}/lifecycle/archived") + elif action == "stale": + stale_days = args.get("stale_days", 90) + return await api_get(f"/api/kb/{kb_name}/lifecycle/stale?stale_days={stale_days}") + else: + return {"error": f"Unknown archive action: {action}"} + else: return {"error": f"Unknown tool: {name}"} diff --git a/mcp_server/server.py b/mcp_server/server.py index 9f64e8f..b0347dc 100644 --- a/mcp_server/server.py +++ b/mcp_server/server.py @@ -671,6 +671,56 @@ async def api_upload_document(kb_name: str, file: UploadFile = File(...)): raise HTTPException(status_code=500, detail=str(e)) +class ContentUploadRequest(BaseModel): + content: str + filename: str + metadata: Optional[Dict[str, Any]] = None + + +@app.post("/api/kb/{kb_name}/documents/content") +async def api_upload_content(kb_name: str, request: ContentUploadRequest): + """REST API: Upload raw text content directly to a knowledge base (no file needed).""" + import tempfile + import os as _os + from app.core.ingestion import ingest_file + + # Validate KB exists + db_manager = get_sqlite_manager() + if not db_manager.collection_exists(kb_name): + raise HTTPException(status_code=404, detail=f"Knowledge base '{kb_name}' not found") + + if not request.content.strip(): + raise HTTPException(status_code=400, detail="Content cannot be empty") + + # Determine file extension from filename + suffix = Path(request.filename).suffix or ".md" + + # Write content to temp file and ingest using existing pipeline + try: + with tempfile.NamedTemporaryFile(delete=False, suffix=suffix, mode="w", encoding="utf-8") as tmp_file: + tmp_file.write(request.content) + tmp_path = tmp_file.name + + result = ingest_file(tmp_path, kb_name, request.filename) + + _os.unlink(tmp_path) + + if result["status"] == "error": + raise HTTPException(status_code=400, detail=result.get("error", "Upload failed")) + + return { + "success": True, + "filename": result["filename"], + "chunks": result["chunks"], + "file_type": result["file_type"] + } + except HTTPException: + raise + except Exception as e: + logger.error(f"Content upload failed: {e}") + raise HTTPException(status_code=500, detail=str(e)) + + @app.post("/api/kb/{kb_name}/import") async def api_import_directory(kb_name: str, directory_path: str): """REST API: Import all files from a directory into a knowledge base.""" @@ -3300,6 +3350,235 @@ async def health_check(): return JSONResponse(content=health_status, status_code=status_code) +# ============================================================================ +# KNOWLEDGE LIFECYCLE ENDPOINTS +# ============================================================================ + +class DedupScanRequest(BaseModel): + threshold: float = 0.85 + max_pairs: int = 100 + +class DedupMergeRequest(BaseModel): + keep_doc_id: str + remove_doc_ids: List[str] + dry_run: bool = False + +class ConsolidateRequest(BaseModel): + doc_ids: List[str] + new_filename: str + dry_run: bool = False + +class ArchiveRequest(BaseModel): + doc_ids: List[str] + reason: str = "manual" + +class RestoreRequest(BaseModel): + doc_ids: List[str] + + +def _get_lifecycle_engine(): + """Lazy import to avoid circular imports.""" + from app.core.knowledge_lifecycle import KnowledgeLifecycleEngine + return KnowledgeLifecycleEngine() + + +def _run_dedup_scan_background(job_id: str, kb_name: str, threshold: float, max_pairs: int): + """Background worker for dedup scan on large KBs.""" + def update_job(status, progress=0, message="", error=None): + with INDEXING_JOBS_LOCK: + INDEXING_JOBS[job_id].update({ + "status": status, "progress": progress, + "message": message, "error": error, "updated_at": time.time() + }) + if status in ("completed", "failed"): + INDEXING_JOBS[job_id]["completed_at"] = time.time() + + try: + update_job("running", 10, "Loading embeddings...") + engine = _get_lifecycle_engine() + update_job("running", 30, "Computing pairwise similarities...") + result = engine.scan_duplicates(kb_name, threshold, max_pairs) + update_job("completed", 100, f"Found {len(result['duplicate_pairs'])} duplicate pairs") + with INDEXING_JOBS_LOCK: + INDEXING_JOBS[job_id]["result"] = result + except Exception as e: + logger.error(f"[Job {job_id}] Dedup scan failed: {e}") + update_job("failed", 0, "", str(e)) + + +def _run_consolidation_background(job_id: str, kb_name: str, doc_ids: list, new_filename: str): + """Background worker for LLM-powered consolidation.""" + def update_job(status, progress=0, message="", error=None): + with INDEXING_JOBS_LOCK: + INDEXING_JOBS[job_id].update({ + "status": status, "progress": progress, + "message": message, "error": error, "updated_at": time.time() + }) + if status in ("completed", "failed"): + INDEXING_JOBS[job_id]["completed_at"] = time.time() + + try: + update_job("running", 10, "Fetching source documents...") + engine = _get_lifecycle_engine() + update_job("running", 30, "Generating consolidated document via LLM...") + result = engine.consolidate_related(kb_name, doc_ids, new_filename, dry_run=False) + if "error" in result: + update_job("failed", 0, "", result["error"]) + else: + update_job("completed", 100, f"Consolidated {result.get('sources_consolidated', 0)} documents") + with INDEXING_JOBS_LOCK: + INDEXING_JOBS[job_id]["result"] = result + except Exception as e: + logger.error(f"[Job {job_id}] Consolidation failed: {e}") + update_job("failed", 0, "", str(e)) + + +@app.post("/api/kb/{kb_name}/lifecycle/dedup-scan") +async def api_lifecycle_dedup_scan(kb_name: str, request: DedupScanRequest, background_tasks: BackgroundTasks): + """Scan KB for duplicate/near-duplicate documents.""" + import uuid + + db_manager = get_sqlite_manager() + if not db_manager.collection_exists(kb_name): + raise HTTPException(status_code=404, detail=f"Knowledge base '{kb_name}' not found") + + doc_count = db_manager.get_collection_count(kb_name) + + # Background mode for large KBs + if doc_count > 500: + _cleanup_old_jobs() + job_id = f"dedup-{kb_name}-{uuid.uuid4().hex[:8]}" + with INDEXING_JOBS_LOCK: + INDEXING_JOBS[job_id] = { + "job_id": job_id, "type": "dedup_scan", "kb_name": kb_name, + "status": "queued", "progress": 0, "message": "Job queued", + "started_at": time.time(), "completed_at": None, "error": None + } + background_tasks.add_task(_run_dedup_scan_background, job_id, kb_name, request.threshold, request.max_pairs) + return {"success": True, "job_id": job_id, "mode": "background", + "message": f"Dedup scan started in background ({doc_count} docs). Check GET /api/jobs/{job_id}"} + + # Sync mode for smaller KBs + engine = _get_lifecycle_engine() + result = engine.scan_duplicates(kb_name, request.threshold, request.max_pairs) + return result + + +@app.post("/api/kb/{kb_name}/lifecycle/dedup-merge") +async def api_lifecycle_dedup_merge(kb_name: str, request: DedupMergeRequest): + """Merge duplicate documents by keeping one and removing the rest.""" + db_manager = get_sqlite_manager() + if not db_manager.collection_exists(kb_name): + raise HTTPException(status_code=404, detail=f"Knowledge base '{kb_name}' not found") + + engine = _get_lifecycle_engine() + result = engine.merge_duplicates(kb_name, request.keep_doc_id, request.remove_doc_ids, request.dry_run) + return result + + +@app.post("/api/kb/{kb_name}/lifecycle/consolidate") +async def api_lifecycle_consolidate(kb_name: str, request: ConsolidateRequest, background_tasks: BackgroundTasks): + """Consolidate related documents into a single merged document using LLM.""" + import uuid + + db_manager = get_sqlite_manager() + if not db_manager.collection_exists(kb_name): + raise HTTPException(status_code=404, detail=f"Knowledge base '{kb_name}' not found") + + if request.dry_run: + engine = _get_lifecycle_engine() + return engine.consolidate_related(kb_name, request.doc_ids, request.new_filename, dry_run=True) + + # Always run consolidation in background (LLM call) + _cleanup_old_jobs() + job_id = f"consolidate-{kb_name}-{uuid.uuid4().hex[:8]}" + with INDEXING_JOBS_LOCK: + INDEXING_JOBS[job_id] = { + "job_id": job_id, "type": "consolidation", "kb_name": kb_name, + "status": "queued", "progress": 0, "message": "Job queued", + "started_at": time.time(), "completed_at": None, "error": None + } + background_tasks.add_task(_run_consolidation_background, job_id, kb_name, request.doc_ids, request.new_filename) + return {"success": True, "job_id": job_id, "mode": "background", + "message": f"Consolidation started. Check GET /api/jobs/{job_id}"} + + +@app.get("/api/kb/{kb_name}/lifecycle/health") +async def api_lifecycle_health(kb_name: str): + """Get comprehensive health report for a knowledge base.""" + db_manager = get_sqlite_manager() + if not db_manager.collection_exists(kb_name): + raise HTTPException(status_code=404, detail=f"Knowledge base '{kb_name}' not found") + + engine = _get_lifecycle_engine() + return engine.get_health_report(kb_name) + + +@app.get("/api/kb/{kb_name}/lifecycle/growth") +async def api_lifecycle_growth(kb_name: str, granularity: str = "month"): + """Get document growth timeline for a knowledge base.""" + db_manager = get_sqlite_manager() + if not db_manager.collection_exists(kb_name): + raise HTTPException(status_code=404, detail=f"Knowledge base '{kb_name}' not found") + + engine = _get_lifecycle_engine() + return engine.get_growth_timeline(kb_name, granularity) + + +@app.post("/api/kb/{kb_name}/lifecycle/archive") +async def api_lifecycle_archive(kb_name: str, request: ArchiveRequest): + """Archive documents in a knowledge base.""" + db_manager = get_sqlite_manager() + if not db_manager.collection_exists(kb_name): + raise HTTPException(status_code=404, detail=f"Knowledge base '{kb_name}' not found") + + engine = _get_lifecycle_engine() + return engine.archive_documents(kb_name, request.doc_ids, request.reason) + + +@app.post("/api/kb/{kb_name}/lifecycle/restore") +async def api_lifecycle_restore(kb_name: str, request: RestoreRequest): + """Restore archived documents in a knowledge base.""" + db_manager = get_sqlite_manager() + if not db_manager.collection_exists(kb_name): + raise HTTPException(status_code=404, detail=f"Knowledge base '{kb_name}' not found") + + engine = _get_lifecycle_engine() + return engine.restore_documents(kb_name, request.doc_ids) + + +@app.get("/api/kb/{kb_name}/lifecycle/archived") +async def api_lifecycle_archived(kb_name: str): + """List archived documents in a knowledge base.""" + db_manager = get_sqlite_manager() + if not db_manager.collection_exists(kb_name): + raise HTTPException(status_code=404, detail=f"Knowledge base '{kb_name}' not found") + + engine = _get_lifecycle_engine() + return engine.list_archived(kb_name) + + +@app.get("/api/kb/{kb_name}/lifecycle/stale") +async def api_lifecycle_stale(kb_name: str, stale_days: int = 90): + """Find stale documents older than the specified threshold.""" + db_manager = get_sqlite_manager() + if not db_manager.collection_exists(kb_name): + raise HTTPException(status_code=404, detail=f"Knowledge base '{kb_name}' not found") + + engine = _get_lifecycle_engine() + return engine.find_stale_documents(kb_name, stale_days) + + +@app.get("/api/kb/{kb_name}/lifecycle/logs") +async def api_lifecycle_logs(kb_name: str, operation_type: Optional[str] = None, limit: int = 50): + """Get lifecycle operation logs for a knowledge base.""" + db_manager = get_sqlite_manager() + if not db_manager.collection_exists(kb_name): + raise HTTPException(status_code=404, detail=f"Knowledge base '{kb_name}' not found") + + return {"logs": db_manager.get_lifecycle_logs(kb_name, operation_type, limit)} + + # ============================================================================ # STARTUP/SHUTDOWN EVENTS # ============================================================================ diff --git a/templates/commands/claude-os-lifecycle.md b/templates/commands/claude-os-lifecycle.md new file mode 100644 index 0000000..f77984d --- /dev/null +++ b/templates/commands/claude-os-lifecycle.md @@ -0,0 +1,310 @@ +--- +name: claude-os-lifecycle +description: Knowledge base lifecycle management - dedup, consolidate, archive, and health analytics +--- + +# Claude OS Knowledge Lifecycle Management + +Manage the health and lifecycle of your knowledge bases: find duplicates, consolidate related docs, archive stale content, and get health reports. + +## Commands + +``` +/claude-os-lifecycle health [kb_name] - Health report with recommendations +/claude-os-lifecycle dedup [kb_name] - Scan and merge duplicate documents +/claude-os-lifecycle consolidate [kb_name] - LLM-powered document merging +/claude-os-lifecycle archive [kb_name] - Find stale docs, archive, restore +/claude-os-lifecycle logs [kb_name] - View operation history +``` + +--- + +## HEALTH REPORT + +### What Happens + +``` +/claude-os-lifecycle health my-project-project_memories +``` + +**Step 1: Get Health Report** +``` +mcp__code-forge__kb_lifecycle_health + kb_name: "my-project-project_memories" +``` + +**Step 2: Display Report** + +``` +โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• +KNOWLEDGE BASE HEALTH REPORT +โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• + +KB: my-project-project_memories +Documents: 47 (42 with embeddings, 3 archived) +Last Updated: 2026-02-05 + +EMBEDDING COVERAGE: + With embeddings: 42/47 (89.4%) + Without: 5 docs need re-indexing + +AGE DISTRIBUTION: + Last 7 days: 8 docs + Last 30 days: 15 docs + Last 90 days: 12 docs + Older: 12 docs + +TOP SIMILAR PAIRS (potential duplicates): + 1. doc-a1b2 <-> doc-c3d4 (0.94 similarity) + 2. doc-e5f6 <-> doc-g7h8 (0.91 similarity) + 3. doc-i9j0 <-> doc-k1l2 (0.88 similarity) + +RECOMMENDATIONS: + [HIGH] Found 3 highly similar document pairs. Run dedup scan. + [MEDIUM] 12 documents older than 90 days. Review for archival. + [MEDIUM] 5 documents lack embeddings. Re-index recommended. + +โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• +``` + +--- + +## DEDUPLICATION + +### Scan for Duplicates + +``` +/claude-os-lifecycle dedup my-project-project_memories +``` + +**Step 1: Run Dedup Scan** +``` +mcp__code-forge__kb_lifecycle_dedup + kb_name: "my-project-project_memories" + action: "scan" + threshold: 0.85 +``` + +**Step 2: Show Results** + +``` +โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• +DUPLICATE SCAN RESULTS +โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• + +Scanned: 47 documents +Duplicate Density: 6.4% +Pairs Found: 3 +Clusters: 2 + +CLUSTER 1 (3 docs): + - doc-a1b2: "Authentication pattern using JWT tokens..." + - doc-c3d4: "JWT authentication implementation pattern..." + - doc-m3n4: "Token-based auth with JWT refresh..." + Similarity: 0.91-0.94 + +CLUSTER 2 (2 docs): + - doc-e5f6: "Redis caching strategy for API responses..." + - doc-g7h8: "API response caching with Redis..." + Similarity: 0.88 + +โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• +``` + +**Step 3: Ask User What to Do** + +For each cluster, offer options: +- **Merge**: Keep the most complete doc, delete others +- **Consolidate**: Use LLM to merge into a single comprehensive doc +- **Skip**: Leave as-is + +``` +Cluster 1: What would you like to do? +[m] Merge (keep best, delete rest) +[c] Consolidate (LLM merge) +[s] Skip +``` + +**If merge:** +``` +mcp__code-forge__kb_lifecycle_dedup + kb_name: "my-project-project_memories" + action: "merge" + keep_doc_id: "doc-a1b2" + remove_doc_ids: ["doc-c3d4", "doc-m3n4"] +``` + +**If consolidate:** +``` +mcp__code-forge__kb_lifecycle_consolidate + kb_name: "my-project-project_memories" + doc_ids: ["doc-a1b2", "doc-c3d4", "doc-m3n4"] + new_filename: "jwt-authentication-pattern-consolidated.md" +``` + +--- + +## CONSOLIDATION + +### LLM-Powered Document Merging + +``` +/claude-os-lifecycle consolidate my-project-project_memories +``` + +**Step 1: Preview (dry run)** +``` +mcp__code-forge__kb_lifecycle_consolidate + kb_name: "my-project-project_memories" + doc_ids: ["doc-a1b2", "doc-c3d4", "doc-m3n4"] + new_filename: "consolidated-auth-patterns.md" + dry_run: true +``` + +**Step 2: Show Preview** +``` +โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• +CONSOLIDATION PREVIEW +โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• + +Sources: 3 documents +Total content: 4,250 characters + +Document 1: "Authentication pattern using JWT..." +Document 2: "JWT authentication implementation..." +Document 3: "Token-based auth with JWT refresh..." + +This will: + - Create: consolidated-auth-patterns.md + - Delete: 3 source documents + - Use LLM to merge content intelligently + +Proceed? [y/n] +โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• +``` + +**Step 3: Execute** +``` +mcp__code-forge__kb_lifecycle_consolidate + kb_name: "my-project-project_memories" + doc_ids: ["doc-a1b2", "doc-c3d4", "doc-m3n4"] + new_filename: "consolidated-auth-patterns.md" + dry_run: false +``` + +--- + +## ARCHIVAL + +### Find and Archive Stale Documents + +``` +/claude-os-lifecycle archive my-project-project_memories +``` + +**Step 1: Find Stale Documents** +``` +mcp__code-forge__kb_lifecycle_archive + kb_name: "my-project-project_memories" + action: "stale" + stale_days: 90 +``` + +**Step 2: Show Stale Documents** +``` +โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• +STALE DOCUMENTS (>90 days old) +โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• + +Found 5 stale documents: + +1. old-pattern-123.md (142 days old) +2. debug-notes-456.md (128 days old) +3. meeting-notes-789.md (115 days old) +4. temp-workaround-012.md (103 days old) +5. initial-setup-345.md (98 days old) + +[a] Archive all [s] Select individually [n] Skip +โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• +``` + +**Step 3: Archive Selected** +``` +mcp__code-forge__kb_lifecycle_archive + kb_name: "my-project-project_memories" + action: "archive" + doc_ids: ["old-pattern-123", "debug-notes-456"] + reason: "stale - over 90 days" +``` + +### List Archived Documents + +``` +mcp__code-forge__kb_lifecycle_archive + kb_name: "my-project-project_memories" + action: "list" +``` + +### Restore Archived Documents + +``` +mcp__code-forge__kb_lifecycle_archive + kb_name: "my-project-project_memories" + action: "restore" + doc_ids: ["old-pattern-123"] +``` + +--- + +## OPERATION LOGS + +### View History + +``` +/claude-os-lifecycle logs my-project-project_memories +``` + +Shows recent lifecycle operations: + +``` +โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• +LIFECYCLE OPERATION LOG +โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• + +1. [2026-02-06 10:30] dedup_scan - completed + Found 3 duplicate pairs in 47 documents + +2. [2026-02-06 10:32] dedup_merge - completed + Kept doc-a1b2, removed 2 documents + +3. [2026-02-06 10:35] archive - completed + Archived 2 documents (reason: stale) + +4. [2026-02-05 14:20] consolidate - completed + Merged 3 documents into consolidated-patterns.md + +โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• +``` + +--- + +## KB NAME RESOLUTION + +If kb_name is not provided, use this strategy: +1. Check if working directory has a Claude OS project registered +2. If so, use `{project}-project_memories` as default +3. If not, list all KBs and ask user to pick one + +``` +mcp__code-forge__list_knowledge_bases +``` + +--- + +## WHY THIS MATTERS + +- **Dedup**: Prevents memory bloat from saving similar insights repeatedly +- **Consolidate**: Turns fragmented notes into comprehensive references +- **Archive**: Keeps active memories relevant without permanent deletion +- **Health**: Proactive monitoring catches issues before they impact recall +- **Logs**: Full audit trail of all lifecycle operations diff --git a/tests/test_knowledge_lifecycle.py b/tests/test_knowledge_lifecycle.py new file mode 100644 index 0000000..2ab05e3 --- /dev/null +++ b/tests/test_knowledge_lifecycle.py @@ -0,0 +1,475 @@ +""" +Tests for Knowledge Lifecycle Engine. +Covers deduplication, archival, analytics, and consolidation. +""" + +import pytest +import tempfile +import json +import uuid +from pathlib import Path +from datetime import datetime, timedelta +from unittest.mock import patch, MagicMock +import numpy as np + +from app.core.sqlite_manager import SQLiteManager +from app.core.kb_types import KBType + + +@pytest.fixture +def lifecycle_db(): + """Create a clean SQLite database with lifecycle support.""" + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as temp_file: + temp_db_path = temp_file.name + + manager = SQLiteManager(temp_db_path) + yield manager + + if Path(temp_db_path).exists(): + Path(temp_db_path).unlink() + + +@pytest.fixture +def populated_kb(lifecycle_db): + """Create a KB with sample documents and embeddings.""" + kb_name = "test-lifecycle-kb" + lifecycle_db.create_collection(kb_name, KBType.GENERIC, description="Test KB") + + # Create documents with embeddings + # Doc 1 and 2 are near-duplicates (similar embeddings) + # Doc 3 is different + np.random.seed(42) + base_vec = np.random.randn(768).astype(np.float32) + base_vec = base_vec / np.linalg.norm(base_vec) + + # Near-duplicate of base (very small perturbation for high similarity) + similar_vec = base_vec + np.random.randn(768).astype(np.float32) * 0.01 + similar_vec = similar_vec / np.linalg.norm(similar_vec) + + # Different vector + different_vec = np.random.randn(768).astype(np.float32) + different_vec = different_vec / np.linalg.norm(different_vec) + + docs = ["Authentication using JWT tokens for API security", + "JWT token authentication pattern for securing APIs", + "Redis caching strategy for database query optimization"] + embeddings = [base_vec.tolist(), similar_vec.tolist(), different_vec.tolist()] + metadatas = [ + {"filename": "auth-jwt.md", "upload_date": datetime.now().isoformat()}, + {"filename": "jwt-auth.md", "upload_date": datetime.now().isoformat()}, + {"filename": "redis-cache.md", "upload_date": (datetime.now() - timedelta(days=120)).isoformat()} + ] + ids = ["doc-auth-1", "doc-auth-2", "doc-redis-1"] + + lifecycle_db.add_documents(kb_name, docs, embeddings, metadatas, ids) + + return kb_name, lifecycle_db + + +@pytest.fixture +def engine(populated_kb): + """Create a KnowledgeLifecycleEngine with populated KB.""" + kb_name, db = populated_kb + + # Patch the singleton to use our test DB + with patch("app.core.knowledge_lifecycle.get_sqlite_manager", return_value=db): + from app.core.knowledge_lifecycle import KnowledgeLifecycleEngine + e = KnowledgeLifecycleEngine() + e.db = db + yield e, kb_name, db + + +@pytest.mark.unit +class TestDeduplication: + """Test duplicate scanning and merging.""" + + def test_scan_empty_kb(self, lifecycle_db): + """Scan with no documents returns empty results.""" + kb_name = "empty-kb" + lifecycle_db.create_collection(kb_name, KBType.GENERIC) + + with patch("app.core.knowledge_lifecycle.get_sqlite_manager", return_value=lifecycle_db): + from app.core.knowledge_lifecycle import KnowledgeLifecycleEngine + engine = KnowledgeLifecycleEngine() + engine.db = lifecycle_db + result = engine.scan_duplicates(kb_name) + + assert result["total_documents"] == 0 + assert result["duplicate_pairs"] == [] + assert result["clusters"] == [] + assert result["duplicate_density"] == 0.0 + + def test_identical_vectors_detected(self, engine): + """Near-duplicate documents are detected above threshold.""" + e, kb_name, _ = engine + result = e.scan_duplicates(kb_name, threshold=0.85) + + assert result["total_documents"] == 3 + assert len(result["duplicate_pairs"]) >= 1 + + # The auth docs should be paired + pair = result["duplicate_pairs"][0] + doc_ids = {pair["doc_a"], pair["doc_b"]} + assert "doc-auth-1" in doc_ids or "doc-auth-2" in doc_ids + assert pair["similarity"] >= 0.85 + + def test_below_threshold_ignored(self, engine): + """Documents below threshold are not flagged.""" + e, kb_name, _ = engine + result = e.scan_duplicates(kb_name, threshold=0.99) + + # At 0.99 threshold, only truly identical docs match + # Our similar_vec has noise so shouldn't match at 0.99 + for pair in result["duplicate_pairs"]: + assert pair["similarity"] >= 0.99 + + def test_union_find_clustering(self, engine): + """Duplicate pairs are clustered correctly.""" + e, kb_name, _ = engine + result = e.scan_duplicates(kb_name, threshold=0.85) + + if result["clusters"]: + cluster = result["clusters"][0] + assert "cluster_id" in cluster + assert "doc_ids" in cluster + assert cluster["size"] >= 2 + + def test_merge_dry_run(self, engine): + """Dry run merge previews without deleting.""" + e, kb_name, db = engine + result = e.merge_duplicates( + kb_name, keep_doc_id="doc-auth-1", + remove_doc_ids=["doc-auth-2"], dry_run=True + ) + + assert result["dry_run"] is True + assert result["keep_doc_id"] == "doc-auth-1" + assert "doc-auth-2" in result["would_remove"] + + # Verify doc still exists + docs = db.get_all_embeddings(kb_name) + doc_ids = [d["doc_id"] for d in docs] + assert "doc-auth-2" in doc_ids + + def test_merge_actual(self, engine): + """Actual merge deletes duplicate documents.""" + e, kb_name, db = engine + result = e.merge_duplicates( + kb_name, keep_doc_id="doc-auth-1", + remove_doc_ids=["doc-auth-2"], dry_run=False + ) + + assert result["dry_run"] is False + assert result["deleted_count"] == 1 + + # Verify doc was deleted + docs = db.get_all_embeddings(kb_name) + doc_ids = [d["doc_id"] for d in docs] + assert "doc-auth-1" in doc_ids + assert "doc-auth-2" not in doc_ids + + +@pytest.mark.unit +class TestArchival: + """Test document archival and restoration.""" + + def test_archive_sets_flag(self, engine): + """Archive sets archived flag in metadata.""" + e, kb_name, db = engine + result = e.archive_documents(kb_name, ["doc-auth-1"], reason="test") + + assert result["archived_count"] == 1 + + docs = db.get_all_embeddings(kb_name, exclude_archived=False) + doc = next(d for d in docs if d["doc_id"] == "doc-auth-1") + assert doc["metadata"]["archived"] is True + assert doc["metadata"]["archive_reason"] == "test" + assert "archived_at" in doc["metadata"] + + def test_restore_clears_flag(self, engine): + """Restore clears archived flag.""" + e, kb_name, db = engine + + # Archive first + e.archive_documents(kb_name, ["doc-auth-1"]) + + # Then restore + result = e.restore_documents(kb_name, ["doc-auth-1"]) + assert result["restored_count"] == 1 + + docs = db.get_all_embeddings(kb_name, exclude_archived=False) + doc = next(d for d in docs if d["doc_id"] == "doc-auth-1") + assert doc["metadata"].get("archived") is False + + def test_exclude_archived_from_embeddings(self, engine): + """Archived docs are excluded when exclude_archived=True.""" + e, kb_name, db = engine + + e.archive_documents(kb_name, ["doc-auth-1"]) + + # With exclude + docs_filtered = db.get_all_embeddings(kb_name, exclude_archived=True) + doc_ids_filtered = [d["doc_id"] for d in docs_filtered] + assert "doc-auth-1" not in doc_ids_filtered + + # Without exclude + docs_all = db.get_all_embeddings(kb_name, exclude_archived=False) + doc_ids_all = [d["doc_id"] for d in docs_all] + assert "doc-auth-1" in doc_ids_all + + def test_find_stale(self, engine): + """Find stale documents correctly identifies old docs.""" + e, kb_name, _ = engine + + with patch("app.core.knowledge_lifecycle.get_documents_metadata") as mock_docs: + mock_docs.return_value = [ + {"filename": "old.md", "upload_date": (datetime.now() - timedelta(days=120)).isoformat()}, + {"filename": "new.md", "upload_date": datetime.now().isoformat()} + ] + result = e.find_stale_documents(kb_name, stale_days=90) + + assert result["stale_count"] == 1 + assert result["stale_documents"][0]["filename"] == "old.md" + assert result["stale_documents"][0]["age_days"] >= 119 + + def test_list_archived(self, engine): + """List archived returns only archived docs.""" + e, kb_name, _ = engine + + # Archive one doc + e.archive_documents(kb_name, ["doc-redis-1"], reason="stale") + + result = e.list_archived(kb_name) + assert result["archived_count"] == 1 + assert result["archived_documents"][0]["doc_id"] == "doc-redis-1" + assert result["archived_documents"][0]["archive_reason"] == "stale" + + +@pytest.mark.unit +class TestAnalytics: + """Test health reports and growth timeline.""" + + def test_health_report_structure(self, engine): + """Health report contains all expected sections.""" + e, kb_name, _ = engine + + with patch("app.core.knowledge_lifecycle.get_collection_stats") as mock_stats, \ + patch("app.core.knowledge_lifecycle.get_documents_metadata") as mock_docs: + mock_stats.return_value = { + "total_documents": 3, "total_chunks": 3, + "last_updated": datetime.now().isoformat() + } + mock_docs.return_value = [ + {"filename": "a.md", "upload_date": datetime.now().isoformat()}, + {"filename": "b.md", "upload_date": datetime.now().isoformat()}, + {"filename": "c.md", "upload_date": (datetime.now() - timedelta(days=100)).isoformat()} + ] + + report = e.get_health_report(kb_name) + + assert "kb_name" in report + assert "stats" in report + assert "embedding_coverage" in report + assert "top_similar_pairs" in report + assert "age_distribution" in report + assert "recommendations" in report + assert report["embedding_coverage"]["total_docs"] == 3 + assert report["embedding_coverage"]["with_embeddings"] == 3 + + def test_growth_timeline_grouping(self, engine): + """Growth timeline groups documents by period.""" + e, kb_name, _ = engine + + with patch("app.core.knowledge_lifecycle.get_documents_metadata") as mock_docs: + now = datetime.now() + mock_docs.return_value = [ + {"filename": "a.md", "upload_date": now.isoformat()}, + {"filename": "b.md", "upload_date": now.isoformat()}, + {"filename": "c.md", "upload_date": (now - timedelta(days=35)).isoformat()} + ] + + result = e.get_growth_timeline(kb_name, granularity="month") + + assert result["kb_name"] == kb_name + assert result["granularity"] == "month" + assert result["total_documents"] == 3 + assert len(result["timeline"]) >= 1 + + # Last entry should have cumulative total = 3 + assert result["timeline"][-1]["total"] == 3 + + def test_density_calculation(self, engine): + """Duplicate density is calculated correctly.""" + e, kb_name, _ = engine + result = e.scan_duplicates(kb_name, threshold=0.85) + + # density = pairs / (n*(n-1)/2) where n=3 => 3 possible pairs + assert 0 <= result["duplicate_density"] <= 1.0 + + +@pytest.mark.unit +class TestLifecycleLog: + """Test lifecycle audit logging.""" + + def test_insert_and_query_log(self, lifecycle_db): + """Insert and retrieve lifecycle logs.""" + kb_name = "test-log-kb" + + log_id = lifecycle_db.insert_lifecycle_log( + kb_name=kb_name, + operation_type="dedup_scan", + status="completed", + input_doc_ids=["doc-1", "doc-2"], + details={"pairs_found": 5} + ) + + assert log_id > 0 + + logs = lifecycle_db.get_lifecycle_logs(kb_name) + assert len(logs) == 1 + assert logs[0]["operation_type"] == "dedup_scan" + assert logs[0]["status"] == "completed" + assert logs[0]["input_doc_ids"] == ["doc-1", "doc-2"] + assert logs[0]["details"]["pairs_found"] == 5 + + def test_update_log_status(self, lifecycle_db): + """Update lifecycle log status and completion.""" + log_id = lifecycle_db.insert_lifecycle_log( + kb_name="test-kb", + operation_type="consolidate", + status="pending" + ) + + success = lifecycle_db.update_lifecycle_log( + log_id, "completed", + output_doc_ids=["new-doc-1"], + details={"merged": 3} + ) + + assert success + + logs = lifecycle_db.get_lifecycle_logs("test-kb") + assert logs[0]["status"] == "completed" + assert logs[0]["output_doc_ids"] == ["new-doc-1"] + assert logs[0]["completed_at"] is not None + + def test_filter_by_operation_type(self, lifecycle_db): + """Filter logs by operation type.""" + lifecycle_db.insert_lifecycle_log("kb1", "dedup_scan", "completed") + lifecycle_db.insert_lifecycle_log("kb1", "archive", "completed") + lifecycle_db.insert_lifecycle_log("kb1", "dedup_merge", "completed") + + dedup_logs = lifecycle_db.get_lifecycle_logs("kb1", operation_type="dedup_scan") + assert len(dedup_logs) == 1 + + all_logs = lifecycle_db.get_lifecycle_logs("kb1") + assert len(all_logs) == 3 + + +@pytest.mark.unit +class TestSQLiteManagerLifecycle: + """Test the new SQLiteManager methods for lifecycle operations.""" + + def test_get_all_embeddings(self, lifecycle_db): + """get_all_embeddings returns docs with parsed embeddings.""" + kb_name = "test-emb-kb" + lifecycle_db.create_collection(kb_name, KBType.GENERIC) + + vec = np.random.randn(768).astype(np.float32) + lifecycle_db.add_documents( + kb_name, + documents=["Test content"], + embeddings=[vec.tolist()], + metadatas=[{"filename": "test.md"}], + ids=["doc-1"] + ) + + results = lifecycle_db.get_all_embeddings(kb_name) + assert len(results) == 1 + assert results[0]["doc_id"] == "doc-1" + assert results[0]["embedding"] is not None + assert len(results[0]["embedding"]) == 768 + + def test_update_document_metadata(self, lifecycle_db): + """update_document_metadata merges new keys into existing metadata.""" + kb_name = "test-meta-kb" + lifecycle_db.create_collection(kb_name, KBType.GENERIC) + + vec = np.random.randn(768).astype(np.float32) + lifecycle_db.add_documents( + kb_name, + documents=["Test content"], + embeddings=[vec.tolist()], + metadatas=[{"filename": "test.md", "tags": ["a"]}], + ids=["doc-1"] + ) + + success = lifecycle_db.update_document_metadata( + kb_name, "doc-1", {"archived": True, "new_key": "value"} + ) + assert success + + docs = lifecycle_db.get_all_embeddings(kb_name, exclude_archived=False) + doc = docs[0] + assert doc["metadata"]["archived"] is True + assert doc["metadata"]["new_key"] == "value" + assert doc["metadata"]["filename"] == "test.md" # original preserved + + def test_delete_documents_by_ids(self, lifecycle_db): + """delete_documents_by_ids removes specified documents.""" + kb_name = "test-delete-kb" + lifecycle_db.create_collection(kb_name, KBType.GENERIC) + + vec = np.random.randn(768).astype(np.float32) + lifecycle_db.add_documents( + kb_name, + documents=["Doc 1", "Doc 2", "Doc 3"], + embeddings=[vec.tolist(), vec.tolist(), vec.tolist()], + metadatas=[{"filename": "1.md"}, {"filename": "2.md"}, {"filename": "3.md"}], + ids=["doc-1", "doc-2", "doc-3"] + ) + + deleted = lifecycle_db.delete_documents_by_ids(kb_name, ["doc-1", "doc-3"]) + assert deleted == 2 + + remaining = lifecycle_db.get_all_embeddings(kb_name) + assert len(remaining) == 1 + assert remaining[0]["doc_id"] == "doc-2" + + +@pytest.mark.integration +class TestConsolidation: + """Integration tests for consolidation (requires Ollama).""" + + def test_consolidate_dry_run(self, engine): + """Dry run shows preview without making changes.""" + e, kb_name, db = engine + + result = e.consolidate_related( + kb_name, + doc_ids=["doc-auth-1", "doc-auth-2"], + new_filename="consolidated-auth.md", + dry_run=True + ) + + assert result["dry_run"] is True + assert result["source_count"] == 2 + assert result["total_chars"] > 0 + + # Verify originals still exist + docs = db.get_all_embeddings(kb_name) + doc_ids = [d["doc_id"] for d in docs] + assert "doc-auth-1" in doc_ids + assert "doc-auth-2" in doc_ids + + def test_consolidate_insufficient_docs(self, engine): + """Consolidation with <2 docs returns error.""" + e, kb_name, _ = engine + + result = e.consolidate_related( + kb_name, + doc_ids=["doc-auth-1"], + new_filename="single.md" + ) + + assert "error" in result