singularity_llm/cli.py

"""CLI interface for Singularity LLM.

Commands:
  python -m singularity_llm chat          — interactive chat REPL
  python -m singularity_llm jarvis        — start built-in voice assistant
  python -m singularity_llm train --data corpus.txt --epochs 10
  python -m singularity_llm serve --port 8548
  python -m singularity_llm stats         — show model stats
  python -m singularity_llm export --format gguf
"""

from __future__ import annotations

import argparse
import json
import logging
import sys
from typing import Any

logger = logging.getLogger(__name__)


def cmd_chat(args: argparse.Namespace) -> None:
    """Interactive chat REPL."""
    from .harness.harness import SingularityHarness
    harness = SingularityHarness()

    # Subscription check — $1/month, free trial, or founder unlock
    access = harness.subscription.check_access()
    if not access["has_access"]:
        print(f"\n⚠ {access['message']}")
        print(f"   Run 'singularity-llm subscribe' to subscribe (${harness.subscription.MONTHLY_PRICE:.2f}/month)")
        print(f"   or 'singularity-llm trial' for a free trial")
        print(f"   or enter founder password for free access\n")
        try:
            password = input("Password (or press Enter to skip): ").strip()
            if password:
                result = harness.subscription.founder_unlock(password)
                if result["success"]:
                    print(f"\n✓ {result['message']}\n")
                    access = {"has_access": True, "status": "founder", "days_remaining": -1}
                else:
                    print(f"\n✗ {result['error']}\n")
                    return
            else:
                return
        except (EOFError, KeyboardInterrupt):
            print()
            return

    # First-run naming flow
    if harness.identity.is_first_run():
        print(f"\n{harness.identity.get_greeting()}")
        try:
            name = input("> ").strip()
            if name:
                harness.identity.set_name(name)
                print(f"\nGreat! I'll be called {name}. How can I help you?\n")
            else:
                harness.identity.set_name("Singularity")
                print(f"\nNo problem, call me Singularity. How can I help you?\n")
        except (EOFError, KeyboardInterrupt):
            harness.identity.set_name("Singularity")
            print(f"\nNo problem, call me Singularity. How can I help you?\n")
    else:
        sub_tag = f" | {access['status']}: {access['days_remaining']}d left" if access["status"] != "active" else ""
        print(f"\n⚡ {harness.identity.get_name()} — Chat Mode")
        print(f"   Model: {harness.model.param_count:,} params | Tier: {harness.settings.tier.value}{sub_tag}")
        print(f"   Type 'exit' to quit, 'stats' for stats\n")

    session_id = f"cli-{int(__import__('time').time())}"

    while True:
        try:
            user_input = input("You: ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\nGoodbye!")
            break

        if not user_input:
            continue
        if user_input.lower() in ("exit", "quit", "bye"):
            print("Goodbye!")
            break
        if user_input.lower() == "stats":
            stats = harness.get_stats()
            print(json.dumps(stats, indent=2, default=str))
            continue

        result = harness.chat(user_input, channel="cli", session_id=session_id)
        name = harness.identity.get_name()
        cached_tag = " [cached]" if result.get("cached") else ""
        print(f"{name}: {result['response']}")
        print(f"  [{result['elapsed_s']}s{cached_tag}]\n")


def cmd_jarvis(args: argparse.Namespace) -> None:
    """Start built-in Jarvis voice assistant."""
    from .voice.jarvis import JarvisAssistant
    jarvis = JarvisAssistant(wake_word=args.wake_word)

    print(f"\n⚡ Jarvis Voice Assistant")
    print(f"   Say '{args.wake_word}' to start talking")
    print(f"   Say '{args.wake_word} stop' to interrupt speech")
    print(f"   Say '{args.wake_word} goodbye' to quit")
    print(f"   Say 'practice mode on' for self-talk training\n")

    # Check if voice backends are available
    if not jarvis.stt.is_available():
        print("⚠ STT not available (install whisper or vosk for voice input)")
        print("  Falling back to text mode. Type your commands.\n")

        session_id = jarvis._session_id
        while True:
            try:
                user_input = input("You: ").strip()
            except (EOFError, KeyboardInterrupt):
                break
            if not user_input:
                continue
            if user_input.lower() in ("exit", "quit", "bye", "goodbye"):
                break
            response = jarvis.text_chat(user_input)
            print(f"Jarvis: {response}\n")

        jarvis.stop()
        stats = jarvis.get_stats()
        print(f"\nConversations: {stats['jarvis']['conversation_count']}")
        return

    # Full voice mode
    try:
        jarvis.run()
    except KeyboardInterrupt:
        jarvis.stop()

    stats = jarvis.get_stats()
    print(f"\nJarvis stats: {json.dumps(stats, indent=2, default=str)}")


def cmd_train(args: argparse.Namespace) -> None:
    """Train the model on text data."""
    from .harness.harness import SingularityHarness
    from .train.train import Trainer
    from .model.quantization import SingularityQuantizer

    harness = SingularityHarness()

    # Train tokenizer first if needed
    print("Training tokenizer...")
    harness.train_tokenizer(args.data)

    # Update model vocab size to match tokenizer
    harness.model.config.vocab_size = harness.tokenizer.actual_vocab_size

    # Get auto-sized training params
    train_params = harness.sizer.get_training_params()
    epochs = args.epochs or train_params["epochs"]
    batch_size = args.batch_size or train_params["batch_size"]
    seq_len = args.seq_len or train_params["seq_len"]

    quantizer = SingularityQuantizer(format=harness.settings.quant.format)

    trainer = Trainer(
        model=harness.model,
        tokenizer=harness.tokenizer,
        lr=train_params["lr"],
        warmup_steps=train_params["warmup_steps"],
        checkpoint_dir=harness.data_dir,
    )

    print(f"Training: epochs={epochs}, batch_size={batch_size}, seq_len={seq_len}")
    stats = trainer.train(
        data_paths=args.data,
        epochs=epochs,
        batch_size=batch_size,
        seq_len=seq_len,
        quantizer=quantizer,
    )

    print(f"\nTraining complete: {json.dumps(stats, indent=2)}")

    # Save model
    harness.save_model()
    print("Model saved.")


def cmd_serve(args: argparse.Namespace) -> None:
    """Start the inference server."""
    from .server.server import run_server
    run_server(host=args.host, port=args.port)


def cmd_stats(args: argparse.Namespace) -> None:
    """Show model stats."""
    from .harness.harness import SingularityHarness
    harness = SingularityHarness()
    stats = harness.get_stats()
    print(json.dumps(stats, indent=2, default=str))


def cmd_export(args: argparse.Namespace) -> None:
    """Export model weights."""
    from .harness.harness import SingularityHarness
    harness = SingularityHarness()
    output_path = args.output or "model_export.npz"
    harness.save_model(output_path)
    print(f"Model exported to {output_path}")


def cmd_goals(args: argparse.Namespace) -> None:
    """Manage goals and projects."""
    from .harness.harness import SingularityHarness
    harness = SingularityHarness()

    if args.goal_action == "create":
        goal_id = harness.create_goal(args.title, args.description, priority=args.priority)
        print(f"Created goal: {goal_id}")
        if args.start_agents:
            harness.start_agents()
            print("Agents started — working on goal.")
    elif args.goal_action == "list":
        goals = harness.get_goals()
        # Also show pending goals
        from .memory.goal_memory import GoalMemory
        import os as _os
        gm = GoalMemory(db_path=_os.path.join(harness.data_dir, "goals.db"))
        pending = gm.list_goals(status="pending")
        for g in pending:
            goals.append(g.as_dict())
        if not goals:
            print("No goals found.")
        for g in goals:
            progress = g.get("progress", 0)
            agent = g.get("assigned_agent", "") or "unassigned"
            print(f"  [{g['priority']}] {g['title']} — {g['status']} ({progress:.0%}) agent: {agent}")
    elif args.goal_action == "agents":
        status = harness.get_agent_status()
        for name, info in status.items():
            current = info.get("current_goal", "idle")
            stats = info.get("stats", {})
            print(f"  {name} ({info['role']}): {'running' if info['running'] else 'stopped'} | current: {current} | completed: {stats.get('goals_completed', 0)}")
    elif args.goal_action == "start-agents":
        harness.start_agents()
        print("All 5 agents started.")
    elif args.goal_action == "stop-agents":
        harness.stop_agents()
        print("All agents stopped.")
    elif args.goal_action == "stats":
        stats = harness.get_stats()
        print(json.dumps({
            "goals": stats.get("goal_memory", {}),
            "agents": stats.get("agents", {}),
            "persistent_memory": stats.get("persistent_memory", {}),
        }, indent=2, default=str))


def cmd_daemon(args: argparse.Namespace) -> None:
    """Start the always-on daemon — agents, self-refinement, skill creation."""
    from .harness.harness import SingularityHarness
    harness = SingularityHarness()

    print(f"\n⚡ Singularity LLM — Always-On Daemon")
    print(f"   5 agents will talk to the LLM when idle")
    print(f"   Self-refinement engine will optimize speed and intelligence")
    print(f"   Skills will be created from agent conversations")
    print(f"   Goals will be progressed automatically")
    print(f"   Press Ctrl+C to stop\n")

    # Create a goal if provided
    if args.goal:
        goal_id = harness.create_goal(args.goal, args.goal, priority="high")
        print(f"Created goal: {goal_id}")

    harness.start_daemon()

    try:
        import time as _time
        while True:
            _time.sleep(10)
            stats = harness.daemon.get_stats()
            print(f"[{_time.strftime('%H:%M:%S')}] idle={stats['idle']} | "
                  f"agent_convs={stats['agent_conversations']} | "
                  f"skills_created={stats['skills_created']} | "
                  f"refinements={stats['refinement_cycles']}")
    except KeyboardInterrupt:
        print("\nStopping daemon...")
        harness.stop_daemon()
        stats = harness.daemon.get_stats()
        print(f"\nDaemon stats: {json.dumps(stats, indent=2, default=str)}")


def cmd_subscribe(args: argparse.Namespace) -> None:
    """Subscribe for $1.00/month — shows payment instructions."""
    from .harness.harness import SingularityHarness
    harness = SingularityHarness()

    print(f"\n💳 Singularity LLM Subscription — ${harness.subscription.MONTHLY_PRICE:.2f}/month")
    print(f"   Duration: {harness.subscription.SUBSCRIPTION_DURATION_DAYS} days")
    print(f"   Auto-renew: enabled\n")

    # Get payment instructions first
    instructions = harness.subscription.get_payment_instructions()
    wallet = instructions["founder_wallet"]
    print(f"📋 Payment Instructions:")
    print(f"   Amount: ${instructions['amount']:.2f} {instructions['currency']}/month")
    print(f"   Network: {instructions['network']}")
    print(f"   Accepted tokens: {', '.join(instructions['accepted_tokens'])}")
    if wallet and wallet != "Fetching from Soulmate OS...":
        print(f"   Founder wallet: {wallet}")
    print(f"   Pay via Google Pay / card: {instructions['soulmate_wallet_url']}")
    print()

    # Activate subscription
    import os as _os
    user_id = _os.getlogin() if hasattr(_os, 'getlogin') else "Singularity-user"
    result = harness.subscription.subscribe(user_id=user_id, token="USDT")
    if result["success"]:
        print(f"✓ {result['message']}")
        print(f"   Months subscribed: {result['months_subscribed']}")
        print(f"   Total paid: ${result['total_paid']:.2f}")
        print(f"   Expires in: {result['days_remaining']} days")
        if result.get("deposit_id"):
            print(f"   Deposit ID: {result['deposit_id']}")
        print()
    else:
        print(f"✗ {result.get('error', 'Subscription failed')}\n")


def cmd_unsubscribe(args: argparse.Namespace) -> None:
    """Cancel subscription."""
    from .harness.harness import SingularityHarness
    harness = SingularityHarness()

    result = harness.subscription.unsubscribe()
    if result["success"]:
        print(f"\n✓ {result['message']}\n")
    else:
        print(f"\n✗ {result.get('error', 'Unsubscribe failed')}\n")


def cmd_trial(args: argparse.Namespace) -> None:
    """Start free trial."""
    from .harness.harness import SingularityHarness
    harness = SingularityHarness()

    print(f"\n🎁 Singularity LLM Free Trial")
    print(f"   Duration: {harness.subscription.TRIAL_DURATION_DAYS} days")
    print(f"   No payment required\n")

    result = harness.subscription.start_trial()
    if result["success"]:
        print(f"✓ {result['message']}")
        print(f"   Days remaining: {result['days_remaining']}\n")
    else:
        print(f"✗ {result.get('error', 'Trial failed')}\n")


def cmd_unlock(args: argparse.Namespace) -> None:
    """Unlock free founder access with password."""
    from .harness.harness import SingularityHarness
    harness = SingularityHarness()

    password = args.password
    if not password:
        print("\n🔐 Founder Unlock — Enter your founder password")
        try:
            password = input("Password: ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\nCancelled.")
            return

    if not password:
        print("No password entered.")
        return

    result = harness.subscription.founder_unlock(password)
    if result["success"]:
        print(f"\n✓ {result['message']}\n")
    else:
        print(f"\n✗ {result['error']}\n")


def cmd_transfer(args: argparse.Namespace) -> None:
    """Process auto-transfers — route $1 payments to founder bank account."""
    from .harness.harness import SingularityHarness
    harness = SingularityHarness()

    print("\n🏦 Auto-Transfer — Routing payments to bank account\n")

    # Show current stats
    stats = harness.subscription.get_auto_transfer_stats()
    print(f"   Auto-transfer: {'enabled' if stats['auto_transfer_enabled'] else 'disabled'}")
    print(f"   Bank routing: {stats['bank_routing']}")
    print(f"   Bank account: {stats['bank_account']}")
    print(f"   Total transferred: ${stats['total_transferred']:.2f}")
    print(f"   Transfer count: {stats['transfer_count']}")
    print(f"   Pending: {stats['pending_transfers']}")
    print(f"   Untransferred payments: {stats['untransferred_payments']}")
    print()

    if not stats["auto_transfer_enabled"]:
        print("⚠ Auto-transfer is disabled. Enable it first.")
        return

    # Process transfers
    result = harness.subscription.process_auto_transfers()
    if result["status"] == "ok":
        if result["transfers_made"] > 0:
            print(f"✓ {result['message']}")
            for t in result["transfers"]:
                print(f"   ${t['amount']:.2f} → {t.get('routing_number', '????')}****/{t.get('account_number', '????')}**** (ref: {t['payment_reference'][:16]}...)")
        else:
            print(f"✓ {result['message']}")
    else:
        print(f"✗ {result.get('message', 'Transfer failed')}")
    print()


def cmd_bank(args: argparse.Namespace) -> None:
    """Show bank account info and auto-transfer settings."""
    from .harness.harness import SingularityHarness
    harness = SingularityHarness()

    print("\n🏦 Founder Bank Account Info\n")
    info = harness.subscription.get_bank_info()
    print(f"   Routing: {info['routing_number']}")
    print(f"   Account: {info['account_number']}")
    print(f"   Auto-transfer: {'enabled' if info['auto_transfer_enabled'] else 'disabled'}")
    print(f"   Total transferred: ${info['total_transferred']:.2f}")
    print(f"   Transfer count: {info['transfer_count']}")

    stats = harness.subscription.get_auto_transfer_stats()
    if stats["recent_transfers"]:
        print(f"\n   Recent transfers:")
        for t in stats["recent_transfers"]:
            print(f"   ${t['amount']:.2f} — {t['status']} — ref: {t['payment_reference'][:16]}...")
    print()

    if info["routing_number"] == "Not set" or info["account_number"] == "Not set":
        print("⚠ Bank routing/account not set!")
        print("   Set environment variables:")
        print("   set INC_LLM_CURRENT_ROUTING=your_routing_number")
        print("   set INC_LLM_CURRENT_ACCOUNT=your_account_number")
        print()


def main() -> None:
    """Main CLI entry point."""
    parser = argparse.ArgumentParser(
        prog="singularity_llm",
        description="Singularity LLM — fast, uncensored, 100% local LLM with voice assistant",
    )
    subparsers = parser.add_subparsers(dest="command", help="Available commands")

    # Chat
    p_chat = subparsers.add_parser("chat", help="Interactive chat REPL")
    p_chat.set_defaults(func=cmd_chat)

    # Jarvis
    p_jarvis = subparsers.add_parser("jarvis", help="Start built-in voice assistant")
    p_jarvis.add_argument("--wake-word", default="jarvis", help="Wake word (default: jarvis)")
    p_jarvis.set_defaults(func=cmd_jarvis)

    # Train
    p_train = subparsers.add_parser("train", help="Train the model on text data")
    p_train.add_argument("--data", required=True, help="Path to .txt file(s) or directory")
    p_train.add_argument("--epochs", type=int, default=0, help="Number of epochs (auto if 0)")
    p_train.add_argument("--batch-size", type=int, default=0, help="Batch size (auto if 0)")
    p_train.add_argument("--seq-len", type=int, default=0, help="Sequence length (auto if 0)")
    p_train.set_defaults(func=cmd_train)

    # Serve
    p_serve = subparsers.add_parser("serve", help="Start inference server")
    p_serve.add_argument("--host", default="0.0.0.0", help="Host (default: 0.0.0.0)")
    p_serve.add_argument("--port", type=int, default=8548, help="Port (default: 8548)")
    p_serve.set_defaults(func=cmd_serve)

    # Stats
    p_stats = subparsers.add_parser("stats", help="Show model stats")
    p_stats.set_defaults(func=cmd_stats)

    # Export
    p_export = subparsers.add_parser("export", help="Export model weights")
    p_export.add_argument("--output", default="", help="Output path")
    p_export.add_argument("--format", default="npz", help="Export format (npz)")
    p_export.set_defaults(func=cmd_export)

    # Goals
    p_goals = subparsers.add_parser("goals", help="Manage goals, projects, and agents")
    p_goals.add_argument("goal_action", choices=["create", "list", "agents", "start-agents", "stop-agents", "stats"],
                         help="Action to perform")
    p_goals.add_argument("--title", default="", help="Goal title (for create)")
    p_goals.add_argument("--description", default="", help="Goal description (for create)")
    p_goals.add_argument("--priority", default="high", choices=["critical", "high", "medium", "low"])
    p_goals.add_argument("--start-agents", action="store_true", help="Start agents after creating goal")
    p_goals.set_defaults(func=cmd_goals)

    # Daemon
    p_daemon = subparsers.add_parser("daemon", help="Start always-on daemon (agents + refinement + skill creation)")
    p_daemon.add_argument("--goal", default="", help="Optional goal to create and work on")
    p_daemon.set_defaults(func=cmd_daemon)

    # Subscribe
    p_sub = subparsers.add_parser("subscribe", help="Subscribe for $1.00/month")
    p_sub.set_defaults(func=cmd_subscribe)

    # Unsubscribe
    p_unsub = subparsers.add_parser("unsubscribe", help="Cancel subscription")
    p_unsub.set_defaults(func=cmd_unsubscribe)

    # Trial
    p_trial = subparsers.add_parser("trial", help="Start 7-day free trial")
    p_trial.set_defaults(func=cmd_trial)

    # Founder unlock
    p_unlock = subparsers.add_parser("unlock", help="Unlock free founder access")
    p_unlock.add_argument("--password", default="", help="Founder password")
    p_unlock.set_defaults(func=cmd_unlock)

    # Auto-transfer
    p_transfer = subparsers.add_parser("transfer", help="Process auto-transfers to bank account")
    p_transfer.set_defaults(func=cmd_transfer)

    # Bank info
    p_bank = subparsers.add_parser("bank", help="Show bank account info")
    p_bank.set_defaults(func=cmd_bank)

    args = parser.parse_args()
    if not args.command:
        parser.print_help()
        sys.exit(1)

    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
        datefmt="%H:%M:%S",
    )

    args.func(args)


if __name__ == "__main__":
    main()