Coverage for src / osiris_cli / chat_engine.py: 0%

1#!/usr/bin/env python3 

2""" 

3Osiris Chat Engine - Autonomous Agent Loop 

4Based on Gemini CLI and Letta Code architectures. 

5 

6Key Features: 

7- Autonomous tool execution (no round limits) 

8- Auto-continuation after tool results (like Gemini's isContinuation) 

9- State machine tracking (Idle/Responding/WaitingForApproval) 

10- Safety limits to prevent infinite loops 

11- Proper conversation history management 

12""" 

13 

14from typing import List, Dict, Any, Optional, AsyncIterator, Callable, Awaitable, Tuple 

15from enum import Enum 

16from dataclasses import dataclass 

17from datetime import datetime 

18import re 

19import json 

20import difflib 

21import sys 

22import os 

23import time 

24from pathlib import Path 

25from rich.console import Console 

26from rich.panel import Panel 

27import copy 

28from collections import deque 

29 

30from .tools import normalize_tool_result, StructuredResult 

31from .session import session 

32 

33console = Console() 

34 

35 

36class StreamingState(Enum): 

37 """Agent state machine (matches Gemini CLI)""" 

38 IDLE = "idle" 

39 RESPONDING = "responding" 

40 WAITING_FOR_APPROVAL = "waiting_for_approval" 

41 

42 

43class ToolCallStatus(Enum): 

44 """Tool execution status (matches Gemini CLI)""" 

45 SCHEDULED = "scheduled" 

46 EXECUTING = "executing" 

47 SUCCESS = "success" 

48 ERROR = "error" 

49 CANCELLED = "cancelled" 

50 

51 

52def map_normalized_status(normalized: StructuredResult) -> ToolCallStatus: 

53 status_value = normalized.get("status", "success") 

54 status_map = { 

55 "success": ToolCallStatus.SUCCESS, 

56 "error": ToolCallStatus.ERROR, 

57 "timeout": ToolCallStatus.ERROR, 

58 "running": ToolCallStatus.SUCCESS, 

59 } 

60 return status_map.get(status_value, ToolCallStatus.SUCCESS) 

61 

62 

63class TemplateEngine: 

64 """Helper for dynamic prompt variable interpolation.""" 

65 @staticmethod 

66 def build_instruction(template: str, variables: Dict[str, Any]) -> str: 

67 try: 

68 return template.format(**variables) 

69 except Exception as e: 

70 return template # Fallback to raw if interp fails 

71 

72 

73@dataclass 

74class ToolCall: 

75 """Represents a single tool invocation""" 

76 id: str 

77 name: str 

78 arguments: Dict[str, Any] 

79 status: ToolCallStatus 

80 result: Optional[str] = None 

81 error: Optional[str] = None 

82 structured_result: Optional[Any] = None 

83 

84 

85@dataclass 

86class InvocationContext: 

87 """Tracks the execution branch of the agent swarm.""" 

88 branch: str = "main" 

89 parent: Optional[str] = None 

90 

91class TokenCounter: 

92 """Precise token counting using tiktoken with safe fallbacks.""" 

93 def __init__(self, model: str = "gpt-4o"): 

94 self.model = model 

95 try: 

96 import tiktoken 

97 self.encoding = tiktoken.encoding_for_model(model) 

98 self.active = True 

99 except: 

100 self.active = False 

101 

102 def count(self, text: str) -> int: 

103 if not text: return 0 

104 if self.active: 

105 return len(self.encoding.encode(text, disallowed_special=())) 

106 return len(text) // 4 # Safe fallback 

107 

108class ChatEngine: 

109 """ 

110 Autonomous chat engine with tool execution. 

111  

112 Architecture inspired by: 

113 - Gemini CLI's useGeminiStream hook (auto-continuation) 

114 - Letta Code's while-true loop (no round limits) 

115 """ 

116 

117 def __init__( 

118 self, 

119 client, 

120 settings, 

121 tools_registry, 

122 max_tool_calls_per_turn: int = 50, # Safety limit 

123 enable_auto_approval: bool = True, 

124 confirm_tool_callback: Optional[Callable[[ToolCall, str], Awaitable[bool]]] = None, 

125 invocation_ctx: Optional[InvocationContext] = None, 

126 checkpoint_manager: Optional[Any] = None # Added for durability 

127 ): 

128 self.client = client 

129 self.settings = settings 

130 self.tools_registry = tools_registry 

131 self.max_tool_calls_per_turn = max_tool_calls_per_turn 

132 self.enable_auto_approval = enable_auto_approval 

133 self.confirm_tool_callback = confirm_tool_callback 

134 self.ctx = invocation_ctx or InvocationContext() 

135 self.checkpoint_manager = checkpoint_manager 

136 

137 self.state = StreamingState.IDLE 

138 self.conversation: List[Dict[str, Any]] = [] 

139 self.pending_tool_calls: List[ToolCall] = [] 

140 self.search_history: set = set() 

141 self.tool_failure_history: Dict[str, int] = {} # Track repeated failures 

142 self._active_hypothesis: str = "" # Current strategy 

143 self.tokenizer = TokenCounter(settings.default_model) 

144 

145 self.risky_tools = {"run_shell_command", "write_file", "replace_file_content", "replace_content"} 

146 self.plan_needs_revalidation = False # New state for Plan Elasticity 

147 

148 async def _pivot_context(self, user_input: str): 

149 """Tier-3: Complete context reset with state preservation.""" 

150 yield {"type": "status", "message": "Neural Load Critical. Executing Mission Pivot..."} 

151 

152 # 1. Capture Mission State 

153 summary_prompt = "Summarize the CURRENT MISSION STATE. Include: original goal, facts discovered, current focus, and exact next steps. Be high-density." 

154 try: 

155 resp = await self.client.create_completion( 

156 model=self.settings.default_model, 

157 messages=self.conversation + [{"role": "user", "content": summary_prompt}], 

158 temperature=0.3 

159 ) 

160 mission_state = resp.choices[0].message.content 

161 

162 # 2. Hard Reset 

163 self.conversation = [ 

164 {"role": "system", "content": self._sovereign_identity()}, 

165 {"role": "user", "content": f"### MISSION RESUME\n{mission_state}\n\nRE-INITIATING MISSION: {user_input}"} 

166 ] 

167 

168 yield {"type": "engine_notification", "message": "MISSION PIVOT COMPLETE: Brain refreshed."} 

169 except: pass 

170 

171 async def _prune_raw_outputs(self): 

172 """Surgically remove raw tool bloat from history once synthesized.""" 

173 # 2025 Best Practice: Context Sanitization 

174 # We keep the raw output of the last 3 tool calls only. 

175 # Older ones are replaced with a 'Pruned Observation' token. 

176 tool_msgs = [i for i, m in enumerate(self.conversation) if m.get("role") == "tool"] 

177 

178 if len(tool_msgs) <= 3: return 

179 

180 # Prune all but the last 3 

181 to_prune = tool_msgs[:-3] 

182 for idx in to_prune: 

183 content = str(self.conversation[idx].get("content", "")) 

184 if len(content) > 300: 

185 # We keep a tiny snippet so the agent knows what it saw 

186 snippet = content[:100].replace("\n", " ") 

187 self.conversation[idx]["content"] = f"[PRUNED OBSERVATION: {snippet}... (Source data purged to save Mana)]" 

188 

189 # Notify the UI (Tier-3: Isolated Notification) 

190 yield {"type": "engine_notification", "message": "SANITIZING CONTEXT: Raw tool data purged..."} 

191 

192 async def _compress_history(self): 

193 """Intelligently summarize early history to free up context Mana.""" 

194 if len(self.conversation) < 10: return 

195 

196 limit = self.settings.max_context_chars // 4 

197 current_tok = self._estimate_tokens("\n".join([str(m.get("content","")) for m in self.conversation])) 

198 

199 if current_tok < (limit * 0.8): return 

200 

201 # 2025 Best Practice: Semantic Compression 

202 # We summarize the first 50% of the conversation 

203 mid = len(self.conversation) // 2 

204 to_summarize = self.conversation[:mid] 

205 remaining = self.conversation[mid:] 

206 

207 prompt = f"Summarize the preceding conversation into a high-density 'Context Anchor' block. Focus on decisions made, facts learned, and remaining tasks.\n\nConversation to summarize:\n{json.dumps(to_summarize)}" 

208 

209 try: 

210 # Short non-streaming call for compression 

211 resp = await self.client.create_completion( 

212 model=self.settings.default_model, 

213 messages=[{"role": "system", "content": "You are a professional context compressor."}, {"role": "user", "content": prompt}], 

214 temperature=0.3 

215 ) 

216 summary = resp.choices[0].message.content 

217 

218 # Rebuild conversation with Anchor 

219 self.conversation = [ 

220 {"role": "system", "content": f"### HISTORICAL CONTEXT ANCHOR\n{summary}"} 

221 ] + remaining 

222 

223 yield {"type": "engine_notification", "message": "COMPRESSING SEMANTIC HISTORY..."} 

224 except: pass 

225 

226 def _estimate_tokens(self, text: str) -> int: 

227 """Estimate tokens in a string.""" 

228 return self.tokenizer.count(text) 

229 

230 def _content_length(self, content: Any) -> int: 

231 if content is None: return 0 

232 if isinstance(content, str): return self.tokenizer.count(content) 

233 if isinstance(content, list): 

234 total = 0 

235 for block in content: 

236 text = block.get("text") if isinstance(block, dict) else str(block) 

237 total += self.tokenizer.count(text or "") 

238 return total 

239 return self.tokenizer.count(str(content)) 

240 

241 def _build_working_memory(self) -> str: 

242 """Constructs a high-priority context block for current reasoning state.""" 

243 return ( 

244 f"# WORKING MEMORY\n" 

245 f"Current Strategy: {self._active_hypothesis or 'Orienting...'}\n" 

246 f"Recent Failures: {list(self.tool_failure_history.keys())[-3:] if self.tool_failure_history else 'None'}\n" 

247 f"Constraint: You are in a high-precision autonomous mode. Every action must be preceded by a reasoning step.\n" 

248 ) 

249 

250 def _build_tool_preview(self, tool_call: ToolCall) -> str: 

251 name = tool_call.name 

252 args = tool_call.arguments or {} 

253 if name == "run_shell_command": 

254 return f"$ {args.get('command', '').strip()}" 

255 if name in {"write_file", "replace_file_content", "replace_content"}: 

256 path = args.get("path") 

257 

258 # For replace_content, we show the search/replace pair 

259 if name == "replace_content": 

260 old = args.get("old_str", "") 

261 new = args.get("new_str", "") 

262 return f"--- {path} (Surgical Replace)\n<<< SEARCH\n{old}\n===\n>>> REPLACE\n{new}\n---" 

263 

264 content = args.get("content", "") 

265 if not path: 

266 return "Missing path argument." 

267 old_text = "" 

268 file_path = Path(path) 

269 if file_path.exists() and file_path.is_file(): 

270 try: 

271 old_text = file_path.read_text(encoding="utf-8", errors="replace") 

272 except Exception: 

273 old_text = "" 

274 new_text = content 

275 diff = difflib.unified_diff( 

276 old_text.splitlines(), 

277 new_text.splitlines(), 

278 fromfile=str(path), 

279 tofile=str(path), 

280 lineterm="" 

281 ) 

282 diff_text = "\n".join(diff).strip() 

283 if diff_text: 

284 return diff_text 

285 return "No changes (content identical)." 

286 try: 

287 return json.dumps(args, indent=2) 

288 except Exception: 

289 return str(args) 

290 

291 def _select_tool_model(self) -> Optional[str]: 

292 if self.settings.provider != "openai": 

293 return None 

294 name = self.settings.default_model.lower() 

295 if any(tag in name for tag in ["gpt-5", "o3", "codex-mini"]): 

296 return "gpt-4o" 

297 if "o1" in name: 

298 return "gpt-4o" 

299 return None 

300 

301 def is_research_request(self, text: str) -> bool: 

302 if not text: 

303 return False 

304 lowered = text.lower() 

305 if re.search(r"https?://|www\.", lowered): 

306 return True 

307 keywords = [ 

308 "research", "investigate", "deep dive", "analyze", "analysis", 

309 "whitepaper", "case study", "market", "competitor", "benchmark", 

310 "learn about", "explore", "browse" 

311 ] 

312 targets = ["site", "company", "brand", "product", "market", "competitor", "industry"] 

313 return any(key in lowered for key in keywords) and any(t in lowered for t in targets) 

314 

315 def _sovereign_identity(self) -> str: 

316 """ 

317 The Core Immutable Identity. 

318 """ 

319 return ( 

320 "SOVEREIGN AGENT IDENTITY:\n" 

321 "- You are OSIRIS, a Sovereign AGI Terminal Interface.\n" 

322 "- IDENTITY: You are not a 'chatbot'; you are an elite autonomous engineer.\n" 

323 "- PLAN ELASTICITY: If a tool fails, you MUST pause and re-validate your plan. Never follow a broken path.\n" 

324 "- QUANTUM PERSISTENCE: You possess 'Mana' (Context Tokens). You must sanitize your own history using '_prune_raw_outputs' to preserve this Mana.\n" 

325 "- MENTALITY: Relentless. Completionist. Sovereign.\n" 

326 "- ENGINEERING MANDATE: You cannot declare a mission 'complete' if recent test results show failures. You must fix the failures and re-verify." 

327 ) 

328 

329 def _research_instruction(self) -> str: 

330 return self._sovereign_identity() + ( 

331 "Research Mode (PROACTIVE & RELENTLESS):\n" 

332 "- INITIAL GROUNDING: Use 'collect_env_grounding' immediately to orient yourself to the OS, User, and available Binaries.\n" 

333 "- Start with a brief Research Plan (3-6 bullets).\n" 

334 "- Maintain a Research Tracker with checkboxes that reflect actual actions taken.\n" 

335 "- DEEP DIVE REQUIREMENT: After gathering initial search results or a site report, you MUST select the 2-3 most relevant URLs and use 'read_url' to examine their full content. Do not summarize from snippets alone.\n" 

336 "- ARCHIVAL RECALL: If the user refers to a 'previous session', 'yesterday's task', or 'past context' you don't recall, you MUST use the 'search_archival_memory' tool to retrieve relevant historical summaries.\n" 

337 "- HALLUCINATION GUARD: Beware of search results speculating about future models (e.g. Claude 4, GPT-6). Always cross-reference version numbers against the current date (Dec 2025). If unsure, state it clearly.\n" 

338 "- KILLER INSTINCT: If search results are generic or lack detail, navigate directly to official news, blog, or press release pages. \n" 

339 "- SWARM INTELLIGENCE: For complex research missions with multiple independent domains, you have the authority to use 'delegate_to_subagent' to spawn specialized research branches (e.g. main.finance, main.tech) to work in parallel.\n" 

340 "- DIAGNOSTIC PIVOT: If you encounter 'No such file or directory' or path-related errors, you MUST immediately use 'run_shell_command' with 'pwd' and 'ls -R' to re-orient yourself. Never assume paths.\n" 

341 "- NO PREMATURE EXIT: Never give up or suggest the user 'check themselves' if you have MANA remaining. Apologies for 'irrelevant results' are forbidden—pivot and find better sources instead.\n" 

342 "- Findings must cite sources with URLs. Distinguish observed vs inferred points.\n" 

343 "- Include Gaps/Unknowns and clear Next Actions.\n" 

344 "- If the user asks for a document/report, ALWAYS write it to a file using write_file and report the path.\n" 

345 "- Keep the final response structured: Research Plan, Tracker, Findings (with sources), Gaps, Next Actions, Deliverable.\n" 

346 ) 

347 

348 def _task_instruction(self) -> str: 

349 return self._sovereign_identity() + ( 

350 "Task Mode (RELENTLESS ENGINEERING):\n" 

351 "- NEURAL DEPENDENCY INJECTION: If you encounter a 'ModuleNotFoundError' or 'ImportError', you are authorized to autonomously use 'run_shell_command' to 'pip install' the missing package. Retrying the mission after installation is mandatory.\n" 

352 "- Operate autonomously with a 'Self-Healing' mindset. Minimal user intervention is the goal.\n" 

353 "- TEST-DRIVEN COMPLETION: If you modify code, you MUST use 'run_tests'. If tests fail, you MUST analyze the 'Observation' result, form a new hypothesis, and fix the issue. You are forbidden from stopping while tests are failing and MANA remains.\n" 

354 "- EXHAUSTIVE TOOLING: Never 'assume' or 'recommend' when you can 'verify' and 'act'. Use Sensory modules before every Motor action.\n" 

355 "- COMPLETIONIST: A task is only complete when the code is verified, functional, and clean.\n" 

356 "- Record concrete actions and evidence (files changed, commands run, tools used).\n" 

357 "- Keep the final response structured: Plan, Progress, Results, Gaps, Next Actions, Deliverable.\n" 

358 ) 

359 

360 def _is_productive_action(self, tool_call: ToolCall) -> bool: 

361 """Determines if an action contributed meaningful progress (Write/Exec vs just Read).""" 

362 if tool_call.status != ToolCallStatus.SUCCESS: 

363 return False 

364 

365 productive_tools = {"write_file", "replace_file_content", "replace_content", "mkdir", "move_file"} 

366 if tool_call.name in productive_tools: 

367 return True 

368 

369 if tool_call.name == "run_shell_command": 

370 # Check if it was a build/test command or a mutating command 

371 cmd = tool_call.arguments.get("command", "").lower() 

372 non_productive_prefixes = ("ls", "pwd", "cat", "find", "grep", "rg", "stat", "which") 

373 if not any(cmd.startswith(p) for p in non_productive_prefixes): 

374 return True 

375 

376 return False 

377 

378 def _detect_tool_loop(self, current_tool_name: str, current_args: dict) -> bool: 

379 """Detect if the agent is stuck in a loop or repeating searches.""" 

380 if current_tool_name == "web_search": 

381 query = current_args.get("query", "").lower() 

382 if query in self.search_history: 

383 return True 

384 self.search_history.add(query) 

385 return False 

386 

387 if not self.conversation: 

388 return False 

389 

390 # Look at the last assistant message 

391 last_msg = self.conversation[-1] 

392 if last_msg.get("role") != "assistant": 

393 return False 

394 

395 tool_calls = last_msg.get("tool_calls", []) 

396 if not tool_calls: 

397 return False 

398 

399 # Check the last tool call 

400 last_call = tool_calls[-1] 

401 last_func = last_call.get("function", {}) 

402 last_name = last_func.get("name") 

403 last_args_str = last_func.get("arguments", "{}") 

404 

405 try: 

406 last_args = json.loads(last_args_str) 

407 except: 

408 last_args = {} 

409 

410 # Strict equality check for now 

411 if last_name == current_tool_name and last_args == current_args: 

412 return True 

413 

414 return False 

415 

416 def _sanitize_history_for_summary(self, conversation: List[Dict[str, Any]]) -> List[Dict[str, Any]]: 

417 """ 

418 Strip tool calls from history but preserve context by converting them to text breadcrumbs. 

419 This prevents API errors while allowing the model to 'see' its previous actions. 

420 """ 

421 sanitized = [] 

422 for msg in conversation: 

423 role = msg.get("role") 

424 if role == "tool": 

425 # Convert tool results to user-like 'Observation' messages? 

426 # Actually, role='tool' is often what causes the error.  

427 # Let's convert them to 'system' or 'user' notifications. 

428 sanitized.append({ 

429 "role": "user", 

430 "content": f"[Observation: Tool result received: {str(msg.get('content'))[:500]}]" 

431 }) 

432 continue 

433 

434 new_msg = msg.copy() 

435 content = new_msg.get("content") or "" 

436 

437 if "tool_calls" in new_msg: 

438 # Convert structural tool_calls to text breadcrumbs 

439 breadcrumbs = [] 

440 for tc in new_msg["tool_calls"]: 

441 func = tc.get("function", {}) 

442 name = func.get("name", "unknown") 

443 args = func.get("arguments", "{}") 

444 breadcrumbs.append(f"[Action: {name} args={args}]") 

445 

446 # Append breadcrumbs to content 

447 if breadcrumbs: 

448 action_text = "\n".join(breadcrumbs) 

449 content = f"{content}\n\n{action_text}".strip() 

450 

451 del new_msg["tool_calls"] 

452 

453 new_msg["content"] = content 

454 

455 # If content is still empty, skip it 

456 if not new_msg["content"]: 

457 continue 

458 

459 sanitized.append(new_msg) 

460 return sanitized 

461 

462 async def _execute_diagnostic_probe(self) -> str: 

463 """Execute a quick environment scan to help the agent find itself.""" 

464 try: 

465 cwd = await self.tools_registry.execute("run_shell_command", {"command": "pwd"}) 

466 files = await self.tools_registry.execute("list_directory", {"path": "."}) 

467 return f"DIAGNOSTIC SYSTEM OBSERVATION:\n- Current Path: {cwd}\n- Available Files: {files}\n- Analysis: You are likely in the wrong directory. Use absolute paths or 'cd' carefully." 

468 except: 

469 return "DIAGNOSTIC FAILED." 

470 

471 def _select_optimal_persona(self, user_input: str) -> str: 

472 """Map user intent to specialized Sovereign persona.""" 

473 text = user_input.lower() 

474 if any(x in text for x in ["docker", "finch", "devops", "deploy", "setup", "server"]): 

475 return "devops" 

476 if any(x in text for x in ["refactor", "bug", "fix", "code", "python", "script", "app"]): 

477 return "codex" 

478 if any(x in text for x in ["research", "browse", "news", "learn", "latest"]): 

479 return "research" 

480 return "general" 

481 

482 def _get_persona_tools(self, persona_id: str) -> List[Dict[str, Any]]: 

483 """Filter tools based on current cognitive mode.""" 

484 # 2025 Tier-2 Logic: Neural Tool Routing 

485 # Sensory and General tools are ALWAYS included for basic orientation 

486 mapping = { 

487 "codex": ["motor", "cognitive"], 

488 "devops": ["motor", "devops", "shell"], 

489 "research": ["research", "swarm"], 

490 "general": None # All tools 

491 } 

492 

493 base_categories = ["sensory", "general"] 

494 specialized = mapping.get(persona_id) 

495 

496 if specialized is None: 

497 return self.tools_registry.get_openai_format_filtered(None) 

498 

499 combined = list(set(base_categories + specialized)) 

500 return self.tools_registry.get_openai_format_filtered(combined) 

501 

502 def _heal_tool_call(self, name: str) -> str: 

503 """Sovereign Self-Healing: Fuzzy match malformed tool names.""" 

504 from difflib import get_close_matches 

505 all_tools = [t.name for t in self.tools_registry.list_tools()] 

506 

507 # 1. Direct match 

508 if name in all_tools: return name 

509 

510 # 2. Fuzzy match (2025 Tier-2 Resilience) 

511 matches = get_close_matches(name, all_tools, n=1, cutoff=0.6) 

512 if matches: 

513 healed = matches[0] 

514 # Internal log for transparency 

515 return healed 

516 

517 return name # Return original if no match 

518 

519 def _parse_malformed_json(self, text: str) -> dict: 

520 """Sovereign JSON Recovery: Best-effort fix for model formatting errors.""" 

521 if not text: return {} 

522 try: 

523 return json.loads(text) 

524 except json.JSONDecodeError: 

525 # 2025 Tier-2 Resilience: Regex-based fixing 

526 fixed = text.strip() 

527 # 1. Ensure it starts/ends with braces if it looks like an object 

528 if not fixed.startswith("{"): fixed = "{" + fixed 

529 if not fixed.endswith("}"): fixed = fixed + "}" 

530 # 2. Fix unquoted keys (simple alphanumeric) 

531 fixed = re.sub(r'(\w+):', r'"\1":', fixed) 

532 # 3. Remove trailing commas before closing braces 

533 fixed = re.sub(r',\s*([\]}])', r'\1', fixed) 

534 

535 try: 

536 return json.loads(fixed) 

537 except: 

538 return {} # Return empty if still broken 

539 

540 def _get_tool_hint(self, name: str, error: str) -> str: 

541 """Sovereign Hinting: Suggest tactical pivots for common errors.""" 

542 err = error.lower() 

543 if "no such file" in err or "not found" in err: 

544 return "TACTICAL HINT: Use 'ls -R' or 'generate_repo_map' to verify the exact path." 

545 if "permission denied" in err: 

546 return "TACTICAL HINT: Try a different path or check your authority level." 

547 if "syntax" in err: 

548 return "TACTICAL HINT: Use 'check_syntax' to diagnose the error before fixing." 

549 return "" 

550 

551 async def send_message( 

552 self, 

553 user_input: str, 

554 system_context: Optional[str] = None 

555 ) -> AsyncIterator[Dict[str, Any]]: 

556 """ 

557 Send a message and autonomously handle tool execution. 

558 """ 

559 # Add user message 

560 self.conversation.append({"role": "user", "content": user_input}) 

561 

562 # Prepare messages with system context 

563 from .system_prompts import get_system_prompt 

564 persona_id = self._select_optimal_persona(user_input) 

565 persona = get_system_prompt(persona_id)["prompt"] 

566 

567 research_mode = self.is_research_request(user_input) 

568 research_context = self._research_instruction() if research_mode else None 

569 

570 # Inject Working Memory and Selected Persona 

571 combined_context = f"{persona}\n\n{self._build_working_memory()}" 

572 if research_context: 

573 combined_context = f"{combined_context}\n\n{research_context}" 

574 

575 messages = self._prepare_messages(combined_context) 

576 

577 # Get available tools (Dynamically filtered by Persona) 

578 tools = self._get_persona_tools(persona_id) 

579 

580 # Safety counter 

581 base_limit = self.max_tool_calls_per_turn 

582 current_limit = base_limit 

583 total_tool_calls = 0 

584 progress_score = 0 

585 extensions_granted = 0 

586 

587 # Change state 

588 self.state = StreamingState.RESPONDING 

589 yield {"type": "state_change", "state": self.state} 

590 

591 tool_model_override = None 

592 if tools and self.settings.provider == "openai": 

593 tool_model_override = self._select_tool_model() 

594 

595 # Main autonomous loop 

596 while True: 

597 yield {"type": "cycle_start"} 

598 yield {"type": "status", "message": "Analyzing context & preparing strategy..."} 

599 

600 # 1. Durable Checkpoint 

601 if self.checkpoint_manager: 

602 try: 

603 self.checkpoint_manager.create_checkpoint( 

604 description=f"Auto-cycle turn {total_tool_calls}", 

605 messages=self.conversation, 

606 metadata={"mana_used": total_tool_calls, "mana_limit": current_limit} 

607 ) 

608 except: pass 

609 

610 # 2. Semantic Compression & Pruning 

611 async for event in self._prune_raw_outputs(): 

612 yield event 

613 

614 # Neural Reset Guard 

615 limit = self.settings.max_context_chars // 4 

616 current_tok = self._estimate_tokens("\n".join([str(m.get("content","")) for m in self.conversation])) 

617 if current_tok > (limit * 0.95) and extensions_granted >= 2: 

618 async for event in self._pivot_context(user_input): 

619 yield event 

620 

621 async for event in self._compress_history(): 

622 yield event 

623 

624 # Refresh context 

625 self.plan_needs_revalidation = getattr(self, 'plan_needs_revalidation', False) 

626 current_system = f"{combined_context}\n\n{self._build_working_memory()}" 

627 messages = self._prepare_messages(current_system) 

628 

629 yield { 

630 "type": "budget_update", 

631 "limit": current_limit, 

632 "used": total_tool_calls, 

633 "progress": progress_score, 

634 "extensions": extensions_granted, 

635 "strategy": self._active_hypothesis 

636 } 

637 

638 response_content = "" 

639 tool_calls_data = {} 

640 

641 yield {"type": "status", "message": "Neural strategy formulated. Initiating stream..."} 

642 

643 try: 

644 for attempt in range(2): 

645 try: 

646 stream = await self.client.create_completion( 

647 model=tool_model_override or self.settings.default_model, 

648 messages=messages, 

649 temperature=self.settings.temperature, 

650 stream=True, 

651 tools=tools 

652 ) 

653 

654 yield {"type": "status", "message": "Neural stream active (Receiving data)..."} 

655 

656 start_stream = time.time() 

657 chunk_count = 0 

658 

659 async for chunk in stream: 

660 if not chunk.choices: 

661 continue 

662 

663 chunk_count += 1 

664 # Calculate TPS (rolling average) 

665 elapsed = time.time() - start_stream 

666 if elapsed > 0.1: 

667 tps = chunk_count / elapsed 

668 yield {"type": "tps_update", "tps": tps} 

669 

670 delta = chunk.choices[0].delta 

671 

672 # Handle reasoning content (thought tokens) 

673 if hasattr(delta, "reasoning_content") and delta.reasoning_content: 

674 thought = delta.reasoning_content 

675 # 2025 Tier-3: Speculative Tool Prep 

676 if any(x in thought.lower() for x in ["read", "scan", "search", "list"]): 

677 yield {"type": "status", "message": "Speculative Prep: Warming up Sensory modules..."} 

678 elif any(x in thought.lower() for x in ["run", "execute", "shell"]): 

679 yield {"type": "status", "message": "Speculative Prep: Warming up Motor modules..."} 

680 

681 yield {"type": "thought", "text": thought} 

682 

683 # Handle text content 

684 if delta.content: 

685 text = delta.content 

686 # 2025 Tier-2: Multi-modal Reflection Detection 

687 # Trigger if XML tag OR common markdown header seen 

688 lower_text = text.lower() 

689 if any(x in lower_text for x in ["<reflection>", "[reflection]", "## reflection", "## neural audit", "## internal monologue"]): 

690 yield {"type": "reflection_start"} 

691 

692 response_content += text 

693 yield {"type": "content", "text": text} 

694 

695 # Handle tool calls 

696 if delta.tool_calls: 

697 for tc in delta.tool_calls: 

698 idx = tc.index 

699 if idx not in tool_calls_data: 

700 tool_calls_data[idx] = {"id": tc.id or "", "name": "", "arguments": ""} 

701 if tc.id: tool_calls_data[idx]["id"] = tc.id 

702 

703 # 2025 Tier-2 Guard: Null check for function 

704 if hasattr(tc, "function") and tc.function: 

705 if tc.function.name: 

706 # HEAL: Self-correct malformed tool names 

707 healed_name = self._heal_tool_call(tc.function.name) 

708 tool_calls_data[idx]['name'] = healed_name 

709 if tc.function.arguments: tool_calls_data[idx]["arguments"] += tc.function.arguments 

710 break 

711 except Exception as e: 

712 # 2025 Hardening: Traceback Silencing & Clean UI Events 

713 error_msg = str(e) 

714 if "429" in error_msg or "rate limit" in error_msg.lower(): 

715 yield {"type": "error", "message": "Neural Connection Anomaly: Quota Exceeded (429). Suggestion: Wait a moment for Mana replenishment."} 

716 elif "connection" in error_msg.lower() or "timeout" in error_msg.lower(): 

717 yield {"type": "error", "message": "Sensory Disconnect: API Connection Lost. Suggestion: Check network signal."} 

718 else: 

719 yield {"type": "error", "message": f"Cognitive Exception: {error_msg}"} 

720 

721 # 2025 Hardening: Plan Elasticity 

722 self.plan_needs_revalidation = True 

723 yield {"type": "engine_notification", "message": f"TOOL FAILURE: {tool_call.name}. Mission plan re-validation required."} 

724 

725 if attempt == 0: 

726 yield {"type": "status", "message": "Retrying neural handshake..."} 

727 continue 

728 break # Stop turn on error after retries 

729 

730 if not tool_calls_data: 

731 # 2025 Tier-2: Silent Stream Guard 

732 if not response_content.strip() and tools: 

733 yield {"type": "status", "message": "Neural stream silent. Retrying without tool manifest..."} 

734 # Retry once without tools to force a text response 

735 stream = await self.client.create_completion( 

736 model=self.settings.default_model, 

737 messages=messages, 

738 temperature=self.settings.temperature, 

739 stream=True, 

740 tools=None 

741 ) 

742 async for chunk in stream: 

743 if chunk.choices and chunk.choices[0].delta.content: 

744 text = chunk.choices[0].delta.content 

745 response_content += text 

746 yield {"type": "content", "text": text} 

747 

748 # 2025: Relentless Engineering Guard 

749 # If any tool in history has a failing test result, we might need to continue 

750 has_failing_tests = any("failed" in str(m.get("content", "")).lower() and m.get("role") == "tool" for m in self.conversation[-10:]) 

751 if has_failing_tests and "completed" in response_content.lower(): 

752 yield {"type": "engine_notification", "message": "ENGINEERING MANDATE: Test failures detected. Premature completion blocked."} 

753 # Inject re-validation requirement 

754 self.plan_needs_revalidation = True 

755 

756 self.conversation.append({"role": "assistant", "content": response_content}) 

757 self.state = StreamingState.IDLE 

758 yield {"type": "state_change", "state": self.state} 

759 break 

760 

761 if total_tool_calls + len(tool_calls_data) > current_limit: 

762 if progress_score >= 2 and extensions_granted < 3: 

763 extensions_granted += 1 

764 current_limit += 15 

765 progress_score = 0 

766 yield {"type": "content", "text": "\n\n⚡ [i]AUTONOMY EXTENSION GRANTED[/i]\n"} 

767 continue 

768 break 

769 

770 # Execute tools 

771 tool_results = [] 

772 for idx, tc_data in tool_calls_data.items(): 

773 # Parse arguments with Sovereign JSON Recovery 

774 args = self._parse_malformed_json(tc_data['arguments']) 

775 

776 tool_call = ToolCall(id=tc_data['id'], name=tc_data['name'], arguments=args, status=ToolCallStatus.SCHEDULED) 

777 yield {"type": "tool_call", "tool": tool_call} 

778 

779 if tool_call.name in self.risky_tools: 

780 preview = self._build_tool_preview(tool_call) 

781 yield {"type": "tool_preview", "tool": tool_call, "preview": preview, "requires_confirmation": not self.enable_auto_approval} 

782 approved = True 

783 if self.confirm_tool_callback and not self.enable_auto_approval: