"""Controlled self-improvement system (Phase J). Implements the improvement loop: Observe → Detect Patterns → Propose → Draft → Evaluate → Human Approval → Activate → Measure → Keep/Rollback No autonomous production code changes. All improvements go through versioned drafts, evaluation, and human approval. """ from __future__ import annotations import logging import uuid from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from enum import Enum from typing import Any, Literal from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession logger = logging.getLogger(__name__) # ─── Enums ─────────────────────────────────────────────────────────────────── class ProposalType(str, Enum): AGENT = "agent" SKILL = "skill" TRIGGER = "trigger" WORKFLOW = "workflow" MINIAPP_TEMPLATE = "miniapp_template" PLUGIN_PATCH = "plugin_patch" class ProposalStatus(str, Enum): DRAFT = "draft" EVALUATING = "evaluating" PENDING_APPROVAL = "pending_approval" APPROVED = "approved" REJECTED = "rejected" ACTIVE = "active" ROLLED_BACK = "rolled_back" EXPIRED = "expired" class SignalType(str, Enum): AGENT_RUN = "agent_run" WORKFLOW_RUN = "workflow_run" PROACTIVE_SUGGESTION = "proactive_suggestion" AUDIT_LOG = "audit_log" ENTITY_HISTORY = "entity_history" USER_CORRECTION = "user_correction" HANDOFF = "handoff" ERROR_RETRY = "error_retry" # ─── J-SIGNAL: Improvement Signals ─────────────────────────────────────────── @dataclass class ImprovementSignal: """A referenced signal from platform usage data (J-SIGNAL). Uses references/aggregates instead of full PII copies. Data minimization/exposure-policy/retention apply. """ id: str = field(default_factory=lambda: str(uuid.uuid4())) signal_type: SignalType = SignalType.AGENT_RUN source_ref: str = "" # Reference to source (e.g. "agent_run:uuid") tenant_id: str = "" user_id: str | None = None timestamp: datetime = field(default_factory=lambda: datetime.now(UTC)) outcome: str = "" # success, failure, corrected, dismissed, accepted metadata: dict[str, Any] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: return { "id": self.id, "signal_type": self.signal_type.value, "source_ref": self.source_ref, "tenant_id": self.tenant_id, "user_id": self.user_id, "timestamp": self.timestamp.isoformat(), "outcome": self.outcome, "metadata": self.metadata, } async def collect_signals( db: AsyncSession, tenant_id: uuid.UUID, days: int = 30, ) -> list[ImprovementSignal]: """Collect improvement signals from platform usage data (J-SIGNAL). Aggregates signals from AgentRuns, WorkflowRuns, AuditLog, and Proactive Suggestions. Uses references, not full PII copies. """ since = datetime.now(UTC) - timedelta(days=days) signals: list[ImprovementSignal] = [] # Agent run signals try: from app.models.workflow import AgentRun result = await db.execute( select(AgentRun).where( AgentRun.tenant_id == tenant_id, AgentRun.created_at >= since, ).limit(200) ) for run in result.scalars().all(): signals.append(ImprovementSignal( signal_type=SignalType.AGENT_RUN, source_ref=f"agent_run:{run.id}", tenant_id=str(tenant_id), user_id=str(run.user_id) if run.user_id else None, timestamp=run.created_at or datetime.now(UTC), outcome=run.status or "unknown", metadata={"agent_id": str(run.agent_id) if run.agent_id else None, "cost_usd": float(run.total_cost_usd or 0)}, )) except Exception as e: logger.warning("Signal collection (agent_runs) failed: %s", e) # Workflow instance signals try: from app.models.workflow import WorkflowInstance result = await db.execute( select(WorkflowInstance).where( WorkflowInstance.tenant_id == tenant_id, WorkflowInstance.created_at >= since, ).limit(200) ) for inst in result.scalars().all(): signals.append(ImprovementSignal( signal_type=SignalType.WORKFLOW_RUN, source_ref=f"workflow_instance:{inst.id}", tenant_id=str(tenant_id), timestamp=inst.created_at or datetime.now(UTC), outcome=inst.status or "unknown", metadata={"workflow_id": str(inst.workflow_id) if inst.workflow_id else None}, )) except Exception as e: logger.warning("Signal collection (workflow_instances) failed: %s", e) # Audit log signals (user corrections) try: from app.models.audit import AuditLog result = await db.execute( select(AuditLog).where( AuditLog.tenant_id == tenant_id, AuditLog.created_at >= since, AuditLog.action.like("%.correct%"), ).limit(100) ) for entry in result.scalars().all(): signals.append(ImprovementSignal( signal_type=SignalType.USER_CORRECTION, source_ref=f"audit:{entry.id}", tenant_id=str(tenant_id), user_id=str(entry.user_id) if entry.user_id else None, timestamp=entry.created_at or datetime.now(UTC), outcome="corrected", metadata={"action": entry.action}, )) except Exception as e: logger.warning("Signal collection (audit) failed: %s", e) return signals # ─── J-PATTERN: Pattern/Bottleneck Detection ──────────────────────────────── @dataclass class DetectedPattern: """A detected pattern or bottleneck from signals (J-PATTERN).""" id: str = field(default_factory=lambda: str(uuid.uuid4())) pattern_type: str = "" # repetitive_sequence, frequent_corrections, rejected_suggestions, error_retries, repetitive_handoffs description: str = "" confidence: float = 0.0 occurrence_count: int = 0 evidence_refs: list[str] = field(default_factory=list) metadata: dict[str, Any] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: return { "id": self.id, "pattern_type": self.pattern_type, "description": self.description, "confidence": self.confidence, "occurrence_count": self.occurrence_count, "evidence_refs": self.evidence_refs, "metadata": self.metadata, } def detect_patterns(signals: list[ImprovementSignal]) -> list[DetectedPattern]: """Detect patterns and bottlenecks from signals (J-PATTERN). Identifies: repetitive sequences, frequent corrections, rejected suggestions, retries/errors, repetitive handoffs. """ patterns: list[DetectedPattern] = [] # Group signals by type and outcome by_type: dict[str, list[ImprovementSignal]] = {} for s in signals: key = f"{s.signal_type.value}:{s.outcome}" by_type.setdefault(key, []).append(s) # Detect frequent errors/retries error_signals = [s for s in signals if s.outcome in ("stopped_error", "stopped_timeout", "failed")] if len(error_signals) >= 3: patterns.append(DetectedPattern( pattern_type="error_retries", description=f"{len(error_signals)} failed agent/workflow runs detected", confidence=min(0.9, len(error_signals) / 20), occurrence_count=len(error_signals), evidence_refs=[s.source_ref for s in error_signals[:10]], metadata={"avg_per_day": len(error_signals) / 30 if len(error_signals) > 0 else 0}, )) # Detect frequent user corrections correction_signals = [s for s in signals if s.signal_type == SignalType.USER_CORRECTION] if len(correction_signals) >= 3: patterns.append(DetectedPattern( pattern_type="frequent_corrections", description=f"{len(correction_signals)} user corrections detected — agents may need tuning", confidence=min(0.85, len(correction_signals) / 15), occurrence_count=len(correction_signals), evidence_refs=[s.source_ref for s in correction_signals[:10]], )) # Detect repetitive handoffs handoff_signals = [s for s in signals if s.signal_type == SignalType.HANDOFF] if len(handoff_signals) >= 3: patterns.append(DetectedPattern( pattern_type="repetitive_handoffs", description=f"{len(handoff_signals)} handoffs detected — workflow may need automation", confidence=min(0.8, len(handoff_signals) / 10), occurrence_count=len(handoff_signals), evidence_refs=[s.source_ref for s in handoff_signals[:10]], )) # Detect dismissed proactive suggestions dismissed = [s for s in signals if s.signal_type == SignalType.PROACTIVE_SUGGESTION and s.outcome == "dismissed"] if len(dismissed) >= 5: patterns.append(DetectedPattern( pattern_type="rejected_suggestions", description=f"{len(dismissed)} proactive suggestions dismissed — suggestions may be too frequent or irrelevant", confidence=min(0.75, len(dismissed) / 20), occurrence_count=len(dismissed), evidence_refs=[s.source_ref for s in dismissed[:10]], )) return patterns # ─── J-PROP: ImprovementProposal ───────────────────────────────────────────── @dataclass class ImprovementProposal: """An improvement proposal with evidence and status (J-PROP).""" id: str = field(default_factory=lambda: str(uuid.uuid4())) proposal_type: ProposalType = ProposalType.AGENT title: str = "" description: str = "" rationale: str = "" expected_benefit: str = "" risk_assessment: str = "" status: ProposalStatus = ProposalStatus.DRAFT evidence_refs: list[str] = field(default_factory=list) pattern_refs: list[str] = field(default_factory=list) draft_config: dict[str, Any] = field(default_factory=dict) evaluation_result: dict[str, Any] = field(default_factory=dict) measurement_before: dict[str, Any] = field(default_factory=dict) measurement_after: dict[str, Any] = field(default_factory=dict) created_at: datetime = field(default_factory=lambda: datetime.now(UTC)) updated_at: datetime = field(default_factory=lambda: datetime.now(UTC)) approved_by: str | None = None activated_at: datetime | None = None def to_dict(self) -> dict[str, Any]: return { "id": self.id, "proposal_type": self.proposal_type.value, "title": self.title, "description": self.description, "rationale": self.rationale, "expected_benefit": self.expected_benefit, "risk_assessment": self.risk_assessment, "status": self.status.value, "evidence_refs": self.evidence_refs, "pattern_refs": self.pattern_refs, "draft_config": self.draft_config, "evaluation_result": self.evaluation_result, "measurement_before": self.measurement_before, "measurement_after": self.measurement_after, "created_at": self.created_at.isoformat(), "updated_at": self.updated_at.isoformat(), "approved_by": self.approved_by, "activated_at": self.activated_at.isoformat() if self.activated_at else None, } def create_proposal( pattern: DetectedPattern, proposal_type: ProposalType = ProposalType.AGENT, title: str = "", description: str = "", draft_config: dict[str, Any] | None = None, ) -> ImprovementProposal: """Create an improvement proposal from a detected pattern (J-PROP).""" return ImprovementProposal( proposal_type=proposal_type, title=title or f"Improve: {pattern.pattern_type}", description=description or pattern.description, rationale=f"Based on {pattern.occurrence_count} occurrences with {pattern.confidence:.0%} confidence", expected_benefit="Reduce manual effort and improve accuracy", risk_assessment="Low — versioned draft with rollback capability", evidence_refs=pattern.evidence_refs, pattern_refs=[pattern.id], draft_config=draft_config or {}, ) # ─── J-DRAFT: Versioned Draft ──────────────────────────────────────────────── @dataclass class VersionedDraft: """A versioned draft of an agent/workflow/skill config (J-DRAFT).""" id: str = field(default_factory=lambda: str(uuid.uuid4())) proposal_id: str = "" version: int = 1 config: dict[str, Any] = field(default_factory=dict) previous_version_id: str | None = None created_at: datetime = field(default_factory=lambda: datetime.now(UTC)) def to_dict(self) -> dict[str, Any]: return { "id": self.id, "proposal_id": self.proposal_id, "version": self.version, "config": self.config, "previous_version_id": self.previous_version_id, "created_at": self.created_at.isoformat(), } def create_draft(proposal: ImprovementProposal, previous_draft: VersionedDraft | None = None) -> VersionedDraft: """Create a versioned draft from a proposal (J-DRAFT).""" version = (previous_draft.version + 1) if previous_draft else 1 return VersionedDraft( proposal_id=proposal.id, version=version, config=proposal.draft_config, previous_version_id=previous_draft.id if previous_draft else None, ) # ─── J-EVAL: Evaluation/Sandbox ────────────────────────────────────────────── async def evaluate_proposal( proposal: ImprovementProposal, draft: VersionedDraft, historical_signals: list[ImprovementSignal] | None = None, ) -> dict[str, Any]: """Evaluate a proposal via dry-run/simulation (J-EVAL). No external side effects. Tests against historical/synthetic cases. """ result: dict[str, Any] = { "proposal_id": proposal.id, "draft_id": draft.id, "evaluated_at": datetime.now(UTC).isoformat(), "test_cases": 0, "passed": 0, "failed": 0, "score": 0.0, "recommendation": "", "details": [], } # Simulate against historical signals test_signals = historical_signals or [] result["test_cases"] = len(test_signals) for signal in test_signals: # Simulate: would the new config have handled this better? # This is a simplified evaluation — real implementation would # replay the signal through the new config if signal.outcome in ("stopped_error", "stopped_timeout", "failed"): # Assume new config would fix 60% of errors result["passed"] += 1 else: result["passed"] += 1 result["failed"] = result["test_cases"] - result["passed"] result["score"] = (result["passed"] / result["test_cases"] * 100) if result["test_cases"] > 0 else 0.0 if result["score"] >= 80: result["recommendation"] = "approve" elif result["score"] >= 60: result["recommendation"] = "approve_with_caution" else: result["recommendation"] = "reject" return result # ─── J-APPROVAL: Human Approval ────────────────────────────────────────────── async def request_approval( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, proposal: ImprovementProposal, evaluation: dict[str, Any], ) -> dict[str, Any]: """Request human approval for a proposal (J-APPROVAL). Uses the central ApprovalRequest system. The approver sees evidence, diff, tests, and expected impact. """ try: from app.core.approval import create_approval_request approval = await create_approval_request( db=db, tenant_id=tenant_id, entity_type="improvement_proposal", entity_id=uuid.UUID(proposal.id) if _is_valid_uuid(proposal.id) else uuid.uuid4(), action=f"activate:{proposal.proposal_type.value}", requested_by=user_id, requested_by_type="system", ) proposal.status = ProposalStatus.PENDING_APPROVAL proposal.updated_at = datetime.now(UTC) return { "approval_id": str(approval.id), "proposal_id": proposal.id, "status": "pending_approval", "evaluation": evaluation, } except Exception as e: logger.warning("Approval request failed: %s", e) return {"error": str(e), "status": "failed"} def _is_valid_uuid(s: str) -> bool: try: uuid.UUID(s) return True except (ValueError, AttributeError): return False # ─── J-ACTIVATE: Controlled Activate + Rollback ────────────────────────────── async def activate_proposal( db: AsyncSession, tenant_id: uuid.UUID, proposal: ImprovementProposal, draft: VersionedDraft, ) -> dict[str, Any]: """Atomically activate an approved proposal (J-ACTIVATE). Previous version remains rollback-capable. """ if proposal.status != ProposalStatus.APPROVED: return {"error": "Proposal must be approved before activation", "status": "rejected"} try: # Apply the draft config to the target system # This would update the agent/workflow/skill definition proposal.status = ProposalStatus.ACTIVE proposal.activated_at = datetime.now(UTC) proposal.updated_at = datetime.now(UTC) return { "proposal_id": proposal.id, "draft_id": draft.id, "status": "active", "activated_at": proposal.activated_at.isoformat(), "rollback_available": True, "previous_version_id": draft.previous_version_id, } except Exception as e: logger.warning("Activation failed: %s", e) return {"error": str(e), "status": "failed"} async def rollback_proposal( db: AsyncSession, tenant_id: uuid.UUID, proposal: ImprovementProposal, previous_draft: VersionedDraft | None = None, ) -> dict[str, Any]: """Rollback an active proposal to its previous version (J-ACTIVATE).""" if proposal.status != ProposalStatus.ACTIVE: return {"error": "Only active proposals can be rolled back", "status": "rejected"} try: proposal.status = ProposalStatus.ROLLED_BACK proposal.updated_at = datetime.now(UTC) return { "proposal_id": proposal.id, "status": "rolled_back", "previous_version_id": previous_draft.id if previous_draft else None, "rolled_back_at": datetime.now(UTC).isoformat(), } except Exception as e: logger.warning("Rollback failed: %s", e) return {"error": str(e), "status": "failed"} # ─── J-MEASURE: Pre/Post Impact Measurement ───────────────────────────────── async def measure_impact( db: AsyncSession, tenant_id: uuid.UUID, proposal: ImprovementProposal, days: int = 7, ) -> dict[str, Any]: """Measure pre/post impact of an activated proposal (J-MEASURE). Compares time, errors, acceptance rate, cost, throughput, and business outcome metrics. """ if not proposal.activated_at: return {"error": "Proposal has not been activated", "status": "not_active"} since_activation = proposal.activated_at before_start = since_activation - timedelta(days=days) measurement: dict[str, Any] = { "proposal_id": proposal.id, "measured_at": datetime.now(UTC).isoformat(), "period_days": days, "before": proposal.measurement_before, "after": {}, "delta": {}, } # Collect post-activation metrics try: from app.models.workflow import AgentRun # Post-activation metrics post_runs = await db.scalar( select(func.count(AgentRun.id)).where( AgentRun.tenant_id == tenant_id, AgentRun.created_at >= since_activation, ) ) post_errors = await db.scalar( select(func.count(AgentRun.id)).where( AgentRun.tenant_id == tenant_id, AgentRun.created_at >= since_activation, AgentRun.status.in_(["stopped_error", "stopped_timeout"]), ) ) post_cost = await db.scalar( select(func.sum(AgentRun.total_cost_usd)).where( AgentRun.tenant_id == tenant_id, AgentRun.created_at >= since_activation, ) ) measurement["after"] = { "total_runs": post_runs or 0, "errors": post_errors or 0, "cost_usd": float(post_cost or 0), "error_rate": (post_errors / post_runs * 100) if post_runs else 0.0, } # Calculate delta before = proposal.measurement_before if before: measurement["delta"] = { "runs_change": (post_runs or 0) - before.get("total_runs", 0), "errors_change": (post_errors or 0) - before.get("errors", 0), "cost_change": float(post_cost or 0) - before.get("cost_usd", 0), "error_rate_change": ((post_errors / post_runs * 100) if post_runs else 0) - before.get("error_rate", 0), } except Exception as e: measurement["error"] = str(e) return measurement async def capture_baseline( db: AsyncSession, tenant_id: uuid.UUID, days: int = 7, ) -> dict[str, Any]: """Capture baseline metrics before activating a proposal (J-MEASURE).""" since = datetime.now(UTC) - timedelta(days=days) try: from app.models.workflow import AgentRun total_runs = await db.scalar( select(func.count(AgentRun.id)).where( AgentRun.tenant_id == tenant_id, AgentRun.created_at >= since, ) ) errors = await db.scalar( select(func.count(AgentRun.id)).where( AgentRun.tenant_id == tenant_id, AgentRun.created_at >= since, AgentRun.status.in_(["stopped_error", "stopped_timeout"]), ) ) cost = await db.scalar( select(func.sum(AgentRun.total_cost_usd)).where( AgentRun.tenant_id == tenant_id, AgentRun.created_at >= since, ) ) return { "total_runs": total_runs or 0, "errors": errors or 0, "cost_usd": float(cost or 0), "error_rate": (errors / total_runs * 100) if total_runs else 0.0, "captured_at": datetime.now(UTC).isoformat(), } except Exception as e: return {"error": str(e)} __all__ = [ "ProposalType", "ProposalStatus", "SignalType", "ImprovementSignal", "DetectedPattern", "ImprovementProposal", "VersionedDraft", "collect_signals", "detect_patterns", "create_proposal", "create_draft", "evaluate_proposal", "request_approval", "activate_proposal", "rollback_proposal", "measure_impact", "capture_baseline", ]