Add bounded venture concurrency and model keys

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 = []
+        started = time.monotonic()
-        for index in range(cohort.cohort_size):
+
+        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"):
+        def research(member_id: str) -> int:
-            research = self.conduct_market_research(member.proposal)
+            member = VentureCohortMember.objects.select_related("proposal").get(id=member_id)
-            sources += len(research.get("sources", []))
+            return len(self.conduct_market_research(member.proposal).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}
+
+        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 = {

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

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={
+            roles=["PROJECT_BRAIN", "PLANNING", "ARCHAEOLOGY_INTERPRETATION", "AGENT_DESIGN", "ESCALATION"],
-                "kind": ResourceKind.MODEL,
+            compute=spark,
-                "provider": "opencode",
+        )
-                "compute": spark,
+        self.update_opencode_resource(
-                "roles": ["PROJECT_BRAIN", "PLANNING", "ARCHAEOLOGY_INTERPRETATION"],
+            model_key="terra",
-                "config": {
+            name=os.environ.get("ARTIFEX_TERRA_RESOURCE_NAME", "GPT-5.6 Terra"),
-                    "transport": os.environ.get("ARTIFEX_SOL_TRANSPORT", "local"),
+            roles=["REASONING", "REVIEW", "PORTFOLIO_IC", "STRATEGY"],
-                    "command": os.environ.get("ARTIFEX_SOL_OPENCODE_COMMAND", "/home/daniel/.opencode/bin/opencode run"),
+            compute=spark,
-                    "use_pty": os.environ.get("ARTIFEX_SOL_USE_PTY", "1") == "1",
+        )
-                    "timeout_seconds": int(os.environ.get("ARTIFEX_SOL_TIMEOUT_SECONDS", "120")),
+        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"),

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:
+    for resource in opencode_resources:
-        providers["sol"] = SolProvider(sol)
+        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

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

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"

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