diff --git a/agents/venture_discovery.py b/agents/venture_discovery.py index 421c186..96da7b9 100644 --- a/agents/venture_discovery.py +++ b/agents/venture_discovery.py @@ -32,7 +32,8 @@ AI_NATIVE_POLICY = {"preference": "Prefer AI-native opportunities where Artifex DEFAULT_HARD_EXCLUSIONS = ["AI RFP response drafting / proposal automation", "SaaS churn prediction / retention playbook generation"] DEFAULT_SOFT_EXCLUSIONS = ["generic contract scanners", "Shopify audit products", "generic AI content generators", "generic productized consulting", "generic one-off audits", "generic chatbot wrappers"] DEFAULT_TERRITORY_ALLOCATION = [OpportunityTerritory.VERTICAL_AI_WRAPPERS, OpportunityTerritory.VERTICAL_AI_WRAPPERS, OpportunityTerritory.ENTERPRISE_WORKFLOW_AUTOMATION, OpportunityTerritory.SMB_AUTOMATION, OpportunityTerritory.DEVELOPER_AI_INFRASTRUCTURE, OpportunityTerritory.INTELLIGENCE_MONITORING, OpportunityTerritory.DATA_DOCUMENT_AUTOMATION, OpportunityTerritory.AI_ENABLED_SERVICE_TO_PLATFORM, OpportunityTerritory.OPEN_CATEGORY, OpportunityTerritory.OPEN_CATEGORY] -DEFAULT_DUPLICATE_POLICY = {"max_near_duplicate_per_thesis_per_cohort": 1, "max_competitive_per_thesis_per_cohort": 2, "max_attempts_per_slot": 3} +DEFAULT_DUPLICATE_POLICY = {"max_near_duplicate_per_thesis_per_cohort": 1, "max_competitive_per_thesis_per_cohort": 2, "max_attempts_per_slot": 3, "cohort_attempt_budget_multiplier": 4} +RESEARCH_CATEGORIES = ["competitors", "pricing", "customer_pain", "market_alternatives", "regulatory_platform_risks"] class VentureDiscoveryService: @@ -98,14 +99,14 @@ class VentureDiscoveryService: self.bus.publish("VENTURE_COMPANY_PROPOSED", payload={"proposal_id": str(proposal.id), "source": source}) return proposal - def conduct_market_research(self, proposal: CompanyProposal, *, graph_run=None) -> dict[str, Any]: - required = ["competitors", "pricing", "customer_pain", "market_alternatives", "regulatory_platform_risks"] + def conduct_market_research(self, proposal: CompanyProposal, *, graph_run=None, depth: str = "light") -> dict[str, Any]: + required = RESEARCH_CATEGORIES research = {"coverage": {category: False for category in required}, "sources": [], "findings": {}, "unverified_categories": required, "research_available": False} - search_sources = self._searxng_sources(proposal, required) if self.web_research_available else [] + search_sources = self._searxng_sources(proposal, required, depth=depth) if self.web_research_available else [] page_corpus = self._page_corpus(search_sources) if search_sources else [] if self.web_research_available and self.router is not None: try: - response = self.router.complete(ModelRequestContract(purpose=ModelCapability.REASONING, model_hint=self.research_model_hint, prompt="Bounded public web market research for exactly one startup pitch. Return JSON with keys: sources (list of {url,title,category,summary}), findings (object keyed by competitors, pricing, customer_pain, market_alternatives, regulatory_platform_risks), and coverage (object with each required key true/false). Use only the provided SearXNG search context and fetched page excerpts for source URLs; do not fabricate URLs. If a category has no source, mark coverage false. Search context: " + json.dumps(search_sources, default=str) + " Page excerpts: " + json.dumps(page_corpus, default=str) + " Pitch: " + json.dumps(proposal.pitch, default=str))) + response = self.router.complete(ModelRequestContract(purpose=ModelCapability.REASONING, model_hint=self.research_model_hint, prompt="Bounded public web market research for exactly one startup pitch. Return JSON with keys: sources (list of {url,title,category,summary}), findings (object keyed by competitors, pricing, customer_pain, market_alternatives, regulatory_platform_risks), and coverage (object with each required key true/false). Use only the provided SearXNG search context and fetched page excerpts for source URLs; do not fabricate URLs. If a category has no strong relevant source, mark coverage false. Research depth: " + depth + ". Search context: " + json.dumps(search_sources, default=str) + " Page excerpts: " + json.dumps(page_corpus, default=str) + " Pitch: " + json.dumps(proposal.pitch, default=str))) parsed = extract_json_object(response.content) if isinstance(parsed, dict): research = self._normalize_research(parsed, required) @@ -119,7 +120,11 @@ class VentureDiscoveryService: research = {**research, "page_corpus": page_corpus, "page_fetch_count": len(page_corpus)} if not research.get("sources"): research = self._missing_research(required, research.get("failure", "public web research unavailable or returned no source-linked evidence")) - source_categories = {str(source.get("category", "")) for source in research.get("sources", []) if source.get("url")} + quality = self.filter_research_sources(proposal, research.get("sources", []), required) + research["sources"] = quality["accepted_sources"] + research["source_rejections"] = quality["rejected_sources"] + research["source_rejection_count"] = len(quality["rejected_sources"]) + source_categories = {str(source.get("category", "")) for source in research.get("sources", []) if source.get("url") and source.get("quality") == "strong"} coverage = dict(research.get("coverage", {})) for category in required: coverage[category] = bool(coverage.get(category)) and category in source_categories @@ -127,15 +132,16 @@ class VentureDiscoveryService: research["unverified_categories"] = [category for category in required if not coverage.get(category)] coverage_ratio = (len(required) - len(research["unverified_categories"])) / len(required) proposal.confidence = round(min(float(proposal.confidence), 0.45 + 0.35 * coverage_ratio), 2) - proposal.market_evidence = [*self._as_list(proposal.market_evidence), *research.get("sources", []), {"type": "research_coverage", "source": "venture_research", "summary": f"Source-linked research coverage: {round(coverage_ratio * 100)}%", "coverage": coverage, "unverified_categories": research["unverified_categories"]}] - proposal.metadata = {**proposal.metadata, "research": {"coverage_ratio": coverage_ratio, "unverified_categories": research["unverified_categories"], "source_count": len(research.get("sources", [])), "page_fetch_count": research.get("page_fetch_count", 0), "provider": research.get("provider", "none"), "search_provider": research.get("search_provider", "none")}} + existing_research = proposal.metadata.get("research", {}) if isinstance(proposal.metadata, dict) else {} + proposal.market_evidence = [*self._as_list(proposal.market_evidence), *research.get("sources", []), {"type": "research_coverage", "source": "venture_research", "summary": f"{depth.title()} source-linked research coverage: {round(coverage_ratio * 100)}%", "coverage": coverage, "unverified_categories": research["unverified_categories"], "depth": depth}] + proposal.metadata = {**proposal.metadata, "research": {**existing_research, "coverage_ratio": coverage_ratio, "coverage": coverage, "unverified_categories": research["unverified_categories"], "source_count": len(research.get("sources", [])), "source_rejection_count": len(quality["rejected_sources"]), "page_fetch_count": research.get("page_fetch_count", 0), "provider": research.get("provider", "none"), "search_provider": research.get("search_provider", "none"), "depth": depth, "explicit_research_failure": research.get("failure", "") if coverage_ratio < 0.8 else "", "before_deep_coverage_ratio": existing_research.get("coverage_ratio") if depth == "deep" else existing_research.get("before_deep_coverage_ratio")}} proposal.evidence_tier = EvidenceTier.TIER_1_PUBLIC_EVIDENCE if coverage_ratio > 0 else EvidenceTier.TIER_0_THESIS if proposal.thesis: proposal.thesis.evidence_tier = proposal.evidence_tier proposal.thesis.save(update_fields=["evidence_tier", "updated_at"]) proposal.pitch = {**proposal.pitch, "Research evidence": research} proposal.save(update_fields=["confidence", "market_evidence", "metadata", "pitch", "evidence_tier", "updated_at"]) - self._artifact(proposal, proposal.mandate, "MARKET_RESEARCH", "Bounded Public Market Research", research, self._readable_research(research), f"{research.get('provider', 'search')}/web research" if research.get("sources") else "research_gap", graph_run=graph_run) + self._artifact(proposal, proposal.mandate, "MARKET_RESEARCH", f"Bounded Public Market Research ({depth})", research, self._readable_research(research), f"{research.get('provider', 'search')}/web research" if research.get("sources") else "research_gap", graph_run=graph_run) return research def board_review(self, proposal: CompanyProposal, *, graph_run=None) -> CompanyBoardReview: @@ -209,7 +215,7 @@ class VentureDiscoveryService: composite = round(sum(scores.values()) / len(scores), 1) decision_type = ICDecisionType.REVISE_AND_RESUBMIT if scores["Demand Evidence"] < 35 or scores["Risk Manageability"] < 35 else ICDecisionType.CONDITIONAL_FUND if composite >= 55 else ICDecisionType.REVISE_AND_RESUBMIT tranche = Decimal("20.00") if decision_type == ICDecisionType.CONDITIONAL_FUND and composite >= 70 else Decimal("10.00") if decision_type == ICDecisionType.CONDITIONAL_FUND else None - decision = ICDecision.objects.create(diligence=diligence, decision=decision_type, component_scores=scores, composite_score=composite, probability_500_within_30_days=scores["Probability of Reaching $500"], initial_tranche=tranche, validation_condition="Obtain 5 credible target-customer responses or 1 explicit willingness-to-pay signal before any build or further spend.", evidence_required=["response transcripts or public thread URLs", "proof of willingness-to-pay signal", "no-spam/no-fabrication compliance note"], recommended_allocation={"initial_tranche": float(tranche or 0), "remaining_reserved": 50 - float(tranche or 0), "no_spend_in_v0": True}, kill_criteria=diligence.final_response.get("kill_criteria", []), next_decision_point="After validation evidence is collected and before any real spend or customer delivery.", metadata={"decision_vocabulary": [item.value for item in ICDecisionType], "no_actual_funding": True, "score_basis": "source_linked_research_and_pitch_attributes", "probability_calibration": calibration}, evidence_tier=diligence.proposal.evidence_tier, raw_probability_500_within_30_days=calibration["raw_probability"], evidence_ceiling=calibration["evidence_ceiling"], probability_explanation=calibration["explanation"], score_definitions=SCORE_DEFINITIONS) + decision, _ = ICDecision.objects.update_or_create(diligence=diligence, defaults={"decision": decision_type, "component_scores": scores, "composite_score": composite, "probability_500_within_30_days": scores["Probability of Reaching $500"], "initial_tranche": tranche, "validation_condition": "Obtain 5 credible target-customer responses or 1 explicit willingness-to-pay signal before any build or further spend.", "evidence_required": ["response transcripts or public thread URLs", "proof of willingness-to-pay signal", "no-spam/no-fabrication compliance note"], "recommended_allocation": {"initial_tranche": float(tranche or 0), "remaining_reserved": 50 - float(tranche or 0), "no_spend_in_v0": True}, "kill_criteria": diligence.final_response.get("kill_criteria", []), "next_decision_point": "After validation evidence is collected and before any real spend or customer delivery.", "metadata": {"decision_vocabulary": [item.value for item in ICDecisionType], "no_actual_funding": True, "score_basis": "source_linked_research_and_pitch_attributes", "probability_calibration": calibration}, "evidence_tier": diligence.proposal.evidence_tier, "raw_probability_500_within_30_days": calibration["raw_probability"], "evidence_ceiling": calibration["evidence_ceiling"], "probability_explanation": calibration["explanation"], "score_definitions": SCORE_DEFINITIONS}) diligence.status = "DECISION" diligence.save(update_fields=["status", "updated_at"]) diligence.proposal.status = CompanyProposalStatus.FUNDED_RECOMMENDED if decision.decision == ICDecisionType.CONDITIONAL_FUND else CompanyProposalStatus.REVISE @@ -251,8 +257,8 @@ class VentureDiscoveryService: def prepare_cohort(self, *, size: int = 10, graph_run=None, concurrency: int = 1) -> VentureCohort: self.seed_dogfood_thesis_registry() mandate = self.create_v0_mandate() - cohort_id = f"VDV03-{timezone.now().strftime('%Y%m%d%H%M%S')}-{hashlib.sha1(str(mandate.id).encode()).hexdigest()[:8]}" - return VentureCohort.objects.create(cohort_id=cohort_id, mandate=mandate, cohort_size=size, graph_run=graph_run, concurrency=concurrency, status="PREPARING", graph_versions={"cohort": "venture_discovery_cohort v2", "company": "venture_discovery v1"}, research_policy={"stage_a": "lightweight_all", "deeper_research": "top_5_if_required", "no_spend": True, "no_customer_outreach": True}, scoring_policy={"dimensions": SCORE_DEFINITIONS, "duplicate_policy": DEFAULT_DUPLICATE_POLICY}, evidence_calibration_policy={tier: ceiling for tier, ceiling in EVIDENCE_CEILINGS.items()}, metadata={"real_spend": 0, "real_customer_outreach": False, "milestone": "VENTURE_DISCOVERY_V0.3"}) + cohort_id = f"VDV04-{timezone.now().strftime('%Y%m%d%H%M%S')}-{hashlib.sha1(str(mandate.id).encode()).hexdigest()[:8]}" + return VentureCohort.objects.create(cohort_id=cohort_id, mandate=mandate, cohort_size=size, graph_run=graph_run, concurrency=concurrency, status="PREPARING", graph_versions={"cohort": "venture_discovery_cohort v3", "company": "venture_discovery v1"}, research_policy={"stage_a": "lightweight_all", "finalist_deeper_research": "top_5", "target_finalist_coverage": 0.8, "no_spend": True, "no_customer_outreach": True}, scoring_policy={"dimensions": SCORE_DEFINITIONS, "duplicate_policy": DEFAULT_DUPLICATE_POLICY}, evidence_calibration_policy={tier: ceiling for tier, ceiling in EVIDENCE_CEILINGS.items()}, metadata={"real_spend": 0, "real_customer_outreach": False, "milestone": "VENTURE_DISCOVERY_V0.4"}) def portfolio_thesis_review(self, cohort: VentureCohort) -> dict[str, Any]: self.seed_dogfood_thesis_registry() @@ -295,45 +301,57 @@ class VentureDiscoveryService: accepted_payloads: list[dict[str, Any]] = [] proposals = [] mandate = getattr(cohort, "ideation_mandate", None) or self.create_ideation_mandate(cohort) - max_attempts = int(cohort.scoring_policy.get("duplicate_policy", DEFAULT_DUPLICATE_POLICY).get("max_attempts_per_slot", 3)) - attempts = hard_rejections = duplicate_rejections = soft_reviews = failed_slots = model_requests = 0 - for index in range(cohort.cohort_size): - territory = mandate.opportunity_territories[index % len(mandate.opportunity_territories)] if mandate.opportunity_territories else OpportunityTerritory.OPEN_CATEGORY - accepted: tuple[dict[str, Any], str, dict[str, Any]] | None = None - for attempt in range(1, max_attempts + 1): + policy = cohort.scoring_policy.get("duplicate_policy", DEFAULT_DUPLICATE_POLICY) + total_budget = max(cohort.cohort_size, int(cohort.cohort_size * int(policy.get("cohort_attempt_budget_multiplier", 4)))) + attempts = hard_rejections = duplicate_rejections = soft_reviews = model_requests = replacement_attempts = 0 + workers = self._effective_concurrency(cohort.concurrency) + peak_concurrency = 0 + generation_records: list[dict[str, Any]] = [] + while len(proposals) < cohort.cohort_size and attempts < total_budget: + batch_size = min(workers, total_budget - attempts, cohort.cohort_size - len(proposals) + workers - 1) + batch_inputs = [] + for _ in range(batch_size): attempts += 1 model_requests += 1 - context = {"hard_exclusions": mandate.hard_exclusions, "soft_exclusions": mandate.soft_exclusions, "territory": territory, "brief": self.ideation_mandate_payload(mandate)} - payload, source = self._company_payload(cohort.mandate, ideation_index=index + 1, ideation_context=context) - gate = self.novelty_gate(payload, cohort, accepted_payloads, mandate, territory=str(territory)) - if gate["decision"] == NoveltyGateDecision.ACCEPT: - accepted = (payload, source, gate) - break + replacement_attempts += 1 if attempts > cohort.cohort_size else 0 + slot_index = len(proposals) + 1 + territory = mandate.opportunity_territories[(attempts - 1) % len(mandate.opportunity_territories)] if mandate.opportunity_territories else OpportunityTerritory.OPEN_CATEGORY + batch_inputs.append({"attempt": attempts, "slot_index": slot_index, "territory": str(territory)}) + + def generate_attempt(item: dict[str, Any]) -> dict[str, Any]: + context = {"hard_exclusions": mandate.hard_exclusions, "soft_exclusions": mandate.soft_exclusions, "territory": item["territory"], "brief": self.ideation_mandate_payload(mandate)} + payload, source = self._company_payload(cohort.mandate, ideation_index=item["attempt"], ideation_context=context) + return {**item, "payload": payload, "source": source} + + for result in self._bounded_map(batch_inputs, generate_attempt, cohort.concurrency): + peak_concurrency = max(peak_concurrency, self._last_bounded_map_peak) + gate = self.novelty_gate(result["payload"], cohort, accepted_payloads, mandate, territory=result["territory"]) + generation_records.append({"attempt": result["attempt"], "decision": gate["decision"], "title": result["payload"].get("title", ""), "territory": result["territory"]}) + if gate["decision"] == NoveltyGateDecision.ACCEPT and len(proposals) < cohort.cohort_size: + proposal = self.generate_single_company(cohort.mandate, ideation_index=len(proposals) + 1, payload=result["payload"], source=result["source"]) + generation_index = len(proposals) + 1 + proposal.metadata = {**proposal.metadata, "cohort_id": cohort.cohort_id, "independent_generation_index": generation_index, "prior_ideas_visible": False, "search_territory": result["territory"], "novelty_gate": {k: v for k, v in gate.items() if k != "matched_thesis"}} + proposal.save(update_fields=["metadata", "updated_at"]) + VentureCohortMember.objects.create(cohort=cohort, proposal=proposal, metadata={"generation_index": generation_index, "source_attempt": result["attempt"]}) + proposals.append(proposal) + accepted_payloads.append(result["payload"]) + continue if gate["decision"] == NoveltyGateDecision.REGENERATE_HARD_EXCLUSION: hard_rejections += 1 elif gate["decision"] == NoveltyGateDecision.REGENERATE_DUPLICATE: duplicate_rejections += 1 elif gate["decision"] == NoveltyGateDecision.REVIEW_SOFT_EXCLUSION: soft_reviews += 1 - VentureGenerationRejection.objects.create(cohort=cohort, slot_index=index + 1, attempt=attempt, decision=gate["decision"], reason=gate["explanation"], candidate=payload, matched_thesis=gate.get("matched_thesis"), similarity_score=float(gate.get("similarity_score", 0.0)), metadata={"territory": str(territory), "classification": gate.get("classification")}) - if accepted is None: - failed_slots += 1 - VentureGenerationRejection.objects.create(cohort=cohort, slot_index=index + 1, attempt=max_attempts, decision=NoveltyGateDecision.FAILED_IDEATION, reason="max ideation attempts exhausted", metadata={"territory": str(territory)}) - continue - payload, source, gate = accepted - proposal = self.generate_single_company(cohort.mandate, ideation_index=index + 1, payload=payload, source=source) - generation_index = index + 1 - proposal.metadata = {**proposal.metadata, "cohort_id": cohort.cohort_id, "independent_generation_index": generation_index, "prior_ideas_visible": False, "search_territory": str(territory), "novelty_gate": {k: v for k, v in gate.items() if k != "matched_thesis"}} - proposal.save(update_fields=["metadata", "updated_at"]) - VentureCohortMember.objects.create(cohort=cohort, proposal=proposal, metadata={"generation_index": generation_index}) - proposals.append(proposal) - accepted_payloads.append(payload) - cohort.status = "PROPOSALS_GENERATED" - cohort.metrics = {**cohort.metrics, "requested_company_count": cohort.cohort_size, "generation_attempts": attempts, "accepted_proposals": len(proposals), "hard_exclusion_rejections": hard_rejections, "duplicate_rejections": duplicate_rejections, "soft_exclusion_reviews": soft_reviews, "failed_slots": failed_slots, "average_attempts_per_company": round(attempts / max(1, len(proposals)), 2), "generation_model_requests": model_requests, "proposal_generation_runtime_seconds": round(time.monotonic() - started, 2), "proposal_generation_peak_concurrency": 1, "peak_concurrency": max(cohort.metrics.get("peak_concurrency", 0), 1)} + VentureGenerationRejection.objects.create(cohort=cohort, slot_index=result["slot_index"], attempt=result["attempt"], decision=gate["decision"], reason=gate["explanation"], candidate=result["payload"], matched_thesis=gate.get("matched_thesis"), similarity_score=float(gate.get("similarity_score", 0.0)), metadata={"territory": result["territory"], "classification": gate.get("classification"), "semantic_cluster_key": gate.get("semantic_cluster_key", "")}) + if len(proposals) >= cohort.cohort_size: + break + failed_slots = max(0, cohort.cohort_size - len(proposals)) + if failed_slots: + for missing in range(len(proposals) + 1, cohort.cohort_size + 1): + VentureGenerationRejection.objects.create(cohort=cohort, slot_index=missing, attempt=attempts, decision=NoveltyGateDecision.FAILED_IDEATION, reason="cohort-level attempt budget exhausted", metadata={"attempt_budget": total_budget}) + cohort.status = "PROPOSALS_GENERATED" if len(proposals) == cohort.cohort_size else "INSUFFICIENT_ACCEPTED_PROPOSALS" + cohort.metrics = {**cohort.metrics, "requested_company_count": cohort.cohort_size, "generation_attempts": attempts, "cohort_attempt_budget": total_budget, "accepted_proposals": len(proposals), "replacement_attempts": replacement_attempts, "hard_exclusion_rejections": hard_rejections, "duplicate_rejections": duplicate_rejections, "soft_exclusion_reviews": soft_reviews, "failed_slots": failed_slots, "average_attempts_per_company": round(attempts / max(1, len(proposals)), 2), "generation_model_requests": model_requests, "proposal_generation_runtime_seconds": round(time.monotonic() - started, 2), "proposal_generation_peak_concurrency": peak_concurrency, "peak_concurrency": max(cohort.metrics.get("peak_concurrency", 0), peak_concurrency), "generation_records": generation_records[-50:]} cohort.save(update_fields=["metrics", "status", "updated_at"]) - if len(proposals) != cohort.cohort_size: - cohort.status = "FAILED_IDEATION" - cohort.save(update_fields=["status", "updated_at"]) return proposals def research_cohort(self, cohort: VentureCohort) -> None: @@ -341,12 +359,13 @@ class VentureDiscoveryService: def research(member_id: str) -> int: member = VentureCohortMember.objects.select_related("proposal").get(id=member_id) - return len(self.conduct_market_research(member.proposal).get("sources", [])) + return len(self.conduct_market_research(member.proposal, depth="light").get("sources", [])) member_ids = [str(member.id) for member in cohort.members.order_by("created_at")] source_counts = self._bounded_map(member_ids, research, cohort.concurrency) peak_concurrency = self._last_bounded_map_peak - cohort.metrics = {**cohort.metrics, "research_runtime_seconds": round(time.monotonic() - started, 2), "total_sources": sum(source_counts), "public_research_queries": len(member_ids), "research_peak_concurrency": peak_concurrency, "peak_concurrency": max(cohort.metrics.get("peak_concurrency", 0), peak_concurrency)} + source_rejections = sum((member.proposal.metadata.get("research", {}) if isinstance(member.proposal.metadata, dict) else {}).get("source_rejection_count", 0) for member in cohort.members.select_related("proposal")) + cohort.metrics = {**cohort.metrics, "research_runtime_seconds": round(time.monotonic() - started, 2), "total_sources": sum(source_counts), "source_rejection_count": source_rejections, "public_research_queries": len(member_ids), "research_peak_concurrency": peak_concurrency, "peak_concurrency": max(cohort.metrics.get("peak_concurrency", 0), peak_concurrency)} cohort.status = "RESEARCHED" cohort.save(update_fields=["metrics", "status", "updated_at"]) @@ -413,6 +432,41 @@ class VentureDiscoveryService: cohort.save(update_fields=["metrics", "status", "updated_at"]) def portfolio_ic(self, cohort: VentureCohort) -> PortfolioICReview: + preliminary_rows = self.portfolio_ranking_rows(cohort) + preliminary_top_5 = preliminary_rows[:5] + before_coverage = {} + for row in preliminary_top_5: + proposal = CompanyProposal.objects.get(id=row["proposal_id"]) + before_coverage[str(proposal.id)] = (proposal.metadata.get("research", {}) if isinstance(proposal.metadata, dict) else {}).get("coverage_ratio", 0.0) + deep_started = time.monotonic() + for row in preliminary_top_5: + proposal = CompanyProposal.objects.get(id=row["proposal_id"]) + self.conduct_market_research(proposal, depth="deep") + diligence = proposal.ic_diligence.order_by("-created_at").first() + if diligence: + self.score_and_decide(diligence) + after_coverage = {} + for row in preliminary_top_5: + proposal = CompanyProposal.objects.get(id=row["proposal_id"]) + after_coverage[str(proposal.id)] = (proposal.metadata.get("research", {}) if isinstance(proposal.metadata, dict) else {}).get("coverage_ratio", 0.0) + rows = self.portfolio_ranking_rows(cohort) + preliminary_rank_by_id = {row["proposal_id"]: index + 1 for index, row in enumerate(preliminary_rows)} + ranking_changes = [{"proposal_id": row["proposal_id"], "company": row["company"], "preliminary_rank": preliminary_rank_by_id.get(row["proposal_id"]), "final_rank": index + 1} for index, row in enumerate(rows) if preliminary_rank_by_id.get(row["proposal_id"]) != index + 1] + for rank, row in enumerate(rows, start=1): + member = cohort.members.get(proposal_id=row["proposal_id"]) + member.rank = rank + member.is_top_3 = rank <= 3 + member.portfolio_score = row["portfolio_score"] + member.save(update_fields=["rank", "is_top_3", "portfolio_score", "updated_at"]) + row["rank"] = rank + concentration = self._portfolio_concentration(cohort) + review, _ = PortfolioICReview.objects.update_or_create(cohort=cohort, defaults={"rankings": rows, "top_3": rows[:3], "concentration": concentration, "metadata": {"no_funding": True, "preliminary_rankings": preliminary_rows, "preliminary_top_5": preliminary_top_5, "finalist_research_coverage_before": before_coverage, "finalist_research_coverage_after": after_coverage, "ranking_changes_after_deep_research": ranking_changes}}) + cohort.metrics = {**cohort.metrics, "finalist_deep_research_count": len(preliminary_top_5), "finalist_deep_research_runtime_seconds": round(time.monotonic() - deep_started, 2), "finalist_research_coverage_before": before_coverage, "finalist_research_coverage_after": after_coverage, "ranking_changes_after_deep_research": ranking_changes} + cohort.status = "PORTFOLIO_IC_COMPLETE" + cohort.save(update_fields=["metrics", "status", "updated_at"]) + return review + + def portfolio_ranking_rows(self, cohort: VentureCohort) -> list[dict[str, Any]]: rows = [] collision_risk = defaultdict(int) for collision in cohort.collisions.exclude(classification=OverlapClassification.NONE): @@ -427,18 +481,7 @@ class VentureDiscoveryService: score = round(decision.composite_score + decision.probability_500_within_30_days * 0.2 + ai_bonus - capability_burden * 1.5 - collision_risk[str(proposal.id)] * 2 - concentration_penalty, 1) rows.append({"proposal_id": str(proposal.id), "company": proposal.title, "thesis": proposal.pitch.get("One-line thesis", proposal.description), "ic_score": decision.composite_score, "probability": decision.probability_500_within_30_days, "decision": decision.decision, "evidence_tier": decision.evidence_tier, "initial_tranche": str(decision.initial_tranche or "0"), "ai_leverage": decision.component_scores.get("AI Leverage", 0), "platformization_potential": decision.component_scores.get("Platformization Potential", 0), "portfolio_score": score, "capability_burden": capability_burden, "collision_risk": collision_risk[str(proposal.id)], "concentration_penalty": concentration_penalty}) rows.sort(key=lambda item: item["portfolio_score"], reverse=True) - for rank, row in enumerate(rows, start=1): - member = cohort.members.get(proposal_id=row["proposal_id"]) - member.rank = rank - member.is_top_3 = rank <= 3 - member.portfolio_score = row["portfolio_score"] - member.save(update_fields=["rank", "is_top_3", "portfolio_score", "updated_at"]) - row["rank"] = rank - concentration = self._portfolio_concentration(cohort) - review, _ = PortfolioICReview.objects.update_or_create(cohort=cohort, defaults={"rankings": rows, "top_3": rows[:3], "concentration": concentration, "metadata": {"no_funding": True}}) - cohort.status = "PORTFOLIO_IC_COMPLETE" - cohort.save(update_fields=["status", "updated_at"]) - return review + return rows def saturation_analysis(self, cohort: VentureCohort) -> list[dict[str, Any]]: rows = [] @@ -546,9 +589,10 @@ class VentureDiscoveryService: clusters = [{"company": cluster.proposal.title, "portfolio_thesis": cluster.portfolio_thesis.canonical_name if cluster.portfolio_thesis else "", "classification": cluster.classification, "similarity_score": cluster.similarity_score, "explanation": cluster.explanation} for cluster in cohort.thesis_clusters.select_related("proposal", "portfolio_thesis").order_by("created_at")] generation_rejections = [{"slot_index": rejection.slot_index, "attempt": rejection.attempt, "decision": rejection.decision, "reason": rejection.reason, "candidate_title": rejection.candidate.get("title", "") if isinstance(rejection.candidate, dict) else "", "similarity_score": rejection.similarity_score} for rejection in cohort.generation_rejections.order_by("slot_index", "attempt")] saturation = [{"thesis": analysis.portfolio_thesis.canonical_name, "proposal_count": analysis.proposal_count, "best_ic_score": analysis.best_ic_score, "score_spread": analysis.score_spread, "status_recommendation": analysis.status_recommendation, "rationale": analysis.rationale} for analysis in cohort.saturation_analyses.select_related("portfolio_thesis").order_by("-proposal_count")] - content = {"cohort_id": cohort.cohort_id, "mandate": cohort.mandate.objective, "ideation_mandate": self.ideation_mandate_payload(ideation_mandate) if ideation_mandate else {}, "hard_exclusions": ideation_mandate.hard_exclusions if ideation_mandate else [], "soft_exclusions": ideation_mandate.soft_exclusions if ideation_mandate else [], "search_territories": ideation_mandate.opportunity_territories if ideation_mandate else [], "generation_rejections": generation_rejections, "thesis_registry_before": (ideation_mandate.registry_snapshot.get("theses", []) if ideation_mandate else []), "thesis_registry_after": cohort.metadata.get("thesis_registry_after", []), "thesis_clusters": clusters, "saturation_analysis": saturation, "idea_diversity_metrics": self.diversity_metrics(cohort), "runtime": cohort.metrics, "total_spend": 0, "customer_outreach": "none", "rankings": review.rankings, "top_3": review.top_3, "collisions": self._collision_summary(cohort), "portfolio_concentration": review.concentration, "capability_demand": review.capability_demand, "top_3_capability_gaps": review.top_3_capability_gaps, "recommended_build_priorities": review.recommended_build_priorities} + accepted_count = cohort.members.count() + content = {"cohort_id": cohort.cohort_id, "mandate": cohort.mandate.objective, "accepted_count": accepted_count, "requested_count": cohort.cohort_size, "ideation_mandate": self.ideation_mandate_payload(ideation_mandate) if ideation_mandate else {}, "hard_exclusions": ideation_mandate.hard_exclusions if ideation_mandate else [], "soft_exclusions": ideation_mandate.soft_exclusions if ideation_mandate else [], "search_territories": ideation_mandate.opportunity_territories if ideation_mandate else [], "generation_rejections": generation_rejections, "thesis_registry_before": (ideation_mandate.registry_snapshot.get("theses", []) if ideation_mandate else []), "thesis_registry_after": cohort.metadata.get("thesis_registry_after", []), "thesis_clusters": clusters, "saturation_analysis": saturation, "idea_diversity_metrics": self.diversity_metrics(cohort), "runtime": cohort.metrics, "total_spend": 0, "customer_outreach": "none", "rankings": review.rankings, "top_3": review.top_3, "collisions": self._collision_summary(cohort), "portfolio_concentration": review.concentration, "capability_demand": review.capability_demand, "top_3_capability_gaps": review.top_3_capability_gaps, "recommended_build_priorities": review.recommended_build_priorities} readable = self._readable_cohort_report(content) - cohort.status = "COMPLETE" + cohort.status = "COMPLETE" if accepted_count == cohort.cohort_size else "PARTIAL_COMPLETE" cohort.save(update_fields=["status", "updated_at"]) return VentureArtifact.objects.create(mandate=cohort.mandate, graph_run=cohort.graph_run, artifact_type="VENTURE_DISCOVERY_COHORT_REPORT", name="Venture Discovery Cohort Report", content=content, readable=readable, generated_by="Portfolio IC") @@ -725,6 +769,10 @@ class VentureDiscoveryService: cohort_match = self.current_cohort_match(payload, accepted_payloads) if cohort_match["classification"] == ThesisMatchClassification.DUPLICATE: return {"decision": NoveltyGateDecision.REGENERATE_DUPLICATE, **cohort_match, "territory": territory} + if cohort_match["classification"] == ThesisMatchClassification.NEAR_DUPLICATE: + cluster_count = sum(1 for item in accepted_payloads if self.semantic_cluster_key(item) == cohort_match.get("semantic_cluster_key")) + if cluster_count >= int(DEFAULT_DUPLICATE_POLICY["max_near_duplicate_per_thesis_per_cohort"]) + 1: + return {"decision": NoveltyGateDecision.REGENERATE_DUPLICATE, **cohort_match, "territory": territory} if cohort_match["classification"] == ThesisMatchClassification.COMPETITIVE: canonical = self.canonical_thesis_name(payload) competitive_count = sum(1 for item in accepted_payloads if self.canonical_thesis_name(item) == canonical) @@ -774,13 +822,15 @@ class VentureDiscoveryService: def current_cohort_match(self, payload: dict[str, Any], accepted_payloads: list[dict[str, Any]]) -> dict[str, Any]: canonical = self.canonical_thesis_name(payload) + cluster_key = self.semantic_cluster_key(payload) best_score = 0.0 best_name = "" for accepted in accepted_payloads: accepted_name = self.canonical_thesis_name(accepted) - if accepted_name == canonical: - return {"matched_thesis": None, "similarity_score": 1.0, "classification": ThesisMatchClassification.DUPLICATE, "explanation": f"Current cohort already accepted canonical thesis {accepted_name}."} - score = 1.0 if accepted_name == canonical else self._jaccard(self.payload_text(payload), self.payload_text(accepted)) + if str(accepted.get("title", "")).strip().lower() == str(payload.get("title", "")).strip().lower(): + return {"matched_thesis": None, "similarity_score": 1.0, "classification": ThesisMatchClassification.DUPLICATE, "explanation": f"Current cohort already accepted exact title {accepted.get('title', '')}.", "semantic_cluster_key": cluster_key} + accepted_cluster_key = self.semantic_cluster_key(accepted) + score = 0.86 if accepted_cluster_key == cluster_key else self._jaccard(self.payload_text(payload), self.payload_text(accepted)) if score > best_score: best_score = score best_name = accepted_name @@ -791,7 +841,7 @@ class VentureDiscoveryService: classification = ThesisMatchClassification.COMPETITIVE elif best_score >= 0.18: classification = ThesisMatchClassification.ADJACENT - return {"matched_thesis": None, "similarity_score": round(best_score, 2), "classification": classification, "explanation": f"Current cohort nearest thesis {best_name or 'none'} at similarity {best_score:.2f}."} + return {"matched_thesis": None, "similarity_score": round(best_score, 2), "classification": classification, "explanation": f"Current cohort nearest thesis {best_name or 'none'} at similarity {best_score:.2f}.", "semantic_cluster_key": cluster_key} def create_portfolio_thesis_from_proposal(self, proposal: CompanyProposal, cohort: VentureCohort) -> PortfolioThesis: fingerprint = self.fingerprint_proposal(proposal) @@ -812,9 +862,26 @@ class VentureDiscoveryService: return "Freelancer contract risk scanning" if "shopify" in text and "audit" in text: return "Shopify compliance/audit" + semantic = self.semantic_cluster_key(payload) + if semantic == "municipal_permit_compliance": + return "Municipal permit compliance automation" + if semantic == "legacy_modernization_triage": + return "Legacy modernization triage copilot" title = str(payload.get("title", "Untitled thesis")).strip() return re.sub(r"\s+", " ", title)[:255] + def semantic_cluster_key(self, payload: dict[str, Any]) -> str: + text = self.payload_text(payload) + if ("permit" in text or "permitdoc" in text) and any(word in text for word in ["compliance", "pre-check", "precheck", "municipal", "contractor"]): + return "municipal_permit_compliance" + if "legacy" in text and any(word in text for word in ["modernization", "modernisation", "triage", "copilot", "agent"]): + return "legacy_modernization_triage" + if "vendor" in text and "onboarding" in text and "compliance" in text: + return "vendor_onboarding_compliance" + title = str(payload.get("title", "")).lower() + tokens = [token for token in re.findall(r"[a-z0-9]{4,}", title) if token not in {"agent", "copilot", "checker", "monitor", "platform", "automation", "workflow", "service", "precheck", "check"}] + return "_".join(tokens[:4]) or self._fingerprint(payload)[:12] + def payload_text(self, payload: dict[str, Any]) -> str: return " ".join(str(payload.get(key, "")) for key in ["title", "one_line_thesis", "description", "problem", "target_customer", "proposed_solution", "business_model", "pricing_hypothesis", "acquisition_strategy", "validation_plan", "differentiation"]).lower() @@ -876,11 +943,12 @@ class VentureDiscoveryService: coverage = {category: bool(dict(payload.get("coverage", {})).get(category)) for category in required} return {"coverage": coverage, "sources": sources, "findings": dict(payload.get("findings", {})), "unverified_categories": []} - def _searxng_sources(self, proposal: CompanyProposal, required: list[str]) -> list[dict[str, Any]]: + def _searxng_sources(self, proposal: CompanyProposal, required: list[str], *, depth: str = "light") -> list[dict[str, Any]]: client = self.search_client or SearxngSearchClient.from_resources() if client is None: return [] sources = [] + limit = 5 if depth == "deep" else 3 query_terms = { "competitors": "competitors alternatives", "pricing": "pricing cost", @@ -889,12 +957,12 @@ class VentureDiscoveryService: "regulatory_platform_risks": "regulatory platform risk compliance", } for category in required: - query = f"{proposal.title} {proposal.target_customer} {query_terms.get(category, category)}" + query = f"{proposal.title} {proposal.problem[:120]} {proposal.target_customer} {query_terms.get(category, category)}" try: - sources.extend(client.search(query, category=category, limit=3)) + sources.extend(client.search(query, category=category, limit=limit)) except Exception: continue - return sources[:15] + return sources[:25 if depth == "deep" else 15] def _page_corpus(self, search_sources: list[dict[str, Any]]) -> list[dict[str, Any]]: if self.page_fetcher is None: @@ -919,6 +987,52 @@ class VentureDiscoveryService: coverage[category] = True return {**research, "coverage": coverage, "sources": merged_sources, "research_available": True, "search_provider": "searxng"} + def filter_research_sources(self, proposal: CompanyProposal, sources: list[dict[str, Any]], required: list[str]) -> dict[str, list[dict[str, Any]]]: + accepted = [] + rejected = [] + seen = set() + for source in sources: + if not isinstance(source, dict): + continue + url = str(source.get("url", "")).strip() + if not url or url in seen: + continue + seen.add(url) + category = str(source.get("category", "")) + quality = self.source_quality(proposal, source, required) + if not quality["accepted"]: + rejected.append({**source, "quality": quality["quality"], "rejection_reason": quality["reason"]}) + continue + accepted.append({**source, "quality": quality["quality"], "relevance_score": quality["score"]}) + return {"accepted_sources": accepted, "rejected_sources": rejected} + + def source_quality(self, proposal: CompanyProposal, source: dict[str, Any], required: list[str]) -> dict[str, Any]: + category = str(source.get("category", "")) + url = str(source.get("url", "")).lower() + source_text = " ".join(str(source.get(key, "")) for key in ["title", "summary", "content", "snippet", "url"]).lower() + if category not in required: + return {"accepted": False, "quality": "weak", "score": 0.0, "reason": "unknown research category"} + if any(noisy in url for noisy in ["webcache", "translate.google", "pinterest", "facebook.com", "instagram.com", "x.com/intent", "archive.org", "web.archive.org"]): + return {"accepted": False, "quality": "weak", "score": 0.0, "reason": "archive/social/noisy source"} + company_text = " ".join([proposal.title, proposal.problem, proposal.target_customer, proposal.proposed_solution]).lower() + category_terms = { + "competitors": "competitor competitors alternative alternatives vendor software platform service", + "pricing": "pricing price cost plan subscription fee rates", + "customer_pain": "problem pain challenge complaint forum reddit customer", + "market_alternatives": "alternative alternatives tools services products market solution", + "regulatory_platform_risks": "regulatory compliance risk policy legal platform permit", + } + company_score = self._jaccard(company_text, source_text) + category_score = self._jaccard(category_terms.get(category, category), source_text) + if category.replace("_", " ") in source_text: + category_score = max(category_score, 0.5) + score = round(company_score * 0.65 + category_score * 0.35, 2) + if score < 0.08: + return {"accepted": False, "quality": "weak", "score": score, "reason": "insufficient semantic relevance"} + if score < 0.16: + return {"accepted": True, "quality": "weak", "score": score, "reason": "accepted as weak context only"} + return {"accepted": True, "quality": "strong", "score": score, "reason": "strong relevant source"} + def _missing_research(self, required: list[str], reason: str) -> dict[str, Any]: return {"coverage": {category: False for category in required}, "sources": [], "findings": {}, "unverified_categories": required, "research_available": False, "failure": reason} diff --git a/graph/venture_cohort.py b/graph/venture_cohort.py index 72e4a88..edd4cce 100644 --- a/graph/venture_cohort.py +++ b/graph/venture_cohort.py @@ -9,7 +9,7 @@ from graph.spec import ExecutionGraphSpec, GraphEdgeSpec, GraphNodeSpec def venture_discovery_cohort_graph_v1() -> ExecutionGraphSpec: nodes = ["prepare_cohort", "portfolio_thesis_review", "create_ideation_mandate", "allocate_search_territories", "generate_independent_proposals", "novelty_gate", "initial_research", "fingerprint_theses", "collision_analysis", "cluster_theses", "run_individual_diligence", "portfolio_compare", "portfolio_ic", "saturation_analysis", "update_thesis_registry", "aggregate_capabilities", "produce_cohort_report", "complete"] - spec = ExecutionGraphSpec(name="venture_discovery_cohort", version=2, graph_type="VENTURE_DISCOVERY_COHORT", entry="prepare_cohort", nodes={node: GraphNodeSpec(node, node if node == "complete" else f"venture_cohort_{node}") for node in nodes}, edges=[GraphEdgeSpec(nodes[index], nodes[index + 1], "success") for index in range(len(nodes) - 1)], terminal_nodes=["complete"], metadata={"description": "Venture Discovery V0.3: portfolio thesis registry, ideation mandate, novelty gates, saturation analysis, and cohort IC."}) + spec = ExecutionGraphSpec(name="venture_discovery_cohort", version=3, graph_type="VENTURE_DISCOVERY_COHORT", entry="prepare_cohort", nodes={node: GraphNodeSpec(node, node if node == "complete" else f"venture_cohort_{node}") for node in nodes}, edges=[GraphEdgeSpec(nodes[index], nodes[index + 1], "success") for index in range(len(nodes) - 1)], terminal_nodes=["complete"], metadata={"description": "Venture Discovery V0.4: concurrent replacement ideation, source-quality filtering, finalist deep research, rescore, and hardened completion status."}) spec.validate() return spec diff --git a/model_router/router.py b/model_router/router.py index b41a9c0..abeb26b 100644 --- a/model_router/router.py +++ b/model_router/router.py @@ -61,7 +61,7 @@ class ModelRouter: provider = self.providers.get(provider_key) if provider is None: raise RuntimeError(f"No model provider configured for {provider_key}") - model_resource = self._resource_for(provider_key, request.purpose) + model_resource = self._resource_for(provider_key, request.purpose) if self.persist_requests else None record = self._start_record(request, provider, model_resource) started = time.monotonic() try: diff --git a/tests/test_venture_discovery_cohort_v02.py b/tests/test_venture_discovery_cohort_v02.py index 853346e..17f0297 100644 --- a/tests/test_venture_discovery_cohort_v02.py +++ b/tests/test_venture_discovery_cohort_v02.py @@ -42,12 +42,24 @@ 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") @@ -58,6 +70,11 @@ def rejecting_service() -> VentureDiscoveryService: 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 @@ -166,6 +183,8 @@ def test_cohort_size_ranking_top3_capability_aggregation_and_graph_lineage() -> 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 @@ -234,6 +253,68 @@ def test_v03_generation_regenerates_rejected_slots_and_preserves_size() -> None: 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