scripts-session-orient

#!/usr/bin/env python3 """

title: "Configuration" component_type: script version: "1.0.0" audience: contributor status: stable summary: "Session Orientation Script - Anti-Forgetting Quick Start" keywords: ['git', 'orient', 'session'] tokens: ~500 created: 2025-12-22 updated: 2025-12-22 script_name: "session-orient.py" language: python executable: true usage: "python3 scripts/session-orient.py [options]" python_version: "3.10+" dependencies: [] modifies_files: false network_access: false requires_auth: false

Session Orientation Script - Anti-Forgetting Quick Start

Automatically orient yourself to recent CODITECT activity at the start of any session. Combines git history, memory system queries, and project status into one comprehensive view.

Usage: python3 scripts/session-orient.py # Full orientation python3 scripts/session-orient.py --quick # Git + stats only python3 scripts/session-orient.py --deep "topic" # Include semantic recall python3 scripts/session-orient.py --json # JSON output """

import subprocess import json import sys import os import urllib.request import urllib.error from pathlib import Path from datetime import datetime, timedelta

Configuration

SCRIPT_DIR = Path(file).parent REPO_ROOT = SCRIPT_DIR.parent CONTEXT_DB_SCRIPT = SCRIPT_DIR / "context-db.py" LICENSE_VALIDATOR = SCRIPT_DIR / "core" / "license_validator.py" VERSION_FILE = REPO_ROOT / "VERSION" VERSION_JSON_URL = "https://storage.googleapis.com/coditect-dist/version.json" VERSION_CACHE_TTL_HOURS = 24

ADR-114 & ADR-118: Use centralized path discovery for user data

sys.path.insert(0, str(SCRIPT_DIR / "core")) try: from paths import get_context_storage_dir CONTEXT_STORAGE = get_context_storage_dir() except ImportError: # Fallback for backward compatibility _user_data = Path.home() / "PROJECTS" / ".coditect-data" / "context-storage" if _user_data.exists(): CONTEXT_STORAGE = _user_data else: CONTEXT_STORAGE = REPO_ROOT / "context-storage"

VERSION_CACHE_FILE = CONTEXT_STORAGE / "version-check.json"

def run_command(cmd, cwd=None, timeout=30): """Run a shell command and return output.""" try: result = subprocess.run( cmd, shell=True, capture_output=True, text=True, cwd=cwd or REPO_ROOT, timeout=timeout ) return result.stdout.strip() except subprocess.TimeoutExpired: return "[timeout]" except Exception as e: return f"[error: {e}]"

def run_context_db(args): """Run context-db.py with given arguments.""" if not CONTEXT_DB_SCRIPT.exists(): return None cmd = f"python3 {CONTEXT_DB_SCRIPT} {args}" return run_command(cmd)

def check_for_updates(): """ Check for available CODITECT updates (C.11.4).

Uses cached version info with 24-hour TTL to avoid
hitting GCS on every /orient invocation.

Returns dict with:
    - current: str - current installed version
    - latest: str - latest available version
    - update_available: bool - whether update is available
    - banner: str - formatted update notification (or None)
    - message: str - changelog/update message
"""
result = {
    "current": None,
    "latest": None,
    "update_available": False,
    "banner": None,
    "message": None,
    "error": None
}

# Read local version
if VERSION_FILE.exists():
    try:
        result["current"] = VERSION_FILE.read_text().strip()
    except Exception as e:
        result["error"] = f"Failed to read VERSION: {e}"
        return result
else:
    result["error"] = "VERSION file not found"
    return result

# Check cache first (C.11.4.2 - rate limit)
cache_valid = False
if VERSION_CACHE_FILE.exists():
    try:
        cache = json.loads(VERSION_CACHE_FILE.read_text())
        checked_at = datetime.fromisoformat(cache.get("checked_at", "2000-01-01"))
        cache_age = datetime.now() - checked_at

        if cache_age < timedelta(hours=VERSION_CACHE_TTL_HOURS):
            # Cache is valid, use cached data
            cache_valid = True
            result["latest"] = cache.get("latest", result["current"])
            result["update_available"] = cache.get("update_available", False)
            result["message"] = cache.get("update_message")
    except Exception:
        pass  # Cache invalid, fetch fresh

# Fetch from GCS if cache invalid/missing
if not cache_valid:
    try:
        req = urllib.request.Request(
            VERSION_JSON_URL,
            headers={"User-Agent": "CODITECT-Orient/1.0"}
        )
        with urllib.request.urlopen(req, timeout=5) as response:
            remote_data = json.loads(response.read().decode())

        result["latest"] = remote_data.get("latest", result["current"])
        result["message"] = remote_data.get("update_message")
        result["update_available"] = _version_compare(
            result["current"], result["latest"]
        ) < 0

        # Update cache
        cache_data = {
            "checked_at": datetime.now().isoformat(),
            "current": result["current"],
            "latest": result["latest"],
            "update_available": result["update_available"],
            "update_message": result["message"],
            "changelog_url": remote_data.get("changelog_url"),
            "min_supported": remote_data.get("min_supported", "1.0.0")
        }
        VERSION_CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
        VERSION_CACHE_FILE.write_text(json.dumps(cache_data, indent=2))

    except urllib.error.URLError:
        # Offline - use cached or assume current
        result["latest"] = result["current"]
        result["update_available"] = False
    except Exception as e:
        result["error"] = f"Version check failed: {e}"
        result["latest"] = result["current"]

# Generate banner if update available
if result["update_available"]:
    result["banner"] = (
        f"⬆️  UPDATE AVAILABLE: v{result['current']} → v{result['latest']}\n"
        f"   Run '/update' to get the latest features and fixes"
    )

return result

def _version_compare(v1: str, v2: str) -> int: """ Compare two semantic versions.

Returns:
    -1 if v1 < v2
     0 if v1 == v2
     1 if v1 > v2
"""
def parse_version(v):
    # Handle versions like "1.0.0", "1.0.0-beta.1"
    parts = v.replace("-", ".").split(".")
    return [int(p) if p.isdigit() else 0 for p in parts[:3]]

p1, p2 = parse_version(v1), parse_version(v2)
for a, b in zip(p1, p2):
    if a < b:
        return -1
    if a > b:
        return 1
return 0

def validate_license(): """ Validate CODITECT license per ADR-067.

Returns dict with:
    - valid: bool - whether license allows operation
    - state: str - ACTIVE/WARNING/GRACE/EXPIRED/MISSING/REVOKED
    - message: str - human-readable status message
    - days_remaining: int - days until expiration (negative if expired)
    - banner: str - formatted banner for display
"""
result = {
    "valid": True,
    "state": "UNKNOWN",
    "message": "",
    "days_remaining": None,
    "banner": None
}

# Check if license validator exists
if not LICENSE_VALIDATOR.exists():
    # Graceful degradation - allow operation if validator not installed
    result["state"] = "UNCONFIGURED"
    result["message"] = "License validator not installed"
    return result

try:
    # Run license validation
    proc = subprocess.run(
        ["python3", str(LICENSE_VALIDATOR), "--json"],
        capture_output=True,
        text=True,
        timeout=10
    )

    if proc.returncode == 0 and proc.stdout.strip():
        data = json.loads(proc.stdout)
        result["state"] = data.get("state", "UNKNOWN")
        result["message"] = data.get("message", "")
        result["days_remaining"] = data.get("days_remaining")
        result["valid"] = result["state"].lower() in ["active", "warning", "grace"]

        # Generate banner based on state (values are lowercase from validator)
        state = result["state"].lower()
        if state == "active":
            days = result["days_remaining"] or 0
            if days > 30:
                result["banner"] = f"🔑 License: Active ({days} days remaining)"
            else:
                result["banner"] = f"🔑 License: Active ({days} days remaining)"
        elif state == "warning":
            days = result["days_remaining"] or 0
            result["banner"] = f"⚠️  LICENSE EXPIRING SOON: {days} days remaining"
        elif state == "grace":
            result["banner"] = f"⚠️  LICENSE EXPIRED - Grace period active"
        elif state == "expired":
            result["valid"] = False
            result["banner"] = "❌ LICENSE EXPIRED - Please renew at https://coditect.ai/renew"
        elif state == "revoked":
            result["valid"] = False
            result["banner"] = "❌ LICENSE REVOKED - Contact support@coditect.ai"
        elif state == "missing":
            result["valid"] = False
            result["banner"] = "❌ NO LICENSE - Activate at https://coditect.ai/activate"
        elif state == "offline_exceeded":
            result["valid"] = False
            result["banner"] = "❌ OFFLINE TOO LONG - Connect to validate license"

    else:
        # Validation failed - check if license file exists
        license_file = REPO_ROOT / "licensing" / "license.json"
        if not license_file.exists():
            result["state"] = "MISSING"
            result["valid"] = False
            result["message"] = "No license file found"
            result["banner"] = "🔒 No license configured (pilot mode)"
        else:
            result["state"] = "ERROR"
            result["message"] = f"Validation error: {proc.stderr[:100]}"

except subprocess.TimeoutExpired:
    result["state"] = "TIMEOUT"
    result["message"] = "License validation timed out"
except json.JSONDecodeError as e:
    result["state"] = "PARSE_ERROR"
    result["message"] = f"Invalid response: {e}"
except Exception as e:
    result["state"] = "ERROR"
    result["message"] = str(e)

return result

def get_git_history(count=10): """Get recent git commits.""" return run_command(f"git log --oneline -{count}")

def get_git_today(): """Get today's commits.""" return run_command('git log --oneline --since="1 day ago"')

def get_git_status(): """Get current git status.""" return run_command("git status --short | head -20")

def get_component_counts(): """Get component counts from config.""" counts_file = REPO_ROOT / "config" / "component-counts.json" if counts_file.exists(): try: with open(counts_file) as f: return json.load(f) except: pass return None

def get_knowledge_stats(): """Get knowledge base statistics.""" output = run_context_db("--knowledge-stats") if output and "Knowledge Base Statistics" in output: return output return None

def get_recent_decisions(limit=10): """Get recent decisions.""" output = run_context_db(f"--decisions --limit {limit}") return output

def get_recent_messages(limit=20): """Get recent messages.""" output = run_context_db(f"--recent {limit}") return output

def get_recall_context(topic): """Get RAG recall for a topic.""" output = run_context_db(f'--recall "{topic}"') return output

def get_task_progress(): """Get task progress from TASKLIST.md.""" tasklist_paths = [ REPO_ROOT / "docs" / "project-management" / "TASKLIST.md", REPO_ROOT / "TASKLIST.md", REPO_ROOT.parent.parent.parent / "docs" / "project-management" / "TASKLIST.md" ]

for path in tasklist_paths:
    if path.exists():
        try:
            with open(path) as f:
                content = f.read()
            completed = content.count("[x]") + content.count("[X]")
            pending = content.count("[ ]")
            total = completed + pending
            if total > 0:
                percent = (completed * 100) // total
                return {"completed": completed, "pending": pending, "total": total, "percent": percent}
        except:
            pass
return None

def format_section(title, content, emoji="📋"): """Format a section for display.""" if not content: return "" return f"\n{emoji} {title}:\n{content}\n"

def orient_quick(as_json=False): """Quick orientation - git + stats only.""" # License validation (ADR-067) license_info = validate_license() # Version check (C.11.4) update_info = check_for_updates()

data = {
    "timestamp": datetime.now().isoformat(),
    "mode": "quick",
    "license": license_info,
    "version": update_info,
    "git_history": get_git_history(10),
    "git_status": get_git_status(),
    "components": get_component_counts(),
}

if as_json:
    print(json.dumps(data, indent=2))
    return

print("🧠 CODITECT Session Orientation (Quick)")
print("━" * 50)

# Show update banner if available (C.11.4.1)
if update_info.get("banner"):
    print(f"\n{update_info['banner']}\n")

# Show license status banner if not active
state_lower = license_info.get("state", "").lower()
if license_info.get("banner") and state_lower not in ["active", "unconfigured"]:
    print(f"\n{license_info['banner']}\n")

if data["git_history"]:
    print(format_section("Recent Git Activity (Last 10)", data["git_history"], "📜"))

if data["git_status"]:
    print(format_section("Uncommitted Changes", data["git_status"], "📝"))

if data["components"]:
    counts = data["components"]
    print(f"\n📦 Component Counts:")
    print(f"   Agents: {counts.get('agents', 'N/A')}")
    print(f"   Commands: {counts.get('commands', 'N/A')}")
    print(f"   Skills: {counts.get('skills', 'N/A')}")
    print(f"   Scripts: {counts.get('scripts', 'N/A')}")

print("\n" + "━" * 50)
print("💡 Run '/orient' for full orientation with memory system")

def orient_full(as_json=False): """Full orientation - git + memory + tasks.""" # License validation (ADR-067) license_info = validate_license() # Version check (C.11.4) update_info = check_for_updates()

data = {
    "timestamp": datetime.now().isoformat(),
    "mode": "full",
    "license": license_info,
    "version": update_info,
    "git_history": get_git_history(10),
    "git_today": get_git_today(),
    "git_status": get_git_status(),
    "components": get_component_counts(),
    "knowledge_stats": get_knowledge_stats(),
    "recent_decisions": get_recent_decisions(10),
    "recent_messages": get_recent_messages(20),
    "task_progress": get_task_progress(),
}

if as_json:
    print(json.dumps(data, indent=2))
    return

print("🧠 CODITECT Session Orientation")
print("━" * 50)

# Show update banner if available (C.11.4.1)
if update_info.get("banner"):
    print(f"\n{update_info['banner']}\n")

# Show license status banner (C.12.4.1)
if license_info.get("banner"):
    print(f"\n{license_info['banner']}")
    state_lower = license_info.get("state", "").lower()
    if state_lower in ["warning", "grace"]:
        print("   Run '/license-status' for details")
    print()

# Git section
if data["git_history"]:
    print(format_section("Recent Git Activity (Last 10)", data["git_history"], "📜"))

if data["git_today"]:
    print(format_section("Today's Changes", data["git_today"], "📅"))

if data["git_status"]:
    print(format_section("Uncommitted Changes", data["git_status"], "📝"))

# Knowledge base section
if data["knowledge_stats"]:
    # Parse and format knowledge stats
    stats = data["knowledge_stats"]
    print(f"\n📊 Knowledge Base Stats:")
    for line in stats.split("\n"):
        if line.strip() and not line.startswith("="):
            print(f"   {line.strip()}")
else:
    print("\n📊 Knowledge Base: Not initialized")
    print("   Run '/cx' then '/cxq --index' to build")

# Decisions section
if data["recent_decisions"]:
    print(format_section("Recent Decisions (Last 10)", data["recent_decisions"][:500], "🎯"))

# Messages section
if data["recent_messages"]:
    # Truncate for display
    messages = data["recent_messages"][:800]
    if len(data["recent_messages"]) > 800:
        messages += "\n   ... (truncated)"
    print(format_section("Recent Activity (Last 20 messages)", messages, "💬"))

# Task progress
if data["task_progress"]:
    tp = data["task_progress"]
    print(f"\n✅ Task Progress: {tp['completed']}/{tp['total']} ({tp['percent']}%)")
    print(f"   Pending: {tp['pending']} tasks")

# Components
if data["components"]:
    counts = data["components"]
    print(f"\n📦 Components: {counts.get('agents', 0)} agents, {counts.get('commands', 0)} commands, {counts.get('skills', 0)} skills")

print("\n" + "━" * 50)
print("✅ Ready to continue work!")
print("💡 End session with '/cx' to preserve context")

def orient_deep(topic, as_json=False): """Deep orientation - full + semantic recall for topic.""" # License validation (ADR-067) license_info = validate_license() # Version check (C.11.4) update_info = check_for_updates()

data = {
    "timestamp": datetime.now().isoformat(),
    "mode": "deep",
    "topic": topic,
    "license": license_info,
    "version": update_info,
    "git_history": get_git_history(10),
    "git_status": get_git_status(),
    "components": get_component_counts(),
    "knowledge_stats": get_knowledge_stats(),
    "recent_decisions": get_recent_decisions(5),
    "recall_context": get_recall_context(topic),
    "task_progress": get_task_progress(),
}

if as_json:
    print(json.dumps(data, indent=2))
    return

print(f"🧠 CODITECT Deep Orientation: '{topic}'")
print("━" * 50)

# Show update banner if available (C.11.4.1)
if update_info.get("banner"):
    print(f"\n{update_info['banner']}\n")

# Git section
if data["git_history"]:
    print(format_section("Recent Git Activity", data["git_history"][:400], "📜"))

# Recall section (the main focus for deep mode)
if data["recall_context"]:
    print(format_section(f"Recalled Context for '{topic}'", data["recall_context"][:1500], "🔍"))
else:
    print(f"\n🔍 No recalled context for '{topic}'")
    print("   Run '/cxq --extract' and '/cxq --embeddings' for semantic search")

# Knowledge stats
if data["knowledge_stats"]:
    # Just show summary line
    for line in data["knowledge_stats"].split("\n"):
        if "Messages:" in line or "Decisions:" in line:
            print(f"   {line.strip()}")

# Recent decisions
if data["recent_decisions"]:
    print(format_section("Recent Decisions", data["recent_decisions"][:300], "🎯"))

print("\n" + "━" * 50)
print(f"✅ Context loaded for '{topic}'")
print("💡 Use '/cxq --recall \"topic\"' for more specific queries")

def main(): import argparse

parser = argparse.ArgumentParser(
    description="CODITECT Session Orientation - Anti-Forgetting Quick Start"
)
parser.add_argument("--quick", action="store_true", help="Quick orientation (git + stats only)")
parser.add_argument("--deep", metavar="TOPIC", help="Deep orientation with semantic recall for topic")
parser.add_argument("--json", action="store_true", help="Output as JSON")

args = parser.parse_args()

if args.deep:
    orient_deep(args.deep, as_json=args.json)
elif args.quick:
    orient_quick(as_json=args.json)
else:
    orient_full(as_json=args.json)

if name == "main": main()