diff --git a/gsap_broker/connectors/ansible.py b/gsap_broker/connectors/ansible.py
new file mode 100644
index 0000000..0379b8d
--- /dev/null
+++ b/gsap_broker/connectors/ansible.py
@@ -0,0 +1,137 @@
+# Copyright 2026 Guildhouse Dev
+# SPDX-License-Identifier: Apache-2.0
+
+"""Ansible connector — fleet management via Ansible playbooks.
+
+The Ansible connector is an OrchestratorConnector: it plans a
+sequence of steps (playbook runs, ad-hoc commands, fact collection)
+and executes them via ansible-runner.
+
+Credential model:
+  Ansible does NOT receive credentials from the broker.  Instead,
+  the connector passes a credential callback that ansible-runner's
+  credential plugin calls at runtime.  This callback resolves
+  credentials via the broker's CredentialResolver for each target
+  host at the moment the task runs — not at planning time.
+
+  This means:
+  - The Ansible inventory file contains NO passwords or keys.
+  - Each host's credentials are resolved just-in-time from the
+    secrets backend (Entra, Vault, etc).
+  - If a credential expires mid-playbook, the next task for that
+    host will resolve a fresh credential.
+
+Real integration:
+  Library: ``ansible-runner`` (execution library)
+  Auth: Per-host credential resolution via custom credential plugin
+  Inventory: Dynamic inventory from Intune device cache or Bascule
+    fleet registry
+
+Stubbed in this sprint — steps return placeholder results.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from gsap_broker.connectors.base import ConnectorContext
+from gsap_broker.connectors.orchestrator import (
+    OrchestratorConnector,
+    WorkflowPlan,
+    WorkflowStep,
+)
+from gsap_broker.credentials.resolver import CredentialResolver
+
+
+class AnsibleConnector(OrchestratorConnector):
+    """Fleet management via Ansible playbooks."""
+
+    connector_id = "ansible"
+    corpus_entry_cid = "sha256:ansible-connector-v1"
+    capability_mask = 0x7  # READ | PROPOSE | MUTATE
+    declared_endpoints = ["ansible://*"]
+    accord_template = "fleet-management"
+    gsap_required = True
+    chronicle_enabled = True
+
+    def __init__(self, credential_resolver: CredentialResolver):
+        super().__init__(credential_resolver)
+
+    async def plan(
+        self, operation: str, parameters: dict[str, Any], context: ConnectorContext
+    ) -> WorkflowPlan:
+        """Map operations to workflow plans.
+
+        Operations:
+          "playbook" → single step running the named playbook
+          "adhoc"    → single step running a module on targets
+          "collect"  → single step gathering facts
+          "role"     → single step applying a role
+        """
+        targets = parameters.get("targets", [])
+        if isinstance(targets, str):
+            targets = [targets]
+
+        if operation == "playbook":
+            return WorkflowPlan(steps=[
+                WorkflowStep(
+                    name=f"playbook:{parameters.get('playbook', 'site.yml')}",
+                    command=parameters.get("playbook", "site.yml"),
+                    targets=targets,
+                    extra_vars=parameters.get("extra_vars", {}),
+                )
+            ])
+
+        if operation == "adhoc":
+            return WorkflowPlan(steps=[
+                WorkflowStep(
+                    name=f"adhoc:{parameters.get('module', 'ping')}",
+                    command=parameters.get("module", "ping"),
+                    targets=targets,
+                    extra_vars=parameters.get("args", {}),
+                )
+            ])
+
+        if operation == "collect":
+            return WorkflowPlan(steps=[
+                WorkflowStep(
+                    name="collect:facts",
+                    command="setup",
+                    targets=targets,
+                    required=False,  # fact collection is best-effort
+                )
+            ])
+
+        if operation == "role":
+            return WorkflowPlan(steps=[
+                WorkflowStep(
+                    name=f"role:{parameters.get('role', '')}",
+                    command=parameters.get("role", ""),
+                    targets=targets,
+                    extra_vars=parameters.get("extra_vars", {}),
+                )
+            ])
+
+        return WorkflowPlan(steps=[
+            WorkflowStep(name=f"unknown:{operation}", command=operation, targets=targets)
+        ])
+
+    async def execute_step(
+        self, step: WorkflowStep, context: ConnectorContext
+    ) -> dict[str, Any]:
+        """Execute a single workflow step via ansible-runner.
+
+        Stubbed — actual ansible-runner integration in a future sprint.
+        """
+        # TODO: use ansible_runner.run() with:
+        #   - playbook=step.command (for playbook operation)
+        #   - module=step.command, module_args=step.extra_vars (for adhoc)
+        #   - inventory from dynamic source (Intune cache, Bascule fleet)
+        #   - credential plugin that calls self._resolver.resolve()
+        #     per-host at runtime
+        return {
+            "success": True,
+            "stub": True,
+            "step": step.name,
+            "targets": step.targets,
+        }
diff --git a/gsap_broker/connectors/orchestrator.py b/gsap_broker/connectors/orchestrator.py
new file mode 100644
index 0000000..d555574
--- /dev/null
+++ b/gsap_broker/connectors/orchestrator.py
@@ -0,0 +1,112 @@
+# Copyright 2026 Guildhouse Dev
+# SPDX-License-Identifier: Apache-2.0
+
+"""Orchestrator connector framework — multi-step workflow execution.
+
+Orchestrator connectors manage workflows that span multiple steps
+and potentially multiple targets.  Each step may acquire its own
+credentials via the CredentialResolver.
+
+Unlike ``SessionConnector`` (single target, single credential,
+single command), an ``OrchestratorConnector``:
+  - Plans a sequence of steps before execution
+  - Executes steps in order, stopping on required-step failure
+  - Reports partial results (which steps completed before failure)
+  - Can target different hosts per step
+
+Rust port note:
+  ``WorkflowStep`` and ``WorkflowPlan`` map to plain structs.
+  ``OrchestratorConnector`` maps to an async trait with
+  ``plan()`` and ``execute_step()`` methods.
+"""
+
+from __future__ import annotations
+
+import logging
+from abc import abstractmethod
+from dataclasses import dataclass, field
+from typing import Any
+
+from gsap_broker.connectors.base import ConnectorContext, ConnectorPlugin, ConnectorResult
+from gsap_broker.credentials.resolver import CredentialResolver
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class WorkflowStep:
+    """A single step in a workflow plan."""
+
+    name: str
+    command: str
+    targets: list[str] = field(default_factory=list)
+    required: bool = True
+    extra_vars: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class WorkflowPlan:
+    """Ordered sequence of steps for a workflow."""
+
+    steps: list[WorkflowStep] = field(default_factory=list)
+
+
+class OrchestratorConnector(ConnectorPlugin):
+    """Base for multi-step workflow connectors (Ansible, Terraform, etc).
+
+    Subclasses implement ``plan()`` to convert an operation + parameters
+    into a ``WorkflowPlan``, and ``execute_step()`` to run each step.
+
+    The base ``invoke()`` handles:
+      - Planning the workflow
+      - Executing steps in order
+      - Stopping on required-step failure
+      - Aggregating results with partial-completion reporting
+    """
+
+    def __init__(self, credential_resolver: CredentialResolver):
+        self._resolver = credential_resolver
+
+    async def invoke(
+        self, operation: str, parameters: dict[str, Any], context: ConnectorContext
+    ) -> ConnectorResult:
+        try:
+            plan = await self.plan(operation, parameters, context)
+        except Exception as e:
+            return ConnectorResult(success=False, error=f"Planning failed: {e}")
+
+        if not plan.steps:
+            return ConnectorResult(success=True, data={"steps": []})
+
+        results: list[dict[str, Any]] = []
+        for step in plan.steps:
+            result = await self.execute_step(step, context)
+            results.append({"step": step.name, **result})
+
+            if not result.get("success") and step.required:
+                return ConnectorResult(
+                    success=False,
+                    data={"completed": results, "failed_at": step.name},
+                )
+
+        return ConnectorResult(success=True, data={"steps": results})
+
+    @abstractmethod
+    async def plan(
+        self, operation: str, parameters: dict[str, Any], context: ConnectorContext
+    ) -> WorkflowPlan:
+        """Convert an operation into a step-by-step execution plan."""
+        ...
+
+    @abstractmethod
+    async def execute_step(
+        self, step: WorkflowStep, context: ConnectorContext
+    ) -> dict[str, Any]:
+        """Execute a single workflow step.
+
+        Returns a dict with at minimum a ``success: bool`` key.
+        """
+        ...
+
+    def health_check(self) -> bool:
+        return True