Coverage for session_buddy / team_knowledge.py: 28.25%

1"""Team Knowledge Base module for shared reflection database with permissions. 

2 

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""" 

9 

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 

22 

23DUCKDB_AVAILABLE = importlib.util.find_spec("duckdb") is not None 

24AIOFILES_AVAILABLE = importlib.util.find_spec("aiofiles") is not None 

25 

26logger = logging.getLogger(__name__) 

27 

28 

29class UserRole(Enum): 

30 """User roles in team knowledge base.""" 

31 

32 VIEWER = "viewer" 

33 CONTRIBUTOR = "contributor" 

34 MODERATOR = "moderator" 

35 ADMIN = "admin" 

36 

37 

38class AccessLevel(Enum): 

39 """Access levels for knowledge base content.""" 

40 

41 PRIVATE = "private" 

42 TEAM = "team" 

43 PROJECT = "project" 

44 PUBLIC = "public" 

45 

46 

47@dataclass 

48class TeamUser: 

49 """Team user information.""" 

50 

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] 

59 

60 

61@dataclass 

62class TeamReflection: 

63 """Team-shared reflection with access control.""" 

64 

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] 

77 

78 

79@dataclass 

80class Team: 

81 """Team information and configuration.""" 

82 

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] 

91 

92 

93class TeamKnowledgeManager: 

94 """Manages team knowledge base with permissions and access control.""" 

95 

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() 

103 

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) 

107 

108 with sqlite3.connect(self.db_path) as conn: 

109 self._create_tables(conn) 

110 self._create_indices(conn) 

111 

112 def _create_tables(self, conn: sqlite3.Connection) -> None: 

113 """Create database tables.""" 

114 self._create_users_table(conn) 

115 self._create_teams_table(conn) 

116 self._create_reflections_table(conn) 

117 self._create_access_logs_table(conn) 

118 

119 def _create_users_table(self, conn: sqlite3.Connection) -> None: 

120 """Create users table.""" 

121 conn.execute(""" 

122 CREATE TABLE IF NOT EXISTS users ( 

123 user_id TEXT PRIMARY KEY, 

124 username TEXT UNIQUE NOT NULL, 

125 email TEXT, 

126 role TEXT NOT NULL, 

127 teams TEXT, -- JSON array 

128 created_at TIMESTAMP, 

129 last_active TIMESTAMP, 

130 permissions TEXT -- JSON object 

131 ) 

132 """) 

133 

134 def _create_teams_table(self, conn: sqlite3.Connection) -> None: 

135 """Create teams table.""" 

136 conn.execute(""" 

137 CREATE TABLE IF NOT EXISTS teams ( 

138 team_id TEXT PRIMARY KEY, 

139 name TEXT NOT NULL, 

140 description TEXT, 

141 owner_id TEXT NOT NULL, 

142 members TEXT, -- JSON array 

143 projects TEXT, -- JSON array 

144 created_at TIMESTAMP, 

145 settings TEXT, -- JSON object 

146 FOREIGN KEY (owner_id) REFERENCES users(user_id) 

147 ) 

148 """) 

149 

150 def _create_reflections_table(self, conn: sqlite3.Connection) -> None: 

151 """Create team_reflections table.""" 

152 conn.execute(""" 

153 CREATE TABLE IF NOT EXISTS team_reflections ( 

154 id TEXT PRIMARY KEY, 

155 content TEXT NOT NULL, 

156 tags TEXT, -- JSON array 

157 access_level TEXT NOT NULL, 

158 team_id TEXT, 

159 project_id TEXT, 

160 author_id TEXT NOT NULL, 

161 created_at TIMESTAMP, 

162 updated_at TIMESTAMP, 

163 votes INTEGER DEFAULT 0, 

164 viewers TEXT, -- JSON array 

165 editors TEXT, -- JSON array 

166 FOREIGN KEY (author_id) REFERENCES users(user_id), 

167 FOREIGN KEY (team_id) REFERENCES teams(team_id) 

168 ) 

169 """) 

170 

171 def _create_access_logs_table(self, conn: sqlite3.Connection) -> None: 

172 """Create access_logs table.""" 

173 conn.execute(""" 

174 CREATE TABLE IF NOT EXISTS access_logs ( 

175 id INTEGER PRIMARY KEY AUTOINCREMENT, 

176 user_id TEXT NOT NULL, 

177 action TEXT NOT NULL, 

178 resource_id TEXT, 

179 resource_type TEXT, 

180 timestamp TIMESTAMP, 

181 details TEXT -- JSON object 

182 ) 

183 """) 

184 

185 def _create_indices(self, conn: sqlite3.Connection) -> None: 

186 """Create database indices.""" 

187 indices = [ 

188 "CREATE INDEX IF NOT EXISTS idx_reflections_team ON team_reflections(team_id)", 

189 "CREATE INDEX IF NOT EXISTS idx_reflections_project ON team_reflections(project_id)", 

190 "CREATE INDEX IF NOT EXISTS idx_reflections_author ON team_reflections(author_id)", 

191 "CREATE INDEX IF NOT EXISTS idx_reflections_access ON team_reflections(access_level)", 

192 "CREATE INDEX IF NOT EXISTS idx_access_logs_user ON access_logs(user_id)", 

193 "CREATE INDEX IF NOT EXISTS idx_access_logs_timestamp ON access_logs(timestamp)", 

194 ] 

195 

196 for index_sql in indices: 

197 conn.execute(index_sql) 

198 

199 async def create_user( 

200 self, 

201 user_id: str, 

202 username: str, 

203 email: str | None = None, 

204 role: UserRole = UserRole.CONTRIBUTOR, 

205 ) -> TeamUser: 

206 """Create a new team user.""" 

207 user = TeamUser( 

208 user_id=user_id, 

209 username=username, 

210 email=email, 

211 role=role, 

212 teams=[], 

213 created_at=datetime.now(), 

214 last_active=datetime.now(), 

215 permissions=self._get_default_permissions(role), 

216 ) 

217 

218 with sqlite3.connect(self.db_path) as conn: 

219 conn.execute( 

220 """ 

221 INSERT INTO users (user_id, username, email, role, teams, created_at, last_active, permissions) 

222 VALUES (?, ?, ?, ?, ?, ?, ?, ?) 

223 """, 

224 ( 

225 user.user_id, 

226 user.username, 

227 user.email, 

228 user.role.value, 

229 json.dumps(user.teams), 

230 user.created_at, 

231 user.last_active, 

232 json.dumps(user.permissions), 

233 ), 

234 ) 

235 

236 await self._log_access( 

237 user_id, 

238 "user_created", 

239 user_id, 

240 "user", 

241 {"role": role.value}, 

242 ) 

243 return user 

244 

245 async def create_team( 

246 self, 

247 team_id: str, 

248 name: str, 

249 description: str, 

250 owner_id: str, 

251 ) -> Team: 

252 """Create a new team.""" 

253 team = Team( 

254 team_id=team_id, 

255 name=name, 

256 description=description, 

257 owner_id=owner_id, 

258 members={owner_id}, 

259 projects=set(), 

260 created_at=datetime.now(), 

261 settings={"auto_approve_members": False, "public_reflections": True}, 

262 ) 

263 

264 with sqlite3.connect(self.db_path) as conn: 

265 conn.execute( 

266 """ 

267 INSERT INTO teams (team_id, name, description, owner_id, members, projects, created_at, settings) 

268 VALUES (?, ?, ?, ?, ?, ?, ?, ?) 

269 """, 

270 ( 

271 team.team_id, 

272 team.name, 

273 team.description, 

274 team.owner_id, 

275 json.dumps(list(team.members)), 

276 json.dumps(list(team.projects)), 

277 team.created_at, 

278 json.dumps(team.settings), 

279 ), 

280 ) 

281 

282 # Add owner to team 

283 await self._add_user_to_team(owner_id, team_id) 

284 await self._log_access( 

285 owner_id, 

286 "team_created", 

287 team_id, 

288 "team", 

289 {"name": name}, 

290 ) 

291 return team 

292 

293 async def add_team_reflection( 

294 self, 

295 content: str, 

296 author_id: str, 

297 tags: list[str] | None = None, 

298 access_level: AccessLevel = AccessLevel.TEAM, 

299 team_id: str | None = None, 

300 project_id: str | None = None, 

301 ) -> str: 

302 """Add reflection to team knowledge base.""" 

303 reflection_id = hashlib.sha256( 

304 f"{content}{author_id}{time.time()}".encode(), 

305 ).hexdigest()[:16] 

306 

307 reflection = TeamReflection( 

308 id=reflection_id, 

309 content=content, 

310 tags=tags or [], 

311 access_level=access_level, 

312 team_id=team_id, 

313 project_id=project_id, 

314 author_id=author_id, 

315 created_at=datetime.now(), 

316 updated_at=datetime.now(), 

317 votes=0, 

318 viewers=set(), 

319 editors=set(), 

320 ) 

321 

322 with sqlite3.connect(self.db_path) as conn: 

323 conn.execute( 

324 """ 

325 INSERT INTO team_reflections 

326 (id, content, tags, access_level, team_id, project_id, author_id, created_at, updated_at, votes, viewers, editors) 

327 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) 

328 """, 

329 ( 

330 reflection.id, 

331 reflection.content, 

332 json.dumps(reflection.tags), 

333 reflection.access_level.value, 

334 reflection.team_id, 

335 reflection.project_id, 

336 reflection.author_id, 

337 reflection.created_at, 

338 reflection.updated_at, 

339 reflection.votes, 

340 json.dumps(list(reflection.viewers)), 

341 json.dumps(list(reflection.editors)), 

342 ), 

343 ) 

344 

345 await self._log_access( 

346 author_id, 

347 "reflection_created", 

348 reflection_id, 

349 "reflection", 

350 {"team_id": team_id, "access_level": access_level.value}, 

351 ) 

352 return reflection_id 

353 

354 async def search_team_reflections( 

355 self, 

356 query: str, 

357 user_id: str, 

358 team_id: str | None = None, 

359 project_id: str | None = None, 

360 tags: list[str] | None = None, 

361 limit: int = 20, 

362 ) -> list[dict[str, Any]]: 

363 """Search team reflections with access control.""" 

364 user_teams = await self._get_user_teams(user_id) 

365 

366 with sqlite3.connect(self.db_path) as conn: 

367 conn.row_factory = sqlite3.Row 

368 

369 query_builder = self._build_search_query( 

370 user_teams, 

371 user_id, 

372 query, 

373 team_id, 

374 project_id, 

375 tags, 

376 limit, 

377 ) 

378 cursor = conn.execute(query_builder.sql, query_builder.params) 

379 results = self._process_search_results(cursor.fetchall()) 

380 

381 await self._log_access( 

382 user_id, 

383 "reflections_searched", 

384 None, 

385 "search", 

386 {"query": query, "results_count": len(results)}, 

387 ) 

388 return results 

389 

390 @dataclass(frozen=True) 

391 class _SearchQueryBuilder: 

392 """Immutable search query builder result.""" 

393 

394 sql: str 

395 params: list[str | int] 

396 

397 def _build_search_query( 

398 self, 

399 user_teams: list[str], 

400 user_id: str, 

401 query: str, 

402 team_id: str | None, 

403 project_id: str | None, 

404 tags: list[str] | None, 

405 limit: int, 

406 ) -> _SearchQueryBuilder: 

407 """Build parameterized search query with conditions.""" 

408 where_conditions = ["1=1"] 

409 params: list[str | int] = [] 

410 

411 # Add access control conditions 

412 access_condition, access_params = self._build_access_condition( 

413 user_teams, 

414 user_id, 

415 ) 

416 where_conditions.append(access_condition) 

417 params.extend(access_params) 

418 

419 # Add content search conditions 

420 if query: 

421 where_conditions.append("(content LIKE ? OR tags LIKE ?)") 

422 params.extend([f"%{query}%", f"%{query}%"]) 

423 

424 # Add filter conditions 

425 self._add_filter_conditions(where_conditions, params, team_id, project_id, tags) 

426 

427 # Build SQL safely - all user input is parameterized via params list 

428 query_sql = ( 

429 "SELECT * FROM team_reflections WHERE " 

430 + " AND ".join(where_conditions) 

431 + " ORDER BY votes DESC, created_at DESC LIMIT ?" 

432 ) 

433 params.append(limit) 

434 

435 return self._SearchQueryBuilder(sql=query_sql, params=params) 

436 

437 def _build_access_condition( 

438 self, 

439 user_teams: list[str], 

440 user_id: str, 

441 ) -> tuple[str, list[str]]: 

442 """Build access control condition for query.""" 

443 placeholders = ",".join("?" * len(user_teams)) 

444 access_condition = f"""( 

445 access_level = 'public' OR 

446 (access_level = 'team' AND team_id IN ({placeholders}) AND team_id IS NOT NULL) OR 

447 author_id = ? 

448 )""" 

449 access_params = [*user_teams, user_id] 

450 return access_condition, access_params 

451 

452 def _add_filter_conditions( 

453 self, 

454 where_conditions: list[str], 

455 params: list[str | int], 

456 team_id: str | None, 

457 project_id: str | None, 

458 tags: list[str] | None, 

459 ) -> None: 

460 """Add filter conditions to query builder.""" 

461 if team_id: 

462 where_conditions.append("team_id = ?") 

463 params.append(team_id) 

464 

465 if project_id: 

466 where_conditions.append("project_id = ?") 

467 params.append(project_id) 

468 

469 if tags: 

470 tag_conditions = [] 

471 for tag in tags: 

472 params.append(f"%{tag}%") # type: ignore[func-returns-value] 

473 tag_conditions.append("tags LIKE ?") 

474 where_conditions.append(f"({' OR '.join(tag_conditions)})") 

475 

476 def _process_search_results(self, rows: list[sqlite3.Row]) -> list[dict[str, Any]]: 

477 """Process and transform search results.""" 

478 return [ 

479 dict(row) 

480 | { 

481 "tags": json.loads(row["tags"] or "[]"), 

482 "viewers": json.loads(row["viewers"] or "[]"), 

483 "editors": json.loads(row["editors"] or "[]"), 

484 } 

485 for row in rows 

486 ] 

487 

488 async def vote_reflection( 

489 self, 

490 reflection_id: str, 

491 user_id: str, 

492 vote_delta: int = 1, 

493 ) -> bool: 

494 """Vote on a team reflection.""" 

495 if not await self._can_access_reflection(reflection_id, user_id): 

496 return False 

497 

498 with sqlite3.connect(self.db_path) as conn: 

499 conn.execute( 

500 """ 

501 UPDATE team_reflections 

502 SET votes = votes + ?, updated_at = ? 

503 WHERE id = ? 

504 """, 

505 (vote_delta, datetime.now(), reflection_id), 

506 ) 

507 

508 await self._log_access( 

509 user_id, 

510 "reflection_voted", 

511 reflection_id, 

512 "reflection", 

513 {"vote_delta": vote_delta}, 

514 ) 

515 return True 

516 

517 async def join_team( 

518 self, 

519 user_id: str, 

520 team_id: str, 

521 requester_id: str | None = None, 

522 ) -> bool: 

523 """Request to join a team or add user to team.""" 

524 team = await self._get_team(team_id) 

525 if not team: 

526 return False 

527 

528 # Check if requester has permission to add users 

529 if requester_id and requester_id != user_id: 

530 if not await self._can_manage_team(requester_id, team_id): 

531 return False 

532 

533 await self._add_user_to_team(user_id, team_id) 

534 await self._log_access( 

535 user_id, 

536 "team_joined", 

537 team_id, 

538 "team", 

539 {"requester_id": requester_id}, 

540 ) 

541 return True 

542 

543 async def get_team_stats(self, team_id: str, user_id: str) -> dict[str, Any] | None: 

544 """Get team statistics and activity.""" 

545 if not await self._can_access_team(user_id, team_id): 

546 return None 

547 

548 with sqlite3.connect(self.db_path) as conn: 

549 conn.row_factory = sqlite3.Row 

550 

551 # Get team info 

552 team_row = conn.execute( 

553 "SELECT * FROM teams WHERE team_id = ?", 

554 (team_id,), 

555 ).fetchone() 

556 if not team_row: 

557 return None 

558 

559 team_data = dict(team_row) 

560 team_data["members"] = json.loads(team_data["members"] or "[]") 

561 team_data["projects"] = json.loads(team_data["projects"] or "[]") 

562 

563 # Get reflection stats 

564 reflection_stats = conn.execute( 

565 """ 

566 SELECT 

567 COUNT(*) as total_reflections, 

568 COUNT(DISTINCT author_id) as active_contributors, 

569 SUM(votes) as total_votes, 

570 AVG(votes) as avg_votes 

571 FROM team_reflections 

572 WHERE team_id = ? 

573 """, 

574 (team_id,), 

575 ).fetchone() 

576 

577 # Get recent activity 

578 recent_activity = conn.execute( 

579 """ 

580 SELECT COUNT(*) as recent_reflections 

581 FROM team_reflections 

582 WHERE team_id = ? AND created_at > ? 

583 """, 

584 (team_id, datetime.now() - timedelta(days=7)), 

585 ).fetchone() 

586 

587 stats = { 

588 "team": team_data.copy(), 

589 "member_count": len(team_data["members"]), 

590 "project_count": len(team_data["projects"]), 

591 "reflection_stats": dict(reflection_stats), 

592 "recent_activity": dict(recent_activity), 

593 } 

594 

595 await self._log_access(user_id, "team_stats_viewed", team_id, "team", {}) 

596 return stats 

597 

598 async def get_user_permissions(self, user_id: str) -> dict[str, Any]: 

599 """Get user's current permissions and team memberships.""" 

600 with sqlite3.connect(self.db_path) as conn: 

601 conn.row_factory = sqlite3.Row 

602 user_row = conn.execute( 

603 "SELECT * FROM users WHERE user_id = ?", 

604 (user_id,), 

605 ).fetchone() 

606 

607 if not user_row: 

608 return {} 

609 

610 user_data = dict(user_row) 

611 user_data["teams"] = json.loads(user_data["teams"] or "[]") 

612 user_data["permissions"] = json.loads(user_data["permissions"] or "{}") 

613 

614 # Get team details 

615 team_details = [] 

616 if user_data["teams"]: 

617 placeholders = ",".join("?" * len(user_data["teams"])) 

618 # Build SQL safely - placeholders generated from list length, not user input 

619 query = ( 

620 "SELECT team_id, name, description FROM teams WHERE team_id IN (" 

621 + placeholders 

622 + ")" 

623 ) 

624 team_rows = conn.execute(query, user_data["teams"]).fetchall() 

625 team_details = [dict(row) for row in team_rows] 

626 

627 return { 

628 "user": user_data, 

629 "teams": team_details, 

630 "can_create_teams": user_data["permissions"].get("create_teams", False), 

631 "can_moderate": user_data["permissions"].get("moderate_content", False), 

632 } 

633 

634 # Private helper methods 

635 

636 def _get_default_permissions(self, role: UserRole) -> dict[str, bool]: 

637 """Get default permissions for user role.""" 

638 base_permissions = { 

639 "read_reflections": True, 

640 "create_reflections": False, 

641 "vote_reflections": False, 

642 "join_teams": False, 

643 "create_teams": False, 

644 "moderate_content": False, 

645 "admin_access": False, 

646 } 

647 

648 if role == UserRole.CONTRIBUTOR: 

649 base_permissions.update( 

650 { 

651 "create_reflections": True, 

652 "vote_reflections": True, 

653 "join_teams": True, 

654 }, 

655 ) 

656 elif role == UserRole.MODERATOR: 

657 base_permissions.update( 

658 { 

659 "create_reflections": True, 

660 "vote_reflections": True, 

661 "join_teams": True, 

662 "create_teams": True, 

663 "moderate_content": True, 

664 }, 

665 ) 

666 elif role == UserRole.ADMIN: 

667 base_permissions.update(dict.fromkeys(base_permissions.keys(), True)) 

668 

669 return base_permissions 

670 

671 async def _add_user_to_team(self, user_id: str, team_id: str) -> None: 

672 """Add user to team.""" 

673 with sqlite3.connect(self.db_path) as conn: 

674 # Update team members 

675 team_row = conn.execute( 

676 "SELECT members FROM teams WHERE team_id = ?", 

677 (team_id,), 

678 ).fetchone() 

679 if team_row: 

680 members = set(json.loads(team_row[0] or "[]")) 

681 members.add(user_id) 

682 conn.execute( 

683 "UPDATE teams SET members = ? WHERE team_id = ?", 

684 (json.dumps(list(members)), team_id), 

685 ) 

686 

687 # Update user teams 

688 user_row = conn.execute( 

689 "SELECT teams FROM users WHERE user_id = ?", 

690 (user_id,), 

691 ).fetchone() 

692 if user_row: 

693 teams = json.loads(user_row[0] or "[]") 

694 if team_id not in teams: 

695 teams.append(team_id) 

696 conn.execute( 

697 "UPDATE users SET teams = ?, last_active = ? WHERE user_id = ?", 

698 (json.dumps(teams), datetime.now(), user_id), 

699 ) 

700 

701 async def _get_user_teams(self, user_id: str) -> list[str]: 

702 """Get teams user belongs to.""" 

703 with sqlite3.connect(self.db_path) as conn: 

704 row = conn.execute( 

705 "SELECT teams FROM users WHERE user_id = ?", 

706 (user_id,), 

707 ).fetchone() 

708 return json.loads(row[0] or "[]") if row else [] 

709 

710 async def _get_team(self, team_id: str) -> dict[str, Any] | None: 

711 """Get team information.""" 

712 with sqlite3.connect(self.db_path) as conn: 

713 conn.row_factory = sqlite3.Row 

714 row = conn.execute( 

715 "SELECT * FROM teams WHERE team_id = ?", 

716 (team_id,), 

717 ).fetchone() 

718 if row: 

719 team_data = dict(row) 

720 team_data["members"] = set(json.loads(team_data["members"] or "[]")) 

721 team_data["projects"] = set(json.loads(team_data["projects"] or "[]")) 

722 team_data["settings"] = json.loads(team_data["settings"] or "{}") 

723 return team_data 

724 return None 

725 

726 async def _can_access_reflection(self, reflection_id: str, user_id: str) -> bool: 

727 """Check if user can access reflection.""" 

728 with sqlite3.connect(self.db_path) as conn: 

729 row = conn.execute( 

730 """ 

731 SELECT access_level, team_id, author_id FROM team_reflections 

732 WHERE id = ? 

733 """, 

734 (reflection_id,), 

735 ).fetchone() 

736 

737 if not row: 

738 return False 

739 

740 access_level, team_id, author_id = row 

741 

742 # Author can always access 

743 if author_id == user_id: 

744 return True 

745 

746 # Public reflections accessible to all 

747 if access_level == AccessLevel.PUBLIC.value: 

748 return True 

749 

750 # Team reflections require team membership 

751 if access_level == AccessLevel.TEAM.value and team_id: 

752 user_teams = await self._get_user_teams(user_id) 

753 return team_id in user_teams 

754 

755 return False 

756 

757 async def _can_access_team(self, user_id: str, team_id: str) -> bool: 

758 """Check if user can access team.""" 

759 user_teams = await self._get_user_teams(user_id) 

760 return team_id in user_teams 

761 

762 async def _can_manage_team(self, user_id: str, team_id: str) -> bool: 

763 """Check if user can manage team.""" 

764 team = await self._get_team(team_id) 

765 if not team: 

766 return False 

767 

768 # Team owner can manage 

769 if team["owner_id"] == user_id: 

770 return True 

771 

772 # Check if user has admin permissions 

773 with sqlite3.connect(self.db_path) as conn: 

774 row = conn.execute( 

775 "SELECT permissions FROM users WHERE user_id = ?", 

776 (user_id,), 

777 ).fetchone() 

778 if row: 

779 permissions = json.loads(row[0] or "{}") 

780 return bool( 

781 permissions.get("admin_access", False) 

782 or permissions.get( 

783 "moderate_content", 

784 False, 

785 ), 

786 ) 

787 

788 return False 

789 

790 async def _log_access( 

791 self, 

792 user_id: str, 

793 action: str, 

794 resource_id: str | None, 

795 resource_type: str, 

796 details: dict[str, Any], 

797 ) -> None: 

798 """Log user access for audit trail.""" 

799 with sqlite3.connect(self.db_path) as conn: 

800 conn.execute( 

801 """ 

802 INSERT INTO access_logs (user_id, action, resource_id, resource_type, timestamp, details) 

803 VALUES (?, ?, ?, ?, ?, ?) 

804 """, 

805 ( 

806 user_id, 

807 action, 

808 resource_id, 

809 resource_type, 

810 datetime.now(), 

811 json.dumps(details), 

812 ), 

813 ) 

814 

815 

816# Global instance 

817_team_knowledge_manager = None 

818 

819 

820def get_team_knowledge_manager() -> TeamKnowledgeManager: 

821 """Get global team knowledge manager instance.""" 

822 global _team_knowledge_manager 

823 if _team_knowledge_manager is None: 

824 _team_knowledge_manager = TeamKnowledgeManager() 

825 return _team_knowledge_manager 

826 

827 

828# Public API functions for MCP tools 

829async def create_team_user( 

830 user_id: str, 

831 username: str, 

832 email: str | None = None, 

833 role: str = "contributor", 

834) -> dict[str, Any]: 

835 """Create a new team user.""" 

836 manager = get_team_knowledge_manager() 

837 user_role = UserRole(role.lower()) 

838 user = await manager.create_user(user_id, username, email, user_role) 

839 return asdict(user) 

840 

841 

842async def create_team( 

843 team_id: str, 

844 name: str, 

845 description: str, 

846 owner_id: str, 

847) -> dict[str, Any]: 

848 """Create a new team.""" 

849 manager = get_team_knowledge_manager() 

850 team = await manager.create_team(team_id, name, description, owner_id) 

851 return { 

852 "team_id": team.team_id, 

853 "name": team.name, 

854 "description": team.description, 

855 "owner_id": team.owner_id, 

856 "member_count": len(team.members), 

857 "project_count": len(team.projects), 

858 "created_at": team.created_at.isoformat(), 

859 "settings": team.settings, 

860 } 

861 

862 

863async def add_team_reflection( 

864 content: str, 

865 author_id: str, 

866 tags: list[str] | None = None, 

867 access_level: str = "team", 

868 team_id: str | None = None, 

869 project_id: str | None = None, 

870) -> str: 

871 """Add reflection to team knowledge base.""" 

872 manager = get_team_knowledge_manager() 

873 level = AccessLevel(access_level.lower()) 

874 return await manager.add_team_reflection( 

875 content,