#!/usr/bin/env python3 """ Session Log Sync Script

Synchronizes machine-specific session logs between local storage and cloud (GCS). Supports conflict-free replication using machine UUID as partition key.

Usage: python3 session-log-sync.py --push # Upload local logs to cloud python3 session-log-sync.py --index # View available logs (all machines) python3 session-log-sync.py --pull # Fetch logs from another machine python3 session-log-sync.py --backup # Create local backup

ADR Reference: ADR-058-machine-specific-session-logs.md

Author: CODITECT Team Version: 1.0.0 Created: 2026-01-06 """

import argparse import json import os import shutil import subprocess import sys from datetime import datetime, timezone from pathlib import Path from typing import Optional, List, Dict

=============================================================================

CONFIGURATION

=============================================================================

HOME = Path.home() CODITECT_DIR = HOME / ".coditect" SESSION_LOGS_DIR = CODITECT_DIR / "session-logs" SESSION_CACHE_DIR = CODITECT_DIR / "session-logs-cache" MACHINE_ID_PATH = CODITECT_DIR / "machine-id.json" BACKUP_DIR = HOME / "PROJECTS" / "BU" / "session-logs"

GCS bucket for cloud sync (configured per-tenant)

GCS_BUCKET = os.environ.get("CODITECT_SESSION_BUCKET", "gs://coditect-session-logs")

Colors for terminal output

Shared Colors module (consolidates 36 duplicate definitions)

_script_dir = Path(file).parent sys.path.insert(0, str(_script_dir / "core")) from colors import Colors

def print_header(text: str) -> None: print(f"\n{Colors.HEADER}{Colors.BOLD}{'='*60}{Colors.ENDC}") print(f"{Colors.HEADER}{Colors.BOLD}{text.center(60)}{Colors.ENDC}") print(f"{Colors.HEADER}{Colors.BOLD}{'='*60}{Colors.ENDC}\n")

def print_success(text: str) -> None: print(f"{Colors.GREEN}✓ {text}{Colors.ENDC}")

def print_warning(text: str) -> None: print(f"{Colors.WARNING}⚠ {text}{Colors.ENDC}")

def print_error(text: str) -> None: print(f"{Colors.FAIL}✗ {text}{Colors.ENDC}")

def print_info(text: str) -> None: print(f"{Colors.BLUE}ℹ {text}{Colors.ENDC}")

=============================================================================

MACHINE ID MANAGEMENT

=============================================================================

def load_machine_id() -> Optional[Dict]: """Load machine ID from file.""" if not MACHINE_ID_PATH.exists(): print_error("Machine ID not found. Run CODITECT-CORE-INITIAL-SETUP.py first.") return None

try:
    with open(MACHINE_ID_PATH) as f:
        return json.load(f)
except Exception as e:
    print_error(f"Failed to load machine ID: {e}")
    return None

=============================================================================

LOCAL OPERATIONS

=============================================================================

def list_local_logs() -> List[Path]: """List all local session logs.""" if not SESSION_LOGS_DIR.exists(): return [] return sorted(SESSION_LOGS_DIR.glob("SESSION-LOG-*.md"))

def get_log_metadata(log_path: Path) -> Dict: """Extract metadata from a session log file.""" try: content = log_path.read_text() metadata = { "filename": log_path.name, "size": log_path.stat().st_size, "modified": datetime.fromtimestamp(log_path.stat().st_mtime).isoformat(), }

    # Extract frontmatter if present
    if content.startswith("---"):
        end_idx = content.find("---", 3)
        if end_idx > 0:
            frontmatter = content[3:end_idx]
            for line in frontmatter.split("\n"):
                if ":" in line:
                    key, value = line.split(":", 1)
                    metadata[key.strip()] = value.strip().strip('"')

    return metadata
except Exception as e:
    return {"filename": log_path.name, "error": str(e)}

def create_local_backup() -> bool: """Create a timestamped local backup of session logs.""" print_header("CREATING LOCAL BACKUP")

machine_id = load_machine_id()
if not machine_id:
    return False

logs = list_local_logs()
if not logs:
    print_info("No session logs to backup")
    return True

# Create timestamped backup directory
timestamp = datetime.now().strftime("%Y-%m-%d-%H%M%S")
backup_path = BACKUP_DIR / timestamp / machine_id["machine_uuid"][:8]
backup_path.mkdir(parents=True, exist_ok=True)

count = 0
for log_path in logs:
    try:
        shutil.copy2(log_path, backup_path / log_path.name)
        count += 1
    except Exception as e:
        print_warning(f"Failed to backup {log_path.name}: {e}")

# Write manifest
manifest = {
    "timestamp": timestamp,
    "machine_uuid": machine_id["machine_uuid"],
    "hostname": machine_id["hostname"],
    "log_count": count,
    "logs": [log.name for log in logs],
    "created_at": datetime.now(timezone.utc).isoformat()
}
with open(backup_path / "BACKUP-MANIFEST.json", "w") as f:
    json.dump(manifest, f, indent=2)

# Update "latest" symlink
latest_link = BACKUP_DIR / "latest"
if latest_link.is_symlink():
    latest_link.unlink()
latest_link.symlink_to(timestamp)

print_success(f"Backed up {count} session log(s) to {backup_path}")
return True

=============================================================================

CLOUD OPERATIONS

=============================================================================

def check_gcloud_auth() -> bool: """Check if gcloud is authenticated.""" try: result = subprocess.run( ["gcloud", "auth", "list", "--filter=status:ACTIVE", "--format=value(account)"], capture_output=True, text=True, check=False ) if result.returncode == 0 and result.stdout.strip(): return True print_warning("Not authenticated with gcloud. Run: gcloud auth login") return False except FileNotFoundError: print_warning("gcloud CLI not found. Install: brew install google-cloud-sdk") return False

def push_to_cloud(dry_run: bool = False) -> bool: """Push local session logs to cloud storage.""" print_header("PUSHING TO CLOUD")

machine_id = load_machine_id()
if not machine_id:
    return False

if not check_gcloud_auth():
    print_info("Falling back to local-only mode")
    return create_local_backup()

logs = list_local_logs()
if not logs:
    print_info("No session logs to push")
    return True

# Build cloud path
cloud_base = f"{GCS_BUCKET}/machines/{machine_id['machine_uuid']}"

if dry_run:
    print_info(f"Would push {len(logs)} logs to {cloud_base}")
    for log in logs:
        print(f"  {log.name}")
    return True

count = 0
for log_path in logs:
    try:
        cloud_path = f"{cloud_base}/{log_path.name}"
        result = subprocess.run(
            ["gsutil", "-q", "cp", str(log_path), cloud_path],
            capture_output=True, text=True, check=False
        )
        if result.returncode == 0:
            count += 1
        else:
            print_warning(f"Failed to upload {log_path.name}: {result.stderr}")
    except Exception as e:
        print_warning(f"Error uploading {log_path.name}: {e}")

# Upload machine ID
try:
    subprocess.run(
        ["gsutil", "-q", "cp", str(MACHINE_ID_PATH), f"{cloud_base}/machine-id.json"],
        capture_output=True, text=True, check=False
    )
except Exception:
    pass

# Update index
update_cloud_index(machine_id, logs)

print_success(f"Pushed {count}/{len(logs)} session log(s) to cloud")
return True

def update_cloud_index(machine_id: Dict, logs: List[Path]) -> None: """Update the cloud index with this machine's logs.""" index_path = SESSION_CACHE_DIR / "index.json"

# Load existing index
if index_path.exists():
    try:
        with open(index_path) as f:
            index = json.load(f)
    except Exception:
        index = {"machines": {}, "updated_at": None}
else:
    index = {"machines": {}, "updated_at": None}

# Update this machine's entry
index["machines"][machine_id["machine_uuid"]] = {
    "hostname": machine_id["hostname"],
    "log_count": len(logs),
    "logs": [log.name for log in logs],
    "last_sync": datetime.now(timezone.utc).isoformat()
}
index["updated_at"] = datetime.now(timezone.utc).isoformat()

# Save index
SESSION_CACHE_DIR.mkdir(parents=True, exist_ok=True)
with open(index_path, "w") as f:
    json.dump(index, f, indent=2)

def pull_index() -> bool: """Pull the session log index from cloud.""" print_header("FETCHING INDEX")

if not check_gcloud_auth():
    # Show local index if available
    index_path = SESSION_CACHE_DIR / "index.json"
    if index_path.exists():
        print_info("Showing cached index (offline mode)")
        show_index()
        return True
    print_error("No cached index available")
    return False

# Download index from cloud
cloud_index = f"{GCS_BUCKET}/index/session-index.json"
local_index = SESSION_CACHE_DIR / "index.json"

try:
    SESSION_CACHE_DIR.mkdir(parents=True, exist_ok=True)
    result = subprocess.run(
        ["gsutil", "-q", "cp", cloud_index, str(local_index)],
        capture_output=True, text=True, check=False
    )
    if result.returncode != 0:
        print_warning("Cloud index not found, showing local only")
except Exception as e:
    print_warning(f"Failed to fetch index: {e}")

show_index()
return True

def show_index() -> None: """Display the session log index.""" index_path = SESSION_CACHE_DIR / "index.json"

if not index_path.exists():
    print_info("No index available")
    return

try:
    with open(index_path) as f:
        index = json.load(f)

    machine_id = load_machine_id()
    current_uuid = machine_id["machine_uuid"] if machine_id else None

    print(f"\n{Colors.CYAN}Available Session Logs:{Colors.ENDC}\n")

    for uuid, info in index.get("machines", {}).items():
        is_current = uuid == current_uuid
        marker = f" {Colors.GREEN}(this machine){Colors.ENDC}" if is_current else ""
        print(f"  {Colors.BOLD}{uuid[:8]}{Colors.ENDC}{marker}")
        print(f"    Hostname: {info.get('hostname', 'unknown')}")
        print(f"    Logs: {info.get('log_count', 0)}")
        print(f"    Last sync: {info.get('last_sync', 'unknown')}")
        print()

    if index.get("updated_at"):
        print(f"{Colors.BLUE}Index updated: {index['updated_at']}{Colors.ENDC}")

except Exception as e:
    print_error(f"Failed to read index: {e}")

def pull_logs(target_uuid: str, target_date: Optional[str] = None) -> bool: """Pull logs from another machine.""" print_header(f"PULLING LOGS FROM {target_uuid[:8]}...")

if not check_gcloud_auth():
    print_error("Cloud authentication required to pull logs")
    return False

# Create cache directory for target machine
cache_dir = SESSION_CACHE_DIR / target_uuid[:8]
cache_dir.mkdir(parents=True, exist_ok=True)

if target_date:
    # Pull specific log
    cloud_path = f"{GCS_BUCKET}/machines/{target_uuid}/SESSION-LOG-{target_date}.md"
    local_path = cache_dir / f"SESSION-LOG-{target_date}.md"

    result = subprocess.run(
        ["gsutil", "-q", "cp", cloud_path, str(local_path)],
        capture_output=True, text=True, check=False
    )
    if result.returncode == 0:
        print_success(f"Downloaded to {local_path}")
        return True
    else:
        print_error(f"Log not found: {result.stderr}")
        return False
else:
    # Pull all logs for machine
    cloud_path = f"{GCS_BUCKET}/machines/{target_uuid}/"

    result = subprocess.run(
        ["gsutil", "-m", "-q", "cp", "-r", cloud_path, str(cache_dir)],
        capture_output=True, text=True, check=False
    )
    if result.returncode == 0:
        count = len(list(cache_dir.glob("SESSION-LOG-*.md")))
        print_success(f"Downloaded {count} log(s) to {cache_dir}")
        return True
    else:
        print_error(f"Failed to pull logs: {result.stderr}")
        return False

=============================================================================

MAIN

=============================================================================

def main(): parser = argparse.ArgumentParser( description="Synchronize machine-specific session logs" )

# Commands
parser.add_argument("--push", action="store_true",
                    help="Push local session logs to cloud")
parser.add_argument("--index", action="store_true",
                    help="View available session logs from all machines")
parser.add_argument("--pull", metavar="UUID",
                    help="Pull session logs from another machine")
parser.add_argument("--backup", action="store_true",
                    help="Create local backup of session logs")
parser.add_argument("--list", action="store_true",
                    help="List local session logs")

# Options
parser.add_argument("--date", metavar="YYYY-MM-DD",
                    help="Specific date for pull operation")
parser.add_argument("--dry-run", action="store_true",
                    help="Show what would be done without doing it")
parser.add_argument("--verbose", "-v", action="store_true",
                    help="Verbose output")

args = parser.parse_args()

# Default to list if no command specified
if not any([args.push, args.index, args.pull, args.backup, args.list]):
    args.list = True

# Execute commands
if args.list:
    print_header("LOCAL SESSION LOGS")
    machine_id = load_machine_id()
    if machine_id:
        print_info(f"Machine: {machine_id['hostname']} ({machine_id['machine_uuid'][:8]}...)")

    logs = list_local_logs()
    if logs:
        print(f"\nFound {len(logs)} session log(s):\n")
        for log in logs:
            meta = get_log_metadata(log)
            print(f"  {log.name}")
            print(f"    Size: {meta.get('size', 0)} bytes")
            print(f"    Modified: {meta.get('modified', 'unknown')}")
            print()
    else:
        print_info("No session logs found")
    return

if args.backup:
    success = create_local_backup()
    sys.exit(0 if success else 1)

if args.push:
    success = push_to_cloud(dry_run=args.dry_run)
    sys.exit(0 if success else 1)

if args.index:
    success = pull_index()
    sys.exit(0 if success else 1)

if args.pull:
    success = pull_logs(args.pull, args.date)
    sys.exit(0 if success else 1)

if name == "main": main()