Implementing Deleted Session Management — A Session Recovery System
When using Claude Control, it's easy to accidentally delete a session. Closing the wrong session mid-task means losing all the context built up to that point. To address this, I implemented a soft-delete and session recovery system.
The Soft-Delete Pattern
Instead of deleting a session immediately, the system marks it as deleted and retains it for a set period.
from datetime import datetime, timedelta
from typing import Optional, List
class SoftDeleteSessionManager:
RETENTION_DAYS = 7 # 삭제된 세션 보관 기간
def __init__(self, redis_store):
self.store = redis_store
async def delete_session(self, session_id: str) -> bool:
session = await self.store.get_session(session_id)
if not session:
return False
# 활성 목록에서 제거
await self.store.redis.srem("claude:sessions:active", session_id)
# 삭제된 세션 목록에 추가
await self.store.redis.sadd("claude:sessions:deleted", session_id)
# 삭제 메타데이터 저장
await self.store.redis.hset(
f"claude:session:{session_id}",
mapping={
"status": "deleted",
"deleted_at": datetime.now().isoformat(),
"delete_reason": "user_action",
"expires_at": (
datetime.now() + timedelta(days=self.RETENTION_DAYS)
).isoformat(),
},
)
# TTL 설정 (7일 후 자동 삭제)
ttl = self.RETENTION_DAYS * 86400
await self.store.redis.expire(
f"claude:session:{session_id}", ttl
)
await self.store.redis.expire(
f"claude:session:{session_id}:logs", ttl
)
return True
Session Recovery
Recovering a deleted session moves it back to the active list and removes its TTL.
async def recover_session(self, session_id: str) -> dict:
# 삭제된 세션인지 확인
is_deleted = await self.store.redis.sismember(
"claude:sessions:deleted", session_id
)
if not is_deleted:
raise ValueError("삭제된 세션이 아닙니다")
session = await self.store.get_session(session_id)
if not session:
raise ValueError("세션 데이터가 만료되었습니다")
# 삭제 목록에서 제거, 활성 목록에 추가
await self.store.redis.srem("claude:sessions:deleted", session_id)
await self.store.redis.sadd("claude:sessions:active", session_id)
# 상태 업데이트 및 TTL 제거
await self.store.redis.hset(
f"claude:session:{session_id}",
mapping={
"status": "recovered",
"recovered_at": datetime.now().isoformat(),
},
)
await self.store.redis.persist(f"claude:session:{session_id}")
await self.store.redis.persist(f"claude:session:{session_id}:logs")
return {"session_id": session_id, "status": "recovered"}
API for Listing Deleted Sessions
@app.get("/api/sessions/deleted")
async def list_deleted_sessions():
deleted_ids = await store.redis.smembers("claude:sessions:deleted")
sessions = []
for sid in deleted_ids:
data = await store.get_session(sid)
if data:
data["session_id"] = sid
ttl = await store.redis.ttl(f"claude:session:{sid}")
data["remaining_hours"] = round(ttl / 3600, 1) if ttl > 0 else 0
sessions.append(data)
sessions.sort(key=lambda x: x.get("deleted_at", ""), reverse=True)
return {"data": sessions}
@app.post("/api/sessions/{session_id}/recover")
async def recover_session(session_id: str):
result = await soft_delete_mgr.recover_session(session_id)
return {"data": result, "message": "세션이 복구되었습니다"}
Cleaning Up Expired Sessions
I also added a background task to fully purge sessions whose retention period has elapsed.
async def cleanup_expired_sessions():
while True:
deleted_ids = await store.redis.smembers("claude:sessions:deleted")
for sid in deleted_ids:
ttl = await store.redis.ttl(f"claude:session:{sid}")
if ttl <= 0:
# TTL 만료: 완전 삭제
await store.redis.srem("claude:sessions:deleted", sid)
await store.redis.delete(f"claude:session:{sid}")
await store.redis.delete(f"claude:session:{sid}:logs")
await store.redis.delete(f"claude:session:{sid}:metrics")
print(f"세션 완전 삭제: {sid}")
await asyncio.sleep(3600) # 1시간마다 실행
Closing Thoughts
The soft-delete pattern is simple, but it provides a meaningful safety net against mistakes. Of the 88 commits in the project, this feature — added near the very end — turned out to be one of the most appreciated in day-to-day use. Along with the final commit, "Implement Autonomous Graph for difficulty-based task execution," it stands as one of the finishing touches on the Claude Control project.