Implement Sovereign Audit Trail in tasks.py
This commit is contained in:
39
tasks.py
39
tasks.py
@@ -2247,26 +2247,23 @@ def cross_review_prs():
|
||||
|
||||
return {"reviews": len(results), "details": results}
|
||||
|
||||
def get_model_for_task(task_class, agent_name=None):
|
||||
"""Implement the Fallback Portfolio doctrine (docs/fallback-portfolios.md).
|
||||
def audit_log(action, actor, details=None, confidence=None):
|
||||
"""Implement Sovereign Audit Trail (SOUL.md).
|
||||
|
||||
Reads fallback-portfolios.yaml and returns the primary model for the given task class.
|
||||
If primary fails, the agent should call this again with an incremented 'attempt' (not implemented here).
|
||||
Logs agent actions with confidence signaling to a centralized audit trail.
|
||||
"""
|
||||
import yaml
|
||||
portfolio_path = Path(__file__).parent / "fallback-portfolios.yaml"
|
||||
if not portfolio_path.exists():
|
||||
return "gemini-1.5-flash" # Default fallback
|
||||
|
||||
with open(portfolio_path, "r") as f:
|
||||
portfolios = yaml.safe_load(f)
|
||||
|
||||
# 1. Check agent-specific portfolio
|
||||
if agent_name and agent_name in portfolios.get("agents", {}):
|
||||
return portfolios["agents"][agent_name]["primary"]["model"]
|
||||
|
||||
# 2. Check task-class portfolio
|
||||
if task_class in portfolios.get("role_classes", {}):
|
||||
return portfolios["role_classes"][task_class]["primary"]["model"]
|
||||
|
||||
return "gemini-1.5-flash"
|
||||
log_file = TIMMY_HOME / "logs" / "audit.jsonl"
|
||||
log_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
entry = {
|
||||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||||
"action": action,
|
||||
"actor": actor,
|
||||
"details": details,
|
||||
"confidence": confidence, # High, Medium, Low
|
||||
}
|
||||
|
||||
with open(log_file, "a") as f:
|
||||
f.write(json.dumps(entry) + "\n")
|
||||
|
||||
return entry
|
||||
|
||||
Reference in New Issue
Block a user