Artifex/tests/test_venture_discovery_cohort_v02.py

from __future__ import annotations

import json
import threading
import time
from io import StringIO

from django.core.management import call_command

from agents.venture_discovery import DEFAULT_HARD_EXCLUSIONS, EVIDENCE_CEILINGS, SCORE_DEFINITIONS, SCORE_DIMENSIONS, VentureDiscoveryService
from control_plane.ventures.models import CohortIdeationMandate, CompanyProposal, EvidenceTier, NoveltyGateDecision, OpportunityTerritory, OverlapClassification, PortfolioICReview, PortfolioSaturationAnalysis, PortfolioThesis, PortfolioThesisCluster, PortfolioThesisStatus, VentureCapabilityDemand, VentureCohort, VentureCollision, VentureGenerationRejection, VentureThesisFingerprint
from graph.bootstrap import champion_venture_discovery_cohort_graph_v1
from graph.langgraph_runtime import LangGraphRuntime
from graph.models import GraphRun, GraphRunStatus
from graph.venture_cohort import venture_discovery_cohort_registry
from model_router.router import ModelProvider, ModelRequestContract, ModelResponseContract, ModelRouter


class SequenceProvider(ModelProvider):
    provider_name = "sol-sequence"

    def __init__(self) -> None:
        self.company_calls = 0
        self.research_calls = 0

    def complete(self, request: ModelRequestContract) -> ModelResponseContract:
        if "Bounded public web market research" in request.prompt:
            self.research_calls += 1
            categories = ["competitors", "pricing", "customer_pain", "market_alternatives", "regulatory_platform_risks"]
            return ModelResponseContract("luna", json.dumps({"sources": [{"url": f"https://example.com/{self.research_calls}/{cat}", "title": cat, "category": cat, "summary": cat} for cat in categories], "findings": {cat: [cat] for cat in categories}, "coverage": {cat: True for cat in categories}}), {})
        self.company_calls += 1
        n = self.company_calls
        industries = ["Shopify", "Developer", "Legal", "Healthcare", "Real Estate", "Restaurant", "Security", "Education", "Logistics", "Finance"]
        industry = industries[(n - 1) % len(industries)]
        return ModelResponseContract("sol", json.dumps({"title": f"{industry} Workflow Monitor {n}", "one_line_thesis": f"Sell a recurring {industry.lower()} workflow monitor to a narrow buyer before building broad software.", "description": f"A focused {industry} workflow monitoring product with AI-assisted weekly exception reports.", "problem": f"{industry} buyers miss recurring operational exceptions that cost time or revenue.", "target_customer": f"Small {industry} operators with active revenue.", "proposed_solution": f"AI-assisted {industry} workflow monitor with prioritized exception reports and lightweight automation.", "business_model": "Recurring AI-enabled service with software automation", "pricing_hypothesis": "$99/month monitor", "acquisition_strategy": "Compliant community posts and direct referrals after approval", "validation_plan": "Collect 5 credible target-customer responses or 1 willingness-to-pay signal before build/spend.", "capital_requested": "50", "time_to_first_dollar_estimate": "3-7 days after outreach approval", "expected_margin": "80-90% gross margin", "build_complexity": "LOW", "market_evidence": [], "differentiation": "Specific recurring workflow data and AI-assisted exception monitoring", "exception_rationale": "vertical specific recurring workflow with proprietary data", "major_risks": ["Demand unproven"], "confidence": 0.65}), {})

    def health(self) -> str:
        return "AVAILABLE"


class RejectThenAcceptProvider(SequenceProvider):
    def complete(self, request: ModelRequestContract) -> ModelResponseContract:
        if "Bounded public web market research" in request.prompt:
            return super().complete(request)
        time.sleep(0.02)
        self.company_calls += 1
        if self.company_calls == 1:
            return ModelResponseContract("sol", json.dumps({"title": "RFP Copilot", "one_line_thesis": "AI RFP response drafting for SaaS teams.", "description": "Automates request for proposal answers.", "problem": "Sales teams hate RFPs.", "target_customer": "B2B SaaS sales teams", "proposed_solution": "Proposal drafting automation", "business_model": "SaaS", "pricing_hypothesis": "$99/month", "acquisition_strategy": "Content", "validation_plan": "Interview later", "capital_requested": "50", "time_to_first_dollar_estimate": "7 days", "expected_margin": "90%", "build_complexity": "LOW", "market_evidence": [], "differentiation": "AI", "major_risks": ["Excluded"], "confidence": 0.5}), {})
        return super().complete(request)


class ManyRejectThenAcceptProvider(SequenceProvider):
    def complete(self, request: ModelRequestContract) -> ModelResponseContract:
        if "Bounded public web market research" in request.prompt:
            return super().complete(request)
        time.sleep(0.02)
        self.company_calls += 1
        if self.company_calls <= 4:
            return ModelResponseContract("sol", json.dumps({"title": f"RFP Copilot {self.company_calls}", "one_line_thesis": "AI RFP response drafting for SaaS teams.", "description": "Automates request for proposal answers.", "problem": "Sales teams hate RFPs.", "target_customer": "B2B SaaS sales teams", "proposed_solution": "Proposal drafting automation", "business_model": "SaaS", "pricing_hypothesis": "$99/month", "acquisition_strategy": "Content", "validation_plan": "Interview later", "capital_requested": "50", "time_to_first_dollar_estimate": "7 days", "expected_margin": "90%", "build_complexity": "LOW", "market_evidence": [], "differentiation": "AI", "major_risks": ["Excluded"], "confidence": 0.5}), {})
        return super().complete(request)


def service() -> VentureDiscoveryService:
    provider = SequenceProvider()
    return VentureDiscoveryService(ModelRouter({"sol": provider, "luna": provider}), web_research_available=True, ideation_model_hint="sol", research_model_hint="luna")


def rejecting_service() -> VentureDiscoveryService:
    provider = RejectThenAcceptProvider()
    return VentureDiscoveryService(ModelRouter({"sol": provider, "luna": provider}), web_research_available=True, ideation_model_hint="sol", research_model_hint="luna")


def many_rejecting_service() -> VentureDiscoveryService:
    provider = ManyRejectThenAcceptProvider()
    return VentureDiscoveryService(ModelRouter({"sol": provider, "luna": provider}), web_research_available=True, ideation_model_hint="sol", research_model_hint="luna")


def test_bounded_map_runs_concurrently_and_preserves_result_order() -> None:
    svc = VentureDiscoveryService()
    svc._effective_concurrency = lambda concurrency: concurrency
    active = 0
    observed_peak = 0
    lock = threading.Lock()

    def worker(value: int) -> int:
        nonlocal active, observed_peak
        with lock:
            active += 1
            observed_peak = max(observed_peak, active)
        time.sleep(0.02)
        with lock:
            active -= 1
        return value * 2

    assert svc._bounded_map([1, 2, 3, 4], worker, 2) == [2, 4, 6, 8]
    assert observed_peak == 2
    assert svc._last_bounded_map_peak == 2


def test_evidence_tiers_probability_ceiling_and_score_orientation() -> None:
    svc = service()
    proposal = svc.generate_single_company(svc.create_v0_mandate())
    assert proposal.evidence_tier == EvidenceTier.TIER_0_THESIS
    assert svc.calibrate_probability(proposal, raw_probability=80)["evidence_adjusted_probability"] == EVIDENCE_CEILINGS[EvidenceTier.TIER_0_THESIS]
    svc.conduct_market_research(proposal)
    assert proposal.evidence_tier == EvidenceTier.TIER_1_PUBLIC_EVIDENCE
    assert svc.calibrate_probability(proposal, raw_probability=80)["evidence_adjusted_probability"] == EVIDENCE_CEILINGS[EvidenceTier.TIER_1_PUBLIC_EVIDENCE]
    diligence = svc.start_ic_diligence(proposal)
    svc.generate_ic_questions(diligence)
    svc.answer_questions(diligence)
    svc.red_team(diligence)
    svc.final_company_response(diligence)
    decision = svc.score_and_decide(diligence)
    assert set(decision.component_scores) == set(SCORE_DIMENSIONS)
    assert "AI Leverage" in decision.component_scores
    assert "Platformization Potential" in decision.component_scores
    assert all("100 = highly attractive" in definition for definition in decision.score_definitions.values())
    assert decision.probability_500_within_30_days <= decision.evidence_ceiling
    assert decision.raw_probability_500_within_30_days >= decision.probability_500_within_30_days


def test_ai_native_platform_opportunities_receive_portfolio_preference() -> None:
    svc = service()
    mandate = svc.create_v0_mandate()
    ai = svc.generate_single_company(mandate)
    ai.title = "AI Support Knowledge Copilot"
    ai.description = "AI agent platform that monitors support tickets, synthesizes answers, and builds a reusable knowledge workflow."
    ai.proposed_solution = "Agent-assisted operational software with recurring automation and data accumulation."
    ai.business_model = "Monthly SaaS platform plus AI-enabled service onboarding."
    ai.differentiation = "Local inference and agent workflows automate most delivery with low marginal labor."
    ai.save(update_fields=["title", "description", "proposed_solution", "business_model", "differentiation", "updated_at"])
    generic = svc.generate_single_company(mandate)
    generic.title = "Emergency Shopify Audit Fix"
    generic.description = "One-time manual Shopify audit and emergency fix service."
    generic.proposed_solution = "Manual one-time consulting audit."
    generic.business_model = "One-time consulting service."
    generic.differentiation = "Fast human review."
    generic.save(update_fields=["title", "description", "proposed_solution", "business_model", "differentiation", "updated_at"])

    assert svc.ai_leverage_score(ai) > svc.ai_leverage_score(generic)
    assert svc.platformization_potential(ai) > svc.platformization_potential(generic)
    assert svc._generic_concentration_penalty(generic) > svc._generic_concentration_penalty(ai)


def test_identity_contamination_detection_and_fingerprint_collision() -> None:
    svc = service()
    proposal = svc.generate_single_company(svc.create_v0_mandate())
    contaminated = {"Company": "Unrelated CRM SaaS", "ICP": "enterprise banks", "Problem": "loan defaults", "Product": "risk platform"}
    result = svc.validate_identity_content(proposal, contaminated)
    assert result["passed"] is False
    fp = svc.fingerprint_proposal(proposal)
    assert isinstance(fp, VentureThesisFingerprint)
    other = svc.generate_single_company(proposal.mandate)
    other.title = proposal.title + " Copy"
    other.target_customer = proposal.target_customer
    other.problem = proposal.problem
    other.proposed_solution = proposal.proposed_solution
    other.business_model = proposal.business_model
    other.save(update_fields=["title", "target_customer", "problem", "proposed_solution", "business_model", "updated_at"])
    overlap = svc.classify_overlap(proposal, other)
    assert overlap["classification"] in {OverlapClassification.NEAR_DUPLICATE, OverlapClassification.DUPLICATE, OverlapClassification.COMPETITIVE}


def test_cohort_size_ranking_top3_capability_aggregation_and_graph_lineage() -> None:
    svc = service()
    version = champion_venture_discovery_cohort_graph_v1()
    graph_run = GraphRun.objects.create(execution_graph_version=version, current_node=version.graph_spec["entry"])
    LangGraphRuntime(venture_discovery_cohort_registry(svc, cohort_size=10, concurrency=2)).run_until_terminal_or_paused(graph_run)
    graph_run.refresh_from_db()
    cohort = VentureCohort.objects.get(id=graph_run.metadata["cohort_id"])

    assert graph_run.status == GraphRunStatus.COMPLETE
    assert cohort.members.count() == 10
    assert CompanyProposal.objects.count() == 10
    assert cohort.members.filter(is_top_3=True).count() == len(cohort.portfolio_review.top_3)
    assert PortfolioICReview.objects.filter(cohort=cohort).exists()
    assert VentureCollision.objects.filter(cohort=cohort).count() == 45
    assert VentureCapabilityDemand.objects.filter(cohort=cohort).exists()
    assert cohort.metadata["real_spend"] == 0
    assert cohort.metadata["real_customer_outreach"] is False
    assert cohort.concurrency == 2
    assert cohort.metrics["peak_concurrency"] == 1
    assert cohort.metrics["proposal_generation_peak_concurrency"] == 1
    assert cohort.metrics["research_peak_concurrency"] == 1
    assert cohort.metrics["individual_diligence_peak_concurrency"] == 1
    assert cohort.metrics["finalist_deep_research_count"] == 5
    assert "ranking_changes_after_deep_research" in cohort.metrics
    report = cohort.mandate.artifacts.get(artifact_type="VENTURE_DISCOVERY_COHORT_REPORT")
    assert len(report.content["rankings"]) == 10
    assert len(report.content["top_3"]) <= 3
    assert report.content["total_spend"] == 0
    assert CohortIdeationMandate.objects.filter(cohort=cohort).exists()
    assert PortfolioThesisCluster.objects.filter(cohort=cohort).count() == 10
    assert PortfolioSaturationAnalysis.objects.filter(cohort=cohort).exists()
    assert "ideation_mandate" in report.content
    assert "thesis_registry_before" in report.content
    assert "thesis_registry_after" in report.content
    assert "saturation_analysis" in report.content
    assert "idea_diversity_metrics" in report.content


def test_v03_thesis_registry_mandate_and_territory_allocation() -> None:
    svc = service()
    cohort = svc.prepare_cohort(size=10)
    review = svc.portfolio_thesis_review(cohort)
    mandate = svc.create_ideation_mandate(cohort)

    assert PortfolioThesis.objects.filter(canonical_name="AI RFP response automation for B2B SaaS", status=PortfolioThesisStatus.SATURATED).exists()
    assert "AI RFP response automation for B2B SaaS" in [row["canonical_name"] for row in review["saturated_thesis_areas"]]
    assert mandate.hard_exclusions == DEFAULT_HARD_EXCLUSIONS
    assert mandate.opportunity_territories[:2] == [OpportunityTerritory.VERTICAL_AI_WRAPPERS, OpportunityTerritory.VERTICAL_AI_WRAPPERS]
    assert mandate.opportunity_territories[-2:] == [OpportunityTerritory.OPEN_CATEGORY, OpportunityTerritory.OPEN_CATEGORY]


def test_v03_registry_seed_historical_metadata_is_json_serializable() -> None:
    svc = service()
    proposal = svc.generate_single_company(svc.create_v0_mandate())
    proposal.title = "RFP Historical Candidate"
    proposal.description = "AI RFP response drafting for SaaS teams."
    proposal.proposed_solution = "RFP proposal automation."
    proposal.save(update_fields=["title", "description", "proposed_solution", "updated_at"])

    svc.seed_dogfood_thesis_registry()

    thesis = PortfolioThesis.objects.get(canonical_name="AI RFP response automation for B2B SaaS")
    assert isinstance(thesis.metadata["historical"]["best_company"], str)


def test_v03_novelty_gate_blocks_hard_exclusion_and_allows_soft_exception() -> None:
    svc = service()
    cohort = svc.prepare_cohort(size=2)
    mandate = svc.create_ideation_mandate(cohort)

    hard_payload = {"title": "RFP Copilot", "one_line_thesis": "AI RFP response drafting", "description": "Drafts proposal responses", "problem": "RFPs", "target_customer": "SaaS sales", "proposed_solution": "RFP automation", "business_model": "SaaS", "pricing_hypothesis": "$99", "acquisition_strategy": "content", "validation_plan": "test", "differentiation": "AI"}
    soft_payload = {**hard_payload, "title": "Regulated Contract Workflow Monitor", "one_line_thesis": "Contract scanner for regulated vendor workflows", "description": "Contract scanner with proprietary recurring workflow data", "problem": "Regulated teams miss renewal obligations", "target_customer": "Compliance teams", "proposed_solution": "Vertical contract workflow monitor", "differentiation": "Vertical proprietary workflow data", "exception_rationale": "specific regulated workflow with recurring data"}

    assert svc.novelty_gate(hard_payload, cohort, [], mandate, territory="OPEN_CATEGORY")["decision"] == NoveltyGateDecision.REGENERATE_HARD_EXCLUSION
    assert svc.novelty_gate(soft_payload, cohort, [], mandate, territory="DATA_DOCUMENT_AUTOMATION")["decision"] == NoveltyGateDecision.ACCEPT


def test_v03_generation_regenerates_rejected_slots_and_preserves_size() -> None:
    svc = rejecting_service()
    cohort = svc.prepare_cohort(size=3)
    svc.portfolio_thesis_review(cohort)
    svc.create_ideation_mandate(cohort)

    proposals = svc.generate_independent_proposals(cohort)

    assert len(proposals) == 3
    assert cohort.members.count() == 3
    assert VentureGenerationRejection.objects.filter(cohort=cohort, decision=NoveltyGateDecision.REGENERATE_HARD_EXCLUSION).count() == 1
    assert cohort.metrics["hard_exclusion_rejections"] == 1
    assert cohort.metrics["failed_slots"] == 0


def test_v04_generation_uses_concurrency_and_cohort_attempt_pool() -> None:
    svc = many_rejecting_service()
    svc._effective_concurrency = lambda concurrency: concurrency
    cohort = svc.prepare_cohort(size=3, concurrency=2)
    cohort.scoring_policy = {**cohort.scoring_policy, "duplicate_policy": {**cohort.scoring_policy["duplicate_policy"], "cohort_attempt_budget_multiplier": 4}}
    cohort.save(update_fields=["scoring_policy", "updated_at"])
    svc.portfolio_thesis_review(cohort)
    svc.create_ideation_mandate(cohort)

    proposals = svc.generate_independent_proposals(cohort)

    assert len(proposals) == 3
    assert cohort.metrics["proposal_generation_peak_concurrency"] == 2
    assert cohort.metrics["replacement_attempts"] > 0
    assert cohort.metrics["generation_attempts"] > cohort.cohort_size
    assert cohort.status == "PROPOSALS_GENERATED"


def test_v04_partial_complete_when_attempt_budget_exhausted() -> None:
    svc = rejecting_service()
    cohort = svc.prepare_cohort(size=3)
    cohort.scoring_policy = {**cohort.scoring_policy, "duplicate_policy": {**cohort.scoring_policy["duplicate_policy"], "cohort_attempt_budget_multiplier": 1}}
    cohort.save(update_fields=["scoring_policy", "updated_at"])
    svc.portfolio_thesis_review(cohort)
    svc.create_ideation_mandate(cohort)

    proposals = svc.generate_independent_proposals(cohort)

    assert len(proposals) < 3
    assert cohort.status == "INSUFFICIENT_ACCEPTED_PROPOSALS"


def test_v04_semantic_duplicate_clusters_known_variants() -> None:
    svc = service()
    cohort = svc.prepare_cohort(size=3)
    mandate = svc.create_ideation_mandate(cohort)
    first = {"title": "PermitDoc Compliance Checker", "one_line_thesis": "Municipal permit compliance checker", "description": "Checks permit compliance for contractors", "problem": "Permit misses", "target_customer": "Residential contractors", "proposed_solution": "Permit compliance checker", "business_model": "SaaS", "pricing_hypothesis": "$99", "acquisition_strategy": "content", "validation_plan": "test", "differentiation": "workflow"}
    second = {**first, "title": "PermitDoc Pre-check", "one_line_thesis": "Municipal permit pre-check for contractors"}
    legacy = {**first, "title": "Legacy Modernization Triage Agent", "one_line_thesis": "Legacy modernization copilot triages risky code migrations", "description": "Legacy modernization triage agent", "problem": "legacy migration risk", "proposed_solution": "triage agent"}

    assert svc.current_cohort_match(second, [first])["classification"] == "NEAR_DUPLICATE"
    assert svc.semantic_cluster_key(first) == svc.semantic_cluster_key(second)
    assert svc.semantic_cluster_key(legacy) == "legacy_modernization_triage"
    assert svc.novelty_gate(second, cohort, [first, second], mandate, territory="OPEN_CATEGORY")["decision"] == NoveltyGateDecision.REGENERATE_DUPLICATE


def test_v04_research_source_quality_filters_irrelevant_pages() -> None:
    svc = service()
    proposal = svc.generate_single_company(svc.create_v0_mandate())
    sources = [
        {"url": "https://example.com/pricing", "title": "Pricing plans for workflow automation", "category": "pricing", "summary": proposal.title + " price cost subscription"},
        {"url": "https://archive.org/noise", "title": "Archived unrelated page", "category": "pricing", "summary": "celebrity news"},
        {"url": "https://example.com/noise", "title": "Sports scores", "category": "pricing", "summary": "football fixtures"},
    ]

    quality = svc.filter_research_sources(proposal, sources, ["pricing"])

    assert len(quality["accepted_sources"]) == 1
    assert quality["accepted_sources"][0]["quality"] == "strong"
    assert len(quality["rejected_sources"]) == 2


def test_run_venture_cohort_management_command_outputs_summary(monkeypatch) -> None:
    import control_plane.ventures.management.commands.run_venture_cohort as command_module

    provider = SequenceProvider()
    monkeypatch.setattr(command_module, "providers_from_resources", lambda: {"qwen": provider})
    stdout = StringIO()

    call_command("run_venture_cohort", "--size", "3", "--concurrency", "2", "--qwen-only", "--no-web-research", stdout=stdout)

    summary = json.loads(stdout.getvalue())
    cohort = VentureCohort.objects.get(cohort_id=summary["cohort_id"])

    assert summary["status"] == GraphRunStatus.COMPLETE
    assert summary["members"] == 3
    assert summary["concurrency"] == 2
    assert summary["fallback_count"] == 0
    assert summary["generation_sources"] == ["qwen"]
    assert len(summary["top_3"]) <= 3
    assert cohort.members.count() == 3


def test_export_venture_cohort_management_command_outputs_all_companies(tmp_path) -> None:
    svc = service()
    version = champion_venture_discovery_cohort_graph_v1()
    graph_run = GraphRun.objects.create(execution_graph_version=version, current_node=version.graph_spec["entry"])
    LangGraphRuntime(venture_discovery_cohort_registry(svc, cohort_size=3, concurrency=1)).run_until_terminal_or_paused(graph_run)
    cohort = VentureCohort.objects.get(id=graph_run.metadata["cohort_id"])
    markdown_path = tmp_path / "cohort.md"
    json_path = tmp_path / "cohort.json"

    call_command("export_venture_cohort", "--cohort", cohort.cohort_id, "--output", str(markdown_path))
    call_command("export_venture_cohort", "--cohort", cohort.cohort_id, "--format", "json", "--output", str(json_path))

    markdown = markdown_path.read_text(encoding="utf-8")
    payload = json.loads(json_path.read_text(encoding="utf-8"))

    assert "## All Company Details" in markdown
    assert markdown.count("### Rank") == 3
    assert "Validation condition:" in markdown
    assert "Autonomous operability:" in markdown
    assert payload["cohort"]["cohort_id"] == cohort.cohort_id
    assert len(payload["companies"]) == 3
    assert payload["companies"][0]["decision"]["decision"]
    assert payload["companies"][0]["autonomous_assessment"]["venture_track"]