59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
"""Session directory management for kfs-answer scripts.
|
|
|
|
Protocol: expects the TRACE_ID env var (unique per agent session, injected by
|
|
the calling agent framework). Without it, exits — skill cannot guarantee
|
|
session isolation.
|
|
|
|
All kfs-answer temporary files (e.g. file_refs.txt) are written under
|
|
./kfs-answer-sessions/{TRACE_ID}/, isolated per session.
|
|
|
|
Cleanup: session dirs with mtime > 24h are removed on each call.
|
|
"""
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import time
|
|
|
|
# Derive project root from script location: scripts/ → kfs-answer/ → skills/ → project root
|
|
_PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", ".."))
|
|
SESSIONS_DIR = os.path.join(_PROJECT_ROOT, "kfs-answer-sessions")
|
|
RETENTION_SEC = 86400 # 24h
|
|
|
|
|
|
def get_session_dir():
|
|
"""Return the session dir for this agent session. Create if missing. Cleanup old."""
|
|
trace_id = os.environ.get("TRACE_ID", "").strip()
|
|
if not trace_id:
|
|
sys.stderr.write(
|
|
"[ERROR: TRACE_ID env var not set. kfs-answer skill requires the calling "
|
|
"agent framework to inject TRACE_ID (unique per agent session) into the "
|
|
"subprocess env. This is a framework integration issue.\n"
|
|
" - catalog-agent: should be injected automatically by HTTP middleware\n"
|
|
" - gbase-agent-service: main.py must set env['TRACE_ID'] before subprocess\n"
|
|
" - Manual testing: run with `TRACE_ID=$(uuidgen) python3 ...`]\n"
|
|
)
|
|
sys.exit(1)
|
|
|
|
session_path = os.path.join(SESSIONS_DIR, trace_id)
|
|
os.makedirs(session_path, exist_ok=True)
|
|
os.utime(session_path, None) # refresh mtime = active marker
|
|
_cleanup_old()
|
|
return session_path
|
|
|
|
|
|
def _cleanup_old():
|
|
if not os.path.isdir(SESSIONS_DIR):
|
|
return
|
|
now = time.time()
|
|
try:
|
|
entries = os.listdir(SESSIONS_DIR)
|
|
except OSError:
|
|
return
|
|
for name in entries:
|
|
full = os.path.join(SESSIONS_DIR, name)
|
|
try:
|
|
if os.path.isdir(full) and now - os.path.getmtime(full) > RETENTION_SEC:
|
|
shutil.rmtree(full, ignore_errors=True)
|
|
except OSError:
|
|
continue
|