feat(mcp): build_so_modify_ui — diff-decorated SO modify card

katana_mcp_server/src/katana_mcp/tools/_modification.py

@@ -586,13 +586,14 @@ def to_tool_result(
         build_modification_preview_ui,
         build_modification_result_ui,
         build_po_modify_ui,
+        build_so_modify_ui,
     )
 
     response_dict = response.model_dump()
 
-    # Per-entity dispatch — PO migrated in #722, BOM in #811; remaining
-    # entities (SO, MO, stock_transfer, item) fall through to the legacy
-    # builders until their respective child PRs land.
+    # Per-entity dispatch — PO migrated in #722, BOM in #811, SO in #723;
+    # remaining entities (MO, stock_transfer, item) fall through to the
+    # legacy builders until their respective child PRs land.
     if response.entity_type == "purchase_order":
         ui = build_po_modify_ui(
             response_dict,
@@ -609,6 +610,14 @@ def to_tool_result(
         )
         return make_tool_result(response, ui=ui)
 
+    if response.entity_type == "sales_order":
+        ui = build_so_modify_ui(
+            response_dict,
+            confirm_request=confirm_request,
+            confirm_tool=confirm_tool,
+        )
+        return make_tool_result(response, ui=ui)
+
     # Legacy path — preserves today's behavior for not-yet-migrated
     # entity types so #722 can land without cross-entity test churn.
     if response.is_preview:

katana_mcp_server/src/katana_mcp/tools/foundation/corrections.py

@@ -171,22 +171,65 @@ def _augment_prior_state_with_snapshot(
 
 async def _run_phases_until_failure(
     phases: list[list[ActionSpec]],
-) -> tuple[list[ActionResult], bool]:
+) -> tuple[list[ActionResult], bool, list[ActionSpec]]:
     """Run each phase via :func:`execute_plan`; halt on the first failed action.
 
-    Returns ``(aggregated_results, failed)`` — ``failed=True`` means an
-    action raised in some phase and subsequent phases were skipped. Callers
-    use the boolean to branch into success vs failure response building.
+    Returns ``(aggregated_results, failed, not_run_specs)`` —
+    ``failed=True`` means an action raised in some phase and subsequent
+    phases were skipped. ``not_run_specs`` carries every :class:`ActionSpec`
+    from the unattempted plan tail (the rest of the current failing phase
+    plus every later phase). Callers use the boolean to branch into success
+    vs failure response building, and the spec tail to synthesize NOT-RUN
+    extras for the per-section row morph in :func:`build_so_modify_ui`
+    (#858 finding B; mirrors ``_modify_sales_order_impl``).
+
     Empty phases are skipped silently.
     """
     aggregated: list[ActionResult] = []
-    for phase in phases:
+    for phase_idx, phase in enumerate(phases):
         if not phase:
             continue
-        aggregated.extend(await execute_plan(phase))
-        if any(a.succeeded is False for a in aggregated):
-            return aggregated, True
-    return aggregated, False
+        phase_results = await execute_plan(phase)
+        aggregated.extend(phase_results)
+        if any(a.succeeded is False for a in phase_results):
+            # Unattempted tail = leftover specs in this phase (after the
+            # failing action) + every spec in every later phase. Mirrors
+            # ``execute_plan``'s fail-fast contract: it returns results
+            # only up through the failed action, so any specs past that
+            # index in the phase never ran.
+            executed_in_phase = len(phase_results)
+            not_run_specs = list(phase[executed_in_phase:])
+            for later_phase in phases[phase_idx + 1 :]:
+                not_run_specs.extend(later_phase)
+            return aggregated, True, not_run_specs
+    return aggregated, False, []
+
+
+def _synthesize_correction_not_run_actions(
+    specs: list[ActionSpec],
+) -> list[dict[str, Any]]:
+    """Build NOT-RUN action dicts for the unattempted phase tail.
+
+    Same shape as :func:`_modify_sales_order_impl`'s NOT-RUN synthesis so
+    :func:`build_so_modify_ui` (via :func:`_so_actions_with_not_run_tail`)
+    can merge them into the per-section row morph without distinguishing
+    apply-vs-correction provenance. ``succeeded=None`` + ``status_label=
+    "NOT RUN"`` sets the "secondary" Badge variant.
+    """
+    return [
+        {
+            "operation": spec.operation,
+            "target_id": spec.target_id,
+            "succeeded": None,
+            "error": None,
+            "changes": [
+                c.model_dump() if hasattr(c, "model_dump") else dict(c)
+                for c in spec.diff
+            ],
+            "status_label": "NOT RUN",
+        }
+        for spec in specs
+    ]
 
 
 def _make_tolerant_patch_apply(
@@ -622,8 +665,11 @@ async def _correct_manufacturing_order_impl(
                 f"({len(full_plan)} action(s))"
             ),
         )
-    aggregated, failed = await _run_phases_until_failure(phases)
+    aggregated, failed, _not_run_specs = await _run_phases_until_failure(phases)
     if failed:
+        # MO modify card doesn't merge NOT-RUN extras yet — drop the spec
+        # tail here. SO failure path below synthesizes them for the SO
+        # modify-card morph (#858 finding B).
         return _build_failure_response(
             request.id, aggregated, prior_state, katana_url, snapshot
         )
@@ -1144,11 +1190,22 @@ async def _correct_sales_order_impl(
                 f"{request.id} ({len(full_plan)} action(s))"
             ),
         )
-    aggregated, failed = await _run_phases_until_failure(phases)
+    aggregated, failed, not_run_specs = await _run_phases_until_failure(phases)
     if failed:
-        return _build_failure_response(
+        response = _build_failure_response(
             request.id, aggregated, prior_state, katana_url, snapshot
         )
+        # Synthesize NOT-RUN entries for the unattempted plan tail so
+        # :func:`build_so_modify_ui` (which handles ``correct_sales_order``
+        # alongside ``modify_sales_order``) renders skipped restore /
+        # recreate / close phases instead of silently overwriting the
+        # preview's full sub-entity rows with only the executed prefix
+        # (#858 finding B — Copilot comment 3312071378). Mirrors the
+        # equivalent synthesis in ``_modify_sales_order_impl``.
+        not_run_actions = _synthesize_correction_not_run_actions(not_run_specs)
+        if not_run_actions:
+            response.extras["not_run_actions"] = not_run_actions
+        return response
 
     return ModificationResponse(
         entity_type="sales_order",
@@ -1585,8 +1642,11 @@ async def _correct_purchase_order_impl(
                 f"{request.id} ({len(full_plan)} action(s))"
             ),
         )
-    aggregated, failed = await _run_phases_until_failure(phases)
+    aggregated, failed, _not_run_specs = await _run_phases_until_failure(phases)
     if failed:
+        # PO modify card doesn't merge NOT-RUN extras yet — drop the spec
+        # tail here. The SO failure path above synthesizes them for the SO
+        # modify-card morph (#858 finding B).
         return _build_failure_response(
             request.id, aggregated, prior_state, katana_url, snapshot
         )

katana_mcp_server/src/katana_mcp/tools/foundation/sales_orders.py

@@ -2417,7 +2417,7 @@ async def _modify_sales_order_impl(
         )
     )
 
-    return await run_modify_plan(
+    response = await run_modify_plan(
         request=request,
         naming=EntityNaming(
             entity_type="sales_order",
@@ -2444,6 +2444,39 @@ async def _modify_sales_order_impl(
         ),
     )
 
+    # Apply-path: synthesize NOT-RUN entries for the unattempted plan tail.
+    # ``execute_plan`` is fail-fast: ``response.actions`` ends at the first
+    # failed action. Without these synthetic entries the modify card's
+    # per-section row morph would silently HIDE every planned action past
+    # the failure — the operator would see "1 succeeded, 1 failed" with
+    # no indication that 3 more actions were never attempted. Mirror
+    # ``_modify_product_bom_impl``'s NOT RUN pattern (#811): emit one
+    # entry per leftover ``ActionSpec`` with ``status_label="NOT RUN"``
+    # and ``succeeded=None`` (so per-row chrome renders the "secondary"
+    # Badge variant via :data:`_SO_SUB_STATUS_VARIANTS`). The renderer
+    # picks them up via ``response.extras["not_run_actions"]`` and merges
+    # them into the per-section row bucketing in
+    # :func:`build_so_modify_ui` — #858 finding B.
+    if not response.is_preview:
+        not_run_specs = plan[len(response.actions) :]
+        not_run_actions = [
+            {
+                "operation": spec.operation,
+                "target_id": spec.target_id,
+                "succeeded": None,
+                "error": None,
+                "changes": [
+                    c.model_dump() if hasattr(c, "model_dump") else dict(c)
+                    for c in spec.diff
+                ],
+                "status_label": "NOT RUN",
+            }
+            for spec in not_run_specs
+        ]
+        if not_run_actions:
+            response.extras["not_run_actions"] = not_run_actions
+    return response
+
 
 @observe_tool
 @unpack_pydantic_params