odysseus-sessions

#!/usr/bin/env python3
"""odysseus-sessions — shell wrapper for chat sessions.

    odysseus-sessions list [--archived] [--folder F] [--limit N]
    odysseus-sessions show SESSION_ID
    odysseus-sessions archive SESSION_ID
    odysseus-sessions unarchive SESSION_ID
    odysseus-sessions delete SESSION_ID         # hard-delete (irreversible)
"""

from __future__ import annotations
import sys
import os, sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "_lib"))
from cli import quiet_logs, emit, fail, common_parser, run, REPO_ROOT as _REPO_ROOT
quiet_logs()

import argparse, json, logging, os, sys
from pathlib import Path

try:
    from core.database import SessionLocal, Session as DbSession
    quiet_logs()
except ModuleNotFoundError as e:
    sys.stderr.write(f"error: {e}\nhint: run from repo root with venv active.\n")
    sys.exit(2)


def _serialize(s: "DbSession") -> dict:
    def _int_or_zero(value) -> int:
        try:
            return int(value or 0)
        except (TypeError, ValueError):
            return 0

    return {
        "id": s.id,
        "name": s.name,
        "model": s.model,
        "endpoint_url": s.endpoint_url,
        "owner": s.owner or "",
        "folder": s.folder or "",
        "archived": bool(s.archived),
        "rag": bool(s.rag),
        "is_important": bool(s.is_important),
        "message_count": _int_or_zero(s.message_count),
        "total_input_tokens": _int_or_zero(s.total_input_tokens),
        "total_output_tokens": _int_or_zero(s.total_output_tokens),
        "last_accessed": s.last_accessed.isoformat() if s.last_accessed else "",
        "created_at": s.created_at.isoformat() if s.created_at else "",
    }


def cmd_list(args):
    db = SessionLocal()
    try:
        q = db.query(DbSession)
        if not args.archived:
            q = q.filter(DbSession.archived == False)  # noqa: E712
        elif args.archived == "only":
            q = q.filter(DbSession.archived == True)  # noqa: E712
        if args.folder:
            q = q.filter(DbSession.folder == args.folder)
        q = q.order_by(DbSession.last_accessed.desc()).limit(args.limit)
        emit([_serialize(s) for s in q.all()], args)
    finally:
        db.close()


def cmd_show(args):
    db = SessionLocal()
    try:
        s = db.get(DbSession, args.id)
        if not s:
            fail(f"no session with id {args.id!r}")
        emit(_serialize(s), args)
    finally:
        db.close()


def cmd_archive(args):
    _set_archived(args, True)


def cmd_unarchive(args):
    _set_archived(args, False)


def _set_archived(args, archived: bool):
    db = SessionLocal()
    try:
        s = db.get(DbSession, args.id)
        if not s:
            fail(f"no session with id {args.id!r}")
        s.archived = archived
        db.commit()
        emit({"ok": True, "id": s.id, "archived": s.archived}, args)
    finally:
        db.close()


def cmd_delete(args):
    if not args.yes:
        fail("delete is irreversible — pass --yes to confirm")
    db = SessionLocal()
    try:
        s = db.get(DbSession, args.id)
        if not s:
            fail(f"no session with id {args.id!r}")
        snap = _serialize(s)
        db.delete(s)
        db.commit()
        emit({"ok": True, "deleted": snap}, args)
    finally:
        db.close()


def _build_parser():
    common = argparse.ArgumentParser(add_help=False)
    common.add_argument("--pretty", action="store_true")

    p = argparse.ArgumentParser(prog="odysseus-sessions", parents=[common])
    sub = p.add_subparsers(dest="cmd", required=True)

    pl = sub.add_parser("list", parents=[common])
    pl.add_argument("--archived", nargs="?", const=True, default=False,
                    help="include archived; --archived only = only archived")
    pl.add_argument("--folder")
    pl.add_argument("--limit", type=int, default=50)
    pl.set_defaults(func=cmd_list)

    psh = sub.add_parser("show", parents=[common])
    psh.add_argument("id")
    psh.set_defaults(func=cmd_show)

    pa = sub.add_parser("archive", parents=[common])
    pa.add_argument("id")
    pa.set_defaults(func=cmd_archive)

    pu = sub.add_parser("unarchive", parents=[common])
    pu.add_argument("id")
    pu.set_defaults(func=cmd_unarchive)

    pd = sub.add_parser("delete", parents=[common])
    pd.add_argument("id")
    pd.add_argument("--yes", action="store_true", help="confirm deletion")
    pd.set_defaults(func=cmd_delete)

    return p


if __name__ == "__main__":
    sys.exit(run(_build_parser()))