Coverage for session_mgmt_mcp/team_knowledge.py: 0.00%
284 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-09-01 05:22 -0700
1"""Team Knowledge Base module for shared reflection database with permissions.
3This module provides team collaboration features including:
4- Shared reflection database with user permissions
5- Team access control and role management
6- Collaborative knowledge sharing
7- Team-specific conversation organization
8"""
10import hashlib
11import importlib.util
12import json
13import logging
14import sqlite3
15import threading
16import time
17from dataclasses import asdict, dataclass
18from datetime import datetime, timedelta
19from enum import Enum
20from pathlib import Path
21from typing import Any
23DUCKDB_AVAILABLE = importlib.util.find_spec("duckdb") is not None
24AIOFILES_AVAILABLE = importlib.util.find_spec("aiofiles") is not None
26logger = logging.getLogger(__name__)
29class UserRole(Enum):
30 """User roles in team knowledge base."""
32 VIEWER = "viewer"
33 CONTRIBUTOR = "contributor"
34 MODERATOR = "moderator"
35 ADMIN = "admin"
38class AccessLevel(Enum):
39 """Access levels for knowledge base content."""
41 PRIVATE = "private"
42 TEAM = "team"
43 PROJECT = "project"
44 PUBLIC = "public"
47@dataclass
48class TeamUser:
49 """Team user information."""
51 user_id: str
52 username: str
53 email: str | None
54 role: UserRole
55 teams: list[str]
56 created_at: datetime
57 last_active: datetime
58 permissions: dict[str, bool]
61@dataclass
62class TeamReflection:
63 """Team-shared reflection with access control."""
65 id: str
66 content: str
67 tags: list[str]
68 access_level: AccessLevel
69 team_id: str | None
70 project_id: str | None
71 author_id: str
72 created_at: datetime
73 updated_at: datetime
74 votes: int
75 viewers: set[str]
76 editors: set[str]
79@dataclass
80class Team:
81 """Team information and configuration."""
83 team_id: str
84 name: str
85 description: str
86 owner_id: str
87 members: set[str]
88 projects: set[str]
89 created_at: datetime
90 settings: dict[str, Any]
93class TeamKnowledgeManager:
94 """Manages team knowledge base with permissions and access control."""
96 def __init__(self, db_path: str | None = None) -> None:
97 """Initialize team knowledge manager."""
98 self.db_path = db_path or str(
99 Path.home() / ".claude" / "data" / "team_knowledge.db",
100 )
101 self._lock = threading.Lock()
102 self._init_database()
104 def _init_database(self) -> None:
105 """Initialize SQLite database for team knowledge."""
106 Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)
108 with sqlite3.connect(self.db_path) as conn:
109 conn.execute("""
110 CREATE TABLE IF NOT EXISTS users (
111 user_id TEXT PRIMARY KEY,
112 username TEXT UNIQUE NOT NULL,
113 email TEXT,
114 role TEXT NOT NULL,
115 teams TEXT, -- JSON array
116 created_at TIMESTAMP,
117 last_active TIMESTAMP,
118 permissions TEXT -- JSON object
119 )
120 """)
122 conn.execute("""
123 CREATE TABLE IF NOT EXISTS teams (
124 team_id TEXT PRIMARY KEY,
125 name TEXT NOT NULL,
126 description TEXT,
127 owner_id TEXT NOT NULL,
128 members TEXT, -- JSON array
129 projects TEXT, -- JSON array
130 created_at TIMESTAMP,
131 settings TEXT, -- JSON object
132 FOREIGN KEY (owner_id) REFERENCES users(user_id)
133 )
134 """)
136 conn.execute("""
137 CREATE TABLE IF NOT EXISTS team_reflections (
138 id TEXT PRIMARY KEY,
139 content TEXT NOT NULL,
140 tags TEXT, -- JSON array
141 access_level TEXT NOT NULL,
142 team_id TEXT,
143 project_id TEXT,
144 author_id TEXT NOT NULL,
145 created_at TIMESTAMP,
146 updated_at TIMESTAMP,
147 votes INTEGER DEFAULT 0,
148 viewers TEXT, -- JSON array
149 editors TEXT, -- JSON array
150 FOREIGN KEY (author_id) REFERENCES users(user_id),
151 FOREIGN KEY (team_id) REFERENCES teams(team_id)
152 )
153 """)
155 conn.execute("""
156 CREATE TABLE IF NOT EXISTS access_logs (
157 id INTEGER PRIMARY KEY AUTOINCREMENT,
158 user_id TEXT NOT NULL,
159 action TEXT NOT NULL,
160 resource_id TEXT,
161 resource_type TEXT,
162 timestamp TIMESTAMP,
163 details TEXT -- JSON object
164 )
165 """)
167 # Create indices
168 conn.execute(
169 "CREATE INDEX IF NOT EXISTS idx_reflections_team ON team_reflections(team_id)",
170 )
171 conn.execute(
172 "CREATE INDEX IF NOT EXISTS idx_reflections_project ON team_reflections(project_id)",
173 )
174 conn.execute(
175 "CREATE INDEX IF NOT EXISTS idx_reflections_author ON team_reflections(author_id)",
176 )
177 conn.execute(
178 "CREATE INDEX IF NOT EXISTS idx_reflections_access ON team_reflections(access_level)",
179 )
180 conn.execute(
181 "CREATE INDEX IF NOT EXISTS idx_access_logs_user ON access_logs(user_id)",
182 )
183 conn.execute(
184 "CREATE INDEX IF NOT EXISTS idx_access_logs_timestamp ON access_logs(timestamp)",
185 )
187 async def create_user(
188 self,
189 user_id: str,
190 username: str,
191 email: str | None = None,
192 role: UserRole = UserRole.CONTRIBUTOR,
193 ) -> TeamUser:
194 """Create a new team user."""
195 user = TeamUser(
196 user_id=user_id,
197 username=username,
198 email=email,
199 role=role,
200 teams=[],
201 created_at=datetime.now(),
202 last_active=datetime.now(),
203 permissions=self._get_default_permissions(role),
204 )
206 with sqlite3.connect(self.db_path) as conn:
207 conn.execute(
208 """
209 INSERT INTO users (user_id, username, email, role, teams, created_at, last_active, permissions)
210 VALUES (?, ?, ?, ?, ?, ?, ?, ?)
211 """,
212 (
213 user.user_id,
214 user.username,
215 user.email,
216 user.role.value,
217 json.dumps(user.teams),
218 user.created_at,
219 user.last_active,
220 json.dumps(user.permissions),
221 ),
222 )
224 await self._log_access(
225 user_id,
226 "user_created",
227 user_id,
228 "user",
229 {"role": role.value},
230 )
231 return user
233 async def create_team(
234 self,
235 team_id: str,
236 name: str,
237 description: str,
238 owner_id: str,
239 ) -> Team:
240 """Create a new team."""
241 team = Team(
242 team_id=team_id,
243 name=name,
244 description=description,
245 owner_id=owner_id,
246 members={owner_id},
247 projects=set(),
248 created_at=datetime.now(),
249 settings={"auto_approve_members": False, "public_reflections": True},
250 )
252 with sqlite3.connect(self.db_path) as conn:
253 conn.execute(
254 """
255 INSERT INTO teams (team_id, name, description, owner_id, members, projects, created_at, settings)
256 VALUES (?, ?, ?, ?, ?, ?, ?, ?)
257 """,
258 (
259 team.team_id,
260 team.name,
261 team.description,
262 team.owner_id,
263 json.dumps(list(team.members)),
264 json.dumps(list(team.projects)),
265 team.created_at,
266 json.dumps(team.settings),
267 ),
268 )
270 # Add owner to team
271 await self._add_user_to_team(owner_id, team_id)
272 await self._log_access(
273 owner_id,
274 "team_created",
275 team_id,
276 "team",
277 {"name": name},
278 )
279 return team
281 async def add_team_reflection(
282 self,
283 content: str,
284 author_id: str,
285 tags: list[str] | None = None,
286 access_level: AccessLevel = AccessLevel.TEAM,
287 team_id: str | None = None,
288 project_id: str | None = None,
289 ) -> str:
290 """Add reflection to team knowledge base."""
291 reflection_id = hashlib.sha256(
292 f"{content}{author_id}{time.time()}".encode(),
293 ).hexdigest()[:16]
295 reflection = TeamReflection(
296 id=reflection_id,
297 content=content,
298 tags=tags or [],
299 access_level=access_level,
300 team_id=team_id,
301 project_id=project_id,
302 author_id=author_id,
303 created_at=datetime.now(),
304 updated_at=datetime.now(),
305 votes=0,
306 viewers=set(),
307 editors=set(),
308 )
310 with sqlite3.connect(self.db_path) as conn:
311 conn.execute(
312 """
313 INSERT INTO team_reflections
314 (id, content, tags, access_level, team_id, project_id, author_id, created_at, updated_at, votes, viewers, editors)
315 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
316 """,
317 (
318 reflection.id,
319 reflection.content,
320 json.dumps(reflection.tags),
321 reflection.access_level.value,
322 reflection.team_id,
323 reflection.project_id,
324 reflection.author_id,
325 reflection.created_at,
326 reflection.updated_at,
327 reflection.votes,
328 json.dumps(list(reflection.viewers)),
329 json.dumps(list(reflection.editors)),
330 ),
331 )
333 await self._log_access(
334 author_id,
335 "reflection_created",
336 reflection_id,
337 "reflection",
338 {"team_id": team_id, "access_level": access_level.value},
339 )
340 return reflection_id
342 async def search_team_reflections(
343 self,
344 query: str,
345 user_id: str,
346 team_id: str | None = None,
347 project_id: str | None = None,
348 tags: list[str] | None = None,
349 limit: int = 20,
350 ) -> list[dict[str, Any]]:
351 """Search team reflections with access control."""
352 user_teams = await self._get_user_teams(user_id)
354 with sqlite3.connect(self.db_path) as conn:
355 conn.row_factory = sqlite3.Row
357 # Build query with access control
358 where_conditions = ["1=1"]
359 params = []
361 # Access control: user can see public, team reflections they have access to, or their own
362 access_condition = """(
363 access_level = 'public' OR
364 (access_level = 'team' AND team_id IN ({}) AND team_id IS NOT NULL) OR
365 author_id = ?
366 )""".format(",".join("?" * len(user_teams)))
367 where_conditions.append(access_condition)
368 params.extend(user_teams)
369 params.append(user_id)
371 # Content search
372 if query:
373 where_conditions.append("(content LIKE ? OR tags LIKE ?)")
374 params.extend([f"%{query}%", f"%{query}%"])
376 # Team filter
377 if team_id:
378 where_conditions.append("team_id = ?")
379 params.append(team_id)
381 # Project filter
382 if project_id:
383 where_conditions.append("project_id = ?")
384 params.append(project_id)
386 # Tag filter
387 if tags:
388 tag_conditions = []
389 for tag in tags:
390 tag_conditions.append("tags LIKE ?")
391 params.append(f"%{tag}%")
392 where_conditions.append(f"({' OR '.join(tag_conditions)})")
394 query_sql = f"""
395 SELECT * FROM team_reflections
396 WHERE {" AND ".join(where_conditions)}
397 ORDER BY votes DESC, created_at DESC
398 LIMIT ?
399 """
400 params.append(limit)
402 cursor = conn.execute(query_sql, params)
403 results = []
405 for row in cursor.fetchall():
406 result = dict(row)
407 result["tags"] = json.loads(result["tags"] or "[]")
408 result["viewers"] = json.loads(result["viewers"] or "[]")
409 result["editors"] = json.loads(result["editors"] or "[]")
410 results.append(result)
412 await self._log_access(
413 user_id,
414 "reflections_searched",
415 None,
416 "search",
417 {"query": query, "results_count": len(results)},
418 )
419 return results
421 async def vote_reflection(
422 self,
423 reflection_id: str,
424 user_id: str,
425 vote_delta: int = 1,
426 ) -> bool:
427 """Vote on a team reflection."""
428 if not await self._can_access_reflection(reflection_id, user_id):
429 return False
431 with sqlite3.connect(self.db_path) as conn:
432 conn.execute(
433 """
434 UPDATE team_reflections
435 SET votes = votes + ?, updated_at = ?
436 WHERE id = ?
437 """,
438 (vote_delta, datetime.now(), reflection_id),
439 )
441 await self._log_access(
442 user_id,
443 "reflection_voted",
444 reflection_id,
445 "reflection",
446 {"vote_delta": vote_delta},
447 )
448 return True
450 async def join_team(
451 self,
452 user_id: str,
453 team_id: str,
454 requester_id: str | None = None,
455 ) -> bool:
456 """Request to join a team or add user to team."""
457 team = await self._get_team(team_id)
458 if not team:
459 return False
461 # Check if requester has permission to add users
462 if requester_id and requester_id != user_id:
463 if not await self._can_manage_team(requester_id, team_id):
464 return False
466 await self._add_user_to_team(user_id, team_id)
467 await self._log_access(
468 user_id,
469 "team_joined",
470 team_id,
471 "team",
472 {"requester_id": requester_id},
473 )
474 return True
476 async def get_team_stats(self, team_id: str, user_id: str) -> dict[str, Any] | None:
477 """Get team statistics and activity."""
478 if not await self._can_access_team(user_id, team_id):
479 return None
481 with sqlite3.connect(self.db_path) as conn:
482 conn.row_factory = sqlite3.Row
484 # Get team info
485 team_row = conn.execute(
486 "SELECT * FROM teams WHERE team_id = ?",
487 (team_id,),
488 ).fetchone()
489 if not team_row:
490 return None
492 team_data = dict(team_row)
493 team_data["members"] = json.loads(team_data["members"] or "[]")
494 team_data["projects"] = json.loads(team_data["projects"] or "[]")
496 # Get reflection stats
497 reflection_stats = conn.execute(
498 """
499 SELECT
500 COUNT(*) as total_reflections,
501 COUNT(DISTINCT author_id) as active_contributors,
502 SUM(votes) as total_votes,
503 AVG(votes) as avg_votes
504 FROM team_reflections
505 WHERE team_id = ?
506 """,
507 (team_id,),
508 ).fetchone()
510 # Get recent activity
511 recent_activity = conn.execute(
512 """
513 SELECT COUNT(*) as recent_reflections
514 FROM team_reflections
515 WHERE team_id = ? AND created_at > ?
516 """,
517 (team_id, datetime.now() - timedelta(days=7)),
518 ).fetchone()
520 stats = {
521 "team": dict(team_data),
522 "member_count": len(team_data["members"]),
523 "project_count": len(team_data["projects"]),
524 "reflection_stats": dict(reflection_stats),
525 "recent_activity": dict(recent_activity),
526 }
528 await self._log_access(user_id, "team_stats_viewed", team_id, "team", {})
529 return stats
531 async def get_user_permissions(self, user_id: str) -> dict[str, Any]:
532 """Get user's current permissions and team memberships."""
533 with sqlite3.connect(self.db_path) as conn:
534 conn.row_factory = sqlite3.Row
535 user_row = conn.execute(
536 "SELECT * FROM users WHERE user_id = ?",
537 (user_id,),
538 ).fetchone()
540 if not user_row:
541 return {}
543 user_data = dict(user_row)
544 user_data["teams"] = json.loads(user_data["teams"] or "[]")
545 user_data["permissions"] = json.loads(user_data["permissions"] or "{}")
547 # Get team details
548 team_details = []
549 if user_data["teams"]:
550 placeholders = ",".join("?" * len(user_data["teams"]))
551 team_rows = conn.execute(
552 f"SELECT team_id, name, description FROM teams WHERE team_id IN ({placeholders})",
553 user_data["teams"],
554 ).fetchall()
555 team_details = [dict(row) for row in team_rows]
557 return {
558 "user": user_data,
559 "teams": team_details,
560 "can_create_teams": user_data["permissions"].get("create_teams", False),
561 "can_moderate": user_data["permissions"].get("moderate_content", False),
562 }
564 # Private helper methods
566 def _get_default_permissions(self, role: UserRole) -> dict[str, bool]:
567 """Get default permissions for user role."""
568 base_permissions = {
569 "read_reflections": True,
570 "create_reflections": False,
571 "vote_reflections": False,
572 "join_teams": False,
573 "create_teams": False,
574 "moderate_content": False,
575 "admin_access": False,
576 }
578 if role == UserRole.CONTRIBUTOR:
579 base_permissions.update(
580 {
581 "create_reflections": True,
582 "vote_reflections": True,
583 "join_teams": True,
584 },
585 )
586 elif role == UserRole.MODERATOR:
587 base_permissions.update(
588 {
589 "create_reflections": True,
590 "vote_reflections": True,
591 "join_teams": True,
592 "create_teams": True,
593 "moderate_content": True,
594 },
595 )
596 elif role == UserRole.ADMIN:
597 base_permissions.update(dict.fromkeys(base_permissions.keys(), True))
599 return base_permissions
601 async def _add_user_to_team(self, user_id: str, team_id: str) -> None:
602 """Add user to team."""
603 with sqlite3.connect(self.db_path) as conn:
604 # Update team members
605 team_row = conn.execute(
606 "SELECT members FROM teams WHERE team_id = ?",
607 (team_id,),
608 ).fetchone()
609 if team_row:
610 members = set(json.loads(team_row[0] or "[]"))
611 members.add(user_id)
612 conn.execute(
613 "UPDATE teams SET members = ? WHERE team_id = ?",
614 (json.dumps(list(members)), team_id),
615 )
617 # Update user teams
618 user_row = conn.execute(
619 "SELECT teams FROM users WHERE user_id = ?",
620 (user_id,),
621 ).fetchone()
622 if user_row:
623 teams = json.loads(user_row[0] or "[]")
624 if team_id not in teams:
625 teams.append(team_id)
626 conn.execute(
627 "UPDATE users SET teams = ?, last_active = ? WHERE user_id = ?",
628 (json.dumps(teams), datetime.now(), user_id),
629 )
631 async def _get_user_teams(self, user_id: str) -> list[str]:
632 """Get teams user belongs to."""
633 with sqlite3.connect(self.db_path) as conn:
634 row = conn.execute(
635 "SELECT teams FROM users WHERE user_id = ?",
636 (user_id,),
637 ).fetchone()
638 return json.loads(row[0] or "[]") if row else []
640 async def _get_team(self, team_id: str) -> dict[str, Any] | None:
641 """Get team information."""
642 with sqlite3.connect(self.db_path) as conn:
643 conn.row_factory = sqlite3.Row
644 row = conn.execute(
645 "SELECT * FROM teams WHERE team_id = ?",
646 (team_id,),
647 ).fetchone()
648 if row:
649 team_data = dict(row)
650 team_data["members"] = set(json.loads(team_data["members"] or "[]"))
651 team_data["projects"] = set(json.loads(team_data["projects"] or "[]"))
652 team_data["settings"] = json.loads(team_data["settings"] or "{}")
653 return team_data
654 return None
656 async def _can_access_reflection(self, reflection_id: str, user_id: str) -> bool:
657 """Check if user can access reflection."""
658 with sqlite3.connect(self.db_path) as conn:
659 row = conn.execute(
660 """
661 SELECT access_level, team_id, author_id FROM team_reflections
662 WHERE id = ?
663 """,
664 (reflection_id,),
665 ).fetchone()
667 if not row:
668 return False
670 access_level, team_id, author_id = row
672 # Author can always access
673 if author_id == user_id:
674 return True
676 # Public reflections accessible to all
677 if access_level == AccessLevel.PUBLIC.value:
678 return True
680 # Team reflections require team membership
681 if access_level == AccessLevel.TEAM.value and team_id:
682 user_teams = await self._get_user_teams(user_id)
683 return team_id in user_teams
685 return False
687 async def _can_access_team(self, user_id: str, team_id: str) -> bool:
688 """Check if user can access team."""
689 user_teams = await self._get_user_teams(user_id)
690 return team_id in user_teams
692 async def _can_manage_team(self, user_id: str, team_id: str) -> bool:
693 """Check if user can manage team."""
694 team = await self._get_team(team_id)
695 if not team:
696 return False
698 # Team owner can manage
699 if team["owner_id"] == user_id:
700 return True
702 # Check if user has admin permissions
703 with sqlite3.connect(self.db_path) as conn:
704 row = conn.execute(
705 "SELECT permissions FROM users WHERE user_id = ?",
706 (user_id,),
707 ).fetchone()
708 if row:
709 permissions = json.loads(row[0] or "{}")
710 return permissions.get("admin_access", False) or permissions.get(
711 "moderate_content",
712 False,
713 )
715 return False
717 async def _log_access(
718 self,
719 user_id: str,
720 action: str,
721 resource_id: str | None,
722 resource_type: str,
723 details: dict[str, Any],
724 ) -> None:
725 """Log user access for audit trail."""
726 with sqlite3.connect(self.db_path) as conn:
727 conn.execute(
728 """
729 INSERT INTO access_logs (user_id, action, resource_id, resource_type, timestamp, details)
730 VALUES (?, ?, ?, ?, ?, ?)
731 """,
732 (
733 user_id,
734 action,
735 resource_id,
736 resource_type,
737 datetime.now(),
738 json.dumps(details),
739 ),
740 )
743# Global instance
744_team_knowledge_manager = None
747def get_team_knowledge_manager() -> TeamKnowledgeManager:
748 """Get global team knowledge manager instance."""
749 global _team_knowledge_manager
750 if _team_knowledge_manager is None:
751 _team_knowledge_manager = TeamKnowledgeManager()
752 return _team_knowledge_manager
755# Public API functions for MCP tools
756async def create_team_user(
757 user_id: str,
758 username: str,
759 email: str | None = None,
760 role: str = "contributor",
761) -> dict[str, Any]:
762 """Create a new team user."""
763 manager = get_team_knowledge_manager()
764 user_role = UserRole(role.lower())
765 user = await manager.create_user(user_id, username, email, user_role)
766 return asdict(user)
769async def create_team(
770 team_id: str,
771 name: str,
772 description: str,
773 owner_id: str,
774) -> dict[str, Any]:
775 """Create a new team."""
776 manager = get_team_knowledge_manager()
777 team = await manager.create_team(team_id, name, description, owner_id)
778 return {
779 "team_id": team.team_id,
780 "name": team.name,
781 "description": team.description,
782 "owner_id": team.owner_id,
783 "member_count": len(team.members),
784 "project_count": len(team.projects),
785 "created_at": team.created_at.isoformat(),
786 "settings": team.settings,
787 }
790async def add_team_reflection(
791 content: str,
792 author_id: str,
793 tags: list[str] | None = None,
794 access_level: str = "team",
795 team_id: str | None = None,
796 project_id: str | None = None,
797) -> str:
798 """Add reflection to team knowledge base."""
799 manager = get_team_knowledge_manager()
800 level = AccessLevel(access_level.lower())
801 return await manager.add_team_reflection(
802 content,
803 author_id,
804 tags,
805 level,
806 team_id,
807 project_id,
808 )
811async def search_team_knowledge(
812 query: str,
813 user_id: str,
814 team_id: str | None = None,
815 project_id: str | None = None,
816 tags: list[str] | None = None,
817 limit: int = 20,
818) -> list[dict[str, Any]]:
819 """Search team reflections with access control."""
820 manager = get_team_knowledge_manager()
821 return await manager.search_team_reflections(
822 query,
823 user_id,
824 team_id,
825 project_id,
826 tags,
827 limit,
828 )
831async def join_team(
832 user_id: str,
833 team_id: str,
834 requester_id: str | None = None,
835) -> bool:
836 """Join a team or add user to team."""
837 manager = get_team_knowledge_manager()
838 return await manager.join_team(user_id, team_id, requester_id)
841async def get_team_statistics(team_id: str, user_id: str) -> dict[str, Any] | None:
842 """Get team statistics and activity."""
843 manager = get_team_knowledge_manager()
844 return await manager.get_team_stats(team_id, user_id)
847async def get_user_team_permissions(user_id: str) -> dict[str, Any]:
848 """Get user's permissions and team memberships."""
849 manager = get_team_knowledge_manager()
850 return await manager.get_user_permissions(user_id)
853async def vote_on_reflection(
854 reflection_id: str,
855 user_id: str,
856 vote_delta: int = 1,
857) -> bool:
858 """Vote on a team reflection."""
859 manager = get_team_knowledge_manager()
860 return await manager.vote_reflection(reflection_id, user_id, vote_delta)