feat(adcp): add force_create_media_buy_arm + force_task_completion controller scenarios

schemas/cache/compliance/comply-test-controller-request.json

@@ -19,7 +19,9 @@
         "seed_pricing_option",
         "seed_creative",
         "seed_plan",
-        "seed_media_buy"
+        "seed_media_buy",
+        "force_create_media_buy_arm",
+        "force_task_completion"
       ],
       "description": "Test scenario to execute. 'list_scenarios' discovers supported scenarios. 'force_*' and 'simulate_*' trigger state transitions. 'seed_*' scenarios pre-populate fixtures (product, pricing option, creative, plan, media buy) so storyboards can reference them by stable ID without the implementer having to guess which IDs the conformance suite expects."
     },
@@ -110,6 +112,21 @@
           "minimum": 0,
           "maximum": 100,
           "description": "Spend to this percentage of budget (0\u2013100). Used by simulate_budget_spend."
+        },
+        "arm": {
+          "type": "string",
+          "enum": ["submitted", "input-required"],
+          "description": "Target arm for the next create_media_buy call. Used by force_create_media_buy_arm."
+        },
+        "task_id": {
+          "type": "string",
+          "maxLength": 128,
+          "description": "Buyer-supplied task ID. Required when arm='submitted' for force_create_media_buy_arm; required for force_task_completion."
+        },
+        "result": {
+          "type": "object",
+          "description": "Completion result payload for force_task_completion. Must be non-empty; 256 KB soft cap.",
+          "additionalProperties": true
         }
       },
       "additionalProperties": true
@@ -396,6 +413,49 @@
           }
         }
       }
+    },
+    {
+      "if": {
+        "properties": {
+          "scenario": {
+            "const": "force_create_media_buy_arm"
+          }
+        }
+      },
+      "then": {
+        "required": [
+          "params"
+        ],
+        "properties": {
+          "params": {
+            "required": [
+              "arm"
+            ]
+          }
+        }
+      }
+    },
+    {
+      "if": {
+        "properties": {
+          "scenario": {
+            "const": "force_task_completion"
+          }
+        }
+      },
+      "then": {
+        "required": [
+          "params"
+        ],
+        "properties": {
+          "params": {
+            "required": [
+              "task_id",
+              "result"
+            ]
+          }
+        }
+      }
     }
   ],
   "additionalProperties": true,

schemas/cache/compliance/comply-test-controller-response.json

@@ -28,7 +28,9 @@
               "seed_pricing_option",
               "seed_creative",
               "seed_plan",
-              "seed_media_buy"
+              "seed_media_buy",
+              "force_create_media_buy_arm",
+              "force_task_completion"
             ]
           },
           "description": "Scenarios this seller has implemented. Runners and sellers MUST accept unknown scenario strings (open-for-extension) \u2014 new scenarios may be added in additive minor bumps."
@@ -173,6 +175,65 @@
         ]
       }
     },
+    {
+      "title": "ForcedDirectiveSuccess",
+      "description": "A force_create_media_buy_arm directive was registered. The next create_media_buy from the same account will be driven into the specified arm.",
+      "type": "object",
+      "properties": {
+        "success": {
+          "type": "boolean",
+          "const": true
+        },
+        "arm": {
+          "type": "string",
+          "enum": ["submitted", "input-required"],
+          "description": "Arm the next create_media_buy will be driven into"
+        },
+        "task_id": {
+          "type": ["string", "null"],
+          "description": "Buyer-supplied task ID registered with the directive (null for input-required arm)"
+        },
+        "message": {
+          "type": ["string", "null"],
+          "description": "Optional message for the input-required arm"
+        },
+        "context": {
+          "$ref": "../core/context.json"
+        },
+        "ext": {
+          "$ref": "../core/ext.json"
+        }
+      },
+      "required": [
+        "success",
+        "arm"
+      ],
+      "additionalProperties": true,
+      "not": {
+        "anyOf": [
+          {
+            "required": [
+              "error"
+            ]
+          },
+          {
+            "required": [
+              "scenarios"
+            ]
+          },
+          {
+            "required": [
+              "previous_state"
+            ]
+          },
+          {
+            "required": [
+              "simulated"
+            ]
+          }
+        ]
+      }
+    },
     {
       "title": "ControllerError",
       "description": "The scenario failed \u2014 invalid transition, unknown entity, unsupported scenario, or invalid params",

src/adcp/server/mcp_tools.py

@@ -849,6 +849,8 @@
                         "force_session_status",
                         "simulate_delivery",
                         "simulate_budget_spend",
+                        "force_create_media_buy_arm",
+                        "force_task_completion",
                     ],
                 },
                 "params": {"type": "object"},

src/adcp/server/test_controller.py

@@ -52,6 +52,8 @@ async def force_account_status(self, account_id, status):
     "force_session_status",
     "simulate_delivery",
     "simulate_budget_spend",
+    "force_create_media_buy_arm",
+    "force_task_completion",
 ]
 
 
@@ -191,6 +193,119 @@ async def simulate_budget_spend(
         """
         raise NotImplementedError
 
+    async def force_create_media_buy_arm(
+        self,
+        arm: str,
+        task_id: str | None = None,
+        message: str | None = None,
+        *,
+        context: ToolContext | None = None,
+    ) -> dict[str, Any]:
+        """Register a single-shot directive for the next create_media_buy call.
+
+        Drives the next ``create_media_buy`` from the same authenticated
+        sandbox account into the specified arm (``submitted`` or
+        ``input-required``). A second registration before consumption
+        overwrites the pending directive.
+
+        Args:
+            arm: Target arm — ``"submitted"`` or ``"input-required"``.
+            task_id: Buyer-supplied task ID. Required when arm=``"submitted"``;
+                max 128 characters.
+            message: Optional message for the input-required arm; max 2000
+                characters.
+            context: Optional ToolContext threaded from the server's
+                context_factory.
+
+        Returns:
+            ForcedDirectiveSuccess shape::
+
+                {"arm": arm, "task_id": task_id, "message": message}
+
+            (``"success": True`` is auto-injected by the dispatcher.)
+
+        Raises:
+            TestControllerError("INVALID_PARAMS", ...): Input validation failed.
+              The dispatcher validates arm values and the task_id/arm
+              constraint before calling this method.
+
+        Example::
+
+            class MySeller(TestControllerStore):
+                def __init__(self):
+                    # (account, principal) → pending directive
+                    self._directives: dict[str, dict] = {}
+
+                async def force_create_media_buy_arm(
+                    self, arm, task_id=None, message=None, *, context=None
+                ):
+                    key = context.caller_identity if context else "default"
+                    self._directives[key] = {"arm": arm, "task_id": task_id, "message": message}
+                    return {"arm": arm, "task_id": task_id, "message": message}
+        """
+        raise NotImplementedError
+
+    async def force_task_completion(
+        self,
+        task_id: str,
+        result: dict[str, Any],
+        *,
+        context: ToolContext | None = None,
+    ) -> dict[str, Any]:
+        """Resolve a previously-submitted task to completed.
+
+        Records ``(task_id, result, ownerKey)`` and transitions the task
+        from ``submitted`` to ``completed``. Cross-account replays return
+        ``NOT_FOUND``; identical-params replays are idempotent; diverging
+        params against a terminal task return ``INVALID_TRANSITION``.
+
+        Args:
+            task_id: Buyer-supplied task ID; max 128 characters.
+            result: Completion result payload (non-empty dict, 256 KB soft
+                cap). Validated against ``async-response-data.json`` by
+                spec-compliant runners.
+            context: Optional ToolContext threaded from the server's
+                context_factory.
+
+        Returns:
+            StateTransitionSuccess shape::
+
+                {"previous_state": "submitted", "current_state": "completed"}
+
+            (``"success": True`` is auto-injected by the dispatcher.)
+
+        Raises:
+            TestControllerError("NOT_FOUND", ...): Cross-account replay — the
+                task_id exists but belongs to a different principal.
+            TestControllerError("INVALID_TRANSITION", ..., current_state="completed"):
+                Diverging-params replay — the task already completed with a
+                different result payload.
+
+        Example::
+
+            class MySeller(TestControllerStore):
+                def __init__(self):
+                    # task_id → {"result": dict, "owner_key": str}
+                    self._completed: dict[str, dict] = {}
+
+                async def force_task_completion(self, task_id, result, *, context=None):
+                    owner = context.caller_identity if context else "default"
+                    existing = self._completed.get(task_id)
+                    if existing:
+                        if existing["owner_key"] != owner:
+                            raise TestControllerError("NOT_FOUND", f"Task {task_id!r} not found")
+                        if existing["result"] == result:
+                            return {"previous_state": "submitted", "current_state": "completed"}
+                        raise TestControllerError(
+                            "INVALID_TRANSITION",
+                            f"Task {task_id!r} already completed with different result",
+                            current_state="completed",
+                        )
+                    self._completed[task_id] = {"result": result, "owner_key": owner}
+                    return {"previous_state": "submitted", "current_state": "completed"}
+        """
+        raise NotImplementedError
+
 
 def _list_scenarios(store: TestControllerStore) -> list[str]:
     """Detect which scenarios a store actually implements.
@@ -353,6 +468,38 @@ async def _handle_test_controller(
                 media_buy_id=scenario_params.get("media_buy_id"),
                 **extra,
             )
+        elif scenario == "force_create_media_buy_arm":
+            arm = scenario_params.get("arm")
+            if not arm:
+                return _controller_error("INVALID_PARAMS", "Missing required parameter: 'arm'")
+            if arm not in ("submitted", "input-required"):
+                return _controller_error(
+                    "INVALID_PARAMS", "arm must be 'submitted' or 'input-required'"
+                )
+            task_id_val = scenario_params.get("task_id")
+            # Strip first so whitespace-only strings are treated as absent
+            task_id_val = task_id_val.strip() if task_id_val else task_id_val
+            if arm == "submitted" and not task_id_val:
+                return _controller_error(
+                    "INVALID_PARAMS", "task_id is required when arm='submitted'"
+                )
+            if task_id_val and len(task_id_val) > 128:
+                return _controller_error("INVALID_PARAMS", "task_id must be ≤128 characters")
+            message_val = scenario_params.get("message")
+            if message_val and len(message_val) > 2000:
+                return _controller_error("INVALID_PARAMS", "message must be ≤2000 characters")
+            result = await method(arm=arm, task_id=task_id_val, message=message_val, **extra)
+        elif scenario == "force_task_completion":
+            task_id_val = scenario_params.get("task_id")
+            task_id_val = task_id_val.strip() if task_id_val else task_id_val
+            if not task_id_val:
+                return _controller_error("INVALID_PARAMS", "Missing required parameter: 'task_id'")
+            if len(task_id_val) > 128:
+                return _controller_error("INVALID_PARAMS", "task_id must be ≤128 characters")
+            result_obj = scenario_params.get("result")
+            if not isinstance(result_obj, dict) or not result_obj:
+                return _controller_error("INVALID_PARAMS", "result must be a non-empty object")
+            result = await method(task_id=task_id_val, result=result_obj, **extra)
         else:
             return _controller_error("UNKNOWN_SCENARIO", f"Unknown scenario: {scenario}")
     except TestControllerError as e:
@@ -438,21 +585,14 @@ async def comply_test_controller(**kwargs: Any) -> str:
         description="Compliance test controller. Sandbox only, not for production use.",
     )
 
-    # Override schema with the proper comply_test_controller inputSchema
+    # Override schema with the proper comply_test_controller inputSchema.
+    # Derived from SCENARIOS so it stays in sync automatically.
     tool.parameters = {
         "type": "object",
         "properties": {
             "scenario": {
                 "type": "string",
-                "enum": [
-                    "list_scenarios",
-                    "force_creative_status",
-                    "force_account_status",
-                    "force_media_buy_status",
-                    "force_session_status",
-                    "simulate_delivery",
-                    "simulate_budget_spend",
-                ],
+                "enum": ["list_scenarios"] + SCENARIOS,
             },
             "params": {"type": "object"},
             "context": {"type": "object"},

src/adcp/types/generated_poc/compliance/comply_test_controller_request.py

@@ -27,6 +27,9 @@ class Scenario(Enum):
     seed_creative = "seed_creative"
     seed_plan = "seed_plan"
     seed_media_buy = "seed_media_buy"
+    # Added for adcp#3104 / adcp#3138 parity with Node training-agent
+    force_create_media_buy_arm = "force_create_media_buy_arm"
+    force_task_completion = "force_task_completion"
 
 
 class ReportedSpend(AdCPBaseModel):
@@ -117,6 +120,25 @@ class Params(AdCPBaseModel):
             le=100.0,
         ),
     ] = None
+    arm: Annotated[
+        str | None,
+        Field(
+            description="Target arm for the next create_media_buy call. Used by force_create_media_buy_arm."
+        ),
+    ] = None
+    task_id: Annotated[
+        str | None,
+        Field(
+            description="Buyer-supplied task ID. Required when arm='submitted' for force_create_media_buy_arm; required for force_task_completion.",
+            max_length=128,
+        ),
+    ] = None
+    result: Annotated[
+        dict[str, Any] | None,
+        Field(
+            description="Completion result payload for force_task_completion. Must be non-empty; 256 KB soft cap."
+        ),
+    ] = None
 
 
 class ComplyTestControllerRequest(AdCPBaseModel):