ai-agent.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Py Agent [v0.9.9.64] - Main CLI Runtime, Workspace Agent & Command Dispatcher [CR.01 CO.01]"""
import json
import os
import re
import shutil
import sqlite3
import subprocess
import sys
import tempfile
import time
from contextlib import closing
CFG_DIR: str = os.path.expanduser("~/.config/py-agent")
VAR_DIR: str = os.path.join(CFG_DIR, "var")
os.makedirs(VAR_DIR, exist_ok=True)
CONTEXT_FILE: str = os.path.join(CFG_DIR, "ai-context.md")
SKILLS_DIR: str = os.path.join(CFG_DIR, "skills")
SESSIONS_DIR: str = os.path.join(CFG_DIR, "projects", ".database")
BASE_PROMPT_CHAT: str = "Active, natural conversational assistant."
BASE_PROMPT_AGENT: str = "Active local workspace developer agent."
# Precompiled hot-path regular expressions
RE_THINK_TAGS: re.Pattern = re.compile(r"<think>[\s\S]*?(?:</think>|$)", re.DOTALL)
RE_AUTO_RUN: re.Pattern = re.compile(
r"Run:\s*((?:trace symbol|blast radius|read function|find symbol)\s+[a-zA-Z0-9_.:#/-]+|architecture overview)"
)
# Requires '/' prefix for single letters (a, b, f) so plain words are not swallowed
RE_THINK_BIN: re.Pattern = re.compile(r"^(?:/(?:[fba]|tk)|/?tk)(?:\s+(\d+))?$", re.IGNORECASE)
RE_SHELL_META: re.Pattern = re.compile(r"[\[\]{}()='\",;|#<>]")
def load_env_file(path: str) -> None:
"""Auto-publishes .env from .env.example and loads variables into os.environ."""
if not os.path.isfile(path):
example_p = os.path.join(os.path.dirname(path), ".env.example")
if os.path.isfile(example_p):
try:
shutil.copy2(example_p, path)
except OSError:
pass
if os.path.isfile(path):
try:
with open(path, "r", encoding="utf-8") as f:
for line in f:
if (l := line.strip()) and not l.startswith("#") and "=" in l:
k, v = l.replace("export ", "", 1).split("=", 1)
if (k := k.strip()) and k not in os.environ:
os.environ[k] = v.split(" #")[0].strip().strip('"').strip("'")
except OSError:
pass
load_env_file(os.path.join(CFG_DIR, ".env"))
if (mod_dir := os.path.join(CFG_DIR, "modules")) not in sys.path:
sys.path.append(mod_dir)
try:
import agent_context as context
from agent_context import STOP_WORDS
except ImportError as e:
sys.stderr.write(f"\033[1;31m[CRITICAL]: Failed to load agent_context: {e}\033[0m\n")
sys.exit(1)
# Lazy references to heavy interactive modules to preserve sub-millisecond shell-hook startup
commands = None
core = None
memories = None
sessions = None
skills = None
tts = None
ui = None
voice = None
def _load_agent_modules() -> None:
"""Loads interactive modules on demand, keeping shell hook invocations sub-millisecond."""
global commands, core, memories, sessions, skills, tts, ui, voice
if commands is not None:
return
try:
import agent_commands as _commands
import agent_core as _core
import agent_memories as _memories
import agent_sessions as _sessions
import agent_skills as _skills
import agent_tts as _tts
import agent_ui as _ui
import agent_voice as _voice
commands, core, memories, sessions, skills, tts, ui, voice = (
_commands,
_core,
_memories,
_sessions,
_skills,
_tts,
_ui,
_voice,
)
except ImportError as e:
sys.stderr.write(f"\033[1;31m[CRITICAL]: Failed to load modules: {e}\033[0m\n")
sys.exit(1)
try:
import readline
readline.parse_and_bind(r'"\e[A": previous-history')
readline.parse_and_bind(r'"\e[B": next-history')
except ImportError:
pass
_transient_cmd_count: int = 0
def _show_transient_status(tag: str) -> None:
"""Displays an inline status tag with breathing room, accumulating for batch removal on next query."""
global _transient_cmd_count
sys.stdout.write(f" \033[2m[{tag}]\033[0m\n\n")
sys.stdout.flush()
_transient_cmd_count += 3
_flash_status = _show_transient_status
def _clear_transient_status(query: str = "") -> None:
"""Erases the ephemeral slash command block and repositions the real query prompt with breathing room."""
global _transient_cmd_count
if _transient_cmd_count > 0:
try:
if query:
sys.stdout.write(f"\033[{_transient_cmd_count + 1}A\r\x1b[0J❯ {query}\n")
else:
sys.stdout.write(f"\033[{_transient_cmd_count}A\r\x1b[0J")
sys.stdout.flush()
except OSError:
pass
_transient_cmd_count = 0
def workspace_db_counts(safe_name: str, workspace_path: str = "") -> tuple[int, int]:
"""Retrieves SQLite turn counts and OKF memory counts without spawning empty databases."""
_load_agent_modules()
db_path = os.path.join(SESSIONS_DIR, f"{safe_name}.db")
t = 0
if os.path.isfile(db_path):
try:
with closing(sqlite3.connect(f"file:{db_path}?mode=ro", uri=True, timeout=1.5)) as conn:
try:
t = conn.execute("SELECT COUNT(*) FROM turns WHERE workspace = ?", (safe_name,)).fetchone()[0]
except (sqlite3.Error, TypeError, IndexError):
t = 0
except sqlite3.Error:
t = 0
m_count = memories.get_memory_count(workspace_path) if workspace_path else 0
return t, m_count
def ensure_clean_agent_dir(workspace_path: str) -> None:
"""Moves root artifacts into .agent/ without clobbering existing destinations."""
if not (ws_name := os.path.basename(workspace_path)):
return
agent_dir = os.path.join(workspace_path, ".agent")
try:
os.makedirs(os.path.join(agent_dir, "memory"), exist_ok=True)
except OSError:
return
targets = [
f"index-map-{ws_name}.txt",
f"index-map-memory-{ws_name}.db",
f"index-map-memory-{ws_name}.db-wal",
f"index-map-memory-{ws_name}.db-shm",
"history.md",
]
for fname in targets:
src = os.path.join(workspace_path, fname)
dst = os.path.join(agent_dir, fname)
if os.path.exists(src):
if not os.path.exists(dst):
try:
os.replace(src, dst)
except OSError:
pass
elif fname == "history.md":
try:
with open(src, "r", encoding="utf-8") as sf, open(dst, "a", encoding="utf-8") as df:
df.write(f"\n\n{sf.read().strip()}\n")
os.remove(src)
except OSError:
pass
def _sweep_dead_session_locks() -> None:
"""Removes .session lockfiles whose owning process is dead. Ignores alive processes."""
sess_dir = os.path.join(CFG_DIR, "var", ".active_sessions")
if not os.path.isdir(sess_dir):
return
try:
entries = os.listdir(sess_dir)
except OSError:
return
if not entries:
return
for f in entries:
if not f.endswith(".session"):
continue
try:
pid = int(f.rsplit("-", 1)[-1].replace(".session", ""))
if pid <= 0:
continue
os.kill(pid, 0)
except ProcessLookupError:
try:
os.remove(os.path.join(sess_dir, f))
except OSError:
pass
except (PermissionError, ValueError, OSError):
pass
def clean_exit(safe_name: str | None = None, workspace_path: str | None = None, sub_id: int | None = None) -> None:
"""Cleans active session locks and worktrees independently, then exits cleanly."""
_clear_transient_status("")
_load_agent_modules()
if safe_name:
if sub_id is not None and sub_id > 0 and workspace_path:
try:
merge = False
if ui and hasattr(ui, "confirm_tool"):
merge = ui.confirm_tool(f"Merge sub-agent #{sub_id} worktree into main branch")
sessions.cleanup_subagent_worktree(workspace_path, sub_id, merge=merge)
except Exception as e:
if os.environ.get("AI_DEBUG") == "1":
sys.stderr.write(f"\r\n[debug] Worktree cleanup failed for sub-agent #{sub_id}: {e}\r\n")
try:
sessions.cleanup_sub_agent(safe_name, os.getpid())
except Exception as e:
if os.environ.get("AI_DEBUG") == "1":
sys.stderr.write(f"\r\n[debug] Session lock cleanup failed: {e}\r\n")
_sweep_dead_session_locks()
if ui and hasattr(ui, "_console"):
ui._console.print("[yellow]Exiting session.[/yellow]")
else:
print("Exiting session.")
sys.exit(0)
def _launch_surface(
script_path: str, is_agent: bool, ws_path: str, skill: str, history: list, args: list[str] | None = None
) -> None:
"""Launches secondary surfaces (TUI, PyCode, WebUI) with state continuity and E2BIG protection."""
_load_agent_modules()
if not os.path.exists(script_path):
ui._console.print(f"[red][sys] Launcher script not found: {script_path}[/red]\n")
return
hist_file = None
try:
hist_data = json.dumps(history)
if len(hist_data) > 65536:
with tempfile.NamedTemporaryFile(
"w", delete=False, prefix="ai_hist_", suffix=".json", encoding="utf-8"
) as tf:
tf.write(hist_data)
hist_file = tf.name
except Exception:
hist_data = ""
env = {
**os.environ,
"AI_IS_AGENT": "1" if is_agent else "0",
"AI_WORKSPACE_PATH": ws_path,
"AI_ACTIVE_SKILL": skill,
"AI_CONFIRM_GATES": os.environ.get("AI_CONFIRM_GATES", "1"),
}
if hist_file:
env["AI_SESSION_HISTORY_FILE"] = hist_file
env["AI_SESSION_HISTORY"] = ""
else:
env["AI_SESSION_HISTORY"] = hist_data
cmd = ([sys.executable, script_path] if script_path.endswith(".py") else ["/bin/bash", script_path]) + (args or [])
try:
proc = subprocess.run(cmd, env=env)
if proc.returncode != 0 and proc.returncode != 130:
ui._console.print(f"[dim yellow][sys] Surface exited with rc={proc.returncode}.[/dim yellow]\n")
st = core.get_state()
os.environ["AI_SHOW_THINKING"] = "1" if st.get("show_thinking", True) else "0"
ui._console.print("[green][sys] Resumed CLI session.[/green]\n")
except Exception as e:
ui._console.print(f"[red][sys] Launch failed: {e}[/red]\n")
finally:
if hist_file and os.path.exists(hist_file):
try:
os.remove(hist_file)
except OSError:
pass
def resolve_default_profile(is_agent: bool = True) -> str:
"""Dynamically resolves default profile from environment, active model, or available profiles."""
if not is_agent:
return "chat"
if env_skill := os.environ.get("AI_ACTIVE_SKILL", "").strip():
primary = env_skill.split()[0].lstrip("-").lower()
if primary and primary != "init":
return primary
try:
m_name = ""
if ui and hasattr(ui, "get_local_model_name"):
m_name = ui.get_local_model_name(max_age=5.0).lower()
if m_name and "not loaded" not in m_name and "offline" not in m_name:
prof_dir = os.path.join(SKILLS_DIR, "profiles")
if os.path.isdir(prof_dir):
available = [os.path.splitext(f)[0] for f in os.listdir(prof_dir) if f.endswith(".md")]
clean_m = re.sub(r"[^a-z0-9]", "", m_name)
for p in available:
clean_p = re.sub(r"[^a-z0-9]", "", p)
if clean_p in clean_m or clean_m in clean_p:
return p
except Exception:
pass
return "agent"
def run_interactive_chat(args: list[str]) -> None:
"""Primary interactive agent and chat engine."""
_load_agent_modules()
is_agent = args[0] == "--talk-chat"
raw_ws = os.environ.get("AI_WORKSPACE_PATH", os.getcwd())
clean_ws = re.sub(r"^[\s]+", "", str(raw_ws)).strip().splitlines()[0].strip() if raw_ws else os.getcwd()
original_workspace_path = os.path.realpath(os.path.expanduser(clean_ws))
workspace_path = original_workspace_path
os.environ["AI_WORKSPACE_PATH"] = workspace_path
home_dir = os.path.expanduser("~")
safe_name = core.workspace_safe_name(workspace_path, home_dir)
try:
os.chdir(workspace_path)
except OSError:
pass
ensure_clean_agent_dir(workspace_path)
cfg_file = os.path.join(workspace_path, ".agent", "config.json")
selected_profile = resolve_default_profile(is_agent)
is_yolo, use_map, is_py, memory_active, adapters_active = False, False, False, False, False
if is_agent:
if not os.path.exists(cfg_file):
selected_profile, is_yolo, use_map, is_py, memory_active, adapters_active = ui.select_workspace_profile(
os.path.basename(workspace_path)
)
try:
os.makedirs(os.path.dirname(cfg_file), exist_ok=True)
with open(cfg_file, "w", encoding="utf-8") as cf:
json.dump(
{
"profile": selected_profile,
"yolo": is_yolo,
"map": use_map,
"py": is_py,
"memory": memory_active,
"adapters": adapters_active,
"created_at": time.strftime("%Y-%m-%d %H:%M"),
},
cf,
indent=2,
)
except OSError:
pass
else:
try:
with open(cfg_file, "r", encoding="utf-8") as cf:
d = json.load(cf)
selected_profile = d.get("profile") or resolve_default_profile(is_agent)
is_yolo = bool(d.get("yolo", False))
use_map = bool(d.get("map", False))
is_py = bool(d.get("py", False))
memory_active = bool(d.get("memory", False))
adapters_active = bool(d.get("adapters", d.get("adp", False)))
if "calm" in d:
core.save_state("calm_mode", bool(d["calm"]))
except (OSError, json.JSONDecodeError):
pass
profile_args = set()
for arg in args:
if arg.startswith("-") and arg not in ("--talk", "--talk-chat"):
selected_profile = arg.lstrip("-").lower()
profile_args.add(arg)
if is_agent:
clean_name = resolve_default_profile(True) if selected_profile in ("init", "pi/pro") else selected_profile
profile_content = skills.load_skill_content(clean_name, SKILLS_DIR, CFG_DIR)
if not profile_content and clean_name not in ("agent", "init"):
ui._console.print(
f"[dim yellow][sys] Skill '{clean_name}' not found. Using minimal agent prompt.[/dim yellow]"
)
active_system_prompt = profile_content or BASE_PROMPT_AGENT
os.environ["AI_ACTIVE_SKILL"] = clean_name
st_init = core.get_state()
reasoning_budget = st_init.get("reasoning_budget", 500)
reasoning_active = st_init.get("reasoning_active", reasoning_budget > 0)
if os.path.isfile(cfg_file):
try:
with open(cfg_file, "r", encoding="utf-8") as cf:
d = json.load(cf)
if "reasoning" in d:
reasoning_active = bool(d["reasoning"])
core.save_state("reasoning_active", reasoning_active)
if "reasoning_budget" in d:
reasoning_budget = int(d["reasoning_budget"])
core.save_state("reasoning_budget", reasoning_budget)
except Exception:
pass
is_purebash = clean_name in ("purebash", "swe", "bash") or "purebash" in clean_name
core.save_state("use_map", use_map)
core.save_state("yolo_mode", is_yolo)
core.save_state("ipython_mode", is_py)
core.save_state("memory_active", memory_active)
core.save_state("adapters_active", adapters_active)
core.save_state("purebash_mode", is_purebash)
os.environ["AI_USE_MAP"] = "1" if use_map else "0"
os.environ["AI_IPYTHON_MODE"] = "1" if is_py else "0"
os.environ["AI_PUREBASH_MODE"] = "1" if is_purebash else "0"
if is_yolo:
os.environ["AI_CONFIRM_GATES"] = "0"
if use_map:
agent_dir = os.path.join(workspace_path, ".agent")
ws_name = os.path.basename(workspace_path)
txt_p = os.path.join(agent_dir, f"index-map-{ws_name}.txt")
db_p = os.path.join(agent_dir, f"index-map-memory-{ws_name}.db")
imap_bin = os.path.join(CFG_DIR, "tools", "index-map", "index-map")
if not os.path.isfile(txt_p) or not os.path.isfile(db_p):
if os.path.isfile(imap_bin):
res_map = subprocess.run(
[sys.executable, imap_bin, "--agent", workspace_path], capture_output=True, text=True
)
if res_map.returncode == 0:
ui._console.print("[dim][sys] Map enabled: compiled index-map.[/dim]")
else:
ui._console.print(
f"[dim yellow][sys] Map compilation failed (rc={res_map.returncode}): {res_map.stderr.strip()[:100]}[/dim yellow]"
)
map_path = next(
(p for p in (txt_p, os.path.join(workspace_path, f"index-map-{ws_name}.txt")) if os.path.exists(p)),
None,
)
if map_path:
try:
with open(map_path, "r", encoding="utf-8") as mf:
new_map = mf.read().strip()
if "### CODESPACE MAP:" not in active_system_prompt:
active_system_prompt += f"\n\n### CODESPACE MAP:\n{new_map}"
except OSError:
pass
else:
clean_name = selected_profile if (selected_profile and selected_profile != "pi/pro") else "chat"
skill_content = skills.load_skill_content(clean_name, SKILLS_DIR, CFG_DIR)
if not skill_content and clean_name != "chat":
ui._console.print(
f"[dim yellow][sys] Skill '{clean_name}' not found. Using minimal chat prompt.[/dim yellow]"
)
active_system_prompt = skill_content or BASE_PROMPT_CHAT
os.environ["AI_ACTIVE_SKILL"] = clean_name
inst_p = os.path.join(SKILLS_DIR, "system_instructions.md")
if os.path.isfile(inst_p) and os.path.getsize(inst_p) > 0:
try:
with open(inst_p, "r", encoding="utf-8") as f:
rules = [l for l in f if l.strip() and not l.strip().startswith("#")]
if active_rules := "".join(rules).strip():
active_system_prompt += f"\n\n{active_rules}"
except OSError:
pass
pending_args = [a for a in args[1:] if a not in profile_args]
pending_query = " ".join(pending_args) if pending_args else None
if pending_query and ("CODEBASE INDEX MAP" in pending_query or "index-map" in pending_query):
if use_map and "### CODESPACE MAP:" not in active_system_prompt:
active_system_prompt += f"\n\n### CODESPACE MAP:\n{pending_query}"
pending_query = None
chat_history = [{"role": "system", "content": active_system_prompt}]
st = core.get_state()
show_stats = st.get("show_stats", True)
if not is_agent:
memory_active = st.get("memory_active", False)
reasoning_budget = st.get("reasoning_budget", 500)
reasoning_active = st.get("reasoning_active", reasoning_budget > 0)
os.environ["AI_RENDER_MARKDOWN"] = "1" if st.get("render_markdown", False) else "0"
os.environ["AI_REASONIX_ACTIVE"] = "1" if st.get("reasonix_active", True) else "0"
os.environ["AI_SHOW_THINKING"] = "1" if st.get("show_thinking", True) else "0"
if is_yolo or st.get("yolo_mode", False):
os.environ["AI_CONFIRM_GATES"] = "0"
db_turns, mem_count = workspace_db_counts(safe_name, workspace_path) if is_agent else (0, 0)
_sweep_dead_session_locks()
sub_id = sessions.get_sub_agent_id(safe_name, os.getpid()) if is_agent else None
if sub_id is not None and sub_id > 0:
try:
new_ws = sessions.create_subagent_worktree(original_workspace_path, sub_id)
os.chdir(new_ws)
workspace_path = new_ws
os.environ["AI_WORKSPACE_PATH"] = workspace_path
except OSError as e:
ui._console.print(f"[red][sys] Sub-agent worktree setup failed: {e}[/red]")
clean_exit(safe_name, workspace_path=original_workspace_path, sub_id=sub_id)
# Wake-on-Key Resonant Ribbon Box
wake_key = ui.draw_session_box(
workspace_path,
home_dir,
is_agent,
db_turns,
mem_count,
memory_active,
active_system_prompt,
clean_name,
sub_id=sub_id,
box_style=st.get("box_style", 1),
wake_on_key=not bool(pending_query),
)
if wake_key:
try:
readline.set_startup_hook(lambda: readline.insert_text(wake_key))
except Exception:
pass
try:
while True:
if pending_query:
query, pending_query = pending_query, None
else:
try:
query = voice.get_prompt_input()
except EOFError:
break
finally:
try:
readline.set_startup_hook(None)
except Exception:
pass
if not query:
continue
ctx = commands.SessionContext(
workspace_path=workspace_path,
home_dir=home_dir,
safe_name=safe_name,
cfg_file=cfg_file,
is_agent=is_agent,
chat_history=chat_history,
active_system_prompt=active_system_prompt,
clean_name=clean_name,
use_map=use_map,
memory_active=memory_active,
is_yolo=is_yolo,
reasoning_active=reasoning_active,
reasoning_budget=reasoning_budget,
show_stats=show_stats,
flash_status=_flash_status,
launch_surface_fn=_launch_surface,
clean_exit_fn=lambda name: clean_exit(
name,
workspace_path=original_workspace_path if (workspace_path != original_workspace_path) else None,
sub_id=sub_id if is_agent else None,
),
)
handled, new_query = commands.dispatch_command(query, ctx)
use_map = ctx.use_map
memory_active = ctx.memory_active
is_yolo = ctx.is_yolo
reasoning_active = ctx.reasoning_active
reasoning_budget = ctx.reasoning_budget
show_stats = ctx.show_stats
if handled:
continue
if new_query:
query = new_query
_clear_transient_status(query)
memory_ctx = memories.get_memory_context(workspace_path) if (is_agent and memory_active) else ""
if RE_THINK_BIN.match(query):
think_bin = next(
(p for p in (f"{CFG_DIR}/modules/agent_chat.py", f"{CFG_DIR}/modules/chat") if os.path.isfile(p)),
None,
)
if think_bin:
try:
res = subprocess.run(
[sys.executable, think_bin, query],
input=json.dumps(chat_history),
text=True,
timeout=120,
)
if res.returncode != 0:
sys.stderr.write(f"\033[1;31m[Warning] chat failed: rc={res.returncode}\033[0m\n")
continue
except Exception as e:
sys.stderr.write(f"\033[1;31m[Warning] chat failed: {e}\033[0m\n")
else:
ui._console.print("[dim yellow][sys] Chat assistant helper not found.[/dim yellow]\n")
continue
sys_ctx = (
""
if query.startswith("init") and "--init" in query
else skills.get_system_context(query, CONTEXT_FILE, STOP_WORDS, SKILLS_DIR, CFG_DIR)
)
if sys_ctx == "__ABORT_TURN__":
ui._console.print("[yellow][sys] Action cancelled.[/yellow]\n")
continue
comb_ctx = "\n\n".join(filter(None, [memory_ctx, sys_ctx]))
prompt = f"<context>\n{comb_ctx}\n</context>\n\n{query}" if comb_ctx else query
chat_history.append({"role": "user", "content": prompt})
try:
readline.add_history(query)
except Exception:
pass
ans = core.stream_response(
chat_history,
prefix="",
show_stats=show_stats,
thinking_budget=reasoning_budget if reasoning_active else 0,
is_agent=is_agent,
)
if not ans:
# Stream failed or interrupted: roll back user turn to prevent context drift
if chat_history and chat_history[-1].get("role") == "user":
chat_history.pop()
continue
clean_ans = RE_THINK_TAGS.sub("", ans).strip()
chat_history.append({"role": "assistant", "content": clean_ans or ans})
try:
tts.speak_response(clean_ans or ans)
except Exception as e:
if os.environ.get("AI_DEBUG") == "1":
sys.stderr.write(f"\r\n[debug] TTS playback error: {e}\r\n")
if not show_stats:
print()
if is_agent:
try:
sessions.log_turn(safe_name, query, ans)
except Exception as e:
if os.environ.get("AI_DEBUG") == "1":
sys.stderr.write(f"\r\n[debug] Failed to log turn to SQLite: {e}\r\n")
if match := RE_AUTO_RUN.search(ans):
try:
readline.set_startup_hook(lambda: readline.insert_text(match.group(1).strip()))
except Exception:
pass
agent_dir = os.path.join(workspace_path, ".agent")
try:
os.makedirs(agent_dir, exist_ok=True)
hist_file = os.path.join(agent_dir, "history.md")
mode = "a" if os.path.exists(hist_file) else "w"
with open(hist_file, mode, encoding="utf-8") as hf:
if mode == "w":
hf.write(f"# Workspace History: {os.path.basename(workspace_path)}\n\n")
hf.write(
f"## [{time.strftime('%Y-%m-%d %H:%M')}] User:\n{query}\n\n### Agent:\n{ans}\n\n---\n\n"
)
except OSError as e:
if os.environ.get("AI_DEBUG") == "1":
sys.stderr.write(f"\r\n[debug] Failed to write history.md: {e}\r\n")
except KeyboardInterrupt:
clean_exit(
safe_name if is_agent else None,
workspace_path=original_workspace_path if (workspace_path != original_workspace_path) else None,
sub_id=sub_id if is_agent else None,
)
def run_direct_query(args: list[str]) -> None:
"""Executes instant single-turn CLI prompt without interactive session loop."""
_load_agent_modules()
query_parts = args[1:]
skill_content = ""
clean_parts: list[str] = []
for part in query_parts:
if part.startswith("-") and part != "-":
loaded = skills.load_skill_content(part.lstrip("-").lower(), SKILLS_DIR, CFG_DIR)
if loaded:
skill_content = loaded
continue
clean_parts.append(part)
query = " ".join(clean_parts).strip()
if query == "-" or (not query and not sys.stdin.isatty()):
try:
query = sys.stdin.read().strip()
except Exception:
query = ""
if not query:
ui._console.print("[dim yellow][sys] Usage: ai --talk <prompt> [-skill][/dim yellow]")
sys.exit(0)
if query.lower() in ("/help", "/h", "help", "--help", "-h"):
ui.show_help()
sys.exit(0)
if "AI_SHOW_THINKING" not in os.environ:
os.environ["AI_SHOW_THINKING"] = "0"
thinking_budget = 0
if env_budget := os.environ.get("AI_THINKING_BUDGET"):
try:
thinking_budget = max(0, int(env_budget.strip()))
except (ValueError, TypeError):
thinking_budget = 0
sys_ctx = skills.get_system_context(query, CONTEXT_FILE, STOP_WORDS, SKILLS_DIR, CFG_DIR)
if sys_ctx == "__ABORT_TURN__":
sys_ctx = ""
active_p = skill_content or BASE_PROMPT_CHAT
inst_p = os.path.join(SKILLS_DIR, "system_instructions.md")
if os.path.isfile(inst_p) and os.path.getsize(inst_p) > 0:
try:
with open(inst_p, "r", encoding="utf-8") as f:
rules = [l for l in f if l.strip() and not l.strip().startswith("#")]
if active_rules := "".join(rules).strip():
active_p += f"\n\n{active_rules}"
except OSError:
pass
messages = [
{"role": "system", "content": active_p},
{"role": "user", "content": f"<context>\n{sys_ctx}\n</context>\n\n{query}" if sys_ctx else query},
]
core.stream_response(messages, prefix="", show_stats=False, thinking_budget=thinking_budget)
sys.exit(0)
def run_matching_search(args: list[str]) -> None:
"""Sub-millisecond shell intent matcher for command not found hook (zero eager module imports)."""
user_input = re.sub(r"[`$]", "", " ".join(args)).strip()
if not user_input or args[0].startswith("--"):
sys.exit(0)
if user_input.lower() in ("/help", "/h"):
_load_agent_modules()
ui.show_help()
sys.exit(0)
shell_name = os.path.basename(os.environ.get("SHELL", "/bin/bash"))
err_msg = f"{'zsh' if 'zsh' in shell_name else 'bash'}: {f'command not found: {user_input}' if 'zsh' in shell_name else f'{user_input}: command not found'}\n"
if RE_SHELL_META.search(user_input):
sys.stderr.write(err_msg)
sys.exit(127)
if matched := context.jaccard_search(user_input, CONTEXT_FILE, STOP_WORDS):
print(
"\n".join(
f"{l.split('|||', 1)[0]}|||{context.clean_tool_prefix(l.split('|||', 1)[1])}"
for l in matched.split("\n")
)
)
sys.exit(0)
sys.stderr.write(err_msg)
sys.exit(127)
def main() -> None:
try:
args = sys.argv[1:]
if not args:
run_direct_query(["--talk"])
elif args[0] == "--interactive" and len(args) >= 2:
_load_agent_modules()
shell_name = os.path.basename(os.environ.get("SHELL", ""))
ui.run_interactive_selection(
" ".join(args[1:]),
lambda q: context.jaccard_search(q, CONTEXT_FILE, STOP_WORDS),
context.clean_tool_prefix,
lambda n: sys.stderr.write(
f"zsh: command not found: {n}\n" if "zsh" in shell_name else f"bash: {n}: command not found\n"
),
lambda: skills.ensure_mysys_exists(SKILLS_DIR, CFG_DIR),
)
sys.exit(0)
elif args[0] in ("--talk", "--talk-chat"):
run_interactive_chat(args) if (args[0] == "--talk-chat" or len(args) == 1) else run_direct_query(args)
sys.exit(0)
else:
run_matching_search(args)
except KeyboardInterrupt:
sys.stderr.write("\nCancelled.\n")
sys.exit(130)
if __name__ == "__main__":
main()