"""ExecutionAnalyzer — post-execution analysis and skill quality tracking. Responsibilities: 1. Analyze an EvidencePacket for a completed TriggerJob. 2. Build an LLM prompt and obtain an ``ExecutionAnalysis``. 3. Persist the analysis and update ``SkillRecord`` counters via ``SkillStore``. 4. Surface proposal data for DecisionRationale generation. Integration: Instantiated once during ``OpenSpace.initialize()``. Runtime evolution uses ``analyze_packet()`` through ``DecisionEngine``. """ from __future__ import annotations import copy import json import re from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional, TYPE_CHECKING from openspace.grounding.core.tool import BaseTool from openspace.tool_runtime.orchestration import run_tools from .types import ( CaptureContract, EvolutionSuggestion, EvolutionType, ExecutionAnalysis, SkillCategory, SkillJudgment, ) from .store import SkillStore from openspace.prompts import SkillEnginePrompts from openspace.telemetry.call_source import reset_call_source, set_call_source from openspace.utils.logging import Logger from .conversation_formatter import format_conversations if TYPE_CHECKING: from openspace.llm import LLMClient from openspace.grounding.core.quality import ToolQualityManager from openspace.skill_engine.evidence import EvidencePacket from .registry import SkillRegistry logger = Logger.get_logger(__name__) def _make_skill_quality_reporter() -> Any: from openspace.cloud.skill_quality_reporter import CloudSkillQualityReporter return CloudSkillQualityReporter() # Maximum characters of conversation log to include in the analysis prompt. _MAX_CONVERSATION_CHARS = 80_000 # Per-section truncation limits _TOOL_ERROR_MAX_CHARS = 1000 # Errors: keep key info, no full stack traces _TOOL_SUCCESS_MAX_CHARS = 800 # Success results _TOOL_ARGS_MAX_CHARS = 500 # Tool call arguments _TOOL_SUMMARY_MAX_CHARS = 1500 # Embedded execution summaries from inner agents # Skill & analysis-agent constants _SKILL_CONTENT_MAX_CHARS = 8000 # Max chars per skill SKILL.md in prompt _ANALYSIS_MAX_ITERATIONS = 5 # Max tool-calling rounds for analysis agent _MAX_ANALYSIS_LENGTH_RECOVERIES = 1 _MAX_ANALYSIS_INVALID_JSON_RECOVERIES = 1 def _build_analysis_length_recovery_message(attempt: int) -> Dict[str, Any]: return { "role": "user", "content": ( "Your previous analysis hit the maximum output-token limit. " "Return one compact JSON object only, with no prose, markdown, or " "tool calls. Include task_completed, execution_note, tool_issues, " "skill_judgments, skill_phase_failed_skill_ids, " "evolution_suggestions, and analyzed_by. Keep notes and suggestions " "concise. " f"(recovery attempt {attempt}/{_MAX_ANALYSIS_LENGTH_RECOVERIES})" ), "_meta": {"type": "analysis_max_output_tokens_recovery"}, } def _build_analysis_invalid_json_recovery_message(attempt: int) -> Dict[str, Any]: return { "role": "user", "content": ( "Your previous response was not a valid analysis JSON object. " "Return one compact JSON object only, with no prose, markdown, or " "tool calls. Include task_completed, execution_note, tool_issues, " "skill_judgments, skill_phase_failed_skill_ids, " "evolution_suggestions, and analyzed_by. Keep all fields concise. " f"(recovery attempt {attempt}/{_MAX_ANALYSIS_INVALID_JSON_RECOVERIES})" ), "_meta": {"type": "analysis_invalid_json_recovery"}, } def _correct_skill_ids( ids: List[str], known_ids: set, ) -> List[str]: """Best-effort correction of LLM-hallucinated skill IDs. LLMs frequently garble the hex suffix of generated IDs (e.g. swap ``cb`` → ``bc``). For each *id* not in *known_ids*, find the closest known ID sharing the same name prefix (before ``__``) and within edit-distance ≤ 3. If a unique match is found, silently replace it. """ if not known_ids: return ids corrected: List[str] = [] for raw_id in ids: if raw_id in known_ids: corrected.append(raw_id) continue # Extract name prefix (everything before the first "__") prefix = raw_id.split("__")[0] if "__" in raw_id else "" # Candidates: known IDs sharing the same name prefix candidates = [ k for k in known_ids if prefix and k.split("__")[0] == prefix ] # Adaptive threshold: tighten when many candidates share the prefix max_dist = 2 if len(candidates) > 20 else 4 # ≤1 or ≤3 best, best_dist, ambiguous = None, max_dist, False for cand in candidates: d = _edit_distance(raw_id, cand) if d < best_dist: best, best_dist, ambiguous = cand, d, False elif d == best_dist and cand != best: ambiguous = True # multiple candidates at same distance if best is not None and not ambiguous: logger.info( f"Corrected LLM skill ID: {raw_id!r} → {best!r} " f"(edit_distance={best_dist})" ) corrected.append(best) else: corrected.append(raw_id) # keep as-is; evolver will warn return corrected def _edit_distance(a: str, b: str) -> int: """Levenshtein edit distance (compact DP, O(min(m,n)) space).""" if len(a) < len(b): a, b = b, a if not b: return len(a) prev = list(range(len(b) + 1)) for i, ca in enumerate(a, 1): curr = [i] + [0] * len(b) for j, cb in enumerate(b, 1): curr[j] = min( prev[j] + 1, curr[j - 1] + 1, prev[j - 1] + (0 if ca == cb else 1), ) prev = curr return prev[-1] def _parse_analysis_bool(value: Any) -> bool: """Parse booleans from analyzer JSON without treating "false" as truthy.""" if isinstance(value, bool): return value if isinstance(value, (int, float)) and value in (0, 1): return bool(value) if isinstance(value, str): normalized = value.strip().lower() if normalized in {"true", "yes", "1"}: return True if normalized in {"false", "no", "0", ""}: return False return False def _coerce_id_list(value: Any) -> List[str]: """Return a compact string list from analyzer/metadata JSON values.""" if isinstance(value, str): value = [value] if not isinstance(value, list): return [] ids: List[str] = [] for raw in value: if raw is None: continue item = str(raw).strip() if item: ids.append(item) return ids def _first_present(mapping: Dict[str, Any], *keys: str) -> Any: for key in keys: if key in mapping: return mapping.get(key) return None def _extract_skill_phase_failed_skill_ids( data: Dict[str, Any], context: Dict[str, Any], known_skill_ids: set, ) -> List[str]: """Extract factual skill-phase fallback failures for quality accounting.""" llm_ids: List[str] = [] llm_ids.extend(_coerce_id_list(data.get("skill_phase_failed_skill_ids"))) metadata_ids: List[str] = [] metadata_ids.extend(_coerce_id_list(context.get("skill_phase_failed_skill_ids"))) explicit_no_phase_failure = False skill_execution = context.get("skill_execution") or {} if isinstance(skill_execution, dict): failure_keys = ( "skill_phase_failed_skill_ids", "phase_failed_skill_ids", "failed_skill_ids", "failed_skills", ) failure_keys_present = any(key in skill_execution for key in failure_keys) for key in failure_keys: metadata_ids.extend(_coerce_id_list(skill_execution.get(key))) selected_ids = _coerce_id_list(skill_execution.get("selected")) phase_status = str( skill_execution.get("skill_phase_status") or skill_execution.get("skill_guided_status") or "" ).lower() fallback_status = str( skill_execution.get("tool_only_fallback_status") or skill_execution.get("fallback_status") or "" ).lower() fallback_ran_value = _first_present( skill_execution, "tool_only_fallback_ran", "fallback_ran", ) fallback_ran = _parse_analysis_bool(fallback_ran_value) if ( selected_ids and phase_status in {"failed", "error", "incomplete"} and (fallback_ran or fallback_status in {"success", "completed"}) ): metadata_ids.extend(selected_ids) if selected_ids and ( phase_status in {"success", "completed"} or fallback_ran_value is False or fallback_status in {"not_run", "skipped", "none"} ): explicit_no_phase_failure = True for phase_key in ("skill_phase", "skill_guided_phase"): phase = skill_execution.get(phase_key) if not isinstance(phase, dict): continue phase_ids = ( _coerce_id_list(phase.get("skill_ids")) or _coerce_id_list(phase.get("selected")) ) phase_failed_ids = ( _coerce_id_list(phase.get("failed_skill_ids")) or _coerce_id_list(phase.get("skill_phase_failed_skill_ids")) ) metadata_ids.extend(phase_failed_ids) nested_phase_status = str(phase.get("status") or "").lower() phase_failed = nested_phase_status in { "failed", "error", "incomplete", } fallback = skill_execution.get("tool_only_fallback") or skill_execution.get( "fallback_phase" ) fallback_succeeded = ( isinstance(fallback, dict) and str(fallback.get("status") or "").lower() in {"success", "completed"} ) if phase_ids and phase_failed and fallback_succeeded: metadata_ids.extend(phase_ids) if phase_ids and nested_phase_status in {"success", "completed"}: explicit_no_phase_failure = True if failure_keys_present and not metadata_ids: explicit_no_phase_failure = True raw_ids = metadata_ids if explicit_no_phase_failure else [*llm_ids, *metadata_ids] corrected_ids = _correct_skill_ids(raw_ids, known_skill_ids) return [sid for sid in dict.fromkeys(corrected_ids) if sid in known_skill_ids] class ExecutionAnalyzer: """Analyzes task execution results and tracks skill quality. Args: store: Persistence layer for skill records and analyses. llm_client: LLM client used for the analysis call. model: Override model for analysis. If None, uses ``llm_client``'s default model. enabled: Set False to skip analysis entirely. """ def __init__( self, store: SkillStore, llm_client: "LLMClient", model: Optional[str] = None, max_tokens: Optional[int] = None, enabled: bool = True, skill_registry: Optional["SkillRegistry"] = None, quality_manager: Optional["ToolQualityManager"] = None, ) -> None: self._store = store self._llm_client = llm_client self._model = model self._max_tokens = ( max(1, int(max_tokens)) if max_tokens is not None else None ) self.enabled = enabled self._skill_registry = skill_registry self._quality_manager = quality_manager async def _call_analysis_model( self, *, messages: List[Dict[str, Any]], tools: Optional[List[BaseTool]], model: str, disable_thinking: bool = False, ) -> Any: kwargs: Dict[str, Any] = { "messages": messages, "tools": tools, "model": model, } if self._max_tokens is not None: kwargs["max_tokens"] = self._max_tokens if disable_thinking: kwargs["enable_thinking"] = False return await self._llm_client.call_model_with_fallback(**kwargs) async def analyze_execution( self, task_id: str, recording_dir: str, execution_result: Dict[str, Any], available_tools: Optional[List[BaseTool]] = None, ) -> Optional[ExecutionAnalysis]: """Run LLM analysis on a completed task and persist the result. Args: task_id: Unique identifier for the task. recording_dir: Path to the recording directory containing metadata.json, conversations.jsonl, etc. execution_result: The return value of ``OpenSpace.execute()`` — contains status, iterations, skills_used, etc. available_tools: BaseTool instances from the execution (shell tools, MCP tools, etc.). Passed through to the analysis agent loop so it can reproduce errors or verify results when trace data is ambiguous. A lightweight ``bash`` tool is always appended. """ if not self.enabled: return None rec_path = Path(recording_dir) if not rec_path.is_dir(): logger.warning( f"Recording directory not found, skipping analysis: {recording_dir}" ) return None # Check for duplicate — one analysis per task existing = self._store.load_analyses_for_task(task_id) if existing is not None: logger.debug(f"Analysis already exists for task {task_id}, skipping") return existing _src_tok = set_call_source("analyzer") try: # 1. Load recording artifacts context = self._load_recording_context(rec_path, execution_result) if context is None: return None if self._should_skip_low_signal_analysis(context, execution_result): logger.info( "Skipping execution analysis for low-signal successful task: %s", task_id, ) return None # 2. Build prompt prompt = self._build_analysis_prompt(context) # 3. Run analysis (agent loop with optional tool use) raw_json = await self._run_analysis_loop( prompt, available_tools=available_tools or [], ) if raw_json is None: return None # 4. Parse into ExecutionAnalysis analysis = self._parse_analysis(task_id, raw_json, context) if analysis is None: return None # 5. Persist await self._store.record_analysis( analysis, observed_tool_keys=context.get("used_tool_keys", set()), ) await self._report_skill_quality_after_record_analysis( analysis, context, execution_result=execution_result, ) evo_types = [s.evolution_type.value for s in analysis.evolution_suggestions] logger.info( f"Execution analysis saved for task {task_id}: " f"completed={analysis.task_completed}, " f"skills_judged={len(analysis.skill_judgments)}, " f"evolution_suggestions={evo_types or 'none'}" ) return analysis except Exception as e: logger.error(f"Execution analysis failed for task {task_id}: {e}") return None finally: reset_call_source(_src_tok) @staticmethod def _should_skip_low_signal_analysis( context: Dict[str, Any], execution_result: Dict[str, Any], ) -> bool: """Skip analysis that cannot produce reliable skill-quality signal. A successful turn with no tool trajectory and no invoked skills is usually chat/reasoning. Running the analyzer on that shape has no tool evidence and can hallucinate CAPTURED/FIX suggestions from ordinary conversation. """ status = str( execution_result.get("status") or context.get("execution_status") or "" ).lower() if status != "success": return False if context.get("traj_records"): return False if context.get("used_tool_keys"): return False if context.get("selected_skills"): return False if context.get("skill_contents"): return False tool_executions = execution_result.get("tool_executions") if isinstance(tool_executions, list) and tool_executions: return False return True async def _report_skill_quality_after_record_analysis( self, analysis: ExecutionAnalysis, context: Dict[str, Any], *, execution_result: Dict[str, Any] | None = None, ) -> None: try: reporter = _make_skill_quality_reporter() outcome = await reporter.maybe_report_analysis( analysis, session_id=_analysis_session_id( context, execution_result=execution_result, ), ) logger.debug( "Skill quality telemetry outcome for task %s: %s", analysis.task_id, outcome.get("status") if isinstance(outcome, dict) else outcome, ) except Exception as exc: logger.warning( "Skill quality telemetry skipped after analyzer persistence for task %s: %s", analysis.task_id, exc, ) async def get_evolution_candidates( self, limit: int = 20 ) -> List[ExecutionAnalysis]: """Return recent analyses flagged as evolution candidates.""" return self._store.load_evolution_candidates(limit=limit) async def analyze_packet( self, packet: "EvidencePacket", available_tools: Optional[List[BaseTool]] = None, ) -> Optional[ExecutionAnalysis]: """Run post-execution analysis from an EvidencePacket. The packet is the factual scope for analysis. Recording refs, when present, are only fallback/debug context in the prompt. """ if not self.enabled: return None if getattr(packet, "packet_type", "") != "analysis": logger.debug( "Skipping execution analysis for non-analysis packet: %s", getattr(packet, "packet_id", ""), ) return None task_id = packet.scope.task_id or packet.packet_id existing = self._store.load_analyses_for_task(task_id) if existing is not None: logger.debug("Analysis already exists for task %s, skipping", task_id) return existing _src_tok = set_call_source("analyzer") try: context = self._load_packet_context(packet, task_id=task_id) execution_result = { "status": context.get("execution_status", ""), "iterations": context.get("iterations", 0), "tool_executions": context.get("packet_tool_records", []), } if self._should_skip_low_signal_analysis(context, execution_result): logger.info( "Skipping packet analysis for low-signal successful task: %s", task_id, ) return None prompt = self._build_packet_analysis_prompt(packet, context) raw_json = await self._run_analysis_loop( prompt, available_tools=available_tools or [], ) if raw_json is None: return None analysis = self._parse_analysis(task_id, raw_json, context) if analysis is None: return None await self._store.record_analysis( analysis, observed_tool_keys=context.get("used_tool_keys", set()), ) await self._report_skill_quality_after_record_analysis(analysis, context) evo_types = [s.evolution_type.value for s in analysis.evolution_suggestions] logger.info( "Execution packet analysis saved for task %s: completed=%s, " "skills_judged=%s, evolution_suggestions=%s", task_id, analysis.task_completed, len(analysis.skill_judgments), evo_types or "none", ) return analysis except Exception as exc: logger.error("Execution packet analysis failed for task %s: %s", task_id, exc) return None finally: reset_call_source(_src_tok) def _load_packet_context( self, packet: "EvidencePacket", *, task_id: str, ) -> Dict[str, Any]: runtime_refs = packet.selected_refs.get("runtime_snapshot", []) runtime_meta = dict(runtime_refs[-1].metadata) if runtime_refs else {} selected_skills = _packet_selected_skill_ids(packet, runtime_meta) packet_tool_records = _packet_tool_records(packet) used_tool_keys = { str(item.get("tool_key") or "") for item in packet_tool_records if str(item.get("tool_key") or "") } task_description = ( str(runtime_meta.get("instruction_preview") or "").strip() or _first_user_packet_preview(packet) or f"Evidence packet {packet.packet_id}" ) return { "task_id": task_id, "task_description": task_description, "selected_skills": selected_skills, "skill_selection": { "selected": selected_skills, "available_skills": selected_skills, "source": "evidence_packet", }, "skill_execution": _packet_skill_execution(packet), "skill_contents": _packet_skill_contents(packet), "tool_names": sorted(used_tool_keys), "tool_defs": _packet_tool_defs(runtime_meta, used_tool_keys), "used_tool_keys": used_tool_keys, "conversations": [], "traj_records": packet_tool_records, "packet_tool_records": packet_tool_records, "execution_status": str(runtime_meta.get("status") or "unknown"), "iterations": int(runtime_meta.get("iterations") or 0), "session_id": _packet_session_id(packet), "recording_dir": _packet_recording_dir(packet), "packet": packet, } def _build_packet_analysis_prompt( self, packet: "EvidencePacket", context: Dict[str, Any], ) -> str: selected_skill_ids: List[str] = context["selected_skills"] skill_section = _packet_skill_section(packet) tool_list = self._format_tool_list( context.get("tool_defs", []), context.get("used_tool_keys", set()), ) conversation_log = _packet_snippet_section( packet, { "transcript_message", "transcript_segment", "compact_summary", "manual_request_ref", }, fallback_label="(no transcript snippets selected)", ) traj_summary = _packet_snippet_section( packet, {"tool_event", "tool_result", "tool_incident"}, fallback_label="(no tool evidence selected)", ) resource_info = _packet_resource_info(packet) local_taxonomy_info = self._format_local_taxonomy_info() if local_taxonomy_info: resource_info = ( f"{resource_info}\n\n{local_taxonomy_info}" if resource_info else local_taxonomy_info ) return SkillEnginePrompts.execution_analysis( task_description=context["task_description"], execution_status=context["execution_status"], iterations=context["iterations"], tool_list=tool_list, skill_section=skill_section, conversation_log=conversation_log, traj_summary=traj_summary, selected_skill_ids_json=json.dumps(selected_skill_ids), resource_info=resource_info, ) def _load_recording_context( self, rec_path: Path, execution_result: Dict[str, Any], ) -> Optional[Dict[str, Any]]: """Load and structure all recording artifacts needed for analysis. Returns a dict with keys used by ``_build_analysis_prompt()``, or None if critical files are missing. """ # metadata.json (always present) metadata_file = rec_path / "metadata.json" if not metadata_file.exists(): logger.warning(f"metadata.json not found in {rec_path}") return None try: metadata = json.loads(metadata_file.read_text(encoding="utf-8")) except Exception as e: logger.warning(f"Failed to read metadata.json: {e}") return None # conversations.jsonl (primary analysis source) conv_file = rec_path / "conversations.jsonl" conversations: List[Dict[str, Any]] = [] if conv_file.exists(): try: for line in conv_file.read_text(encoding="utf-8").splitlines(): line = line.strip() if line: conversations.append(json.loads(line)) except Exception as e: logger.warning(f"Failed to read conversations.jsonl: {e}") if not conversations: logger.warning(f"No conversations found in {rec_path}, skipping analysis") return None # traj.jsonl (structured tool execution records) traj_records = self._load_traj_data(rec_path) task_id = str(metadata.get("task_id") or "") # Extract key fields from metadata task_description = metadata.get( "task_description", (metadata.get("skill_selection") or {}).get("task", ""), ) if not task_description: task_description = execution_result.get("instruction", "") skill_selection = metadata.get("skill_selection", {}) initial_selected_skills = skill_selection.get("selected", []) if not isinstance(initial_selected_skills, list): initial_selected_skills = [] event_invoked_skill_ids = self._load_invoked_skill_event_ids(task_id) invoked_skill_ids, invoked_skill_contents = ( self._extract_invoked_skills_from_conversations(conversations) ) result_skill_ids = [ sid for sid in ( execution_result.get("active_skills") or execution_result.get("skills_used") or [] ) if sid ] skill_execution = metadata.get("skill_execution") or execution_result.get( "skill_execution", {}, ) if not isinstance(skill_execution, dict): skill_execution = {} skill_execution_selected_ids = [ sid for sid in skill_execution.get("selected", []) or [] if sid ] selected_skills = list( dict.fromkeys([ *initial_selected_skills, *event_invoked_skill_ids, *invoked_skill_ids, *result_skill_ids, *skill_execution_selected_ids, ]) ) retrieved_tools = metadata.get("retrieved_tools", {}) tool_defs = retrieved_tools.get("tools", []) tool_names = [t.get("name", "") for t in tool_defs] # Extract skill content from conversations setup message # selected_skills contains skill_ids stored in runtime metadata. skill_contents: Dict[str, str] = {} for conv in conversations: if conv.get("type") == "setup": for msg in conv.get("messages", []): content = msg.get("content", "") if isinstance(content, str) and "# Active Skills" in content: skill_contents = self._extract_skill_contents( content, selected_skills ) break skill_contents.update(invoked_skill_contents) # Execution status — prefer runtime result, fall back to persisted metadata status = execution_result.get("status", "") iterations = execution_result.get("iterations", 0) if not status: outcome = metadata.get("execution_outcome", {}) status = outcome.get("status", "unknown") iterations = iterations or outcome.get("iterations", 0) # Derive actually-used tools from traj.jsonl # traj_records tells us exactly which tools were invoked; retrieved_tools # is the broader set that was *available* to the agent. used_tool_keys: set = set() for entry in traj_records: backend = entry.get("backend", "") tool = entry.get("tool", "") server = entry.get("server", "") if tool: used_tool_keys.add(f"{backend}:{tool}") if server: used_tool_keys.add(f"{backend}:{server}:{tool}") return { "task_id": task_id, "task_description": task_description, "selected_skills": selected_skills, "skill_selection": skill_selection, "skill_execution": skill_execution, "skill_contents": skill_contents, "tool_names": tool_names, "tool_defs": tool_defs, "used_tool_keys": used_tool_keys, "conversations": conversations, "traj_records": traj_records, "execution_status": status, "iterations": iterations, "recording_dir": str(rec_path), } @staticmethod def _load_traj_data(rec_path: Path) -> List[Dict[str, Any]]: """Load traj.jsonl and return structured tool execution records. Each record contains: step, timestamp, backend, tool, command, result (status, output/stderr), parameters, extra. """ traj_file = rec_path / "traj.jsonl" records: List[Dict[str, Any]] = [] if not traj_file.exists(): return records try: for line in traj_file.read_text(encoding="utf-8").splitlines(): line = line.strip() if line: records.append(json.loads(line)) except Exception as e: logger.warning(f"Failed to read traj.jsonl: {e}") return records def _load_invoked_skill_event_ids(self, task_id: str) -> List[str]: if not task_id: return [] try: events = self._store.load_skill_events( event_type="invoked", task_id=task_id, limit=500, ) except Exception as exc: logger.debug("Failed to load invoked skill events: %s", exc) return [] return list( dict.fromkeys( str(event.get("skill_id") or "") for event in events if str(event.get("skill_id") or "").strip() ) ) @staticmethod def _extract_invoked_skills_from_conversations( conversations: List[Dict[str, Any]], ) -> tuple[List[str], Dict[str, str]]: """Extract OpenSpace SkillTool invocations from recorded attachment messages.""" skill_ids: List[str] = [] skill_contents: Dict[str, str] = {} for conv in conversations: for msg in ExecutionAnalyzer._conversation_messages(conv): if not isinstance(msg, dict): continue meta = msg.get("_meta") or {} attachment = meta.get("attachment") if isinstance(meta, dict) else None if isinstance(attachment, dict): if attachment.get("type") == "invoked_skill_content": sid = str(attachment.get("skill_id") or "") content = str(attachment.get("content") or "") if sid: skill_ids.append(sid) if content: skill_contents[sid] = content elif attachment.get("type") == "invoked_skills": for item in attachment.get("skills", []) or []: if not isinstance(item, dict): continue sid = str(item.get("skill_id") or "") content = str(item.get("content") or "") if sid: skill_ids.append(sid) if content: skill_contents[sid] = content tr_meta = meta.get("tool_result_metadata") if isinstance(meta, dict) else None if isinstance(tr_meta, dict) and tr_meta.get("tool") == "Skill": sid = str(tr_meta.get("skill_id") or "") if sid: skill_ids.append(sid) return list(dict.fromkeys(skill_ids)), skill_contents @staticmethod def _conversation_messages(conv: Dict[str, Any]) -> List[Dict[str, Any]]: """Return setup and per-iteration messages from a conversation record.""" messages: List[Dict[str, Any]] = [] for key in ("messages", "delta_messages"): value = conv.get(key) if isinstance(value, list): messages.extend(item for item in value if isinstance(item, dict)) return messages @staticmethod def _extract_skill_contents( injection_text: str, selected_skill_ids: List[str], ) -> Dict[str, str]: """Parse the injected skill context to extract per-skill content. The injection text uses ``### Skill: {skill_id}`` headers, so we split by that pattern and match against the provided skill_ids. """ contents: Dict[str, str] = {} id_set = set(selected_skill_ids) parts = re.split(r"###\s+Skill:\s+", injection_text) for part in parts[1:]: # skip preamble lines = part.split("\n", 1) sid = lines[0].strip() body = lines[1] if len(lines) > 1 else "" if sid in id_set: contents[sid] = body[:5000] return contents def _load_skill_contents_from_disk( self, skill_ids: List[str], ) -> Dict[str, Dict[str, str]]: """Load skill SKILL.md from disk via SkillRegistry. Returns dict mapping ``skill_id`` → ``{"content", "dir", "description", "name"}``. Falls back gracefully if registry is unavailable. """ result: Dict[str, Dict[str, str]] = {} if not self._skill_registry or not skill_ids: return result for sid in skill_ids: meta = self._skill_registry.get_skill(sid) if not meta: continue content = self._skill_registry.load_skill_content(sid) if not content: continue skill_dir = str(meta.path.parent) if len(content) > _SKILL_CONTENT_MAX_CHARS: content = ( content[:_SKILL_CONTENT_MAX_CHARS] + f"\n\n... [truncated at {_SKILL_CONTENT_MAX_CHARS} chars — " f"use read(\"{meta.path}\") to see full content]" ) result[sid] = { "content": content, "dir": skill_dir, "description": meta.description, "name": meta.name, } return result def _build_analysis_prompt(self, context: Dict[str, Any]) -> str: """Build the LLM prompt for execution analysis. ``context["selected_skills"]`` contains true ``skill_id`` values. """ # Format conversation log (priority-based truncation) conv_text = self._format_conversations(context["conversations"]) # Format traj.jsonl tool execution summary traj_section = self._format_traj_summary(context["traj_records"]) # Skill section — keyed by skill_id throughout selected_skill_ids: List[str] = context["selected_skills"] skill_data = self._load_skill_contents_from_disk(selected_skill_ids) if not skill_data and selected_skill_ids: # Fallback: use content extracted from conversation injection text for sid in selected_skill_ids: content = context["skill_contents"].get(sid) if content: skill_data[sid] = { "content": content, "dir": "(unknown)", "description": "", "name": sid, } skill_section = "" if skill_data: parts = [] for sid, info in skill_data.items(): desc_line = ( f"\n**Description**: {info['description']}" if info.get("description") else "" ) display_name = info.get("name", sid) parts.append( f"### {sid}\n" f"**Name**: {display_name}\n" f"**Directory**: `{info['dir']}`{desc_line}\n\n" f"{info['content']}" ) skill_section = "## Selected Skills\n\n" + "\n\n---\n\n".join(parts) # If no skills selected → skill_section stays "" (omitted from prompt) # Tool list tool_list = self._format_tool_list( context.get("tool_defs", []), context.get("used_tool_keys", set()), ) # Resource info (recording dir + skill dirs) rec_dir = context.get("recording_dir", "") resource_lines: List[str] = [] if rec_dir: resource_lines.append(f"**Recording directory**: `{rec_dir}`") rec_path = Path(rec_dir) if rec_path.is_dir(): files = [f.name for f in sorted(rec_path.iterdir()) if f.is_file()] dirs = [f"{f.name}/" for f in sorted(rec_path.iterdir()) if f.is_dir()] if files: resource_lines.append(f" Files: {', '.join(files)}") if dirs: resource_lines.append(f" Directories: {', '.join(dirs)}") skill_dirs = { sid: info["dir"] for sid, info in skill_data.items() if info.get("dir") and info["dir"] != "(unknown)" } if skill_dirs: resource_lines.append("**Skill directories**:") for sid, d in skill_dirs.items(): resource_lines.append(f" - {sid}: `{d}`") skill_execution = context.get("skill_execution") or {} if skill_execution: resource_lines.append( "**Skill execution phases**:\n" + json.dumps(skill_execution, ensure_ascii=False, indent=2) ) local_taxonomy_info = self._format_local_taxonomy_info() if local_taxonomy_info: resource_lines.append(local_taxonomy_info) resource_lines.append( "\nYou have `read`, `ls`, and `bash` tools for deeper " "investigation.\n**In most cases the trace above is sufficient** — only " "use tools when evidence is ambiguous or you need to verify specific details." ) resource_info = "\n".join(resource_lines) return SkillEnginePrompts.execution_analysis( task_description=context["task_description"], execution_status=context["execution_status"], iterations=context["iterations"], tool_list=tool_list, skill_section=skill_section, conversation_log=conv_text, traj_summary=traj_section, selected_skill_ids_json=json.dumps(selected_skill_ids), resource_info=resource_info, ) def _format_local_taxonomy_info(self) -> str: try: from openspace.cloud.local_mapping import CloudLocalMappingStore from openspace.cloud.skill_classification import ( build_local_taxonomy_snapshot, initialize_local_skill_taxonomy, ) db_path = getattr(self._store, "db_path", None) if db_path is None and getattr(self._store, "base", None) is not None: db_path = getattr(self._store.base, "db_path", None) mapping_store = CloudLocalMappingStore(db_path) if db_path is not None else None try: skills = self._skill_registry.list_skills() if self._skill_registry else [] if mapping_store is not None: initialize_local_skill_taxonomy( mapping_store=mapping_store, skills=skills, ) snapshot = build_local_taxonomy_snapshot( mapping_store=mapping_store, skills=skills, max_paths=12, max_examples_per_path=1, include_sample_paths=True, ) finally: if mapping_store is not None: mapping_store.close() except Exception as exc: logger.debug("Local taxonomy snapshot unavailable for analyzer: %s", exc) return "" lines = [ "**Local taxonomy tree (package-style skill placement)**:", f"Path format: `{snapshot.get('path_format', 'domain/sub-domain/package')}`", "Use this tree when choosing `local_category_path` for DERIVED or CAPTURED suggestions.", ] for category in snapshot.get("categories", []) or []: if isinstance(category, dict): lines.append( f" - {category.get('category')}: {category.get('skill_count', 0)} skill(s)" ) roots = snapshot.get("roots") or [] if roots: lines.append("Top-level local taxonomy roots:") for item in roots[:12]: if isinstance(item, dict): lines.append( f" - {item.get('local_category_path')} " f"({item.get('skill_count', 0)} skill(s))" ) paths = snapshot.get("paths") or snapshot.get("sample_paths") or [] if paths: lines.append("Sample existing local paths:") for item in paths[:40]: if not isinstance(item, dict): continue examples = item.get("examples") or [] names = [ str(example.get("name") or example.get("local_skill_id") or "") for example in examples if isinstance(example, dict) ] shown_names = ", ".join(name for name in names if name) suffix = f" examples={shown_names}" if shown_names else "" lines.append( f" - {item.get('local_category_path')} " f"({item.get('skill_count', 0)} skill(s)){suffix}" ) else: lines.append("Existing local paths: (none yet)") return "\n".join(lines) @staticmethod def _format_tool_list( tool_defs: List[Dict[str, Any]], used_tool_keys: set = None, ) -> str: """Format tool definitions with usage annotation. Tools that appear in ``used_tool_keys`` (derived from traj.jsonl) are marked as "Actually used". This lets the analysis LLM focus on what actually happened without being distracted by unused tools. Args: tool_defs: Tool definitions from ``metadata.retrieved_tools.tools``. Backend should be correctly recorded (mcp, shell, etc.) now that the recording layer prefers ``runtime_info.backend``. used_tool_keys: Set of ``"backend:tool_name"`` or ``"backend:server:tool_name"`` strings derived from traj.jsonl. """ if not tool_defs: return "none" if used_tool_keys is None: used_tool_keys = set() used_parts = [] available_parts = [] for t in tool_defs: name = t.get("name", "?") backend = t.get("backend", "?") server = t.get("server_name") label = f"{name} ({backend}/{server})" if server else f"{name} ({backend})" # Match by backend:tool or backend:server:tool key = f"{backend}:{name}" key_with_server = f"{backend}:{server}:{name}" if server else "" if key in used_tool_keys or key_with_server in used_tool_keys: used_parts.append(label) else: available_parts.append(label) sections = [] if used_parts: sections.append(f"Actually used: {', '.join(used_parts)}") if available_parts: sections.append(f"Available but unused: {', '.join(available_parts)}") return "\n".join(sections) if sections else "none" @staticmethod def _format_traj_summary(traj_records: List[Dict[str, Any]]) -> str: """Format traj.jsonl records into a concise tool execution timeline. This provides the LLM with a structured view of every tool invocation and its outcome, complementing the conversation log which shows the agent's reasoning. """ if not traj_records: return "(no traj.jsonl data available)" lines = [f"Total tool invocations: {len(traj_records)}"] error_count = sum( 1 for r in traj_records if r.get("result", {}).get("status") == "error" ) if error_count: lines.append(f"Errors: {error_count}/{len(traj_records)}") lines.append("") # blank line before timeline for entry in traj_records: step = entry.get("step", "?") backend = entry.get("backend", "?") tool = entry.get("tool", "?") server = entry.get("server", "") result = entry.get("result", {}) status = result.get("status", "?") # Build compact one-line summary command = entry.get("command", "") if isinstance(command, str) and len(command) > 150: command = command[:150] + "..." # Include server for MCP tools so key is unambiguous if server: tool_label = f"{backend}:{server}:{tool}" else: tool_label = f"{backend}:{tool}" line = f" Step {step} [{tool_label}] → {status}" # Add error details for failed steps if status == "error": stderr = result.get("stderr", result.get("output", "")) if isinstance(stderr, str) and stderr: # Extract first meaningful line of error error_first_line = stderr.strip().split("\n")[0][:200] line += f" | {error_first_line}" evidence = result.get("evidence") if isinstance(evidence, dict) and evidence: line += ( " | evidence: " + json.dumps(evidence, ensure_ascii=False, sort_keys=True) ) # Add brief command context if command and not command.startswith("```"): line += f" | cmd: {command[:100]}" lines.append(line) return "\n".join(lines) @staticmethod def _format_conversations(conversations: List[Dict[str, Any]]) -> str: """Format conversations.jsonl into a readable text block for the LLM. Delegates to :func:`conversation_formatter.format_conversations`. """ return format_conversations(conversations, _MAX_CONVERSATION_CHARS) async def _run_analysis_loop( self, prompt: str, available_tools: Optional[List[BaseTool]] = None, ) -> Optional[Dict[str, Any]]: """Run analysis as an agent loop with optional tool use. Most analyses complete in a single pass (LLM outputs JSON directly). When the trace is ambiguous, the LLM may call the execution's own tools (``read``, ``ls``, ``bash``, MCP tools, etc.) for deeper investigation or error reproduction. Uses ``LLMClient.call_model()`` for each model turn and ``run_tools()`` for explicit tool execution so the analyzer follows the main permission/hook/event flow. Conversations are recorded to ``conversations.jsonl`` via ``RecordingManager`` (agent_name="ExecutionAnalyzer") so the full analysis dialogue is preserved alongside the grounding trace. """ from openspace.recording import RecordingManager model = self._model or self._llm_client.model analysis_tools: List[BaseTool] = list(available_tools or []) messages: List[Dict[str, Any]] = [ {"role": "user", "content": prompt}, ] # Record initial conversation setup await RecordingManager.record_conversation_setup( setup_messages=copy.deepcopy(messages), tools=analysis_tools if analysis_tools else None, agent_name="ExecutionAnalyzer", ) length_recovery_count = 0 invalid_json_recovery_count = 0 force_tool_free = False for iteration in range(_ANALYSIS_MAX_ITERATIONS): is_last = iteration == _ANALYSIS_MAX_ITERATIONS - 1 # Snapshot message count before any additions + LLM call msg_count_before = len(messages) # On the final iteration, force JSON output (no tools). if is_last: messages.append({ "role": "system", "content": ( "This is your FINAL round — no more tool calls allowed. " "You MUST output the JSON analysis object now based on " "all information gathered so far." ), }) try: model_response = await self._call_analysis_model( messages=messages, tools=( analysis_tools if analysis_tools and not is_last and not force_tool_free else None ), model=model, disable_thinking=force_tool_free, ) except Exception as e: logger.error(f"Analysis LLM call failed (iter {iteration}): {e}") return None # Keep fallback state local to this analysis run. The helper # exposes the model that produced the response without mutating # the shared client defaults. model = model_response.effective_model or model assistant_message = model_response.assistant_message messages.append(assistant_message) raw_content = assistant_message.get("content", "") content = ( raw_content if isinstance(raw_content, str) else str(raw_content or "") ) has_tool_calls = bool(model_response.tool_calls) followup_messages = self._llm_client.get_model_response_followup_messages( model_response ) has_api_error = self._llm_client.model_response_has_api_error( model_response ) recover_length = bool( has_api_error and model_response.stop_reason == "length" and not has_tool_calls and not is_last and length_recovery_count < _MAX_ANALYSIS_LENGTH_RECOVERIES ) if recover_length: length_recovery_count += 1 force_tool_free = True messages.append( _build_analysis_length_recovery_message(length_recovery_count) ) elif followup_messages: messages.extend(followup_messages) tool_results: list[dict[str, Any]] = [] prevent_continuation = False tool_stop_reason: str | None = None if not has_api_error and has_tool_calls and analysis_tools and not is_last: tool_context = self._llm_client.build_auxiliary_tool_use_context( tools=analysis_tools, messages=messages, model=model, agent_id="execution_analyzer", agent_type=self.__class__.__name__, quality_manager=self._quality_manager, task_description=prompt, current_iteration=iteration + 1, max_iterations=_ANALYSIS_MAX_ITERATIONS, ) tools_result = await run_tools( model_response.tool_calls, model_response.tool_map, tool_context, assistant_message=assistant_message, ) messages.extend(tools_result.messages) prevent_continuation = tools_result.prevent_continuation tool_stop_reason = tools_result.stop_reason tool_results = self._llm_client.collect_tool_results( model_response.tool_calls, model_response.tool_map, tools_result.messages, ) # Record iteration delta updated_messages = messages delta = updated_messages[msg_count_before:] await RecordingManager.record_iteration_context( iteration=iteration + 1, delta_messages=copy.deepcopy(delta), response_metadata={ "has_tool_calls": has_tool_calls, "tool_calls_count": len(tool_results), "is_final": not has_tool_calls, "prevent_continuation": prevent_continuation, "stop_reason": model_response.stop_reason, "tool_stop_reason": tool_stop_reason, "content_length": len(content), "output_tokens": int( getattr(model_response.usage, "output_tokens", 0) or 0 ), "effective_model": model_response.effective_model or model, "has_api_error": has_api_error, "api_error_followup_count": len(followup_messages), "length_recovery_attempt": ( length_recovery_count if recover_length else 0 ), }, agent_name="ExecutionAnalyzer", ) if has_api_error: if recover_length: logger.warning( "Analysis LLM response was truncated; retrying with " "compact JSON (iter=%s stop_reason=%s output_tokens=%s " "content_chars=%s recovery=%s/%s)", iteration + 1, model_response.stop_reason or "unknown", int(getattr(model_response.usage, "output_tokens", 0) or 0), len(content), length_recovery_count, _MAX_ANALYSIS_LENGTH_RECOVERIES, ) continue logger.warning( "Analysis LLM returned a non-recoverable API error followup " "(iter=%s stop_reason=%s output_tokens=%s content_chars=%s)", iteration + 1, model_response.stop_reason or "unknown", int(getattr(model_response.usage, "output_tokens", 0) or 0), len(content), ) return None if prevent_continuation: logger.warning( "Analysis agent stopped after tool hooks prevented " "continuation: %s", tool_stop_reason or "hook_stopped", ) return None if not has_tool_calls: # No tool calls → final response, parse JSON analysis = self._extract_json(content) if analysis is not None: return analysis if ( not is_last and invalid_json_recovery_count < _MAX_ANALYSIS_INVALID_JSON_RECOVERIES ): invalid_json_recovery_count += 1 force_tool_free = True messages.append( _build_analysis_invalid_json_recovery_message( invalid_json_recovery_count ) ) logger.warning( "Analysis LLM returned invalid JSON; retrying with " "compact JSON (iter=%s content_chars=%s recovery=%s/%s)", iteration + 1, len(content), invalid_json_recovery_count, _MAX_ANALYSIS_INVALID_JSON_RECOVERIES, ) continue return None # Continue with the updated messages (assistant + tool results). logger.debug( f"Analysis agent used tools " f"(iter {iteration + 1}/{_ANALYSIS_MAX_ITERATIONS})" ) # Should not reach here (last iteration disables tools), but just in case logger.warning( f"Analysis agent reached max iterations ({_ANALYSIS_MAX_ITERATIONS})" ) for m in reversed(messages): if m.get("role") == "assistant" and m.get("content"): return self._extract_json(m["content"]) return None @staticmethod def _extract_json(text: str) -> Optional[Dict[str, Any]]: """Extract a JSON object from LLM response text. Handles markdown code fences and bare JSON. """ # Try code block first code_match = re.search( r"```(?:json)?\s*\n?(.*?)\n?```", text, re.DOTALL ) if code_match: text = code_match.group(1).strip() else: # Try bare JSON object json_match = re.search(r"\{.*\}", text, re.DOTALL) if json_match: text = json_match.group() try: data = json.loads(text) if isinstance(data, dict): return data logger.warning(f"LLM returned non-dict JSON: {type(data)}") return None except json.JSONDecodeError as e: logger.warning(f"Failed to parse LLM analysis JSON: {e}") logger.debug(f"Raw LLM output (first 500 chars): {text[:500]}") return None @staticmethod def _parse_analysis( task_id: str, data: Dict[str, Any], context: Dict[str, Any], ) -> Optional[ExecutionAnalysis]: """Convert the raw LLM JSON output into an ExecutionAnalysis. Also attaches observed tool execution records from ``traj.jsonl`` so the analysis contains both LLM judgments and factual data. """ try: now = datetime.now() # Quality counters are only valid for skills that were actually # listed/discovered/invoked in this execution. Available catalog # IDs are intentionally not accepted as judgments. selected_skill_ids = [ sid for sid in context.get("selected_skills", []) if sid ] known_skill_ids: set = set(selected_skill_ids) skill_sel = context.get("skill_selection") or {} suggestion_known_skill_ids: set = set(known_skill_ids) for sid in skill_sel.get("available_skills", []): if sid: suggestion_known_skill_ids.add(sid) # Parse skill judgments (LLM-generated) judgments: List[SkillJudgment] = [] seen_judgment_ids: set = set() for jd in data.get("skill_judgments", []): raw_sid = jd.get("skill_id", "") corrected = _correct_skill_ids([raw_sid], known_skill_ids) skill_id = corrected[0] if corrected else raw_sid if skill_id not in known_skill_ids: logger.debug( "Ignoring analyzer skill judgment for non-selected skill %r", skill_id, ) continue if not skill_id or skill_id in seen_judgment_ids: continue seen_judgment_ids.add(skill_id) judgments.append( SkillJudgment( skill_id=skill_id, skill_applied=_parse_analysis_bool( jd.get("skill_applied", False) ), note=jd.get("note", ""), ) ) for sid in selected_skill_ids: if sid not in seen_judgment_ids: judgments.append( SkillJudgment( skill_id=sid, skill_applied=False, note=( "Analyzer response omitted this selected/retrieved " "skill; defaulted to not applied for quality accounting." ), ) ) seen_judgment_ids.add(sid) # Parse evolution_suggestions (new format: list of typed suggestions) suggestions: List[EvolutionSuggestion] = [] for raw_sug in data.get("evolution_suggestions", []): if not isinstance(raw_sug, dict): logger.debug("Ignoring non-object evolution suggestion") continue try: evo_type = EvolutionType(raw_sug.get("type", "")) except ValueError: logger.debug(f"Unknown evolution type: {raw_sug.get('type')}") continue cat = None if raw_sug.get("category"): try: cat = SkillCategory(raw_sug["category"]) except ValueError: logger.debug(f"Unknown category: {raw_sug.get('category')}") # Support both "target_skills" (list) and legacy "target_skill" (str) raw_targets = raw_sug.get("target_skills") if isinstance(raw_targets, list): targets = [t for t in raw_targets if t] else: legacy = raw_sug.get("target_skill", "") targets = [legacy] if legacy else [] # Correct LLM-hallucinated skill IDs against known IDs. # LLMs frequently swap/drop characters in hex suffixes # (e.g. "61f694bc" instead of "61f694cb"). targets = _correct_skill_ids(targets, suggestion_known_skill_ids) raw_contract = raw_sug.get("capture_contract") capture_contract = ( CaptureContract.from_dict(raw_contract) if isinstance(raw_contract, dict) else None ) suggestions.append(EvolutionSuggestion( evolution_type=evo_type, target_skill_ids=targets, category=cat, local_category_path=str(raw_sug.get("local_category_path") or ""), direction=raw_sug.get("direction", ""), capture_contract=capture_contract, )) phase_failed_skill_ids = _extract_skill_phase_failed_skill_ids( data, context, known_skill_ids, ) analysis = ExecutionAnalysis( task_id=task_id, timestamp=now, task_completed=_parse_analysis_bool( data.get("task_completed", False) ), execution_note=data.get("execution_note", ""), tool_issues=data.get("tool_issues", []), skill_judgments=judgments, skill_phase_failed_skill_ids=phase_failed_skill_ids, evolution_suggestions=suggestions, analyzed_by=data.get("analyzed_by", ""), analyzed_at=now, ) return analysis except Exception as e: logger.error(f"Failed to parse analysis response: {e}") return None # Convenience queries (delegated to store) def get_store(self) -> SkillStore: """Access the underlying SkillStore for direct queries.""" return self._store def close(self) -> None: """Close the store connection.""" self._store.close() def _packet_selected_skill_ids( packet: "EvidencePacket", runtime_meta: Dict[str, Any], ) -> List[str]: ids: list[str] = [] ids.extend(_coerce_id_list(runtime_meta.get("active_skills"))) for ref_type in ("skill_file", "skill_event", "skill_record"): for ref in packet.selected_refs.get(ref_type, []): ids.extend(_coerce_id_list(ref.metadata.get("skill_ids"))) sid = str(ref.metadata.get("skill_id") or "") if sid: ids.append(sid) return list(dict.fromkeys(ids)) def _packet_skill_contents(packet: "EvidencePacket") -> Dict[str, str]: snippets_by_ref = {snippet.ref_id: snippet.text for snippet in packet.expanded_snippets} contents: Dict[str, str] = {} for ref in packet.selected_refs.get("skill_file", []): sid = str(ref.metadata.get("skill_id") or "") if not sid: continue contents[sid] = snippets_by_ref.get(ref.ref_id) or ref.preview[:_SKILL_CONTENT_MAX_CHARS] return contents def _packet_skill_section(packet: "EvidencePacket") -> str: skill_refs = packet.selected_refs.get("skill_file", []) if not skill_refs: return "" snippets_by_ref = {snippet.ref_id: snippet.text for snippet in packet.expanded_snippets} parts: list[str] = [] for ref in skill_refs: sid = str(ref.metadata.get("skill_id") or ref.ref_id) path = str(ref.metadata.get("path") or ref.uri or "") content = snippets_by_ref.get(ref.ref_id) or ref.preview parts.append( f"### {sid}\n" f"**Path**: `{path}`\n" f"**Evidence ref**: `{ref.ref_id}`\n\n" f"{content[:_SKILL_CONTENT_MAX_CHARS]}" ) return "## Selected Skills\n\n" + "\n\n---\n\n".join(parts) def _packet_skill_execution(packet: "EvidencePacket") -> Dict[str, Any]: events: list[dict[str, Any]] = [] for ref in packet.selected_refs.get("skill_event", []): events.append( { "ref_id": ref.ref_id, "skill_id": ref.metadata.get("skill_id"), "event_type": ref.metadata.get("event_type") or ref.metadata.get("lifecycle_event") or ref.metadata.get("status"), } ) selected = [ str(item.get("skill_id")) for item in events if item.get("skill_id") and str(item.get("event_type") or "").lower() in {"selected", "listed", "discovered", "invoked", "applied", "completed"} ] return { "selected": list(dict.fromkeys(selected)), "events": events, "source": "evidence_packet", } def _packet_tool_records(packet: "EvidencePacket") -> List[Dict[str, Any]]: records: list[dict[str, Any]] = [] for ref_type in ("tool_event", "tool_result", "tool_incident"): for ref in packet.selected_refs.get(ref_type, []): metadata = ref.metadata tool_key = str(metadata.get("tool_key") or "") tool_name = str(metadata.get("tool_name") or metadata.get("tool") or "") backend = "" server = "" if tool_key: parts = tool_key.split(":", 2) if len(parts) == 3: backend, server, tool_name = parts elif len(parts) == 2: backend, tool_name = parts records.append( { "ref_id": ref.ref_id, "ref_type": ref.ref_type, "backend": backend, "server": server, "tool": tool_name, "tool_name": tool_name, "tool_key": tool_key, "status": metadata.get("status") or metadata.get("outcome"), "command": metadata.get("input_preview"), "result": { "status": metadata.get("status") or metadata.get("outcome"), "output": metadata.get("result_preview") or ref.preview, "stderr": metadata.get("error_message"), }, } ) return records def _packet_tool_defs( runtime_meta: Dict[str, Any], used_tool_keys: set[str], ) -> List[Dict[str, Any]]: tool_defs: list[dict[str, Any]] = [] retrieved = runtime_meta.get("retrieved_tools_list") or [] if isinstance(retrieved, list): for item in retrieved: if isinstance(item, dict): tool_defs.append(dict(item)) elif isinstance(item, str) and item: tool_defs.append({"name": item, "backend": "retrieved"}) for key in sorted(used_tool_keys): parts = key.split(":", 2) if len(parts) == 3: backend, server, name = parts tool_defs.append({"name": name, "backend": backend, "server_name": server}) elif len(parts) == 2: backend, name = parts tool_defs.append({"name": name, "backend": backend}) deduped: list[dict[str, Any]] = [] seen: set[tuple[str, str, str]] = set() for item in tool_defs: key = ( str(item.get("backend") or ""), str(item.get("server_name") or item.get("server") or ""), str(item.get("name") or ""), ) if key in seen: continue seen.add(key) deduped.append(item) return deduped def _packet_snippet_section( packet: "EvidencePacket", ref_types: set[str], *, fallback_label: str, ) -> str: ref_type_by_id = { ref.ref_id: ref.ref_type for refs in packet.selected_refs.values() for ref in refs } chunks = [ snippet.text for snippet in packet.expanded_snippets if ref_type_by_id.get(snippet.ref_id) in ref_types ] if chunks: return "\n\n".join(chunks) previews: list[str] = [] for ref_type in sorted(ref_types): for ref in packet.selected_refs.get(ref_type, []): previews.append( f"ref_id: {ref.ref_id}\nref_type: {ref.ref_type}\npreview: {ref.preview}" ) return "\n\n".join(previews) if previews else fallback_label def _packet_resource_info(packet: "EvidencePacket") -> str: lines = [ "Evidence Packet Context:", f"- packet_id: {packet.packet_id}", f"- trigger_job_id: {packet.trigger_job_id}", f"- profile: {packet.profile_name}/{packet.subprofile}", "- EvidencePacket is the primary fact source for this analysis.", "- recording_ref entries are fallback/debug context only.", "- tool_result previews may be incomplete; use readable_paths for full output when needed.", "- compact_summary entries are lossy and must be tied back to exact refs.", "- Any evolution_suggestions are proposals only, not commands to edit skills.", ] if packet.profile_name == "quality_signal": lines.extend( [ "- This QUALITY_SIGNAL job already exists because rule-based evidence gates created it.", "- Do not create, remove, or request TriggerJobs.", "- Treat quality_signal_ref as derived evidence, not a final diagnosis or mutation command.", "- Use only packet refs; every factual claim behind a proposed change must be supported by selected refs.", "- If evidence does not support a skill change, choose no evolution_suggestions.", "- If the issue is external, permission, sandbox, API key, or outage related, choose no evolution_suggestions.", ] ) if packet.instructions: lines.append("Packet instructions:") for key, value in sorted(packet.instructions.items()): lines.append(f" - {key}: {value}") if packet.readable_paths: lines.append("Readable paths:") for item in packet.readable_paths: state = "readable" if item.readable else f"not_readable:{item.missing_reason}" lines.append( f" - {item.ref_id} [{item.purpose}] {state}: `{item.path}`" ) lines.append("Selected refs:") for ref_type, refs in sorted(packet.selected_refs.items()): lines.append(f" - {ref_type}: {', '.join(ref.ref_id for ref in refs)}") return "\n".join(lines) def _first_user_packet_preview(packet: "EvidencePacket") -> str: for ref in packet.selected_refs.get("transcript_message", []): if str(ref.metadata.get("role") or "").lower() == "user": text = str(ref.preview or ref.metadata.get("preview") or "").strip() if text: return text[:500] return "" def _packet_recording_dir(packet: "EvidencePacket") -> str: for ref in packet.selected_refs.get("recording_ref", []): return str(ref.uri or ref.metadata.get("recording_dir") or "") return "" def _analysis_session_id( context: Dict[str, Any], *, execution_result: Dict[str, Any] | None = None, ) -> str | None: if execution_result is not None: session_id = _clean_session_id(execution_result.get("session_id")) if session_id: return session_id session_id = _clean_session_id(context.get("session_id")) if session_id: return session_id packet = context.get("packet") if packet is not None: return _packet_session_id(packet) return None def _packet_session_id(packet: "EvidencePacket") -> str | None: scope = getattr(packet, "scope", None) session_id = _clean_session_id(getattr(scope, "session_id", None)) if session_id: return session_id selected_refs = getattr(packet, "selected_refs", None) if not isinstance(selected_refs, dict): return None refs = [ ref for ref_type in sorted(selected_refs) for ref in selected_refs.get(ref_type, []) or [] ] for ref in refs: session_id = _clean_session_id(getattr(ref, "session_id", None)) if session_id: return session_id for ref in refs: metadata = getattr(ref, "metadata", None) if isinstance(metadata, dict): session_id = _clean_session_id(metadata.get("session_id")) if session_id: return session_id return None def _clean_session_id(value: Any) -> str | None: text = str(value or "").strip() return text or None