fix(providers): pass agent system_prompt as native system parameter in Claude provider

docs/providers/claude.md

@@ -8,6 +8,7 @@ The Claude provider enables Conductor workflows to use Anthropic's Claude models
 - [API Key Setup](#api-key-setup)
 - [Model Selection](#model-selection)
 - [Runtime Configuration](#runtime-configuration)
+- [System Prompt](#system-prompt)
 - [Streaming Limitations](#streaming-limitations)
 - [Extended Thinking](#extended-thinking)
 - [Troubleshooting](#troubleshooting)
@@ -240,6 +241,15 @@ agents:
       - to: $end
 ```
 
+## System Prompt
+
+When an agent defines a `system_prompt`, the Claude provider forwards this value as the native top-level `system` parameter in the Anthropic Messages API. 
+
+Key details of this integration:
+- **Consistent Application**: The `system_prompt` is sent on every API call in the agent's execution path, including the main loop, tool-use iterations, parse recovery, interrupt partial output requests, and retries.
+- **Empty Prompts**: Any empty or whitespace-only `system_prompt` is normalized to `None` and is not sent to the API.
+- **Caching**: Anthropic `cache_control` support for the `system` parameter is not implemented yet and is planned as a follow-up.
+
 ## Streaming Limitations
 
 **Phase 1 Implementation Status**: The Claude provider currently does NOT support real-time streaming.

plugins/conductor/skills/conductor/references/execution.md

@@ -221,7 +221,7 @@ Performs both schema and **semantic** checks:
 - Parallel group agent references
 - For-each `source` format and reserved names
 - Stale agent references and undeclared explicit-mode dependencies in `prompt`, `system_prompt`, `command`, `args`, `working_dir`, `input_mapping`, parallel-group inputs, and workflow `output:` templates
-- Warning when an agent defines `system_prompt` but no `prompt:` (unusual — system prompts are only meaningful alongside a user prompt)
+- Warning when an agent defines `system_prompt` but no `prompt:` (unusual because the user-authored task prompt would be empty)
 
 The success summary table includes Parallel Groups and For-each Groups counts.
 

src/conductor/config/validator.py

@@ -182,12 +182,9 @@ def validate_workflow_config(
             errors.extend(tool_errors)
 
         # Warn when an LLM agent has system_prompt but no (non-empty) prompt.
-        # The Copilot provider concatenates `agent.system_prompt` into the prompt,
-        # but providers that ignore system_prompt entirely (e.g., Claude) would
-        # leave such agents with an empty user message. Even with the executor
-        # rendering system_prompt, omitting `prompt:` is a portability hazard
-        # and almost always indicates the author meant to include a `prompt:`
-        # block alongside the persona/methodology in `system_prompt:`.
+        # Omitting `prompt:` leaves the user-authored task prompt empty, which
+        # almost always means dynamic, must-execute content belongs in `prompt:`
+        # alongside the persona/methodology in `system_prompt:`.
         if (
             agent.type in (None, "agent")
             and agent.system_prompt
@@ -196,12 +193,11 @@ def validate_workflow_config(
             warnings.append(
                 f"Agent '{agent.name}' defines `system_prompt` but no `prompt` "
                 "(or only whitespace). "
-                "Some providers (e.g., Claude) ignore `system_prompt` entirely, "
-                "which would send an empty user message to the model. "
-                "Even with the Copilot provider, the model often responds poorly "
-                "to a missing user prompt. Move the dynamic, must-execute content "
-                "(input references, instructions) into a `prompt:` block; keep the "
-                "persona and static methodology in `system_prompt:`."
+                "The user-authored task prompt is empty, which almost always "
+                "indicates a latent authoring mistake. Move "
+                "the dynamic, must-execute content (input references, instructions) "
+                "into a `prompt:` block; keep the persona and static methodology "
+                "in `system_prompt:`."
             )
 
     # Validate parallel groups

src/conductor/providers/claude.py

@@ -977,6 +977,13 @@ async def _execute_with_retry(
         # Done before the per-model max_tokens warning so the warning logic
         # accounts for thinking-aware caps.
         thinking = self._resolve_thinking_for_agent(agent, model)
+        # Use strip() only as a blank-prompt predicate: the rendered value is
+        # forwarded verbatim for cross-provider parity (Copilot, Claude Agent
+        # SDK, and Hermes all pass agent.system_prompt through unchanged).
+        raw_system_prompt = agent.system_prompt
+        system_prompt = (
+            raw_system_prompt if raw_system_prompt and raw_system_prompt.strip() else None
+        )
 
         # Validate max_tokens against model-specific limits.
         # Skip the warning when extended thinking is enabled — the per-call
@@ -1039,6 +1046,7 @@ async def _execute_with_retry(
                     event_callback=event_callback,
                     thinking=thinking,
                     max_parse_recovery_attempts=config.max_parse_recovery_attempts,
+                    system_prompt=system_prompt,
                 )
 
                 # Handle partial output from mid-agent interrupt
@@ -1313,6 +1321,7 @@ async def _execute_api_call(
         max_tokens: int,
         tools: list[dict[str, Any]] | None = None,
         thinking: dict[str, Any] | None = None,
+        system_prompt: str | None = None,
     ) -> ClaudeResponse:
         """Execute non-streaming Claude API call using AsyncAnthropic.
 
@@ -1329,6 +1338,8 @@ async def _execute_api_call(
                 supplied, ``temperature`` is forced to 1.0 and ``max_tokens``
                 is bumped to satisfy the API constraint
                 ``max_tokens > budget_tokens``.
+            system_prompt: Optional rendered system prompt passed as the
+                top-level Anthropic ``system`` parameter.
 
         Returns:
             Claude API response object with content blocks and usage metadata.
@@ -1363,6 +1374,9 @@ async def _execute_api_call(
         if thinking is not None:
             kwargs["thinking"] = thinking
 
+        if system_prompt:
+            kwargs["system"] = system_prompt
+
         # Execute non-streaming API call (async)
         logger.debug(
             f"Executing non-streaming Claude API call: model={model}, "
@@ -1388,6 +1402,7 @@ async def _execute_agentic_loop(
         event_callback: EventCallback | None = None,
         thinking: dict[str, Any] | None = None,
         max_parse_recovery_attempts: int | None = None,
+        system_prompt: str | None = None,
     ) -> tuple[ClaudeResponse, int | None, bool]:
         """Execute an agentic loop that handles MCP tool calls.
 
@@ -1414,8 +1429,10 @@ async def _execute_agentic_loop(
                 None means no time limit.
             interrupt_signal: Optional event that signals a mid-agent interrupt.
             event_callback: Optional callback for streaming SDK events upstream.
+            thinking: Optional extended-thinking kwarg forwarded to every API call.
             max_parse_recovery_attempts: Resolved per-agent parse recovery limit.
                 None means use the provider-level default.
+            system_prompt: Optional rendered system prompt forwarded to every API call.
 
         Returns:
             Tuple of (final_response, total_tokens_used, is_partial).
@@ -1464,6 +1481,7 @@ async def _execute_agentic_loop(
                     tools=tools,
                     has_output_schema=has_output_schema,
                     thinking=thinking,
+                    system_prompt=system_prompt,
                 )
                 total_tokens += interrupt_tokens
                 return interrupt_response, total_tokens, True
@@ -1491,6 +1509,7 @@ async def _execute_agentic_loop(
                             output_schema=output_schema,
                             thinking=thinking,
                             max_parse_recovery_attempts=max_parse_recovery_attempts,
+                            system_prompt=system_prompt,
                         )
                     )
                 else:
@@ -1502,6 +1521,7 @@ async def _execute_agentic_loop(
                             max_tokens=max_tokens,
                             tools=tools,
                             thinking=thinking,
+                            system_prompt=system_prompt,
                         )
                     )
                 interrupt_task = asyncio.create_task(interrupt_signal.wait())
@@ -1533,6 +1553,7 @@ async def _execute_agentic_loop(
                         tools=tools,
                         has_output_schema=has_output_schema,
                         thinking=thinking,
+                        system_prompt=system_prompt,
                     )
                     total_tokens += partial_tokens
                     return partial_resp, total_tokens, True
@@ -1548,6 +1569,7 @@ async def _execute_agentic_loop(
                     output_schema=output_schema,
                     thinking=thinking,
                     max_parse_recovery_attempts=max_parse_recovery_attempts,
+                    system_prompt=system_prompt,
                 )
             else:
                 response = await self._execute_api_call(
@@ -1557,6 +1579,7 @@ async def _execute_agentic_loop(
                     max_tokens=max_tokens,
                     tools=tools,
                     thinking=thinking,
+                    system_prompt=system_prompt,
                 )
 
             # Accumulate token usage
@@ -1764,6 +1787,7 @@ async def _request_partial_output(
         tools: list[dict[str, Any]] | None,
         has_output_schema: bool,
         thinking: dict[str, Any] | None = None,
+        system_prompt: str | None = None,
     ) -> tuple[Any, int]:
         """Send a final API call requesting partial output after interrupt.
 
@@ -1781,6 +1805,8 @@ async def _request_partial_output(
             max_tokens: Maximum output tokens.
             tools: Tool definitions (may include emit_output).
             has_output_schema: Whether the agent defines an output schema.
+            thinking: Optional extended-thinking kwarg forwarded to the API call.
+            system_prompt: Optional rendered system prompt forwarded to the API call.
 
         Returns:
             Tuple of (response, tokens_used_in_this_call).
@@ -1808,6 +1834,7 @@ async def _request_partial_output(
             max_tokens=max_tokens,
             tools=tools,
             thinking=thinking,
+            system_prompt=system_prompt,
         )
 
         call_tokens = 0
@@ -1828,6 +1855,7 @@ async def _execute_with_parse_recovery(
         output_schema: dict[str, OutputField] | None,
         thinking: dict[str, Any] | None = None,
         max_parse_recovery_attempts: int | None = None,
+        system_prompt: str | None = None,
     ) -> ClaudeResponse:
         """Execute API call with parse recovery for malformed JSON responses.
 
@@ -1842,8 +1870,10 @@ async def _execute_with_parse_recovery(
             max_tokens: Maximum output tokens.
             tools: Tool definitions for structured output.
             output_schema: Expected output schema (None if no schema).
+            thinking: Optional extended-thinking kwarg forwarded to every API call.
             max_parse_recovery_attempts: Resolved per-agent parse recovery limit.
                 None means use the provider-level default.
+            system_prompt: Optional rendered system prompt forwarded to every API call.
 
         Returns:
             Claude API response.
@@ -1867,6 +1897,7 @@ async def _execute_with_parse_recovery(
             max_tokens=max_tokens,
             tools=tools,
             thinking=thinking,
+            system_prompt=system_prompt,
         )
 
         # If no output schema, return immediately (no recovery needed)
@@ -1932,6 +1963,7 @@ async def _execute_with_parse_recovery(
                 max_tokens=max_tokens,
                 tools=tools,
                 thinking=thinking,
+                system_prompt=system_prompt,
             )
 
             # Check if recovery succeeded (tool_use)

tests/test_config/test_validator.py

@@ -66,10 +66,9 @@ def test_valid_multi_agent_config(self) -> None:
     def test_warns_on_system_prompt_without_prompt(self) -> None:
         """Agent with system_prompt but no prompt: should produce a warning.
 
-        The Copilot provider concatenates system_prompt with the user prompt,
-        so a missing prompt means an empty user message — almost always a
-        latent author mistake. Other providers (Claude) ignore system_prompt
-        entirely, so the agent would have no instructions at all.
+        A missing prompt leaves the user-authored task prompt empty, which
+        is almost always a latent authoring mistake even when system_prompt is
+        configured correctly.
         """
         config = WorkflowConfig(
             workflow=WorkflowDef(name="test", entry_point="lonely"),
@@ -87,6 +86,12 @@ def test_warns_on_system_prompt_without_prompt(self) -> None:
         assert any(
             "lonely" in w and "system_prompt" in w and "no `prompt`" in w for w in warnings
         ), f"expected system_prompt-without-prompt warning; got: {warnings!r}"
+        warning_text = "\n".join(warnings).lower()
+        assert "user-authored task prompt is empty" in warning_text
+        assert "ignore" not in warning_text
+        assert "entirely" not in warning_text
+        assert "empty user message" not in warning_text
+        assert "on every provider" not in warning_text
 
     def test_no_warning_when_prompt_present_alongside_system_prompt(self) -> None:
         """Having both system_prompt and prompt is the expected pattern — no warning."""