Prevent duplicate research idea portfolios

argus_skill/core/vertical_contract.py

@@ -151,6 +151,7 @@ class VerticalLibraryContext:
     model: str | None
     emit: Callable[[dict], None]
     required_skill_paths: list[str] = field(default_factory=list)
+    prompt_blocks: list[str] = field(default_factory=list)
 
 
 @dataclass(frozen=True)

argus_skill/skills/loop_skill_library.py

@@ -18,19 +18,27 @@
 
 class SkillLibraryMixin:
     def _prepare_skill_libraries(self, mission: MissionContext) -> SkillLibraryState:
-        required_skill_paths = self._prepare_vertical_libraries(mission)
+        required_skill_paths, prompt_blocks = self._prepare_vertical_libraries(mission)
         state = SkillLibraryState()
         state.skill_libraries = self.engineer_mission.libraries(
             task=mission.skill_task,
             required_relative_paths=required_skill_paths,
         )
-        state.skill_text = state.skill_libraries.block
+        state.skill_text = "\n\n".join(
+            block
+            for block in (state.skill_libraries.block, *prompt_blocks)
+            if block
+        )
         state.reviewer_skill_block = self.reviewer.mission.libraries().block
         return state
 
-    def _prepare_vertical_libraries(self, mission: MissionContext) -> tuple[str, ...]:
+    def _prepare_vertical_libraries(
+        self,
+        mission: MissionContext,
+    ) -> tuple[tuple[str, ...], tuple[str, ...]]:
         """Let the provider run optional domain setup with explicit inputs."""
         required_skill_paths: list[str] = []
+        prompt_blocks: list[str] = []
         try:
             from ..core.pipeline_state import pipeline_state_exists
             from ..verticals._base import load_vertical_contract
@@ -66,10 +74,14 @@ def _prepare_vertical_libraries(self, mission: MissionContext) -> tuple[str, ...
                 model=self.config.engineer_model,
                 emit=self._emit,
                 required_skill_paths=required_skill_paths,
+                prompt_blocks=prompt_blocks,
             ))
         except Exception:  # noqa: BLE001 — optional domain preparation is non-blocking
             log.debug("vertical Skill-library preparation skipped", exc_info=True)
-        return tuple(dict.fromkeys(required_skill_paths))
+        return (
+            tuple(dict.fromkeys(required_skill_paths)),
+            tuple(dict.fromkeys(prompt_blocks)),
+        )
 
     def _adapt_after_rejections(
         self,

argus_skill/verticals/research/library_preparation.py

@@ -51,6 +51,20 @@ def prepare_skill_libraries(context: VerticalLibraryContext) -> None:
         direction=context.direction,
         state_root=context.state_root,
     )
+    try:
+        display_root = team_root.relative_to(context.workdir)
+    except ValueError:
+        display_root = team_root
+    context.prompt_blocks.append(
+        "## Canonical research idea portfolio\n"
+        f"- The runtime has already formed the only authorized Idea portfolio at "
+        f"`{display_root}`.\n"
+        "- Inspect and settle that exact team. Do not call `team form`, create a "
+        "second portfolio, or use a different `.argus/teams/...` path.\n"
+        "- If the mission contract names another team path, that path is stale and "
+        "does not authorize a replacement; the canonical runtime-owned path above "
+        "takes precedence."
+    )
     selection = idea_portfolio_selection(
         context.workdir,
         state_root=context.state_root,

argus_skill/verticals/research/stages.py

@@ -689,6 +689,50 @@ def search_altitude_context(project_root: object) -> str:
     return ""
 
 
+def planner_task_issues(
+    stage: str,
+    project_root: Path,
+    task: object,
+) -> tuple[str, ...]:
+    """Keep Planner tasks from competing with the runtime-owned Idea portfolio."""
+    _ = project_root
+    if str(stage or "").strip().lower() != "idea":
+        return ()
+    owns_paths = tuple(
+        str(path or "").strip().replace("\\", "/")
+        for path in (getattr(task, "owns_paths", ()) or ())
+    )
+    if not any(
+        path == ".argus/teams" or path.startswith(".argus/teams/")
+        for path in owns_paths
+    ):
+        return ()
+    contract = " ".join(
+        [
+            *(
+                str(getattr(task, field, "") or "")
+                for field in ("title", "objective", "acceptance_check")
+            ),
+            *owns_paths,
+        ]
+    ).lower()
+    explicitly_portfolio = any(
+        marker in contract
+        for marker in ("portfolio", "tournament", "idea-pipeline")
+    )
+    portfolio_shaped = (
+        "route" in contract
+        and ("review" in contract or "selector" in contract)
+        and ("twelve" in contract or "12 " in contract)
+    )
+    if not explicitly_portfolio and not portfolio_shaped:
+        return ()
+    return (
+        "the research runtime owns the canonical Idea portfolio; omit all "
+        "`.argus/teams/...` paths and let the runtime-provided portfolio complete",
+    )
+
+
 def role_banner(role: str = "engineer") -> str:
     if role == "engineer" and os.environ.get(_TEAM_TASK_ENV, "").strip():
         return _ENGINEER_TEAM_RESEARCH_EXECUTION
@@ -718,6 +762,7 @@ def role_banner(role: str = "engineer") -> str:
     "ENGINEER_STAGE_OPERATIONS",
     "REQUIRE_INDEPENDENT_REVIEW",
     "role_banner",
+    "planner_task_issues",
     "import_legacy_state",
     "search_altitude_context",
     "render_role_prompt_fragment",

tests/skills/test_research_idea_portfolio.py

@@ -3,8 +3,11 @@
 import json
 import time
 from pathlib import Path
+from types import SimpleNamespace
 
 from argus_skill.core.vertical_contract import VerticalLibraryContext
+from argus_skill.skills.loop_skill_library import SkillLibraryMixin
+from argus_skill.skills.loop_state import MissionContext
 from argus_skill.skills.vertical_select import reset_stage_for_new_intent
 from argus_skill.team import task_board
 from argus_skill.verticals.research.idea_portfolio import (
@@ -18,6 +21,7 @@
 from argus_skill.verticals.research.library_preparation import (
     prepare_skill_libraries,
 )
+from argus_skill.verticals.research.stages import planner_task_issues
 
 
 def _state(root: Path) -> None:
@@ -307,6 +311,111 @@ def test_direct_idea_only_research_does_not_prepare_a_paper_portfolio(
     assert idea_portfolio_completion_issues(tmp_path) == ()
 
 
+def test_preparation_exposes_the_only_canonical_portfolio_to_engineer(
+    tmp_path: Path,
+) -> None:
+    _state(tmp_path)
+    prompt_blocks: list[str] = []
+
+    prepare_skill_libraries(
+        VerticalLibraryContext(
+            workdir=tmp_path,
+            state_root=tmp_path,
+            stage="idea",
+            objective="select one strong idea",
+            direction="reliable agents",
+            workflow_mode="staged",
+            paper_mission=True,
+            team_task_id=None,
+            runner=None,
+            model=None,
+            emit=lambda _event: None,
+            prompt_blocks=prompt_blocks,
+        )
+    )
+
+    assert len(prompt_blocks) == 1
+    assert f".argus/teams/{TEAM_ID}-g1" in prompt_blocks[0]
+    assert "only authorized Idea portfolio" in prompt_blocks[0]
+    assert "Do not call `team form`" in prompt_blocks[0]
+
+
+def test_skill_library_state_includes_vertical_prompt_blocks() -> None:
+    class Harness(SkillLibraryMixin):
+        engineer_mission = SimpleNamespace(
+            libraries=lambda **_kwargs: SimpleNamespace(block="library index")
+        )
+        reviewer = SimpleNamespace(
+            mission=SimpleNamespace(
+                libraries=lambda: SimpleNamespace(block="reviewer index")
+            )
+        )
+
+        def _prepare_vertical_libraries(
+            self,
+            _mission: MissionContext,
+        ) -> tuple[tuple[str, ...], tuple[str, ...]]:
+            return (("required.md",), ("runtime-owned portfolio",))
+
+    state = Harness()._prepare_skill_libraries(
+        MissionContext(
+            workdir=Path("/project"),
+            run_id="run-1",
+            task="task",
+            skill_task="task",
+            request_anchor="request",
+            active_vertical="research",
+            engineer_role_banner="",
+            seed_thread_id=None,
+            scope="",
+        )
+    )
+
+    assert state.skill_text == "library index\n\nruntime-owned portfolio"
+    assert state.reviewer_skill_block == "reviewer index"
+
+
+def test_planner_cannot_claim_a_second_idea_portfolio_path(
+    tmp_path: Path,
+) -> None:
+    task = SimpleNamespace(
+        title="Select an idea tournament",
+        objective="Produce twelve routes and twelve independent reviews.",
+        acceptance_check="The portfolio has one selector.",
+        owns_paths=[".argus/teams/idea-tournament-20260913", "RESEARCH_NOTES.md"],
+    )
+
+    issues = planner_task_issues("idea", tmp_path, task)
+
+    assert len(issues) == 1
+    assert "runtime owns the canonical Idea portfolio" in issues[0]
+    assert planner_task_issues("experiment", tmp_path, task) == ()
+
+
+def test_planner_may_use_nonportfolio_team_in_idea_stage(tmp_path: Path) -> None:
+    task = SimpleNamespace(
+        title="Audit one source",
+        objective="Delegate one bounded citation audit.",
+        acceptance_check="The citation is checked.",
+        owns_paths=[".argus/teams/citation-audit"],
+    )
+
+    assert planner_task_issues("idea", tmp_path, task) == ()
+
+
+def test_planner_may_benchmark_twelve_items_in_a_nonportfolio_team(
+    tmp_path: Path,
+) -> None:
+    task = SimpleNamespace(
+        title="Benchmark kernels",
+        objective="Benchmark 12 kernels in parallel.",
+        acceptance_check="All measurements are recorded.",
+        owns_paths=[".argus/teams/kernel-benchmark"],
+    )
+
+    assert planner_task_issues("idea", tmp_path, task) == ()
+
+
 def test_locked_paper_idea_uses_playbook_without_reselection(
     tmp_path: Path,
 ) -> None: