diff --git a/docs/providers/claude.md b/docs/providers/claude.md
index 1e06a964..90538218 100644
--- a/docs/providers/claude.md
+++ b/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.
diff --git a/plugins/conductor/skills/conductor/references/execution.md b/plugins/conductor/skills/conductor/references/execution.md
index e5a8bd77..4d02ba45 100644
--- a/plugins/conductor/skills/conductor/references/execution.md
+++ b/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.
 
diff --git a/src/conductor/config/validator.py b/src/conductor/config/validator.py
index 101447d5..81b52bcb 100644
--- a/src/conductor/config/validator.py
+++ b/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
diff --git a/src/conductor/providers/claude.py b/src/conductor/providers/claude.py
index 9b158530..7d30db6f 100644
--- a/src/conductor/providers/claude.py
+++ b/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)
diff --git a/tests/test_config/test_validator.py b/tests/test_config/test_validator.py
index 234d1bde..db8ce25d 100644
--- a/tests/test_config/test_validator.py
+++ b/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."""
diff --git a/tests/test_providers/test_claude_system_prompt.py b/tests/test_providers/test_claude_system_prompt.py
new file mode 100644
index 00000000..8f0df0ca
--- /dev/null
+++ b/tests/test_providers/test_claude_system_prompt.py
@@ -0,0 +1,357 @@
+"""Regression tests for Claude provider system prompt forwarding."""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+from types import SimpleNamespace
+from typing import TYPE_CHECKING
+from unittest.mock import AsyncMock, Mock, patch
+
+import pytest
+
+from conductor.config.schema import AgentDef, OutputField, ValidatorConfig
+from conductor.engine.validator import VALIDATOR_SYSTEM_PROMPT, OutputValidator
+from conductor.exceptions import ProviderError
+
+if TYPE_CHECKING:
+    from conductor.providers.claude import ClaudeProvider, RetryConfig
+
+
+def _text_response(text: str = "Done") -> Mock:
+    response = Mock()
+    response.content = [SimpleNamespace(type="text", text=text)]
+    response.usage = SimpleNamespace(input_tokens=10, output_tokens=20)
+    return response
+
+
+def _tool_response(
+    name: str,
+    input_data: dict[str, str | bool | list[str]],
+    tool_id: str = "tool-1",
+) -> Mock:
+    response = Mock()
+    response.content = [SimpleNamespace(type="tool_use", name=name, input=input_data, id=tool_id)]
+    response.usage = SimpleNamespace(input_tokens=10, output_tokens=20)
+    return response
+
+
+def _mock_claude_provider(
+    mock_anthropic_module: Mock,
+    mock_anthropic_class: Mock,
+    retry_config: RetryConfig | None = None,
+) -> tuple[ClaudeProvider, Mock]:
+    mock_anthropic_module.__version__ = "0.77.0"
+    mock_client = Mock()
+    mock_client.models.list = AsyncMock(return_value=Mock(data=[]))
+    mock_anthropic_class.return_value = mock_client
+
+    from conductor.providers.claude import ClaudeProvider
+
+    provider = ClaudeProvider(api_key="test-key", retry_config=retry_config)
+    return provider, mock_client
+
+
+class TestClaudeSystemPromptForwarding:
+    """Tests for top-level system kwarg forwarding to Anthropic Messages API."""
+
+    @patch("conductor.providers.claude.ANTHROPIC_SDK_AVAILABLE", True)
+    @patch("conductor.providers.claude.AsyncAnthropic")
+    @patch("conductor.providers.claude.anthropic")
+    @pytest.mark.asyncio
+    async def test_plain_execute_passes_system_prompt_to_sdk(
+        self, mock_anthropic_module: Mock, mock_anthropic_class: Mock
+    ) -> None:
+        """Requirement AC-A(a): plain Claude execution sends rendered system_prompt as system."""
+        provider, mock_client = _mock_claude_provider(mock_anthropic_module, mock_anthropic_class)
+        mock_client.messages.create = AsyncMock(return_value=_text_response())
+
+        agent = AgentDef.model_validate(
+            {
+                "name": "plain",
+                "prompt": "Test prompt",
+                "system_prompt": "Rendered system instructions",
+            }
+        )
+
+        await provider.execute(agent=agent, context={}, rendered_prompt="Test prompt")
+
+        call_kwargs = mock_client.messages.create.call_args.kwargs
+        assert call_kwargs["system"] == "Rendered system instructions"
+
+    @patch("conductor.providers.claude.ANTHROPIC_SDK_AVAILABLE", True)
+    @patch("conductor.providers.claude.AsyncAnthropic")
+    @patch("conductor.providers.claude.anthropic")
+    @pytest.mark.asyncio
+    async def test_tool_use_loop_passes_system_prompt_on_every_sdk_call(
+        self, mock_anthropic_module: Mock, mock_anthropic_class: Mock
+    ) -> None:
+        """Requirement AC-A(b): every SDK call in a tool-use loop sends system_prompt."""
+        provider, mock_client = _mock_claude_provider(mock_anthropic_module, mock_anthropic_class)
+        mock_client.messages.create = AsyncMock(
+            side_effect=[
+                _tool_response("lookup", {"query": "status"}),
+                _text_response("Final answer"),
+            ]
+        )
+        mock_mcp = Mock()
+        mock_mcp.has_servers.return_value = True
+        mock_mcp.get_all_tools.return_value = [
+            {
+                "name": "lookup",
+                "description": "Lookup data",
+                "input_schema": {"type": "object", "properties": {"query": {"type": "string"}}},
+            }
+        ]
+        mock_mcp.call_tool = AsyncMock(return_value="tool result")
+        provider._mcp_manager = mock_mcp
+
+        agent = AgentDef.model_validate(
+            {
+                "name": "tool_agent",
+                "prompt": "Test prompt",
+                "system_prompt": "Rendered tool-loop instructions",
+            }
+        )
+
+        await provider.execute(agent=agent, context={}, rendered_prompt="Test prompt")
+
+        assert mock_client.messages.create.call_count == 2
+        for call in mock_client.messages.create.call_args_list:
+            assert call.kwargs["system"] == "Rendered tool-loop instructions"
+
+    @patch("conductor.providers.claude.ANTHROPIC_SDK_AVAILABLE", True)
+    @patch("conductor.providers.claude.AsyncAnthropic")
+    @patch("conductor.providers.claude.anthropic")
+    @pytest.mark.asyncio
+    async def test_parse_recovery_passes_system_prompt_on_initial_and_recovery_calls(
+        self, mock_anthropic_module: Mock, mock_anthropic_class: Mock
+    ) -> None:
+        """Requirement AC-A(c): parse recovery retries send system_prompt on each SDK call."""
+        provider, mock_client = _mock_claude_provider(mock_anthropic_module, mock_anthropic_class)
+        mock_client.messages.create = AsyncMock(
+            side_effect=[
+                _text_response("not json"),
+                _tool_response("emit_output", {"result": "recovered"}),
+            ]
+        )
+
+        agent = AgentDef.model_validate(
+            {
+                "name": "structured",
+                "prompt": "Test prompt",
+                "system_prompt": "Rendered recovery instructions",
+                "output": {"result": OutputField(type="string")},
+            }
+        )
+
+        await provider.execute(agent=agent, context={}, rendered_prompt="Test prompt")
+
+        assert mock_client.messages.create.call_count == 2
+        for call in mock_client.messages.create.call_args_list:
+            assert call.kwargs["system"] == "Rendered recovery instructions"
+
+    @patch("conductor.providers.claude.ANTHROPIC_SDK_AVAILABLE", True)
+    @patch("conductor.providers.claude.AsyncAnthropic")
+    @patch("conductor.providers.claude.anthropic")
+    @pytest.mark.asyncio
+    async def test_retry_loop_passes_system_prompt_on_every_attempt(
+        self, mock_anthropic_module: Mock, mock_anthropic_class: Mock
+    ) -> None:
+        """Requirement AC-A(e): outer retry loop resends the same system_prompt per attempt."""
+        from conductor.providers.claude import RetryConfig
+
+        retry_config = RetryConfig(max_attempts=2, base_delay=0.0, max_delay=0.0, jitter=0.0)
+        provider, mock_client = _mock_claude_provider(
+            mock_anthropic_module, mock_anthropic_class, retry_config=retry_config
+        )
+        mock_client.messages.create = AsyncMock(
+            side_effect=[
+                ProviderError("transient Claude failure", status_code=503),
+                _text_response("Final answer"),
+            ]
+        )
+
+        verbatim_prompt = "  Preserve me exactly\n"
+        agent = AgentDef.model_validate(
+            {
+                "name": "retrying",
+                "prompt": "Test prompt",
+                "system_prompt": verbatim_prompt,
+            }
+        )
+
+        with patch("conductor.providers.claude.asyncio.sleep", new_callable=AsyncMock):
+            await provider.execute(agent=agent, context={}, rendered_prompt="Test prompt")
+
+        assert mock_client.messages.create.call_count == 2
+        for call in mock_client.messages.create.call_args_list:
+            assert call.kwargs["system"] == verbatim_prompt
+        assert len(provider.get_retry_history()) == 1
+
+    @patch("conductor.providers.claude.ANTHROPIC_SDK_AVAILABLE", True)
+    @patch("conductor.providers.claude.AsyncAnthropic")
+    @patch("conductor.providers.claude.anthropic")
+    @pytest.mark.asyncio
+    async def test_interrupt_partial_output_passes_system_prompt(
+        self, mock_anthropic_module: Mock, mock_anthropic_class: Mock
+    ) -> None:
+        """Requirement AC-A(d): interrupt partial-output SDK call sends system_prompt."""
+        provider, mock_client = _mock_claude_provider(mock_anthropic_module, mock_anthropic_class)
+        mock_client.messages.create = AsyncMock(
+            return_value=_tool_response("emit_output", {"result": "partial"})
+        )
+        interrupt_signal = asyncio.Event()
+        interrupt_signal.set()
+
+        agent = AgentDef.model_validate(
+            {
+                "name": "interruptible",
+                "prompt": "Test prompt",
+                "system_prompt": "Rendered interrupt instructions",
+                "output": {"result": OutputField(type="string")},
+            }
+        )
+
+        await provider.execute(
+            agent=agent,
+            context={},
+            rendered_prompt="Test prompt",
+            interrupt_signal=interrupt_signal,
+        )
+
+        call_kwargs = mock_client.messages.create.call_args.kwargs
+        assert call_kwargs["system"] == "Rendered interrupt instructions"
+
+    @patch("conductor.providers.claude.ANTHROPIC_SDK_AVAILABLE", True)
+    @patch("conductor.providers.claude.AsyncAnthropic")
+    @patch("conductor.providers.claude.anthropic")
+    @pytest.mark.asyncio
+    async def test_in_flight_interrupt_partial_output_passes_system_prompt(
+        self, mock_anthropic_module: Mock, mock_anthropic_class: Mock
+    ) -> None:
+        """Requirement AC-A(f): interrupt during a blocked API call keeps system_prompt.
+
+        Covers the in-flight race branch (asyncio.wait FIRST_COMPLETED), unlike
+        the pre-set event test which only covers the between-turn branch.
+        """
+        provider, mock_client = _mock_claude_provider(mock_anthropic_module, mock_anthropic_class)
+
+        call_started = asyncio.Event()
+        never_release = asyncio.Event()
+        interrupt_signal = asyncio.Event()
+        call_number = 0
+
+        async def create_side_effect(**kwargs: object) -> Mock:
+            nonlocal call_number
+            call_number += 1
+            if call_number == 1:
+                call_started.set()
+                # Block until cancelled by the provider's interrupt race; the
+                # CancelledError must propagate so the partial-output path runs.
+                await never_release.wait()
+            return _tool_response("emit_output", {"result": "partial"})
+
+        mock_client.messages.create = AsyncMock(side_effect=create_side_effect)
+
+        verbatim_prompt = "Rendered interrupt instructions\n"
+        agent = AgentDef.model_validate(
+            {
+                "name": "interruptible",
+                "prompt": "Test prompt",
+                "system_prompt": verbatim_prompt,
+                "output": {"result": OutputField(type="string")},
+            }
+        )
+
+        execute_task = asyncio.create_task(
+            provider.execute(
+                agent=agent,
+                context={},
+                rendered_prompt="Test prompt",
+                interrupt_signal=interrupt_signal,
+            )
+        )
+
+        try:
+            await asyncio.wait_for(call_started.wait(), timeout=5)
+            interrupt_signal.set()
+            output = await asyncio.wait_for(execute_task, timeout=5)
+        finally:
+            if not execute_task.done():
+                execute_task.cancel()
+                with contextlib.suppress(asyncio.CancelledError):
+                    await execute_task
+
+        assert output.partial is True
+        assert mock_client.messages.create.call_count == 2
+        partial_kwargs = mock_client.messages.create.call_args_list[1].kwargs
+        assert partial_kwargs["system"] == verbatim_prompt
+        assert "interrupted" in partial_kwargs["messages"][-1]["content"].lower()
+
+
+class TestSystemKwargAbsent:
+    """Tests for omitting the system kwarg when no usable system prompt exists."""
+
+    @patch("conductor.providers.claude.ANTHROPIC_SDK_AVAILABLE", True)
+    @patch("conductor.providers.claude.AsyncAnthropic")
+    @patch("conductor.providers.claude.anthropic")
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("system_prompt", [None, "", "   "])
+    async def test_system_kwarg_absent_when_system_prompt_empty(
+        self,
+        mock_anthropic_module: Mock,
+        mock_anthropic_class: Mock,
+        system_prompt: str | None,
+    ) -> None:
+        """Requirement AC-B: None, empty, and whitespace system_prompt omit system kwarg."""
+        provider, mock_client = _mock_claude_provider(mock_anthropic_module, mock_anthropic_class)
+        mock_client.messages.create = AsyncMock(return_value=_text_response())
+
+        agent = AgentDef.model_validate(
+            {"name": "empty_system", "prompt": "Test prompt", "system_prompt": system_prompt}
+        )
+
+        await provider.execute(agent=agent, context={}, rendered_prompt="Test prompt")
+
+        call_kwargs = mock_client.messages.create.call_args.kwargs
+        assert "system" not in call_kwargs
+
+
+class TestValidatorSystemPromptForwarding:
+    """Tests for OutputValidator synthetic-agent system prompt forwarding on Claude."""
+
+    @patch("conductor.providers.claude.ANTHROPIC_SDK_AVAILABLE", True)
+    @patch("conductor.providers.claude.AsyncAnthropic")
+    @patch("conductor.providers.claude.anthropic")
+    @pytest.mark.asyncio
+    async def test_output_validator_sends_formatted_rubric_as_system_prompt(
+        self, mock_anthropic_module: Mock, mock_anthropic_class: Mock
+    ) -> None:
+        """Requirement AC-C: validator rubric reaches Claude as formatted system kwarg."""
+        provider, mock_client = _mock_claude_provider(mock_anthropic_module, mock_anthropic_class)
+        mock_client.messages.create = AsyncMock(
+            return_value=_tool_response("emit_output", {"passed": True, "issues": []})
+        )
+        criteria = "Answer must mention the verified source."
+        agent = AgentDef.model_validate(
+            {
+                "name": "reviewer",
+                "prompt": "Review the answer",
+                "model": "claude-3-5-sonnet-latest",
+                "validator": ValidatorConfig(criteria=criteria),
+            }
+        )
+
+        outcome = await OutputValidator().validate(
+            agent=agent,
+            primary_prompt="Review the answer",
+            primary_output={"result": "Verified source included."},
+            provider=provider,
+        )
+
+        assert outcome.passed is True
+        call_kwargs = mock_client.messages.create.call_args.kwargs
+        # system_prompt is forwarded verbatim (no trim) for cross-provider parity.
+        assert call_kwargs["system"] == VALIDATOR_SYSTEM_PROMPT.format(criteria=criteria)
+        assert "{{" not in call_kwargs["system"]