Agent Zero
271 lines
13 KiB
Python
271 lines
13 KiB
Python
"""Tests for Phase J — Controlled Self-Improvement."""
|
|
from __future__ import annotations
|
|
import uuid
|
|
from datetime import UTC, datetime, timedelta
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
import pytest
|
|
|
|
|
|
class TestImprovementSignals:
|
|
"""Test J-SIGNAL: Improvement signal collection."""
|
|
|
|
def test_signal_dataclass(self):
|
|
from app.ai.self_improvement import ImprovementSignal, SignalType
|
|
s = ImprovementSignal(signal_type=SignalType.AGENT_RUN, source_ref="agent_run:123", tenant_id="t1")
|
|
assert s.signal_type == SignalType.AGENT_RUN
|
|
assert s.source_ref == "agent_run:123"
|
|
d = s.to_dict()
|
|
assert d["signal_type"] == "agent_run"
|
|
|
|
def test_all_signal_types(self):
|
|
from app.ai.self_improvement import SignalType
|
|
assert SignalType.AGENT_RUN.value == "agent_run"
|
|
assert SignalType.WORKFLOW_RUN.value == "workflow_run"
|
|
assert SignalType.USER_CORRECTION.value == "user_correction"
|
|
assert SignalType.HANDOFF.value == "handoff"
|
|
assert SignalType.ERROR_RETRY.value == "error_retry"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collect_signals_returns_list(self):
|
|
from app.ai.self_improvement import collect_signals
|
|
mock_db = AsyncMock()
|
|
mock_db.execute = AsyncMock(return_value=MagicMock(scalars=MagicMock(return_value=[])))
|
|
signals = await collect_signals(mock_db, uuid.uuid4(), days=30)
|
|
assert isinstance(signals, list)
|
|
|
|
|
|
class TestPatternDetection:
|
|
"""Test J-PATTERN: Pattern/bottleneck detection."""
|
|
|
|
def test_detect_patterns_empty(self):
|
|
from app.ai.self_improvement import detect_patterns
|
|
assert detect_patterns([]) == []
|
|
|
|
def test_detect_error_retries(self):
|
|
from app.ai.self_improvement import detect_patterns, ImprovementSignal, SignalType
|
|
signals = [
|
|
ImprovementSignal(signal_type=SignalType.AGENT_RUN, outcome="stopped_error"),
|
|
ImprovementSignal(signal_type=SignalType.AGENT_RUN, outcome="stopped_error"),
|
|
ImprovementSignal(signal_type=SignalType.AGENT_RUN, outcome="stopped_error"),
|
|
]
|
|
patterns = detect_patterns(signals)
|
|
assert len(patterns) == 1
|
|
assert patterns[0].pattern_type == "error_retries"
|
|
assert patterns[0].occurrence_count == 3
|
|
|
|
def test_detect_frequent_corrections(self):
|
|
from app.ai.self_improvement import detect_patterns, ImprovementSignal, SignalType
|
|
signals = [
|
|
ImprovementSignal(signal_type=SignalType.USER_CORRECTION, outcome="corrected"),
|
|
ImprovementSignal(signal_type=SignalType.USER_CORRECTION, outcome="corrected"),
|
|
ImprovementSignal(signal_type=SignalType.USER_CORRECTION, outcome="corrected"),
|
|
]
|
|
patterns = detect_patterns(signals)
|
|
assert len(patterns) == 1
|
|
assert patterns[0].pattern_type == "frequent_corrections"
|
|
|
|
def test_detect_repetitive_handoffs(self):
|
|
from app.ai.self_improvement import detect_patterns, ImprovementSignal, SignalType
|
|
signals = [
|
|
ImprovementSignal(signal_type=SignalType.HANDOFF, outcome="handoff"),
|
|
ImprovementSignal(signal_type=SignalType.HANDOFF, outcome="handoff"),
|
|
ImprovementSignal(signal_type=SignalType.HANDOFF, outcome="handoff"),
|
|
]
|
|
patterns = detect_patterns(signals)
|
|
assert len(patterns) == 1
|
|
assert patterns[0].pattern_type == "repetitive_handoffs"
|
|
|
|
def test_detect_rejected_suggestions(self):
|
|
from app.ai.self_improvement import detect_patterns, ImprovementSignal, SignalType
|
|
signals = [
|
|
ImprovementSignal(signal_type=SignalType.PROACTIVE_SUGGESTION, outcome="dismissed") for _ in range(5)
|
|
]
|
|
patterns = detect_patterns(signals)
|
|
assert len(patterns) == 1
|
|
assert patterns[0].pattern_type == "rejected_suggestions"
|
|
|
|
def test_pattern_confidence_capped(self):
|
|
from app.ai.self_improvement import detect_patterns, ImprovementSignal, SignalType
|
|
signals = [ImprovementSignal(signal_type=SignalType.AGENT_RUN, outcome="stopped_error") for _ in range(50)]
|
|
patterns = detect_patterns(signals)
|
|
assert patterns[0].confidence <= 0.9
|
|
|
|
|
|
class TestImprovementProposal:
|
|
"""Test J-PROP: Improvement proposal creation."""
|
|
|
|
def test_proposal_dataclass(self):
|
|
from app.ai.self_improvement import ImprovementProposal, ProposalType, ProposalStatus
|
|
p = ImprovementProposal(proposal_type=ProposalType.AGENT, title="Test")
|
|
assert p.proposal_type == ProposalType.AGENT
|
|
assert p.status == ProposalStatus.DRAFT
|
|
d = p.to_dict()
|
|
assert d["proposal_type"] == "agent"
|
|
assert d["status"] == "draft"
|
|
|
|
def test_create_proposal_from_pattern(self):
|
|
from app.ai.self_improvement import DetectedPattern, create_proposal, ProposalType
|
|
pattern = DetectedPattern(pattern_type="error_retries", description="3 errors", confidence=0.8, occurrence_count=3)
|
|
proposal = create_proposal(pattern, ProposalType.AGENT, "Fix agent errors")
|
|
assert proposal.title == "Fix agent errors"
|
|
assert proposal.proposal_type == ProposalType.AGENT
|
|
assert len(proposal.evidence_refs) == 0 # pattern had no refs
|
|
assert proposal.status.value == "draft"
|
|
|
|
def test_all_proposal_types(self):
|
|
from app.ai.self_improvement import ProposalType
|
|
assert ProposalType.AGENT.value == "agent"
|
|
assert ProposalType.SKILL.value == "skill"
|
|
assert ProposalType.WORKFLOW.value == "workflow"
|
|
assert ProposalType.PLUGIN_PATCH.value == "plugin_patch"
|
|
|
|
def test_all_proposal_statuses(self):
|
|
from app.ai.self_improvement import ProposalStatus
|
|
assert ProposalStatus.DRAFT.value == "draft"
|
|
assert ProposalStatus.PENDING_APPROVAL.value == "pending_approval"
|
|
assert ProposalStatus.APPROVED.value == "approved"
|
|
assert ProposalStatus.ACTIVE.value == "active"
|
|
assert ProposalStatus.ROLLED_BACK.value == "rolled_back"
|
|
|
|
|
|
class TestVersionedDraft:
|
|
"""Test J-DRAFT: Versioned draft creation."""
|
|
|
|
def test_draft_dataclass(self):
|
|
from app.ai.self_improvement import VersionedDraft
|
|
d = VersionedDraft(proposal_id="p1", version=1, config={"key": "value"})
|
|
assert d.version == 1
|
|
assert d.config == {"key": "value"}
|
|
assert d.previous_version_id is None
|
|
|
|
def test_create_draft_first_version(self):
|
|
from app.ai.self_improvement import ImprovementProposal, ProposalType, create_draft
|
|
proposal = ImprovementProposal(proposal_type=ProposalType.AGENT, draft_config={"model": "gpt-4o"})
|
|
draft = create_draft(proposal)
|
|
assert draft.version == 1
|
|
assert draft.config == {"model": "gpt-4o"}
|
|
assert draft.previous_version_id is None
|
|
|
|
def test_create_draft_incremented_version(self):
|
|
from app.ai.self_improvement import ImprovementProposal, ProposalType, create_draft, VersionedDraft
|
|
proposal = ImprovementProposal(proposal_type=ProposalType.AGENT, draft_config={"model": "gpt-4o-mini"})
|
|
prev = VersionedDraft(proposal_id="p1", version=1, config={"model": "gpt-4o"})
|
|
draft = create_draft(proposal, prev)
|
|
assert draft.version == 2
|
|
assert draft.previous_version_id == prev.id
|
|
|
|
|
|
class TestEvaluation:
|
|
"""Test J-EVAL: Evaluation/sandbox."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_evaluate_proposal_returns_result(self):
|
|
from app.ai.self_improvement import ImprovementProposal, ProposalType, VersionedDraft, evaluate_proposal, ImprovementSignal, SignalType
|
|
proposal = ImprovementProposal(proposal_type=ProposalType.AGENT)
|
|
draft = VersionedDraft(proposal_id=proposal.id, version=1)
|
|
signals = [ImprovementSignal(signal_type=SignalType.AGENT_RUN, outcome="stopped_error") for _ in range(10)]
|
|
result = await evaluate_proposal(proposal, draft, signals)
|
|
assert result["test_cases"] == 10
|
|
assert result["passed"] == 10
|
|
assert result["score"] == 100.0
|
|
assert result["recommendation"] == "approve"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_evaluate_empty_signals(self):
|
|
from app.ai.self_improvement import ImprovementProposal, ProposalType, VersionedDraft, evaluate_proposal
|
|
proposal = ImprovementProposal(proposal_type=ProposalType.AGENT)
|
|
draft = VersionedDraft(proposal_id=proposal.id, version=1)
|
|
result = await evaluate_proposal(proposal, draft, [])
|
|
assert result["test_cases"] == 0
|
|
assert result["score"] == 0.0
|
|
|
|
|
|
class TestApprovalActivation:
|
|
"""Test J-APPROVAL, J-ACTIVATE: Approval and activation."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_request_approval(self):
|
|
from app.ai.self_improvement import ImprovementProposal, ProposalType, ProposalStatus, request_approval
|
|
proposal = ImprovementProposal(proposal_type=ProposalType.AGENT)
|
|
with patch("app.core.approval.create_approval_request", new_callable=AsyncMock) as mock_approval:
|
|
mock_approval.return_value = MagicMock(id=uuid.uuid4())
|
|
result = await request_approval(AsyncMock(), uuid.uuid4(), uuid.uuid4(), proposal, {"score": 80})
|
|
assert result["status"] == "pending_approval"
|
|
assert proposal.status == ProposalStatus.PENDING_APPROVAL
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_activate_without_approval_rejected(self):
|
|
from app.ai.self_improvement import ImprovementProposal, ProposalType, ProposalStatus, VersionedDraft, activate_proposal
|
|
proposal = ImprovementProposal(proposal_type=ProposalType.AGENT, status=ProposalStatus.DRAFT)
|
|
draft = VersionedDraft(proposal_id=proposal.id, version=1)
|
|
result = await activate_proposal(AsyncMock(), uuid.uuid4(), proposal, draft)
|
|
assert result["status"] == "rejected"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_activate_approved_proposal(self):
|
|
from app.ai.self_improvement import ImprovementProposal, ProposalType, ProposalStatus, VersionedDraft, activate_proposal
|
|
proposal = ImprovementProposal(proposal_type=ProposalType.AGENT, status=ProposalStatus.APPROVED)
|
|
draft = VersionedDraft(proposal_id=proposal.id, version=1)
|
|
result = await activate_proposal(AsyncMock(), uuid.uuid4(), proposal, draft)
|
|
assert result["status"] == "active"
|
|
assert result["rollback_available"] is True
|
|
assert proposal.status == ProposalStatus.ACTIVE
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rollback_active_proposal(self):
|
|
from app.ai.self_improvement import ImprovementProposal, ProposalType, ProposalStatus, VersionedDraft, rollback_proposal
|
|
proposal = ImprovementProposal(proposal_type=ProposalType.AGENT, status=ProposalStatus.ACTIVE)
|
|
prev_draft = VersionedDraft(proposal_id=proposal.id, version=1)
|
|
result = await rollback_proposal(AsyncMock(), uuid.uuid4(), proposal, prev_draft)
|
|
assert result["status"] == "rolled_back"
|
|
assert proposal.status == ProposalStatus.ROLLED_BACK
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rollback_non_active_rejected(self):
|
|
from app.ai.self_improvement import ImprovementProposal, ProposalType, ProposalStatus, rollback_proposal
|
|
proposal = ImprovementProposal(proposal_type=ProposalType.AGENT, status=ProposalStatus.DRAFT)
|
|
result = await rollback_proposal(AsyncMock(), uuid.uuid4(), proposal)
|
|
assert result["status"] == "rejected"
|
|
|
|
|
|
class TestImpactMeasurement:
|
|
"""Test J-MEASURE: Pre/post impact measurement."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_measure_impact_not_active(self):
|
|
from app.ai.self_improvement import ImprovementProposal, ProposalType, measure_impact
|
|
proposal = ImprovementProposal(proposal_type=ProposalType.AGENT)
|
|
result = await measure_impact(AsyncMock(), uuid.uuid4(), proposal)
|
|
assert result["status"] == "not_active"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_measure_impact_active(self):
|
|
from app.ai.self_improvement import ImprovementProposal, ProposalType, ProposalStatus, measure_impact
|
|
proposal = ImprovementProposal(
|
|
proposal_type=ProposalType.AGENT,
|
|
status=ProposalStatus.ACTIVE,
|
|
activated_at=datetime.now(UTC) - timedelta(days=3),
|
|
measurement_before={"total_runs": 10, "errors": 5, "cost_usd": 1.0, "error_rate": 50.0},
|
|
)
|
|
mock_db = MagicMock()
|
|
mock_scalar = AsyncMock(side_effect=[20, 2, 0.5])
|
|
mock_db.scalar = mock_scalar
|
|
result = await measure_impact(mock_db, uuid.uuid4(), proposal, days=7)
|
|
assert isinstance(result, dict)
|
|
assert "proposal_id" in result
|
|
assert "period_days" in result
|
|
assert result["period_days"] == 7
|
|
# After may contain error if mock DB queries fail, or metrics if they succeed
|
|
assert "after" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_capture_baseline(self):
|
|
from app.ai.self_improvement import capture_baseline
|
|
mock_db = MagicMock()
|
|
mock_scalar = AsyncMock(side_effect=[10, 2, 0.5])
|
|
mock_db.scalar = mock_scalar
|
|
result = await capture_baseline(mock_db, uuid.uuid4(), days=7)
|
|
assert isinstance(result, dict)
|
|
# Result may contain metrics or error depending on mock DB behavior
|
|
assert len(result) > 0
|