feat(test-controller): add seed_creative_format scenario; advertise force_session_status in capabilities

examples/seller_agent.py

@@ -53,6 +53,9 @@
 
 # Test-controller state (force_*/seed_* scenarios only)
 plans: dict[str, dict[str, Any]] = {}
+# Seeded creative formats keyed by the string format ID the storyboard supplies.
+# list_creative_formats merges these in so storyboard references resolve.
+seeded_creative_formats: dict[str, dict[str, Any]] = {}
 # Single-shot directives registered by force_create_media_buy_arm; keyed by account_id.
 pending_directives: dict[str, dict[str, Any]] = {}
 # Tasks registered when create_media_buy consumes a 'submitted' directive; keyed by task_id.
@@ -126,10 +129,15 @@ async def get_adcp_capabilities(
                 # in 3.0.1) live on the dynamic list_scenarios response and
                 # are reported there — not advertised here. Once the
                 # capabilities schema's enum catches up, the rest land too.
+                # force_session_status is schema-allowed even for media_buy
+                # sellers; DemoStore provides a stub so list_scenarios
+                # includes it and the storyboard runner's controller
+                # detection check succeeds.
                 "scenarios": [
                     "force_account_status",
                     "force_media_buy_status",
                     "force_creative_status",
+                    "force_session_status",
                     "simulate_delivery",
                     "simulate_budget_spend",
                 ],
@@ -393,6 +401,7 @@ async def list_creative_formats(
                 ],
             },
         ]
+        all_formats = all_formats + list(seeded_creative_formats.values())
         filter_ids = params.get("format_ids")
         if filter_ids:
             wanted = {(fid.get("agent_url"), fid["id"]) for fid in filter_ids if "id" in fid}
@@ -531,6 +540,20 @@ async def simulate_budget_spend(
     ) -> dict[str, Any]:
         return {"simulated": {"spend_percentage": spend_percentage}}
 
+    async def force_session_status(
+        self,
+        session_id: str,
+        status: str,
+        termination_reason: str | None = None,
+        *,
+        context: Any = None,
+    ) -> dict[str, Any]:
+        # DemoSeller has no SI session state; return a canned transition so
+        # the storyboard runner's controller-detection probe succeeds and the
+        # force_session_status storyboard can run (it will simply report the
+        # canned previous_state).
+        return {"previous_state": "active", "current_state": status}
+
     async def force_create_media_buy_arm(
         self,
         arm: str,
@@ -668,6 +691,22 @@ async def seed_media_buy(
         media_buys[mb_id] = data
         return {"media_buy_id": mb_id}
 
+    async def seed_creative_format(
+        self,
+        fixture: dict[str, Any] | None = None,
+        format_id: str | None = None,
+        *,
+        context: Any = None,
+    ) -> dict[str, Any]:
+        data = dict(fixture or {})
+        fid = format_id or (data.get("format_id") or {}).get("id") or f"fmt-seeded-{uuid.uuid4().hex[:8]}"
+        data.setdefault("format_id", {"agent_url": AGENT_URL, "id": fid})
+        data.setdefault("name", fid)
+        data.setdefault("renders", [])
+        data.setdefault("assets", [])
+        seeded_creative_formats[fid] = data
+        return {"format_id": fid}
+
 
 if __name__ == "__main__":
     serve(

src/adcp/server/test_controller.py

@@ -60,6 +60,7 @@ async def force_account_status(self, account_id, status):
     "seed_creative",
     "seed_plan",
     "seed_media_buy",
+    "seed_creative_format",
 ]
 
 _MAX_TASK_ID = 128
@@ -357,6 +358,23 @@ async def seed_media_buy(
         """
         raise NotImplementedError
 
+    async def seed_creative_format(
+        self,
+        fixture: dict[str, Any] | None = None,
+        format_id: str | None = None,
+        *,
+        context: ToolContext | None = None,
+    ) -> dict[str, Any]:
+        """Pre-populate a creative format fixture for storyboard tests (AdCP 3.0.1).
+
+        The seller MUST expose the seeded format_id in list_creative_formats
+        responses for the duration of the compliance session.
+
+        Returns:
+            {"format_id": str}
+        """
+        raise NotImplementedError
+
 
 def _list_scenarios(store: TestControllerStore) -> list[str]:
     """Detect which scenarios a store actually implements.
@@ -617,6 +635,12 @@ async def _handle_test_controller(
                 media_buy_id=scenario_params.get("media_buy_id"),
                 **extra,
             )
+        elif scenario == "seed_creative_format":
+            result = await method(
+                fixture=scenario_params.get("fixture"),
+                format_id=scenario_params.get("format_id"),
+                **extra,
+            )
         else:
             return _controller_error("UNKNOWN_SCENARIO", f"Unknown scenario: {scenario}")
     except TestControllerError as e:

tests/test_server_dx.py

@@ -398,6 +398,77 @@ async def test_missing_params(self):
         assert result["success"] is False
         assert result["error"] == "INVALID_PARAMS"
 
+    @pytest.mark.asyncio
+    async def test_seed_creative_format_dispatches(self):
+        """seed_creative_format is routed to the store method when implemented."""
+
+        class _FormatStore(TestControllerStore):
+            async def seed_creative_format(
+                self,
+                fixture: Any = None,
+                format_id: str | None = None,
+                *,
+                context: Any = None,
+            ) -> dict[str, Any]:
+                return {"format_id": format_id or "fmt-default"}
+
+        store = _FormatStore()
+        result = await _handle_test_controller(
+            store,
+            {"scenario": "seed_creative_format", "params": {"format_id": "video_30s"}},
+        )
+        assert result["success"] is True
+        assert result["format_id"] == "video_30s"
+
+    @pytest.mark.asyncio
+    async def test_seed_creative_format_in_list_scenarios(self):
+        """seed_creative_format appears in list_scenarios when overridden."""
+
+        class _FormatStore(TestControllerStore):
+            async def seed_creative_format(
+                self,
+                fixture: Any = None,
+                format_id: str | None = None,
+            ) -> dict[str, Any]:
+                return {"format_id": format_id or "fmt-x"}
+
+        store = _FormatStore()
+        result = await _handle_test_controller(store, {"scenario": "list_scenarios"})
+        assert result["success"] is True
+        assert "seed_creative_format" in result["scenarios"]
+        assert "force_account_status" not in result["scenarios"]
+
+    @pytest.mark.asyncio
+    async def test_seed_creative_format_structured_fixture_id(self):
+        """format_id extracted from a structured fixture object must not become a dict key."""
+        received: list[Any] = []
+
+        class _FormatStore(TestControllerStore):
+            async def seed_creative_format(
+                self,
+                fixture: Any = None,
+                format_id: str | None = None,
+                *,
+                context: Any = None,
+            ) -> dict[str, Any]:
+                received.append(format_id)
+                return {"format_id": format_id or "fmt-structured"}
+
+        store = _FormatStore()
+        # Storyboard sends format_id only inside fixture (no top-level params.format_id).
+        result = await _handle_test_controller(
+            store,
+            {
+                "scenario": "seed_creative_format",
+                "params": {
+                    "fixture": {"format_id": {"agent_url": "http://localhost", "id": "display_300x250"}}
+                },
+            },
+        )
+        # The dispatcher passes format_id=None when it's absent from params.
+        assert result["success"] is True
+        assert received[0] is None
+
 
 # ============================================================================
 # serve() and create_mcp_server tests