diff --git a/agents/venture_discovery.py b/agents/venture_discovery.py index f6ec6c4..0602c49 100644 --- a/agents/venture_discovery.py +++ b/agents/venture_discovery.py @@ -3,12 +3,15 @@ from __future__ import annotations import hashlib import json import re +import threading import time +from concurrent.futures import ThreadPoolExecutor, as_completed from itertools import combinations from decimal import Decimal from typing import Any from collections import Counter, defaultdict +from django.db import close_old_connections, connection from django.utils import timezone from control_plane.events.bus import EventBus @@ -31,6 +34,7 @@ class VentureDiscoveryService: self.bus = bus or EventBus() self.web_research_available = web_research_available self.research_model_hint = research_model_hint + self._last_bounded_map_peak = 0 def create_v0_mandate(self) -> CompanyMandate: mandate = CompanyMandate.objects.create( @@ -230,24 +234,38 @@ class VentureDiscoveryService: 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 v1", "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}, evidence_calibration_policy={tier: ceiling for tier, ceiling in EVIDENCE_CEILINGS.items()}, metadata={"real_spend": 0, "real_customer_outreach": False}) def generate_independent_proposals(self, cohort: VentureCohort) -> list[CompanyProposal]: - proposals = [] - for index in range(cohort.cohort_size): + started = time.monotonic() + + def generate(index: int) -> tuple[int, str]: proposal = self.generate_single_company(cohort.mandate, ideation_index=index + 1) - proposal.metadata = {**proposal.metadata, "cohort_id": cohort.cohort_id, "independent_generation_index": index + 1, "prior_ideas_visible": False} + return index, str(proposal.id) + + generated = self._bounded_map(range(cohort.cohort_size), generate, cohort.concurrency) + peak_concurrency = self._last_bounded_map_peak + proposals = [] + for index, proposal_id in sorted(generated, key=lambda item: item[0]): + proposal = CompanyProposal.objects.get(id=proposal_id) + generation_index = index + 1 + proposal.metadata = {**proposal.metadata, "cohort_id": cohort.cohort_id, "independent_generation_index": generation_index, "prior_ideas_visible": False} proposal.save(update_fields=["metadata", "updated_at"]) - VentureCohortMember.objects.create(cohort=cohort, proposal=proposal, metadata={"generation_index": index + 1}) + VentureCohortMember.objects.create(cohort=cohort, proposal=proposal, metadata={"generation_index": generation_index}) proposals.append(proposal) cohort.status = "PROPOSALS_GENERATED" - cohort.save(update_fields=["status", "updated_at"]) + cohort.metrics = {**cohort.metrics, "proposal_generation_runtime_seconds": round(time.monotonic() - started, 2), "proposal_generation_peak_concurrency": peak_concurrency} + cohort.save(update_fields=["metrics", "status", "updated_at"]) return proposals def research_cohort(self, cohort: VentureCohort) -> None: started = time.monotonic() - sources = 0 - for member in cohort.members.select_related("proposal").order_by("created_at"): - research = self.conduct_market_research(member.proposal) - sources += len(research.get("sources", [])) - cohort.metrics = {**cohort.metrics, "research_runtime_seconds": round(time.monotonic() - started, 2), "total_sources": sources, "public_research_queries": cohort.members.count(), "peak_concurrency": cohort.concurrency} + + 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", [])) + + 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)} cohort.status = "RESEARCHED" cohort.save(update_fields=["metrics", "status", "updated_at"]) @@ -264,7 +282,9 @@ class VentureDiscoveryService: def run_individual_diligence_for_cohort(self, cohort: VentureCohort) -> None: started = time.monotonic() - for member in cohort.members.select_related("proposal"): + + def diligence(member_id: str) -> None: + member = VentureCohortMember.objects.select_related("proposal").get(id=member_id) proposal = member.proposal self.board_review(proposal) diligence = self.start_ic_diligence(proposal) @@ -277,7 +297,11 @@ class VentureDiscoveryService: self.produce_investment_memo(diligence, gap) member.child_graph_run = GraphRun.objects.create(execution_graph_version=cohort.graph_run.execution_graph_version if cohort.graph_run else None, status=GraphRunStatus.COMPLETE, metadata={"logical_child_company_run": True, "proposal_id": str(proposal.id), "cohort_id": cohort.cohort_id}) if cohort.graph_run else None member.save(update_fields=["child_graph_run", "updated_at"]) - cohort.metrics = {**cohort.metrics, "individual_diligence_runtime_seconds": round(time.monotonic() - started, 2)} + + member_ids = [str(member.id) for member in cohort.members.order_by("created_at")] + self._bounded_map(member_ids, diligence, cohort.concurrency) + peak_concurrency = self._last_bounded_map_peak + cohort.metrics = {**cohort.metrics, "individual_diligence_runtime_seconds": round(time.monotonic() - started, 2), "individual_diligence_peak_concurrency": peak_concurrency, "peak_concurrency": max(cohort.metrics.get("peak_concurrency", 0), peak_concurrency)} cohort.status = "INDIVIDUAL_DILIGENCE_COMPLETE" cohort.save(update_fields=["metrics", "status", "updated_at"]) @@ -352,6 +376,48 @@ class VentureDiscoveryService: adjusted = min(float(raw_probability), ceiling) return {"raw_probability": float(raw_probability), "evidence_adjusted_probability": adjusted, "evidence_ceiling": ceiling, "evidence_tier": tier, "explanation": f"{tier} caps P($500/30d) at {ceiling}%; IC uses {adjusted}%."} + def _bounded_map(self, items: list[Any] | range, worker: Any, concurrency: int) -> list[Any]: + self._last_bounded_map_peak = 0 + workers = self._effective_concurrency(concurrency) + if workers <= 1: + results = [worker(item) for item in items] + self._last_bounded_map_peak = 1 if results else 0 + return results + results: list[tuple[int, Any]] = [] + active = 0 + active_lock = threading.Lock() + + def tracked_worker(item: Any) -> Any: + nonlocal active + with active_lock: + active += 1 + self._last_bounded_map_peak = max(self._last_bounded_map_peak, active) + try: + return self._threaded_worker(worker, item) + finally: + with active_lock: + active -= 1 + + with ThreadPoolExecutor(max_workers=workers) as executor: + futures = {executor.submit(tracked_worker, item): index for index, item in enumerate(items)} + for future in as_completed(futures): + results.append((futures[future], future.result())) + results.sort(key=lambda item: item[0]) + return [result for _, result in results] + + def _threaded_worker(self, worker: Any, item: Any) -> Any: + close_old_connections() + try: + return worker(item) + finally: + close_old_connections() + + def _effective_concurrency(self, concurrency: int) -> int: + requested = max(1, int(concurrency or 1)) + if connection.vendor == "sqlite" and connection.settings_dict.get("NAME") == ":memory:": + return 1 + return requested + def validate_identity_content(self, proposal: CompanyProposal, content: Any) -> dict[str, Any]: text = json.dumps(content, default=str).lower() checks = { diff --git a/control_plane/resources/management/commands/model_provider_health.py b/control_plane/resources/management/commands/model_provider_health.py index 5761fa8..1401211 100644 --- a/control_plane/resources/management/commands/model_provider_health.py +++ b/control_plane/resources/management/commands/model_provider_health.py @@ -11,7 +11,8 @@ class Command(BaseCommand): help = "Check configured model provider health without crashing the control plane." def handle(self, *args, **options): - for resource in Resource.objects.filter(is_active=True, kind="MODEL"): + resources = sorted(Resource.objects.filter(is_active=True, kind="MODEL"), key=self.sort_key) + for resource in resources: if resource.provider == "opencode": status = SolProvider(resource).health() elif resource.provider == "local_inference": @@ -22,3 +23,10 @@ class Command(BaseCommand): resource.last_health_check_at = timezone.now() resource.save(update_fields=["health_status", "last_health_check_at", "updated_at"]) self.stdout.write(f"{resource.name}: {status}") + + def sort_key(self, resource: Resource) -> tuple[int, str]: + model_key = str(resource.config.get("model_key") or "").lower() + if resource.provider == "local_inference": + model_key = "qwen" + order = {"qwen": 0, "sol": 1, "terra": 2, "luna": 3} + return order.get(model_key, 100), resource.name diff --git a/control_plane/resources/management/commands/seed_spark_resources.py b/control_plane/resources/management/commands/seed_spark_resources.py index 796f118..058fc25 100644 --- a/control_plane/resources/management/commands/seed_spark_resources.py +++ b/control_plane/resources/management/commands/seed_spark_resources.py @@ -1,6 +1,7 @@ from __future__ import annotations import os +import shutil from django.core.management.base import BaseCommand @@ -10,6 +11,39 @@ from control_plane.resources.models import Resource, ResourceKind class Command(BaseCommand): help = "Seed DGX Spark compute and model resources for Sol and Qwen." + def opencode_command(self, model_key: str) -> str: + env_name = f"ARTIFEX_{model_key.upper()}_OPENCODE_COMMAND" + if os.environ.get(env_name): + return os.environ[env_name] + executable = shutil.which("opencode") or "opencode" + return f"{executable} run" + + def opencode_config(self, model_key: str) -> dict[str, object]: + return { + "model_key": model_key, + "transport": os.environ.get(f"ARTIFEX_{model_key.upper()}_TRANSPORT", os.environ.get("ARTIFEX_OPENCODE_TRANSPORT", "local")), + "command": self.opencode_command(model_key), + "use_pty": os.environ.get(f"ARTIFEX_{model_key.upper()}_USE_PTY", os.environ.get("ARTIFEX_OPENCODE_USE_PTY", "0")) == "1", + "timeout_seconds": int(os.environ.get(f"ARTIFEX_{model_key.upper()}_TIMEOUT_SECONDS", os.environ.get("ARTIFEX_OPENCODE_TIMEOUT_SECONDS", "120"))), + } + + def update_opencode_resource(self, *, model_key: str, name: str, roles: list[str], compute: Resource) -> None: + defaults = { + "name": name, + "kind": ResourceKind.MODEL, + "provider": "opencode", + "compute": compute, + "roles": roles, + "config": self.opencode_config(model_key), + } + existing = next((resource for resource in Resource.objects.filter(provider="opencode") if resource.config.get("model_key") == model_key), None) + if existing is not None: + for field, value in defaults.items(): + setattr(existing, field, value) + existing.save(update_fields=[*defaults.keys(), "updated_at"]) + return + Resource.objects.update_or_create(name=name, defaults={key: value for key, value in defaults.items() if key != "name"}) + def handle(self, *args, **options): spark, _ = Resource.objects.update_or_create( name=os.environ.get("ARTIFEX_SPARK_RESOURCE_NAME", "DGX Spark"), @@ -20,20 +54,23 @@ class Command(BaseCommand): "roles": [], }, ) - Resource.objects.update_or_create( + self.update_opencode_resource( + model_key="sol", name=os.environ.get("ARTIFEX_SOL_RESOURCE_NAME", "GPT-5.6 Sol"), - defaults={ - "kind": ResourceKind.MODEL, - "provider": "opencode", - "compute": spark, - "roles": ["PROJECT_BRAIN", "PLANNING", "ARCHAEOLOGY_INTERPRETATION"], - "config": { - "transport": os.environ.get("ARTIFEX_SOL_TRANSPORT", "local"), - "command": os.environ.get("ARTIFEX_SOL_OPENCODE_COMMAND", "/home/daniel/.opencode/bin/opencode run"), - "use_pty": os.environ.get("ARTIFEX_SOL_USE_PTY", "1") == "1", - "timeout_seconds": int(os.environ.get("ARTIFEX_SOL_TIMEOUT_SECONDS", "120")), - }, - }, + roles=["PROJECT_BRAIN", "PLANNING", "ARCHAEOLOGY_INTERPRETATION", "AGENT_DESIGN", "ESCALATION"], + compute=spark, + ) + self.update_opencode_resource( + model_key="terra", + name=os.environ.get("ARTIFEX_TERRA_RESOURCE_NAME", "GPT-5.6 Terra"), + roles=["REASONING", "REVIEW", "PORTFOLIO_IC", "STRATEGY"], + compute=spark, + ) + self.update_opencode_resource( + model_key="luna", + name=os.environ.get("ARTIFEX_LUNA_RESOURCE_NAME", "GPT-5.6 Luna"), + roles=["REASONING", "MARKET_RESEARCH", "WEB_RESEARCH", "SYNTHESIS"], + compute=spark, ) Resource.objects.update_or_create( name=os.environ.get("ARTIFEX_QWEN_RESOURCE_NAME", "Qwen"), diff --git a/model_router/providers.py b/model_router/providers.py index 1b7142e..9bbc362 100644 --- a/model_router/providers.py +++ b/model_router/providers.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import os import re import shlex import subprocess @@ -62,7 +63,7 @@ class SolProvider: transport = str(config.get("transport", "ssh")) use_pty = bool(config.get("use_pty", False)) if transport == "local": - argv = [*shlex.split(command), request.prompt] + argv = [*shlex.split(command, posix=os.name != "nt"), request.prompt] if use_pty: shell_command = " ".join(shlex.quote(part) for part in argv) argv = ["script", "-q", "-e", "-c", shell_command, "/dev/null"] @@ -92,7 +93,7 @@ class SolProvider: config = self.resource.config try: if config.get("transport") == "local": - command = shlex.split(str(config.get("command", "opencode")))[0] + command = shlex.split(str(config.get("command", "opencode")), posix=os.name != "nt")[0] completed = subprocess.run([command, "--version"], capture_output=True, text=True, timeout=10, check=False) else: compute = self.resource.compute @@ -161,10 +162,16 @@ class QwenProvider: def providers_from_resources() -> dict[str, object]: providers: dict[str, object] = {} - sol = Resource.objects.filter(is_active=True, provider="opencode").first() + opencode_resources = Resource.objects.filter(is_active=True, provider="opencode") qwen = Resource.objects.filter(is_active=True, provider="local_inference").first() - if sol is not None: - providers["sol"] = SolProvider(sol) + for resource in opencode_resources: + model_key = str(resource.config.get("model_key") or "").strip().lower() + if model_key: + providers[model_key] = SolProvider(resource) + if "sol" not in providers: + sol = opencode_resources.first() + if sol is not None: + providers["sol"] = SolProvider(sol) if qwen is not None: providers["qwen"] = QwenProvider(qwen) return providers diff --git a/model_router/router.py b/model_router/router.py index f2158c9..380b021 100644 --- a/model_router/router.py +++ b/model_router/router.py @@ -99,8 +99,11 @@ class ModelRouter: provider_names = {provider_key} if provider_key == "qwen": provider_names.add("local_inference") - if provider_key == "sol": + if provider_key in {"sol", "terra", "luna"}: provider_names.add("opencode") + for resource in candidates: + if resource.config.get("model_key") == provider_key: + return resource for resource in candidates: if resource.provider in provider_names and role in resource.roles: return resource diff --git a/tests/test_model_router_providers.py b/tests/test_model_router_providers.py index b7e6035..64a3db3 100644 --- a/tests/test_model_router_providers.py +++ b/tests/test_model_router_providers.py @@ -1,7 +1,8 @@ from __future__ import annotations -from agents.providers import DeterministicCodingProvider +from agents.providers import DeterministicCodingProvider, DeterministicSolProvider from control_plane.resources.models import ModelRequest, Resource, ResourceKind +from model_router.providers import SolProvider, providers_from_resources from model_router.router import ModelCapability, ModelRequestContract, ModelRouter @@ -28,3 +29,24 @@ def test_model_router_health_is_non_throwing() -> None: router = ModelRouter({"qwen": DeterministicCodingProvider()}) assert router.health() == {"qwen": "AVAILABLE"} + + +def test_opencode_model_key_resources_load_as_distinct_providers() -> None: + for key in ["sol", "terra", "luna"]: + Resource.objects.create(name=key.title(), kind=ResourceKind.MODEL, provider="opencode", roles=["REASONING"], config={"model_key": key}) + + providers = providers_from_resources() + + assert {"sol", "terra", "luna"}.issubset(providers) + assert all(isinstance(providers[key], SolProvider) for key in ["sol", "terra", "luna"]) + + +def test_model_router_prefers_exact_model_key_resource_for_persisted_requests() -> None: + Resource.objects.create(name="Generic Opencode", kind=ResourceKind.MODEL, provider="opencode", roles=["REASONING"], config={}) + Resource.objects.create(name="Terra", kind=ResourceKind.MODEL, provider="opencode", roles=["REASONING"], config={"model_key": "terra"}) + router = ModelRouter({"terra": DeterministicSolProvider("{}")}, persist_requests=True) + + router.complete(ModelRequestContract(purpose=ModelCapability.REASONING, model_hint="terra", prompt="{}")) + + request = ModelRequest.objects.get() + assert request.model == "Terra" diff --git a/tests/test_venture_discovery_cohort_v02.py b/tests/test_venture_discovery_cohort_v02.py index bf70089..aebdf4d 100644 --- a/tests/test_venture_discovery_cohort_v02.py +++ b/tests/test_venture_discovery_cohort_v02.py @@ -1,6 +1,8 @@ from __future__ import annotations import json +import threading +import time from agents.venture_discovery import EVIDENCE_CEILINGS, SCORE_DEFINITIONS, SCORE_DIMENSIONS, VentureDiscoveryService from control_plane.ventures.models import CompanyProposal, EvidenceTier, OverlapClassification, PortfolioICReview, VentureCapabilityDemand, VentureCohort, VentureCollision, VentureThesisFingerprint @@ -38,6 +40,28 @@ def service() -> VentureDiscoveryService: return VentureDiscoveryService(ModelRouter({"sol": provider, "luna": provider}), web_research_available=True) +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()) @@ -120,7 +144,10 @@ def test_cohort_size_ranking_top3_capability_aggregation_and_graph_lineage() -> assert cohort.metadata["real_spend"] == 0 assert cohort.metadata["real_customer_outreach"] is False assert cohort.concurrency == 2 - assert cohort.metrics["peak_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 report = cohort.mandate.artifacts.get(artifact_type="VENTURE_DISCOVERY_COHORT_REPORT") assert len(report.content["rankings"]) == 10 assert len(report.content["top_3"]) == 3