feat(mcp): add list_orders_in_workflow + document list_orders gotchas (#50)

Three follow-up tweaks identified after the F3 probe revealed how
sparsely populated the workflow status field actually is in this tenant
(84% of non-cancelled orders have no status set).

### New tool: list_orders_in_workflow

Returns only orders that StatusPro is actively tracking through a
workflow stage — i.e. orders with `status` assigned to one of the
tenant's defined status codes. Internally fans out one
`list_orders(status_code=…)` call per defined status (concurrency
capped at 10) and merges the results, deduplicated by id.

This was the operationally-useful filter that didn't exist:

- `list_orders()` returned all 478 (includes cancelled)
- `list_orders(exclude_cancelled=True)` returned 475 (84% have no status)
- `list_orders(exclude_cancelled=True, status_code=X)` only one stage at a time

`list_orders_in_workflow()` returns the 77 orders StatusPro is actually
tracking — the right starting point for reconciliation flows.

Live verified against the dev tenant: 77 unique orders across 7 active
status codes (Ride Wrap Installation has 0 orders, correctly omitted).
Optional `search` parameter applied within each per-status call.

### Doc tweaks on list_orders

The tool description now calls out two gotchas surfaced during the
F3 probes:

1. Many orders have NO status assigned. An agent calling list_orders
   and trying to inspect status.code in code will silently miss the
   84% with no status. Description points to list_orders_in_workflow
   for the in-progress subset.

2. financial_status, fulfillment_status, and tags filter server-side
   but are NOT echoed back on results. The OrderListItem schema doesn't
   include them, so list-then-inspect produces wrong answers — agents
   should filter for what they need.

Same gotchas captured in help.py's list_orders row.

### Files

- statuspro_mcp_server/src/statuspro_mcp/tools/orders.py — new tool +
  description update + module docstring tool count (10 -> 11)
- statuspro_mcp_server/src/statuspro_mcp/server.py — list_orders_in_workflow
  added to _READ_ONLY_TOOLS
- statuspro_mcp_server/src/statuspro_mcp/resources/help.py — new row
  + gotcha note on existing list_orders row

Author: dougborg

statuspro_mcp_server/src/statuspro_mcp/resources/help.py

@@ -11,7 +11,8 @@
 
 | Tool | Endpoint | Purpose |
 | ---- | -------- | ------- |
-| `list_orders` | `GET /orders` | Paginated list with filters (search, status, tags, due-date range). Auto-paginates. `search` matches order number, name, or customer fields — use it to find an order from just an order number. |
+| `list_orders` | `GET /orders` | Paginated list with filters (search, status, tags, due-date range). Auto-paginates. `search` matches order number, name, or customer fields — use it to find an order from just an order number. **Gotchas:** (1) many orders may have NO `status` set — use `list_orders_in_workflow` if you only want orders StatusPro is actively tracking; (2) `financial_status`, `fulfillment_status`, and `tags` filter server-side but are NOT echoed back on results — filter for what you need rather than list-then-inspect. |
+| `list_orders_in_workflow` | `GET /orders?status_code=…` xN (parallel, capped 10 concurrent) | Return only orders with a workflow status assigned. Iterates per defined status_code; merges results. Use when you want the operationally-tracked subset of orders, not the full list (which includes orders with no status set). |
 | `get_order` | `GET /orders/{id}` | Full detail for one order, with the most recent `history_limit` history entries (default 50). When `history_truncated` is true, use `get_order_history` for older entries. |
 | `get_order_history` | `GET /orders/{id}` (client-side paged) | Page through the full history timeline of one order. Use when `get_order` indicated truncation. |
 | `get_orders_batch` | `GET /orders/{id}` xN (parallel fan-out) | Fetch up to 50 orders by id in one tool call. Returns per-id found/not-found results. Useful when an external system hands you a list of ids. |

statuspro_mcp_server/src/statuspro_mcp/server.py

@@ -222,6 +222,7 @@ def _build_auth() -> "AuthProvider | None":
 # Add response caching middleware with TTLs for read-only tools
 _READ_ONLY_TOOLS = [
     "list_orders",
+    "list_orders_in_workflow",
     "get_order",
     "get_order_history",
     "get_orders_batch",

statuspro_mcp_server/src/statuspro_mcp/tools/orders.py

@@ -1,11 +1,12 @@
 """MCP tools for StatusPro orders.
 
-10 tools covering the ``/orders*`` endpoints plus three batch read tools
-(``get_orders_batch``, ``lookup_orders_batch``, ``summarize_active_orders``)
-that fan out parallel calls under the hood when the StatusPro server has
-no native batch endpoint. Mutations use a two-step confirm pattern: call
-with ``confirm=False`` to see a preview, then ``confirm=True`` to execute
-(the client host elicits explicit user approval via ``ctx.elicit``).
+11 tools covering the ``/orders*`` endpoints plus a set of batch / aggregation
+read tools (``get_orders_batch``, ``lookup_orders_batch``,
+``list_orders_in_workflow``, ``summarize_active_orders``) that fan out
+parallel calls under the hood when the StatusPro server has no native batch
+endpoint. Mutations use a two-step confirm pattern: call with
+``confirm=False`` to see a preview, then ``confirm=True`` to execute (the
+client host elicits explicit user approval via ``ctx.elicit``).
 
 All read-and-mutation tools (``list_orders``, ``get_order``,
 ``update_order_status``, ``add_order_comment``, ``update_order_due_date``,
@@ -23,6 +24,7 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 from typing import Annotated, Any
 
 from fastmcp import Context, FastMCP
@@ -90,6 +92,8 @@
 )
 from statuspro_public_api_client.utils import is_success, unwrap
 
+logger = logging.getLogger(__name__)
+
 
 def _to_summary(order: Any) -> OrderSummary:
     """Convert a domain Order or attrs model to an OrderSummary."""
@@ -213,6 +217,27 @@ async def _count_fulfillment(
     return value.value, int(getattr(getattr(parsed, "meta", None), "total", None) or 0)
 
 
+def _merge_unique_by_id[T](batches: list[list[T]]) -> list[T]:
+    """Flatten ``batches`` preserving first-seen order, deduplicated by ``.id``.
+
+    Used by ``list_orders_in_workflow`` to merge per-status_code results.
+    Output is deterministic: orders within a batch keep API order, and
+    earlier batches take precedence over later ones for the same id.
+
+    Each item in each batch must have an ``.id`` attribute.
+    """
+    seen_ids: set[Any] = set()
+    out: list[T] = []
+    for batch in batches:
+        for item in batch:
+            item_id = getattr(item, "id", None)
+            if item_id is None or item_id in seen_ids:
+                continue
+            seen_ids.add(item_id)
+            out.append(item)
+    return out
+
+
 def _build_batch_response(
     requested_count: int, results: list[BatchOrderResult]
 ) -> BatchOrderResponse:
@@ -324,7 +349,19 @@ def register_tools(mcp: FastMCP) -> None:
 
     @mcp.tool(
         name="list_orders",
-        description="List orders with optional filters. Auto-paginates when page is unset.",
+        description=(
+            "List orders with optional filters. Auto-paginates when page is "
+            "unset.\n\n"
+            "Two gotchas worth knowing about:\n"
+            "1. Many orders may have NO `status` set at all (newly created, "
+            "not yet moved into the workflow). If you only want orders "
+            "StatusPro is actively tracking through workflow stages, use "
+            "`list_orders_in_workflow` — it iterates per known status_code "
+            "and returns only orders with a status assigned.\n"
+            "2. The `financial_status`, `fulfillment_status`, and `tags` "
+            "filters work server-side but those fields are NOT echoed back "
+            "on results. Filter for what you need; don't list-then-inspect."
+        ),
         meta=UI_META,
     )
     async def list_orders(
@@ -423,6 +460,123 @@ async def list_orders(
             orders_table=orders_table,
         )
 
+    @mcp.tool(
+        name="list_orders_in_workflow",
+        description=(
+            "Return only orders that StatusPro is actively tracking through "
+            "a workflow stage — i.e. orders with a `status` set to one of "
+            "the tenant's defined status codes. Excludes cancelled orders "
+            "and orders with no status assigned (which can be a large "
+            "fraction of the total — see `list_orders` description for "
+            "context). Internally fans out one `list_orders(status_code=…)` "
+            "call per defined status (concurrency capped at 10) and merges "
+            "the results."
+        ),
+        meta=UI_META,
+    )
+    async def list_orders_in_workflow(
+        context: Context,
+        search: Annotated[
+            str | None,
+            Field(
+                description=(
+                    "Optional full-text search applied within each "
+                    "status_code call (matches order number, name, or "
+                    "customer fields)."
+                ),
+            ),
+        ] = None,
+    ) -> ToolResult:
+        services = get_services(context)
+
+        # Fetch the status catalog first, then fetch the per-status order
+        # lists concurrently via _bounded_gather (capped at 10 in-flight,
+        # captures per-row exceptions). Each per-status call auto-paginates
+        # (no `page` set), so we get the full set per status code without
+        # ceiling at per_page=100.
+        statuses_list = await services.client.statuses.list()
+        active_codes: list[str] = []
+        for s in statuses_list:
+            code = getattr(s, "code", None)
+            if isinstance(code, str) and code:
+                active_codes.append(code)
+
+        async def fetch_for_code(code: str) -> list[Any]:
+            kwargs: dict[str, Any] = {
+                "exclude_cancelled": True,
+                "status_code": code,
+                "per_page": 100,
+            }
+            if search:
+                kwargs["search"] = search
+            return await services.client.orders.list(**kwargs)
+
+        # _bounded_gather catches Exception (not BaseException) so individual
+        # failures (rate limit, transport error, auth) don't kill the whole
+        # call — we proceed with whatever buckets succeeded. CancelledError
+        # propagates correctly.
+        gathered = await _bounded_gather(
+            [fetch_for_code(code) for code in active_codes]
+        )
+        successful_batches: list[list[Any]] = []
+        failed_codes: list[str] = []
+        for code, result in zip(active_codes, gathered, strict=True):
+            if isinstance(result, Exception):
+                failed_codes.append(code)
+                logger.warning(
+                    "list_orders_in_workflow: status_code=%s fetch failed: %s",
+                    code,
+                    _classify_error(result, what=f"status_code={code}"),
+                )
+            else:
+                successful_batches.append(result)
+
+        orders = _merge_unique_by_id(successful_batches)
+
+        summaries = [_to_summary(o) for o in orders]
+        summary_dicts = [s.model_dump() for s in summaries]
+        filters_line = f"in workflow ({len(active_codes)} statuses)" + (
+            f", search={search!r}" if search else ""
+        )
+
+        response = OrderList(
+            orders=summaries,
+            total=len(summaries),
+            filters={
+                "in_workflow": True,
+                "status_codes": [code for code, _, _ in active_codes if code],
+                "search": search,
+            },
+        )
+        app = build_orders_table_ui(
+            summary_dicts,
+            total=len(summaries),
+            filters_line=filters_line,
+        )
+        orders_table = (
+            format_md_table(
+                headers=["Order #", "Customer", "Status", "Due"],
+                rows=[
+                    [
+                        d.get("order_number") or "—",
+                        d.get("customer_name") or "—",
+                        d.get("status_name") or "—",
+                        d.get("due_date") or "—",
+                    ]
+                    for d in summary_dicts
+                ],
+            )
+            or "_(no orders in workflow)_"
+        )
+        return make_tool_result(
+            response,
+            template_name="orders_list",
+            ui=app,
+            total=len(summaries),
+            filters_line=f"Filters: {filters_line}",
+            orders_table=orders_table,
+        )
+
     @mcp.tool(
         name="get_order",
         description=(

statuspro_mcp_server/tests/tools/test_batch_tools.py

@@ -251,6 +251,73 @@ def test_mixed_outcomes(self):
         assert resp.error_count == 2  # ambiguous + ConnectionError
 
 
+@pytest.mark.unit
+class TestMergeUniqueById:
+    """`_merge_unique_by_id` is the dedupe helper used by `list_orders_in_workflow`.
+
+    It must produce deterministic output (so callers can rely on stable
+    ordering across calls), preserve API order within batches, and skip
+    items missing an id.
+    """
+
+    def test_empty_input(self):
+        from statuspro_mcp.tools.orders import _merge_unique_by_id
+
+        assert _merge_unique_by_id([]) == []
+
+    def test_single_batch_passthrough(self):
+        from statuspro_mcp.tools.orders import _merge_unique_by_id
+
+        a = OrderSummary(id=1)
+        b = OrderSummary(id=2)
+        result = _merge_unique_by_id([[a, b]])
+        assert [o.id for o in result] == [1, 2]
+
+    def test_dedupe_across_batches(self):
+        """Order appearing in two status_code buckets is included exactly once."""
+        from statuspro_mcp.tools.orders import _merge_unique_by_id
+
+        a = OrderSummary(id=1)
+        b = OrderSummary(id=2)
+        result = _merge_unique_by_id([[a, b], [a]])
+        assert [o.id for o in result] == [1, 2]
+
+    def test_first_seen_wins(self):
+        """When the same id appears in two batches, the first batch's item wins.
+
+        Batches are passed in catalog order — so the order's "primary" status
+        bucket (whichever comes first in the catalog) is what we surface.
+        """
+        from statuspro_mcp.tools.orders import _merge_unique_by_id
+
+        a_v1 = OrderSummary(id=1, status_name="In Production")
+        a_v2 = OrderSummary(id=1, status_name="Shipped")
+        result = _merge_unique_by_id([[a_v1], [a_v2]])
+        assert len(result) == 1
+        assert result[0].status_name == "In Production"
+
+    def test_preserves_api_order_within_batch(self):
+        from statuspro_mcp.tools.orders import _merge_unique_by_id
+
+        items = [OrderSummary(id=i) for i in (5, 2, 8, 1)]
+        result = _merge_unique_by_id([items])
+        assert [o.id for o in result] == [5, 2, 8, 1]
+
+    def test_skips_items_with_no_id(self):
+        """Defensive: if an item has no `.id`, skip it rather than crash."""
+        from statuspro_mcp.tools.orders import _merge_unique_by_id
+
+        # OrderSummary requires id, so use a stand-in that has no id attr.
+        class Anonymous:
+            pass
+
+        a = OrderSummary(id=1)
+        b = Anonymous()
+        c = OrderSummary(id=3)
+        result = _merge_unique_by_id([[a, b, c]])
+        assert [getattr(o, "id", None) for o in result] == [1, 3]
+
+
 @pytest.mark.unit
 class TestExactMatchDisambiguationNoneSafety:
     """Edge case: rows with None order_number/name must not match."""