From a1a1307e4a3a94d3a62cb94fa620d90623cf2d04 Mon Sep 17 00:00:00 2001 From: rosspeili Date: Wed, 17 Jun 2026 12:58:55 +0300 Subject: [PATCH] feat: add structured skill execution session logging Capture per-turn skill execution events in session history and transcript output while preserving human-readable agent replies and shielding tool logs from model context/routing. --- rooms/agent.py | 60 ++++++++++++++++++++++------ rooms/session.py | 32 ++++++++++++++- rooms/storage.py | 28 +++++++++++-- tests/test_session.py | 93 +++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 196 insertions(+), 17 deletions(-) diff --git a/rooms/agent.py b/rooms/agent.py index a6b2bd7..43ccea9 100644 --- a/rooms/agent.py +++ b/rooms/agent.py @@ -25,6 +25,7 @@ def __init__(self, config: AgentConfig): self.system_prompt = config.system_prompt self.expertise = config.expertise self._skill_runtime = SkillRuntime(config) + self._last_skill_events: List[Dict[str, Any]] = [] def _execute_custom_function(self, messages: List[Dict[str, str]]) -> str: """Dynamically loads and invokes a custom python function for inference.""" @@ -62,11 +63,20 @@ def _execute_custom_function(self, messages: List[Dict[str, str]]) -> str: logger.error(f"Error executing custom function '{func_name}' in {file_path}: {e}") return f"[Error: Custom function failed. Details: {str(e)}]" - def generate_response(self, context_messages: List[Dict[str, str]], override_params: Optional[Dict[str, Any]] = None) -> str: + def consume_last_skill_events(self) -> List[Dict[str, Any]]: + """Return and clear structured skill events for the last generation call.""" + events = list(self._last_skill_events) + self._last_skill_events = [] + return events + + def generate_response_with_events( + self, context_messages: List[Dict[str, str]], override_params: Optional[Dict[str, Any]] = None + ) -> Dict[str, Any]: """ - Generate a response using LiteLLM or a custom function. + Generate a response and structured skill execution events. """ params = override_params or {} + self._last_skill_events = [] # Build the system message for this agent full_system_prompt = self.config.system_prompt @@ -85,7 +95,8 @@ def generate_response(self, context_messages: List[Dict[str, str]], override_par messages.extend(context_messages) if self.model_type == ModelType.CUSTOM_FUNCTION: - return self._execute_custom_function(messages) + content = self._execute_custom_function(messages) + return {"content": content, "skill_events": []} try: # LiteLLM handling @@ -101,7 +112,7 @@ def generate_response(self, context_messages: List[Dict[str, str]], override_par litellm_params.update(params) tools = self._skill_runtime.get_tools() if self._skill_runtime.has_skills and self._skill_runtime.load_error: - return f"[Error: {self._skill_runtime.load_error}]" + return {"content": f"[Error: {self._skill_runtime.load_error}]", "skill_events": []} if tools: litellm_params["tools"] = tools litellm_params["tool_choice"] = "auto" @@ -112,13 +123,17 @@ def generate_response(self, context_messages: List[Dict[str, str]], override_par if tools and tool_calls: if len(tool_calls) > self.config.max_skill_calls_per_turn: - return ( - "[Error: Model requested too many tool calls in one turn " - f"({len(tool_calls)} > {self.config.max_skill_calls_per_turn})]" - ) + return { + "content": ( + "[Error: Model requested too many tool calls in one turn " + f"({len(tool_calls)} > {self.config.max_skill_calls_per_turn})]" + ), + "skill_events": [], + } tool_call_payload: List[Dict[str, Any]] = [] tool_results: List[Dict[str, Any]] = [] + skill_events: List[Dict[str, Any]] = [] for tc in tool_calls: func = getattr(tc, "function", None) tool_name = getattr(func, "name", "") @@ -128,6 +143,15 @@ def generate_response(self, context_messages: List[Dict[str, str]], override_par except Exception: # noqa: BLE001 parsed_args = {} execution = self._skill_runtime.execute_tool(tool_name, parsed_args, self.config.timeout) + skill_events.append( + { + "event_type": "skill_execution", + "tool_name": tool_name, + "arguments": parsed_args, + "result": execution, + "ok": bool(execution.get("ok")), + } + ) tc_id = getattr(tc, "id", f"call_{len(tool_call_payload)}") tool_call_payload.append( @@ -149,16 +173,28 @@ def generate_response(self, context_messages: List[Dict[str, str]], override_par second_params = dict(litellm_params) second_params["messages"] = messages second_response = litellm.completion(**second_params) - return (second_response.choices[0].message.content or "").strip() + content = (second_response.choices[0].message.content or "").strip() + self._last_skill_events = skill_events + return {"content": content, "skill_events": skill_events} - return (first_message.content or "").strip() + return {"content": (first_message.content or "").strip(), "skill_events": []} except litellm.Timeout as e: logger.error(f"Timeout logic executed for agent '{self.name}' on model '{self.model}': {e}") - return f"[Timeout Error: The model '{self.model}' took too long to respond ({self.config.timeout}s)]" + return { + "content": f"[Timeout Error: The model '{self.model}' took too long to respond ({self.config.timeout}s)]", + "skill_events": [], + } except Exception as e: logger.error(f"Error getting response from agent '{self.name}' on model '{self.model}': {e}") - return f"[Error: Could not generate response. Details: {str(e)}]" + return {"content": f"[Error: Could not generate response. Details: {str(e)}]", "skill_events": []} + + def generate_response(self, context_messages: List[Dict[str, str]], override_params: Optional[Dict[str, Any]] = None) -> str: + """ + Backward-compatible wrapper that returns only human-readable content. + """ + result = self.generate_response_with_events(context_messages, override_params=override_params) + return str(result.get("content", "")).strip() def __repr__(self): return f"" diff --git a/rooms/session.py b/rooms/session.py index 8c2b7ce..bdf3a94 100644 --- a/rooms/session.py +++ b/rooms/session.py @@ -55,7 +55,7 @@ def __init__(self, config: SessionConfig, agents: List[Agent], user_profile: Opt self.config = config self.agents = agents self.user_profile = user_profile # {"name": "...", "background": "..."} - self.history: List[Dict[str, str]] = [] + self.history: List[Dict[str, Any]] = [] self.turn_count = 0 self._last_orchestrator_turn = -1 self._forced_next_agent: Optional[Agent] = None # Locked next agent from @mention or user direction @@ -91,6 +91,9 @@ def get_agent_context(self, current_agent: Agent) -> List[Dict[str, str]]: """Format history into an LLM context including system prompt.""" context = [] for msg in self.history: + if msg.get("role") == "skill": + # Keep tool logs in session history/transcripts, but out of model context. + continue role = "user" if msg["role"] == current_agent.name: role = "assistant" @@ -104,6 +107,27 @@ def get_agent_context(self, current_agent: Agent) -> List[Dict[str, str]]: context.append({"role": role, "content": content}) return context + def _append_skill_events(self, agent: Agent, skill_events: List[Dict[str, Any]]) -> None: + """Store structured skill execution events in session history.""" + for event in skill_events: + result = event.get("result", {}) + tool_name = event.get("tool_name", "unknown_tool") + status = "ok" if event.get("ok") else "error" + content = f"{agent.name} used {tool_name} ({status})" + self.history.append( + { + "role": "skill", + "agent": agent.name, + "event_type": event.get("event_type", "skill_execution"), + "tool_name": tool_name, + "arguments": event.get("arguments", {}), + "result": result, + "status": status, + "content": content, + "timestamp": _now(), + } + ) + def generate_next_turn(self) -> Optional[Dict[str, str]]: """Determine next agent, get response, and log it.""" if self.turn_count >= self.config.max_turns: @@ -141,6 +165,9 @@ def generate_next_turn(self) -> Optional[Dict[str, str]]: context = self.get_agent_context(agent) response_text = agent.generate_response(context) + skill_events = agent.consume_last_skill_events() if hasattr(agent, "consume_last_skill_events") else [] + if skill_events: + self._append_skill_events(agent, skill_events) # Handle PASS: agent has nothing to add — silently skip turn if response_text.strip().upper() == "PASS": @@ -171,7 +198,8 @@ def _select_next_agent(self) -> Agent: elif self.config.session_type == SessionType.DYNAMIC: # Build context text from recent history for scoring - recent = " ".join(m["content"] for m in self.history[-5:]) + recent_messages = [m for m in self.history if m.get("role") != "skill"] + recent = " ".join(m.get("content", "") for m in recent_messages[-5:]) # 1. Check for @mention or name reference in last user/agent message if self.history: diff --git a/rooms/storage.py b/rooms/storage.py index 8234086..fe809d3 100644 --- a/rooms/storage.py +++ b/rooms/storage.py @@ -1,6 +1,7 @@ import csv +import json import os -from typing import List, Dict +from typing import List, Dict, Any def slugify_topic(topic: str, max_words: int = 5) -> str: @@ -12,7 +13,7 @@ def slugify_topic(topic: str, max_words: int = 5) -> str: return slug or "session" -def save_transcript(history: List[Dict[str, str]], filepath: str, format: str = "markdown"): +def save_transcript(history: List[Dict[str, Any]], filepath: str, format: str = "markdown"): """ Save the conversation history to filepath. Format: 'markdown' or 'csv'. @@ -28,10 +29,22 @@ def save_transcript(history: List[Dict[str, str]], filepath: str, format: str = writer = csv.writer(f) writer.writerow(["Timestamp", "Speaker", "Message"]) for msg in public_history: + if msg.get("role") == "skill": + payload = { + "event_type": msg.get("event_type", "skill_execution"), + "agent": msg.get("agent", ""), + "tool_name": msg.get("tool_name", ""), + "status": msg.get("status", ""), + "arguments": msg.get("arguments", {}), + "result": msg.get("result", {}), + } + message = json.dumps(payload, ensure_ascii=True, separators=(",", ":")) + else: + message = msg.get("content", "").replace("\n", " ") writer.writerow([ msg.get("timestamp", ""), msg.get("role", ""), - msg.get("content", "").replace("\n", " ") + message ]) else: # Markdown @@ -42,4 +55,13 @@ def save_transcript(history: List[Dict[str, str]], filepath: str, format: str = content = msg.get("content", "") ts = msg.get("timestamp", "") ts_str = f" _{ts}_" if ts else "" + if role == "skill": + content = ( + f"- agent: {msg.get('agent', '')}\n" + f"- tool: {msg.get('tool_name', '')}\n" + f"- status: {msg.get('status', '')}\n" + f"- arguments: `{json.dumps(msg.get('arguments', {}), ensure_ascii=True)}`\n" + f"- result: `{json.dumps(msg.get('result', {}), ensure_ascii=True)}`" + ) + role = "skill event" f.write(f"### {role.strip().capitalize()}{ts_str}\n\n{content}\n\n---\n\n") diff --git a/tests/test_session.py b/tests/test_session.py index f93cc23..06cb7c7 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -1,8 +1,11 @@ import pytest +import tempfile +from pathlib import Path from unittest.mock import MagicMock from rooms.config import SessionConfig, AgentConfig, SessionType from rooms.agent import Agent from rooms.session import Session, _score_agent_expertise +from rooms.storage import save_transcript def test_round_robin_session(): # Setup @@ -318,3 +321,93 @@ def test_hitl_trigger_only_once_per_message(): session.add_user_message("Theo", "Hello") assert session.needs_human_input() is False + + +def test_session_logs_structured_skill_events_and_keeps_reply_human_readable(): + config = SessionConfig( + topic="Skill Logging Test", + agents=[AgentConfig(name="AgentA", system_prompt="sys")], + session_type=SessionType.ROUND_ROBIN, + max_turns=1, + ) + agent_a = Agent(config.agents[0]) + agent_a.generate_response = MagicMock(return_value="Human-readable answer") + agent_a.consume_last_skill_events = MagicMock( + return_value=[ + { + "event_type": "skill_execution", + "tool_name": "finance_wallet_screening", + "arguments": {"wallet": "0xabc"}, + "result": {"ok": True, "data": {"flagged": True}}, + "ok": True, + } + ] + ) + session = Session(config, [agent_a]) + + turn = session.generate_next_turn() + assert turn["content"] == "Human-readable answer" + + skill_logs = [m for m in session.history if m.get("role") == "skill"] + assert len(skill_logs) == 1 + assert skill_logs[0]["tool_name"] == "finance_wallet_screening" + assert skill_logs[0]["status"] == "ok" + + +def test_skill_events_are_excluded_from_agent_context(): + config = SessionConfig( + topic="Skill Context Isolation", + agents=[AgentConfig(name="AgentA", system_prompt="sys")], + session_type=SessionType.ROUND_ROBIN, + max_turns=1, + ) + agent_a = Agent(config.agents[0]) + session = Session(config, [agent_a]) + session.history.append( + { + "role": "skill", + "agent": "AgentA", + "event_type": "skill_execution", + "tool_name": "finance_wallet_screening", + "arguments": {"wallet": "0xabc"}, + "result": {"ok": True}, + "status": "ok", + "content": "AgentA used finance_wallet_screening (ok)", + "timestamp": "2026-06-17 12:00:00", + } + ) + + context = session.get_agent_context(agent_a) + assert all("finance_wallet_screening" not in msg["content"] for msg in context) + + +def test_transcript_writer_persists_skill_events(): + history = [ + {"role": "system", "content": "bootstrap", "timestamp": "2026-06-17 12:00:00"}, + {"role": "AgentA", "content": "Normal answer", "timestamp": "2026-06-17 12:00:01"}, + { + "role": "skill", + "agent": "AgentA", + "event_type": "skill_execution", + "tool_name": "finance_wallet_screening", + "arguments": {"wallet": "0xabc"}, + "result": {"ok": True, "data": {"flagged": True}}, + "status": "ok", + "content": "AgentA used finance_wallet_screening (ok)", + "timestamp": "2026-06-17 12:00:02", + }, + ] + + with tempfile.TemporaryDirectory() as td: + md_path = Path(td) / "session.md" + csv_path = Path(td) / "session.csv" + + save_transcript(history, str(md_path), format="markdown") + save_transcript(history, str(csv_path), format="csv") + + md_text = md_path.read_text(encoding="utf-8") + csv_text = csv_path.read_text(encoding="utf-8") + + assert "Skill event" in md_text + assert "finance_wallet_screening" in md_text + assert 'tool_name"":""finance_wallet_screening' in csv_text