Rectify: Dispatch Envelope Shape Discrimination Immunity (#4118)

## Summary

The `_fmt_dispatch_food_truck` formatter conflates two structurally
distinct `success=False` response shapes — `DispatchRejected`
(subprocess never started) and `DispatchCompleted(success=False)`
(subprocess ran and failed) — under a single early-return branch. This
strips critical fields (`dispatched_session_id`, `dispatch_status`,
`reason`, `token_usage`, `elapsed_seconds`) from failed dispatches that
actually ran, causing the orchestrator to believe no session ever
existed. The architectural solution introduces a typed `kind`
discriminator field at the envelope layer, adds the field to the
existing `DispatchEnvelopeResult` TypedDict, and adds per-shape
formatter contract tests that structurally prevent any formatter from
silently dropping fields for a given envelope shape.

## Implementation Plan

Plan file:
`/home/talon/projects/autoskillit-runs/remediation-20260626-152751-009727/.autoskillit/temp/rectify/rectify_dispatch_envelope_shape_discrimination_2026-06-26_153500.md`

🤖 Generated with [Claude Code](https://claude.com/claude-code) via
AutoSkillit
<!-- autoskillit:pipeline-signature
steps=prepare_pr,run_arch_lenses,compose_pr,annotate_pr_diff,review_pr
-->

Closes #4111

## Token Usage Summary

| Step | Model | count | uncached | output | cache_read | peak_ctx |
turns | cache_write | time |

|------|-------|-------|----------|--------|------------|----------|-------|-------------|------|
| rectify* | opus[1m] | 1 | 4.9k | 16.0k | 1.3M | 109.9k | 52 | 90.8k |
13m 50s |
| review_approach* | sonnet | 1 | 60 | 8.3k | 257.1k | 51.2k | 17 |
37.9k | 6m 15s |
| dry_walkthrough* | opus | 1 | 46 | 9.5k | 1.3M | 86.7k | 52 | 87.5k |
9m 8s |
| implement* | MiniMax-M3 | 1 | 108.6k | 21.8k | 3.0M | 0 | 93 | 0 | 5m
48s |
| assess* | opus[1m] | 1 | 48 | 4.6k | 1.0M | 58.7k | 43 | 39.6k | 5m 9s
|
| audit_impl* | sonnet | 1 | 539 | 11.2k | 284.0k | 47.9k | 20 | 41.0k |
6m 21s |
| prepare_pr* | MiniMax-M3 | 1 | 43.4k | 2.3k | 152.5k | 0 | 14 | 0 |
30s |
| compose_pr* | MiniMax-M3 | 1 | 36.8k | 1.4k | 171.8k | 0 | 11 | 0 |
31s |
| review_pr* | sonnet | 3 | 510 | 112.4k | 4.0M | 99.9k | 152 | 225.2k |
28m 58s |
| ci_conflict_fix* | opus[1m] | 1 | 26 | 2.1k | 306.4k | 45.6k | 19 |
26.6k | 1m 14s |
| diagnose_ci* | MiniMax-M3 | 1 | 54.4k | 2.3k | 297.6k | 0 | 20 | 0 |
53s |
| resolve_ci* | opus[1m] | 1 | 54 | 6.1k | 1.7M | 77.7k | 49 | 58.5k |
4m 29s |
| **Total** | | | 249.4k | 198.0k | 13.8M | 109.9k | | 607.0k | 1h 23m |

\* *Step used a non-Anthropic provider; caching behavior may differ.*

## Token Efficiency

| Step | LoC Changed | cache_read/LoC | cache_write/LoC | output/LoC |
|------|-------------|----------------|-----------------|------------|
| rectify | 0 | — | — | — |
| review_approach | 0 | — | — | — |
| dry_walkthrough | 0 | — | — | — |
| implement | 288 | 10507.4 | 0.0 | 75.7 |
| assess | 3 | 344689.7 | 13199.0 | 1535.7 |
| audit_impl | 0 | — | — | — |
| prepare_pr | 0 | — | — | — |
| compose_pr | 0 | — | — | — |
| review_pr | 0 | — | — | — |
| ci_conflict_fix | 381 | 804.2 | 69.8 | 5.5 |
| diagnose_ci | 0 | — | — | — |
| resolve_ci | 1 | 1706644.0 | 58532.0 | 6130.0 |
| **Total** | **673** | 20433.8 | 902.0 | 294.2 |

## Model Usage Breakdown

| Model | steps | uncached | output | cache_read | cache_write | time |
|-------|-------|----------|--------|------------|-------------|------|
| opus[1m] | 4 | 5.0k | 28.8k | 4.3M | 215.5k | 24m 43s |
| sonnet | 3 | 1.1k | 131.8k | 4.5M | 304.0k | 41m 36s |
| opus | 1 | 46 | 9.5k | 1.3M | 87.5k | 9m 8s |
| MiniMax-M3 | 4 | 243.2k | 27.8k | 3.6M | 0 | 7m 42s |

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: Trecek

src/autoskillit/fleet/state_types.py

@@ -374,6 +374,7 @@ class DispatchRejected:
 
     def to_envelope(self) -> str:
         d: dict[str, Any] = {
+            "kind": "rejected",
             "success": False,
             "error": self.error_code,
             "user_visible_message": self.message,
@@ -406,6 +407,7 @@ class DispatchCompleted:
 
     def to_envelope(self) -> str:
         d: dict[str, Any] = {
+            "kind": "completed",
             "success": self.success,
             "dispatch_status": self.dispatch_status.value,
             "dispatch_id": self.dispatch_id,

src/autoskillit/hooks/formatters/_fmt_dispatch.py

@@ -23,18 +23,19 @@ def _fmt_dispatch_food_truck(data: dict, _pipeline: bool) -> str:
     """Format dispatch_food_truck result as Markdown-KV.
 
     Renders both success (DispatchCompleted) and error (DispatchRejected /
-    fleet_error) response shapes — discriminates on ``data.get("success")``.
+    fleet_error) response shapes — discriminates on ``data.get("kind")``.
     Nested structured fields (l3_payload, health_report, token_usage,
     resume_checkpoint) are rendered in full without size-based truncation
     to preserve dispatch_plan visibility for downstream orchestrators.
     """
     success = data.get("success", False)
     mark = _CHECK_MARK if success else _CROSS_MARK
     status = "OK" if success else "FAIL"
+    kind = data.get("kind")
 
     lines = [f"## dispatch_food_truck {mark} {status}", ""]
 
-    if not success:
+    if kind == "rejected" or (not success and kind is None):
         # Error path — DispatchRejected / fleet_error shape
         error = data.get("error", "unknown")
         lines.append(f"error: {error}")
@@ -154,4 +155,19 @@ def _fmt_dispatch_food_truck(data: dict, _pipeline: bool) -> str:
         "details",
     }
 )
-_FMT_DISPATCH_FOOD_TRUCK_SUPPRESSED: frozenset[str] = frozenset()
+_FMT_DISPATCH_FOOD_TRUCK_SUPPRESSED: frozenset[str] = frozenset({"kind"})
+
+_FMT_DISPATCH_COMPLETED_REQUIRED: frozenset[str] = frozenset(
+    {
+        "dispatch_status",
+        "dispatch_id",
+        "dispatched_session_id",
+        "reason",
+    }
+)
+_FMT_DISPATCH_REJECTED_REQUIRED: frozenset[str] = frozenset(
+    {
+        "error",
+        "user_visible_message",
+    }
+)

src/autoskillit/server/tools/_types.py

@@ -176,6 +176,7 @@ class DispatchEnvelopeResult(TypedDict, total=False):
     """
 
     success: bool
+    kind: str
     dispatch_status: str
     dispatch_id: str
     dispatched_session_id: str

tests/fleet/test_dispatch_envelope_fields.py

@@ -136,6 +136,70 @@ async def dispatch_food_truck(self, orchestrator_prompt, cwd, *, on_spawn=None,
         assert result["dispatch_status"] == "resumable"
 
 
+class TestKindDiscriminatorEnvelopeField:
+    def test_kind_discriminator_present_in_both_envelopes(self):
+        """Both DispatchCompleted and DispatchRejected must carry a 'kind' discriminator
+        in their envelope. This explicit discriminator is what allows the formatter to
+        distinguish them when both have success=False.
+        """
+        from autoskillit.core import FleetErrorCode
+        from autoskillit.fleet import DispatchCompleted, DispatchRejected, DispatchStatus
+
+        completed = DispatchCompleted(
+            success=True,
+            dispatch_status=DispatchStatus.SUCCESS,
+            dispatch_id="d-c",
+            dispatched_session_id="s-c",
+            reason="ok",
+        )
+        completed_env = json.loads(completed.to_envelope())
+        assert completed_env.get("kind") == "completed", (
+            f"DispatchCompleted.to_envelope() must emit 'kind': 'completed': {completed_env}"
+        )
+
+        rejected = DispatchRejected(
+            error_code=FleetErrorCode.FLEET_QUOTA_EXHAUSTED,
+            message="quota hit",
+            dispatch_id="d-r",
+        )
+        rejected_env = json.loads(rejected.to_envelope())
+        assert rejected_env.get("kind") == "rejected", (
+            f"DispatchRejected.to_envelope() must emit 'kind': 'rejected': {rejected_env}"
+        )
+
+    def test_rejected_envelope_never_emits_completed_exclusive_fields(self):
+        """DispatchRejected.to_envelope() must never emit DispatchCompleted-exclusive
+        fields. This is a structural exclusion guard that protects the discriminator
+        invariant: if a future change adds dispatch_status to DispatchRejected, this
+        test fails immediately.
+        """
+        from autoskillit.core import FleetErrorCode
+        from autoskillit.fleet import DispatchRejected
+
+        rejected = DispatchRejected(
+            error_code=FleetErrorCode.FLEET_QUOTA_EXHAUSTED,
+            message="quota hit",
+            dispatch_id="d-r",
+        )
+        env = json.loads(rejected.to_envelope())
+        excluded_keys = {
+            "dispatch_status",
+            "dispatched_session_id",
+            "reason",
+            "token_usage",
+            "l3_payload",
+            "l3_parse_source",
+            "lifespan_started",
+            "stderr",
+            "elapsed_seconds",
+        }
+        leaked = excluded_keys & set(env.keys())
+        assert not leaked, (
+            f"DispatchRejected.to_envelope() leaked DispatchCompleted-exclusive fields: "
+            f"{sorted(leaked)}. The 'kind' discriminator invariant is broken."
+        )
+
+
 class TestHealthReportEnvelopeField:
     def test_envelope_includes_health_report_when_present(self):
         from autoskillit.fleet import DispatchCompleted, DispatchStatus

tests/infra/conftest.py

@@ -59,6 +59,13 @@ def _dispatch_food_truck_json_producer() -> dict:
         resume_checkpoint={"step": 1, "completed_items": ["a"]},
         health_report={"status": "healthy", "findings": []},
     )
+    completed_failure = DispatchCompleted(
+        success=False,
+        dispatch_status=DispatchStatus.FAILURE,
+        dispatch_id="d3",
+        dispatched_session_id="s3",
+        reason="fleet_l3_no_result_block",
+    )
     rejected = DispatchRejected(
         error_code=FleetErrorCode.FLEET_QUOTA_EXHAUSTED,
         message="quota limit hit",
@@ -70,6 +77,7 @@ def _dispatch_food_truck_json_producer() -> dict:
     for envelope_str in (
         base.to_envelope(),
         with_optionals.to_envelope(),
+        completed_failure.to_envelope(),
         rejected.to_envelope(),
         error_str,
     ):

tests/infra/test_pretty_output_formatters.py

@@ -866,6 +866,7 @@ def test_dispatch_food_truck_six_group_plan_preserved():
     from tests.infra._pretty_output_helpers import _wrap_for_claude_code
 
     payload = {
+        "kind": "completed",
         "success": True,
         "dispatch_status": "completed_clean",
         "dispatch_id": "d-test-001",
@@ -912,6 +913,7 @@ def test_dispatch_food_truck_error_envelope_preserves_diagnostics():
     from tests.infra._pretty_output_helpers import _wrap_for_claude_code
 
     payload = {
+        "kind": "rejected",
         "success": False,
         "error": "fleet_quota_exhausted",
         "user_visible_message": "quota limit hit for this session",
@@ -927,3 +929,82 @@ def test_dispatch_food_truck_error_envelope_preserves_diagnostics():
     assert "fleet_quota_exhausted" in text
     assert "quota limit hit" in text
     assert "dispatch_id: d-rejected-001" in text
+
+
+# PHK-FMT-DISPATCH-COMPLETED-FAIL: DispatchCompleted(success=False) must not be conflated
+def test_completed_failure_through_formatter_preserves_all_fields():
+    """A DispatchCompleted(success=False) envelope carries critical fields that
+    distinguish it from DispatchRejected: dispatched_session_id, dispatch_status,
+    reason, token_usage, elapsed_seconds, lifespan_started. These fields must ALL
+    survive formatting — they prove the subprocess actually ran. The bug: the
+    formatter's `if not success:` branch strips these fields because it conflates
+    DispatchCompleted(success=False) with DispatchRejected.
+    """
+    from tests.infra._pretty_output_helpers import _wrap_for_claude_code
+
+    payload = {
+        "kind": "completed",
+        "success": False,
+        "dispatch_status": "failure",
+        "dispatch_id": "d-completed-fail-001",
+        "dispatched_session_id": "sess-xyz",
+        "reason": "fleet_l3_no_result_block",
+        "token_usage": {"input": 1250, "output": 82381},
+        "lifespan_started": True,
+        "elapsed_seconds": 9774.74,
+    }
+    event = {
+        "tool_name": "mcp__plugin_autoskillit_autoskillit__dispatch_food_truck",
+        "tool_response": _wrap_for_claude_code(payload),
+    }
+    out, _ = _run_hook(event=event)
+    text = json.loads(out)["hookSpecificOutput"]["updatedMCPToolOutput"]
+
+    assert "FAIL" in text, f"Header must show FAIL on failure: {text!r}"
+    assert "dispatch_status: failure" in text, (
+        f"dispatch_status must appear in formatted output: {text!r}"
+    )
+    assert "dispatched_session_id: sess-xyz" in text, (
+        f"dispatched_session_id must appear in formatted output: {text!r}"
+    )
+    assert "reason: fleet_l3_no_result_block" in text, (
+        f"reason must appear in formatted output: {text!r}"
+    )
+    assert "token_usage:" in text, f"token_usage section must appear in formatted output: {text!r}"
+    assert "elapsed_seconds: 9774.74" in text, (
+        f"elapsed_seconds must appear in formatted output: {text!r}"
+    )
+    assert "lifespan_started: True" in text, (
+        f"lifespan_started must appear in formatted output: {text!r}"
+    )
+    assert "dispatch_id: d-completed-fail-001" in text, (
+        f"dispatch_id must appear in formatted output: {text!r}"
+    )
+
+
+# PHK-FMT-DISPATCH-FLEET-ERROR: fleet_error() envelope has no kind field — fallback path required
+def test_fleet_error_through_formatter_renders_error_fields():
+    """fleet_error() produces envelopes without a kind field. The formatter must
+    handle these via a fallback path (kind is None + not success → rejection rendering).
+    This protects backward compatibility for the legacy error envelope shape.
+    """
+    from tests.infra._pretty_output_helpers import _wrap_for_claude_code
+
+    payload = {
+        "success": False,
+        "error": "fleet_quota_exhausted",
+        "user_visible_message": "quota limit hit",
+        "details": None,
+    }
+    event = {
+        "tool_name": "mcp__plugin_autoskillit_autoskillit__dispatch_food_truck",
+        "tool_response": _wrap_for_claude_code(payload),
+    }
+    out, _ = _run_hook(event=event)
+    text = json.loads(out)["hookSpecificOutput"]["updatedMCPToolOutput"]
+    assert "fleet_quota_exhausted" in text, (
+        f"fleet_error() error code must appear in output: {text!r}"
+    )
+    assert "quota limit hit" in text, (
+        f"fleet_error() user_visible_message must appear in output: {text!r}"
+    )

tests/infra/test_pretty_output_hook_infra.py

@@ -368,3 +368,119 @@ def test_conftest_import_does_not_load_server():
     assert result.returncode == 0, (
         f"Importing tests.infra.conftest triggered autoskillit.server import:\n{result.stderr}"
     )
+
+
+# ---------------------------------------------------------------------------
+# PHK-FMT-DISPATCH-SHAPE: Per-shape field obligation tests
+# ---------------------------------------------------------------------------
+
+
+def test_shape_field_obligations_per_kind():
+    """Per-shape required-field sets must be enforced for dispatch_food_truck.
+
+    The _FMT_DISPATCH_FOOD_TRUCK_RENDERED frozenset is a flat union of all 18 fields
+    and cannot express per-shape obligations. This test defines required-field sets
+    per shape ('completed' vs 'rejected') and asserts the formatter renders them
+    when given an envelope with the matching 'kind' discriminator.
+
+    This is the shape-aware coverage contract that protects against future
+    regressions where a formatter silently drops fields for a specific shape.
+    """
+    from autoskillit.hooks.formatters._fmt_dispatch import (
+        _FMT_DISPATCH_COMPLETED_REQUIRED,
+        _FMT_DISPATCH_REJECTED_REQUIRED,
+    )
+    from tests.infra._pretty_output_helpers import _wrap_for_claude_code
+
+    completed_required = _FMT_DISPATCH_COMPLETED_REQUIRED
+    rejected_required = _FMT_DISPATCH_REJECTED_REQUIRED
+
+    completed_payload = {
+        "kind": "completed",
+        "success": True,
+        "dispatch_status": "success",
+        "dispatch_id": "d-shape-c",
+        "dispatched_session_id": "s-shape-c",
+        "reason": "ok",
+        "token_usage": {},
+        "l3_payload": None,
+        "l3_parse_source": "",
+        "lifespan_started": False,
+        "stderr": "",
+        "elapsed_seconds": 1.0,
+    }
+    event_c = {
+        "tool_name": "mcp__plugin_autoskillit_autoskillit__dispatch_food_truck",
+        "tool_response": _wrap_for_claude_code(completed_payload),
+    }
+    out_c, _ = _run_hook(event=event_c)
+    text_c = json.loads(out_c)["hookSpecificOutput"]["updatedMCPToolOutput"]
+    completed_expected = {
+        "dispatch_status": "dispatch_status: success",
+        "dispatch_id": "dispatch_id: d-shape-c",
+        "dispatched_session_id": "dispatched_session_id: s-shape-c",
+        "reason": "reason: ok",
+    }
+    for field in completed_required:
+        assert field in completed_expected, (
+            f"Field {field!r} from _FMT_DISPATCH_COMPLETED_REQUIRED has no expected value — "
+            f"add it to completed_expected and completed_payload"
+        )
+        assert completed_expected[field] in text_c, (
+            f"Completed shape missing {field!r}: expected {completed_expected[field]!r} in output"
+        )
+
+    rejected_payload = {
+        "kind": "rejected",
+        "success": False,
+        "error": "fleet_quota_exhausted",
+        "user_visible_message": "quota hit",
+        "details": None,
+    }
+    event_r = {
+        "tool_name": "mcp__plugin_autoskillit_autoskillit__dispatch_food_truck",
+        "tool_response": _wrap_for_claude_code(rejected_payload),
+    }
+    out_r, _ = _run_hook(event=event_r)
+    text_r = json.loads(out_r)["hookSpecificOutput"]["updatedMCPToolOutput"]
+    rejected_expected = {
+        "error": "fleet_quota_exhausted",
+        "user_visible_message": "quota hit",
+    }
+    for field in rejected_required:
+        assert field in rejected_expected, (
+            f"Field {field!r} from _FMT_DISPATCH_REJECTED_REQUIRED has no expected value — "
+            f"add it to rejected_expected and rejected_payload"
+        )
+        assert rejected_expected[field] in text_r, (
+            f"Rejected shape missing {field!r}: expected {rejected_expected[field]!r} in output"
+        )
+
+
+def test_json_producer_includes_completed_failure():
+    """_dispatch_food_truck_json_producer must include a DispatchCompleted(success=False)
+    shape so the coverage contract exercises both success states and the union of keys
+    includes 'kind'.
+
+    Asserts on the individual completed_failure envelope (not the merged dict)
+    to catch regressions where this specific shape drops 'kind'.
+    """
+    from autoskillit.fleet.state_types import DispatchCompleted, DispatchStatus
+    from tests.infra.conftest import _dispatch_food_truck_json_producer
+
+    keys = _dispatch_food_truck_json_producer().keys()
+    assert "kind" in keys, (
+        f"_dispatch_food_truck_json_producer must emit 'kind' field: {sorted(keys)}"
+    )
+    failure = DispatchCompleted(
+        success=False,
+        dispatch_status=DispatchStatus.FAILURE,
+        dispatch_id="d-verify",
+        dispatched_session_id="s-verify",
+        reason="verify_kind",
+    )
+    failure_env = json.loads(failure.to_envelope())
+    assert "kind" in failure_env, (
+        f"DispatchCompleted(success=False).to_envelope() must emit 'kind': "
+        f"{sorted(failure_env.keys())}"
+    )