garvitsachdeva/SpindleFlow-RL

1"""2Delegation trace — audit trail for regulated industries.3Every delegation decision is logged. generate_explanation() produces4human-readable audit text.5"""6 7from __future__ import annotations8from dataclasses import dataclass, field9from datetime import datetime10from env.delegation_graph import DelegationEdge11 12 13@dataclass14class DelegationTrace:15    """Complete audit record for one episode."""16    episode_id: str17    task_description: str18    task_complexity: str19    start_time: str = field(default_factory=lambda: datetime.utcnow().isoformat())20    delegation_edges: list[DelegationEdge] = field(default_factory=list)21    scratchpad_entries: list[dict] = field(default_factory=list)22    final_reward: float = 0.023    approved_by_policy: bool = True24 25    def record_edge(self, edge: DelegationEdge) -> None:26        self.delegation_edges.append(edge)27 28    def record_scratchpad(self, author_id: str, content: str, step: int) -> None:29        self.scratchpad_entries.append({30            "author": author_id,31            "step": step,32            "content_preview": content[:200],33        })34 35    def generate_explanation(self) -> str:36        """37        Generate a human-readable audit trail.38        Suitable for compliance export.39        """40        lines = [41            "=== DELEGATION AUDIT TRAIL ===",42            f"Episode: {self.episode_id}",43            f"Time: {self.start_time}",44            f"Task: {self.task_description}",45            f"Complexity: {self.task_complexity}",46            f"Final Reward: {self.final_reward:.3f}",47            "",48            "Delegation Sequence:",49        ]50 51        for i, edge in enumerate(self.delegation_edges):52            lines.append(53                f"  Step {i+1}: {edge.caller_id} -> {edge.callee_id} "54                f"[mode: {edge.delegation_mode}]"55            )56 57        lines.extend([58            "",59            f"Total specialists called: {len(self.delegation_edges)}",60            f"Max delegation depth reached: "61            f"{max((e.depth for e in self.delegation_edges), default=0)}",62            "=== END AUDIT TRAIL ===",63        ])64 65        return "\n".join(lines)66 67    def to_dict(self) -> dict:68        return {69            "episode_id": self.episode_id,70            "task": self.task_description,71            "complexity": self.task_complexity,72            "start_time": self.start_time,73            "delegation_steps": [74                {75                    "caller": e.caller_id,76                    "callee": e.callee_id,77                    "mode": e.delegation_mode,78                    "depth": e.depth,79                }80                for e in self.delegation_edges81            ],82            "reward": self.final_reward,83        }84