#
# @Author: Bhaskar S
# @Blog:   https://polarsparc.github.io
# @Date:   07 Aug 2026
# @Ver:    1.0
#

agent:
  provider: llama.cpp
  model: qwen3.6-a3b
  temperature: 0.9
  top_p: 0.95
  top_k: 64
  base_url: http://192.168.1.25:8000

#
# @Author: Bhaskar S
# @Blog:   https://polarsparc.github.io
# @Date:   07 Aug 2026
# @Ver:    1.0
#

import logging
import os
import yaml
from pydantic import BaseModel

logging.basicConfig(format='%(levelname)s %(asctime)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)

# Default path to the settings file
DEFAULT_SETTINGS_YAML = './settings.yaml'

class LLMConfig(BaseModel):
  """Holds the LLM parameters used to build the chat model"""
  provider: str = 'ollama'
  model: str = 'qwen-3.6'
  temperature: float = 0.0
  top_p: float = 0.5
  top_k: int = 32
  base_url: str = 'http://localhost:11434'

def load_config(path: str = DEFAULT_SETTINGS_YAML) -> LLMConfig:
  """Load LLMConfig from a YAML file, falling back to defaults"""
  data = {}
  logger.info(f'Config settings: {path}')
  if os.path.exists(path):
    with open(path, 'r') as settings:
      data = yaml.safe_load(settings) or {}
    logger.info(f'Loaded settings: {data}')
  else:
    logger.info(f'No settings file {path} not found, using defaults')

  return LLMConfig(**data['agent'])

if __name__ == '__main__':
  config = load_config()
  print(config)

#
# @Author: Bhaskar S
# @Blog:   https://polarsparc.github.io
# @Date:   07 Aug 2026
# @Ver:    1.0
#

import logging
from langchain_ollama import ChatOllama
from langchain_openai import ChatOpenAI
from config import LLMConfig

logging.basicConfig(format='%(levelname)s %(asctime)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)

def build_chat_model(config: LLMConfig):
  """Return a chat model instance matching config.provider"""
  if config.provider == 'ollama':
    logger.info(f'Ollama server: base-url -> {config.base_url}, model -> {config.model}')
    return ChatOllama(
      model=config.model,
      base_url=config.base_url,
      temperature=config.temperature,
      top_p=config.top_p,
      top_k=config.top_k,
    )

  if config.provider == 'llama.cpp':
    logger.info(f'Llama.cpp server: base-url -> {config.base_url}, model -> {config.model}')
    # llama.cpp's server exposes an OpenAI-compatible HTTP API, so we use ChatOpenAI
    return ChatOpenAI(
      model=config.model,
      base_url=config.base_url,
      api_key='llama.cpp',
      temperature=config.temperature,
      top_p=config.top_p,
      extra_body={'top_k': config.top_k},
    )

  raise ValueError(f'Unsupported provider -> {config.provider}')

#
# @Author: Bhaskar S
# @Blog:   https://polarsparc.github.io
# @Date:   07 Aug 2026
# @Ver:    1.0
#

import logging
from langchain_core.messages import AIMessage, AIMessageChunk, HumanMessage, SystemMessage
from langchain.agents import create_agent
from rich.console import Console
from rich.prompt import Prompt
from config import load_config
from llm import build_chat_model

DEFAULT_SYSTEM_MESSAGE = SystemMessage(content='You are a helpful AI task assistant')

logging.basicConfig(format='%(levelname)s %(asctime)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)

console = Console()

def initial_history() -> list:
    """Return a clean conversation history containing only the system message"""
    return [DEFAULT_SYSTEM_MESSAGE]

def handle_prompt(agent, history: list, message: str):
    """Send the prompt message to the agent and return (response, new_history) the response as tokens arrive"""
    history = history + [HumanMessage(content=message)]

    response = ''
    for chunk, _ in agent.stream({'messages': history}, stream_mode='messages'):
        if isinstance(chunk, AIMessageChunk) and chunk.content:
            console.print(chunk.content, end='')
            response += chunk.content
    console.print()

    new_history = history + [AIMessage(content=response)]
    return response, new_history

def process_line(line: str, agent, history: list):
    """Dispatch the user prompt and return (output_message, new_history, should_continue) tuple.
    If /exit, exit the program"""
    line = line.strip()

    if line == '/exit':
        logger.info("Exiting bcode!")
        return None, history, False

    if line == '/clear':
        logger.info('Context clear command')
        return 'Context Cleared', initial_history(), True

    if line.startswith('/prompt'):
        message = line[len('/prompt'):].strip()
        if not message:
            return 'Usage: /prompt <message>', history, True
        try:
            response, new_history = handle_prompt(agent, history, message)
            return response, new_history, True
        except Exception as exc:
            logger.error(f'Error processing prompt: {exc}')
            return f'Error: {exc}', history, True

    return 'Unknown Command', history, True

def run() -> None:
    """Start the agent ReACT loop"""
    logger.info('starting bcode agent CLI')

    config = load_config()
    llm = build_chat_model(config)
    # No tools are registered in this version; the agent simply reasons and responds
    agent = create_agent(llm, tools=[])
    history = initial_history()

    console.print('[bold magenta3]-----> bcode v1.0 <-----[/bold magenta3]')
    while True:
        try:
            line = Prompt.ask('[bold dark_orange]bc[/bold dark_orange]', console=console, show_default=False)
        except (EOFError, KeyboardInterrupt):
            console.print()
            break

        if not line.strip():
            continue

        output, history, should_continue = process_line(line, agent, history)
        if output is not None:
            console.print(output)
        if not should_continue:
            break

    logger.info('bcode agent stopped')

if __name__ == "__main__":
    run()

#
# @Author: Bhaskar S
# @Blog:   https://polarsparc.github.io
# @Date:   07 Aug 2026
# @Ver:    1.1
#

agent:
  provider: llama.cpp
  model: qwen3.6-a3b
  temperature: 0.9
  top_p: 0.95
  top_k: 64
  base_url: http://192.168.1.25:8000
  tools: True

tools:
  bash: ask
  edit: allow
  grep: allow
  read: allow
  websearch: ask
  write: allow

#
# @Author: Bhaskar S
# @Blog:   https://polarsparc.github.io
# @Date:   07 Aug 2026
# @Ver:    1.1
#

import logging
import os
import yaml
from pydantic import BaseModel

logging.basicConfig(format='%(levelname)s %(asctime)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)

# Default path to the settings file
DEFAULT_SETTINGS_YAML = './settings.yaml'

class LLMConfig(BaseModel):
  """Holds the LLM parameters used to build the chat model"""
  provider: str = 'ollama'
  model: str = 'qwen-3.6'
  temperature: float = 0.0
  top_p: float = 0.5
  top_k: int = 32
  base_url: str = 'http://localhost:11434'
  tools: bool = False

class LLMTools(BaseModel):
  """Holds the LLM tools along with their permissions used to build the chat model"""
  tool_perm: dict[str, str] = {}

def load_config(path: str = DEFAULT_SETTINGS_YAML) -> LLMConfig:
  """Load LLMConfig from a YAML file, falling back to defaults"""
  params = {}
  logger.info(f'Config settings: {path}')
  if os.path.exists(path):
    with open(path, 'r') as settings:
      params = yaml.safe_load(settings) or {}
    logger.info(f'Loaded settings: {params}')
  else:
    logger.info(f'No settings file {path} not found, using defaults')

  return LLMConfig(**params['agent']), LLMTools(tool_perm=params['tools'])

if __name__ == '__main__':
  config, tool_permissions = load_config()
  logger.info(f'config: {config}, tools: {tool_permissions}')

#
# @Author: Bhaskar S
# @Blog:   https://polarsparc.github.io
# @Date:   08 Aug 2026
# @Ver:    1.1
#

import logging
import os
import re
import subprocess
from ddgs import DDGS
from langchain_core.tools import tool
from rich.prompt import Confirm
from config import LLMTools

logging.basicConfig(format='%(levelname)s %(asctime)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)

# Any operation targeting /etc or anything under it must be blocked
_ETC_PATH = os.path.realpath('/etc')

def _guard_etc_path(path: str) -> None:
  """Raise PermissionError if path resolves to /etc or anything under it"""
  resolved = os.path.realpath(os.path.abspath(os.path.expanduser(path)))
  if resolved == _ETC_PATH or resolved.startswith(_ETC_PATH + os.sep):
    raise PermissionError(f'Access to system files under /etc is not permitted: {path}')

def _guard_etc_in_command(command: str) -> None:
  """Best-effort textual block of /etc references in a shell command"""
  if '/etc' in command:
    raise PermissionError('Shell commands referencing /etc are not permitted')

@tool
def bash(command: str) -> str:
  """Execute a shell command or executable and return its output"""
  try:
    _guard_etc_in_command(command)
    result = subprocess.run(
      command, shell=True, capture_output=True, text=True, timeout=30
    )
    output = result.stdout + result.stderr
    return f'(Exit code from bash {result.returncode})\n{output}'
  except subprocess.TimeoutExpired:
    return 'Bash Error: command timed out after 30 seconds'
  except Exception as exc:
    return f'Bash Error: {exc}'

@tool
def edit(path: str, old_string: str, new_string: str) -> str:
  """Replace the first occurrence of old_string with new_string in a file"""
  try:
    _guard_etc_path(path)
    with open(path, 'r') as f:
      content = f.read()
    if old_string not in content:
      return f'Edit Error: old_string not found in {path}'
    content = content.replace(old_string, new_string, 1)
    with open(path, 'w') as f:
      f.write(content)
    return f'Edited {path}'
  except Exception as exc:
    return f'Edit Error: {exc}'

@tool
def grep(pattern: str, path: str = '.') -> str:
  """Search for lines matching a regex pattern in a file or directory"""
  try:
    _guard_etc_path(path)

    if os.path.isfile(path):
      file_paths = [path]
    else:
      file_paths = []
      for root, _, filenames in os.walk(path):
        for filename in filenames:
          file_paths.append(os.path.join(root, filename))

    regex = re.compile(pattern)

    matches = []
    for file_path in file_paths:
      try:
        _guard_etc_path(file_path)
        with open(file_path, 'r', errors='ignore') as f:
          for line_number, line in enumerate(f, start=1):
            if regex.search(line):
              matches.append(f'{file_path}:{line_number}: {line.rstrip()}')
      except (PermissionError, OSError):
        continue

    if not matches:
      return "No matches found."

    max_matches = 100
    if len(matches) > max_matches:
      matches = matches[:max_matches] + [f'... [truncated, {len(matches)} total matches]']
    return '\n'.join(matches)
  except Exception as exc:
    return f'Grep Error: {exc}'

@tool
def read(path: str) -> str:
  """Read and return the contents of a file"""
  try:
    _guard_etc_path(path)
    with open(path, 'r') as f:
      content = f.read()
    max_chars = 1024 * 16
    if len(content) > max_chars:
      return content[:max_chars] + f'\n... [truncated, {len(content)} total characters]'
    return content
  except Exception as exc:
    return f'Read Error: {exc}'

@tool
def websearch(query: str) -> str:
  """Search the web and return a short list of matching results"""
  try:
    results = DDGS().text(query, max_results=5)
    if not results:
      return 'No web search results found'
    lines = []
    for r in results:
      lines.append(f"{r.get('title', '')}\n{r.get('href', '')}\n{r.get('body', '')}")
    return '\n\n'.join(lines)
  except Exception as exc:
    return f'Websearch Error: {exc}'

@tool
def write(path: str, content: str) -> str:
  """Write content to a file, overwriting it if it already exists"""
  try:
    _guard_etc_path(path)
    with open(path, 'w') as f:
      f.write(content)
    return f'Wrote {len(content)} characters to {path}'
  except Exception as exc:
    return f'Write Error: {exc}'

TOOL_REGISTRY = {
  "bash": bash,
  "edit": edit,
  "grep": grep,
  "read": read,
  "websearch": websearch,
  "write": write,
}

def _is_tool_allowed(name: str, permissions: LLMTools) -> bool:
  """Decide whether tool `name` is allowed"""
  perm = permissions.tool_perm.get(name)
  if perm == 'allow':
    return True
  if perm == 'ask':
    return Confirm.ask(f'Allow tool [bold bright_red]{name}[/bold bright_red]?', default=False)
  return False

def _wrap_with_permission_check(name, tool_ref, permissions: LLMTools):
  """Return a wrapper on a tool that checks _is_tool_allowed before each call"""
  logger.info(f"Wrapping tool function {tool_ref.func} for '{name}' ")

  original_func = tool_ref.func

  def guarded_func(**kwargs):
    if not _is_tool_allowed(name, permissions):
      logger.info(f"Tool '{name}' call blocked")
      raise PermissionError(f"Tool '{name}' call blocked")
    return original_func(**kwargs)

  return tool_ref.model_copy(update={'func': guarded_func})

def get_tools(permissions: LLMTools):
  """Return all tools from the registry, each wrapped by _wrap_with_permission_check"""
  return [
    _wrap_with_permission_check(name, tool_ref, permissions) for name, tool_ref in TOOL_REGISTRY.items()
  ]

if __name__ == '__main__':
  logger.info(get_tools(LLMTools()))

#
# @Author: Bhaskar S
# @Blog:   https://polarsparc.github.io
# @Date:   08 Aug 2026
# @Ver:    1.1
#

import logging
from langchain_core.messages import AIMessage, AIMessageChunk, HumanMessage, SystemMessage
from langchain.agents import create_agent
from rich.console import Console
from rich.prompt import Prompt
from config import load_config
from llm import build_chat_model
from tools import get_tools

DEFAULT_SYSTEM_MESSAGE = SystemMessage(content='You are a helpful AI task assistant')

logging.basicConfig(format='%(levelname)s %(asctime)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)

console = Console()

def initial_history() -> list:
  """Return a clean conversation history containing only the system message"""
  return [DEFAULT_SYSTEM_MESSAGE]

def handle_prompt(agent, history: list, message: str):
  """Send the prompt message to the agent and return (response, new_history) the response as tokens arrive"""
  history = history + [HumanMessage(content=message)]

  response = ''
  for chunk, _ in agent.stream({'messages': history}, stream_mode='messages'):
    if isinstance(chunk, AIMessageChunk) and chunk.content:
      console.print(chunk.content, end='')
      response += chunk.content
  console.print()

  new_history = history + [AIMessage(content=response)]
  return response, new_history

def process_line(line: str, agent, history: list):
  """Dispatch the user prompt and return (output_message, new_history, should_continue) tuple.
  If /exit, exit the program"""
  line = line.strip()

  if line == '/exit':
    logger.info("Exiting bcode!")
    return None, history, False

  if line == '/clear':
    logger.info('Context clear command')
    return 'Context Cleared', initial_history(), True

  if line.startswith('/prompt'):
    message = line[len('/prompt'):].strip()
    if not message:
      return 'Usage: /prompt <message>', history, True
    try:
      response, new_history = handle_prompt(agent, history, message)
      return response, new_history, True
    except Exception as exc:
      logger.error(f'Error processing prompt: {exc}')
      return f'Error: {exc}', history, True

  return 'Unknown Command', history, True

def run() -> None:
  """Start the agent ReACT loop"""
  logger.info('starting bcode agent CLI')

  config, permissions = load_config()
  llm = build_chat_model(config)

  if config.tools:
    tools = get_tools(permissions)
  else:
    tools = []

  agent = create_agent(llm, tools=tools)
  history = initial_history()

  console.print('[bold magenta3]-----> bcode v1.1 <-----[/bold magenta3]')
  while True:
    try:
      line = Prompt.ask('[bold dark_orange]bc[/bold dark_orange]', console=console, show_default=False)
    except (EOFError, KeyboardInterrupt):
      console.print()
      break

    if not line.strip():
      continue

    output, history, should_continue = process_line(line, agent, history)
    if output is not None:
      console.print(output)
    if not should_continue:
      break

  logger.info('bcode agent stopped')

if __name__ == "__main__":
  run()