var SAFE_COMMANDSThe type of the None singleton.
-def ask(self,
user_input,
dryrun=False,
chat_history=None,
status=None,
debug=False,
stream=True)
+def ask(self,
user_input,
dryrun=False,
chat_history=None,
status=None,
debug=False,
stream=True,
session_id=None)
@MethodHook
-def ask(self, user_input, dryrun=False, chat_history=None, status=None, debug=False, stream=True):
+def ask(self, user_input, dryrun=False, chat_history=None, status=None, debug=False, stream=True, session_id=None):
if chat_history is None: chat_history = []
+
+ # Load session if provided and history is empty
+ if session_id and not chat_history:
+ session_data = self.load_session_data(session_id)
+ if session_data:
+ chat_history = session_data.get("history", [])
+ # If we loaded history, the caller might need it back
+ # But typically ask() is called in a loop with an external history object
+
usage = {"input": 0, "output": 0, "total": 0}
# 1. Selector de Rol inicial (Sticky Brain)
@@ -1756,15 +1896,20 @@ def ask(self, user_input, dryrun=False, chat_history=None, status=None, debug=Fa
model = self.architect_model if current_brain == "architect" else self.engineer_model
key = self.architect_key if current_brain == "architect" else self.engineer_key
- # Estructura optimizada para Prompt Caching
- if "claude" in model.lower():
+ # Estructura optimizada para Prompt Caching (Solo para Anthropic directo, Vertex tiene reglas distintas)
+ if "claude" in model.lower() and "vertex" not in model.lower():
messages = [{"role": "system", "content": [{"type": "text", "text": system_prompt, "cache_control": {"type": "ephemeral"}}]}]
else:
messages = [{"role": "system", "content": system_prompt}]
# Interleaving de historial
last_role = "system"
- for msg in chat_history[-self.max_history:]:
+ # Sanitize history if the current target model is not compatible with cache_control
+ history_to_process = chat_history[-self.max_history:]
+ if "claude" not in model.lower() or "vertex" in model.lower():
+ history_to_process = self._sanitize_messages(history_to_process)
+
+ for msg in history_to_process:
m = msg if isinstance(msg, dict) else msg.model_dump(exclude_none=True)
role = m.get('role')
if role == last_role and role == 'user':
@@ -1792,7 +1937,7 @@ def ask(self, user_input, dryrun=False, chat_history=None, status=None, debug=Fa
console.print(f"[yellow] You can press Ctrl+C to interrupt and get a summary of progress.[/yellow]")
soft_limit_warned = True
- label = "[bold purple]Architect" if current_brain == "architect" else "[bold blue]Engineer"
+ label = "[bold medium_purple]Architect" if current_brain == "architect" else "[bold blue]Engineer"
if status: status.update(f"{label} is thinking... (step {iteration})")
streamed_response = False
@@ -1837,7 +1982,7 @@ def ask(self, user_input, dryrun=False, chat_history=None, status=None, debug=Fa
messages.append(msg_dict)
if debug and resp_msg.content:
- console.print(Panel(Markdown(resp_msg.content), title=f"{label} Reasoning", border_style="purple" if current_brain == "architect" else "blue"))
+ console.print(Panel(Markdown(resp_msg.content), title=f"{label} Reasoning", border_style="medium_purple" if current_brain == "architect" else "blue"))
if not resp_msg.tool_calls: break
@@ -1854,8 +1999,8 @@ def ask(self, user_input, dryrun=False, chat_history=None, status=None, debug=Fa
continue
if status:
- if fn == "delegate_to_engineer": status.update(f"[bold purple]Architect: [DELEGATING MISSION] {args.get('task','')[:40]}...")
- elif fn == "manage_memory_tool": status.update(f"[bold purple]Architect: [UPDATING MEMORY]")
+ if fn == "delegate_to_engineer": status.update(f"[bold medium_purple]Architect: [DELEGATING MISSION] {args.get('task','')[:40]}...")
+ elif fn == "manage_memory_tool": status.update(f"[bold medium_purple]Architect: [UPDATING MEMORY]")
if debug: console.print(Panel(Text(json.dumps(args, indent=2)), title=f"{label} Decision: {fn}", border_style="white"))
@@ -1863,7 +2008,7 @@ def ask(self, user_input, dryrun=False, chat_history=None, status=None, debug=Fa
obs, eng_usage = self._engineer_loop(args["task"], status=status, debug=debug, chat_history=messages[:-1])
usage["input"] += eng_usage["input"]; usage["output"] += eng_usage["output"]; usage["total"] += eng_usage["total"]
elif fn == "consult_architect":
- if status: status.update("[bold purple]Engineer consulting Architect...")
+ if status: status.update("[bold medium_purple]Engineer consulting Architect...")
try:
# Consultation only - Engineer stays in control
claude_resp = completion(
@@ -1876,13 +2021,13 @@ def ask(self, user_input, dryrun=False, chat_history=None, status=None, debug=Fa
num_retries=3
)
obs = claude_resp.choices[0].message.content
- if debug: console.print(Panel(Markdown(obs), title="[bold purple]Architect Consultation[/bold purple]", border_style="purple"))
+ if debug: console.print(Panel(Markdown(obs), title="[bold medium_purple]Architect Consultation[/bold medium_purple]", border_style="medium_purple"))
except Exception as e:
if status: status.update("[bold orange3]Architect unavailable! Engineer continuing alone...")
obs = f"Architect unavailable ({str(e)}). Proceeding with your best technical judgment."
elif fn == "escalate_to_architect":
- if status: status.update("[bold purple]Transferring control to Architect...")
+ if status: status.update("[bold medium_purple]Transferring control to Architect...")
# Full escalation - Architect takes over
current_brain = "architect"
model = self.architect_model
@@ -1893,7 +2038,7 @@ def ask(self, user_input, dryrun=False, chat_history=None, status=None, debug=Fa
handover_msg = f"HANDOVER FROM EXECUTION ENGINE\n\nReason: {args['reason']}\n\nContext: {args['context']}\n\nYou are now in control of this conversation."
pending_user_message = handover_msg
obs = "Control transferred to Architect. Handover context will be provided."
- if debug: console.print(Panel(Text(handover_msg), title="[bold purple]Escalation to Architect[/bold purple]", border_style="purple"))
+ if debug: console.print(Panel(Text(handover_msg), title="[bold medium_purple]Escalation to Architect[/bold medium_purple]", border_style="medium_purple"))
elif fn == "return_to_engineer":
if status: status.update("[bold blue]Transferring control back to Engineer...")
@@ -1951,19 +2096,8 @@ def ask(self, user_input, dryrun=False, chat_history=None, status=None, debug=Fa
messages.append(resp_msg.model_dump(exclude_none=True))
except Exception: pass
finally:
- try:
- log_dir = self.config.defaultdir
- os.makedirs(log_dir, exist_ok=True)
- log_path = os.path.join(log_dir, "ai_debug.json")
- hist = []
- if os.path.exists(log_path):
- try:
- with open(log_path, "r") as f: hist = json.load(f)
- except (IOError, json.JSONDecodeError): hist = []
- hist.append({"timestamp": datetime.datetime.now().isoformat(), "roles": {"strategic_engine": self.architect_model, "execution_engine": self.engineer_model}, "session": messages})
- with open(log_path, "w") as f: json.dump(hist[-10:], f, indent=4)
- except Exception as e:
- if debug: console.print(f"[dim red]Debug log failed: {e}[/dim red]")
+ # Auto-save session
+ self.save_session(messages, model=model)
return {
"response": messages[-1].get("content"),
@@ -1989,6 +2123,40 @@ def confirm(self, user_input): return True
+def delete_session(self, session_id)
+def delete_session(self, session_id):
+ """Deletes a session by ID."""
+ path = os.path.join(self.sessions_dir, f"{session_id}.json")
+ if os.path.exists(path):
+ os.remove(path)
+ printer.success(f"Session {session_id} deleted.")
+ else:
+ printer.error(f"Session {session_id} not found.")Deletes a session by ID.
+def get_last_session_id(self)
+def get_last_session_id(self):
+ """Returns the ID of the most recent session."""
+ sessions = self._get_sessions()
+ return sessions[0]["id"] if sessions else NoneReturns the ID of the most recent session.
def get_node_info_tool(self, node_name)
List nodes matching the filter pattern. Returns metadata for <=5 nodes, names only for more.
+def list_sessions(self)
+def list_sessions(self):
+ """Prints a list of sessions using printer.table."""
+ sessions = self._get_sessions()
+ if not sessions:
+ printer.info("No saved AI sessions found.")
+ return
+
+ columns = ["ID", "Title", "Created At", "Model"]
+ rows = [[s["id"], s["title"], s["created_at"], s["model"]] for s in sessions]
+ printer.table("AI Persisted Sessions", columns, rows)Prints a list of sessions using printer.table.
+def load_session_data(self, session_id)
+def load_session_data(self, session_id):
+ """Loads a session's raw data by ID."""
+ path = os.path.join(self.sessions_dir, f"{session_id}.json")
+ if os.path.exists(path):
+ try:
+ with open(path, "r") as f:
+ data = json.load(f)
+ self.session_id = session_id
+ self.session_path = path
+ return data
+ except Exception as e:
+ printer.error(f"Failed to load session {session_id}: {e}")
+ return NoneLoads a session's raw data by ID.
def manage_memory_tool(self, content, action='append')
Execute commands on nodes matching the filter. Native interactive confirmation for unsafe commands.
+def save_session(self, history, title=None, model=None)
+def save_session(self, history, title=None, model=None):
+ """Saves current history to the session file."""
+ if not self.session_id:
+ # Generate ID from first user query if available
+ first_user_msg = next((m["content"] for m in history if m["role"] == "user"), "new-session")
+ self.session_id = self._generate_session_id(first_user_msg)
+ self.session_path = os.path.join(self.sessions_dir, f"{self.session_id}.json")
+
+ # If it's a new file, we might want to set a better title
+ if not os.path.exists(self.session_path) and not title:
+ raw_title = next((m["content"] for m in history if m["role"] == "user"), "New Session")
+ # Clean title: remove newlines, multiple spaces
+ clean_title = " ".join(raw_title.split())
+ if len(clean_title) > 40:
+ title = clean_title[:37].strip() + "..."
+ else:
+ title = clean_title
+
+ try:
+ # Read existing metadata if it exists
+ metadata = {}
+ if os.path.exists(self.session_path):
+ with open(self.session_path, "r") as f:
+ metadata = json.load(f)
+
+ metadata.update({
+ "id": self.session_id,
+ "title": title or metadata.get("title", "New Session"),
+ "created_at": metadata.get("created_at", datetime.datetime.now().isoformat()),
+ "updated_at": datetime.datetime.now().isoformat(),
+ "model": model or metadata.get("model", self.engineer_model),
+ "history": history
+ })
+
+ with open(self.session_path, "w") as f:
+ json.dump(metadata, f, indent=4)
+ except Exception as e:
+ printer.error(f"Failed to save session: {e}")
+
+ except Exception as e:
+ printer.error(f"Failed to save session: {e}")Saves current history to the session file.
@@ -2248,27 +2513,31 @@ class configfile:
'''
home = os.path.expanduser("~")
defaultdir = home + '/.config/conn'
- self.defaultdir = defaultdir
- Path(defaultdir).mkdir(parents=True, exist_ok=True)
- Path(f"{defaultdir}/plugins").mkdir(parents=True, exist_ok=True)
- pathfile = defaultdir + '/.folder'
- try:
- with open(pathfile, "r") as f:
- configdir = f.read().strip()
- except (FileNotFoundError, IOError):
- with open(pathfile, "w") as f:
- f.write(str(defaultdir))
- configdir = defaultdir
- defaultfile = configdir + '/config.yaml'
- self.cachefile = configdir + '/.config.cache.json'
- self.fzf_cachefile = configdir + '/.fzf_nodes_cache.txt'
- self.folders_cachefile = configdir + '/.folders_cache.txt'
- self.profiles_cachefile = configdir + '/.profiles_cache.txt'
- defaultkey = configdir + '/.osk'
- if conf == None:
- self.file = defaultfile
+
+ if conf is None:
+ # Standard path: use ~/.config/conn and respect .folder redirection
+ self.anchor_path = defaultdir
+ self.defaultdir = defaultdir
+ Path(defaultdir).mkdir(parents=True, exist_ok=True)
- # Backwards compatibility: Migrate from JSON to YAML
+ pathfile = defaultdir + '/.folder'
+ try:
+ with open(pathfile, "r") as f:
+ configdir = f.read().strip()
+ except (FileNotFoundError, IOError):
+ with open(pathfile, "w") as f:
+ f.write(str(defaultdir))
+ configdir = defaultdir
+
+ self.defaultdir = configdir
+ self.file = configdir + '/config.yaml'
+ self.key = key or (configdir + '/.osk')
+
+ # Ensure redirected directories exist
+ Path(configdir).mkdir(parents=True, exist_ok=True)
+ Path(f"{configdir}/plugins").mkdir(parents=True, exist_ok=True)
+
+ # Backwards compatibility: Migrate from JSON to YAML only for default path
legacy_json = configdir + '/config.json'
legacy_noext = configdir + '/config'
legacy_file = None
@@ -2291,38 +2560,44 @@ class configfile:
os.remove(self.file)
printer.warning("YAML verification failed after migration, keeping legacy config.")
else:
- with open(self.cachefile, 'w') as f:
+ # Note: cachefile is derived later, we use temp one for migration sync
+ temp_cache = configdir + '/.config.cache.json'
+ with open(temp_cache, 'w') as f:
json.dump(old_data, f)
shutil.move(legacy_file, legacy_file + ".backup")
printer.success(f"Migrated legacy config ({len(old_data.get('connections',{}))} folders/nodes) into YAML and Cache successfully!")
except Exception as e:
- # Clean up partial YAML if it was created
if os.path.exists(self.file):
- try:
- os.remove(self.file)
- except OSError:
- pass
+ try: os.remove(self.file)
+ except OSError: pass
printer.warning(f"Failed to migrate legacy config: {e}")
else:
- self.file = conf
-
- if key == None:
- self.key = defaultkey
- else:
- self.key = key
+ # Custom path (common in tests): isolate everything to the conf parent directory
+ self.file = os.path.abspath(conf)
+ configdir = os.path.dirname(self.file)
+ self.anchor_path = configdir
+ self.defaultdir = configdir
+ self.key = os.path.abspath(key) if key else (configdir + '/.osk')
+
+ # Sidecar files always live next to the config file (or in the redirected configdir)
+ self.cachefile = configdir + '/.config.cache.json'
+ self.fzf_cachefile = configdir + '/.fzf_nodes_cache.txt'
+ self.folders_cachefile = configdir + '/.folders_cache.txt'
+ self.profiles_cachefile = configdir + '/.profiles_cache.txt'
if os.path.exists(self.file):
config = self._loadconfig(self.file)
else:
config = self._createconfig(self.file)
+
self.config = config["config"]
self.connections = config["connections"]
self.profiles = config["profiles"]
+
if not os.path.exists(self.key):
self._createkey(self.key)
with open(self.key) as f:
self.privatekey = RSA.import_key(f.read())
- f.close()
self.publickey = self.privatekey.publickey()
# Self-heal text caches if they are missing
@@ -4724,12 +4999,17 @@ def test(self, commands, expected, vars = None,*, prompt = None, parallel = 10,
architect_system_promptaskconfirmdelete_sessionengineer_system_promptget_last_session_idget_node_info_toollist_nodes_toollist_sessionsload_session_datamanage_memory_toolregister_ai_toolrun_commands_toolsave_session
@@ -4761,7 +5041,7 @@ def test(self, commands, expected, vars = None,*, prompt = None, parallel = 10,