diff --git a/argus_skill/core/vertical_contract.py b/argus_skill/core/vertical_contract.py index 14e13bb35..72b4b9bcc 100644 --- a/argus_skill/core/vertical_contract.py +++ b/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) diff --git a/argus_skill/skills/loop_skill_library.py b/argus_skill/skills/loop_skill_library.py index ea4ad3162..60b5fff64 100644 --- a/argus_skill/skills/loop_skill_library.py +++ b/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, diff --git a/argus_skill/verticals/research/library_preparation.py b/argus_skill/verticals/research/library_preparation.py index aad74113d..1d3ffc00f 100644 --- a/argus_skill/verticals/research/library_preparation.py +++ b/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, diff --git a/argus_skill/verticals/research/stages.py b/argus_skill/verticals/research/stages.py index ebc37398b..0bc58cd14 100644 --- a/argus_skill/verticals/research/stages.py +++ b/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", diff --git a/tests/skills/test_research_idea_portfolio.py b/tests/skills/test_research_idea_portfolio.py index df4ffb8a1..46c4b21ad 100644 --- a/tests/skills/test_research_idea_portfolio.py +++ b/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: