Create a new session for a project, closing any existing one. Args: project_name: Name of the project project_dir: Absolute path to the project directory
(project_name: str, project_dir: Path)
| 442 | |
| 443 | |
| 444 | async def create_session(project_name: str, project_dir: Path) -> SpecChatSession: |
| 445 | """Create a new session for a project, closing any existing one. |
| 446 | |
| 447 | Args: |
| 448 | project_name: Name of the project |
| 449 | project_dir: Absolute path to the project directory |
| 450 | """ |
| 451 | old_session: Optional[SpecChatSession] = None |
| 452 | |
| 453 | with _sessions_lock: |
| 454 | # Get existing session to close later (outside the lock) |
| 455 | old_session = _sessions.pop(project_name, None) |
| 456 | session = SpecChatSession(project_name, project_dir) |
| 457 | _sessions[project_name] = session |
| 458 | |
| 459 | # Close old session outside the lock to avoid blocking |
| 460 | if old_session: |
| 461 | try: |
| 462 | await old_session.close() |
| 463 | except Exception as e: |
| 464 | logger.warning(f"Error closing old session for {project_name}: {e}") |
| 465 | |
| 466 | return session |
| 467 | |
| 468 | |
| 469 | async def remove_session(project_name: str) -> None: |
no test coverage detected