fix(subagent): structured subagent_status field over text parsing (#3146) (#3154)

* fix(subagent): structured subagent_status field over text parsing

Closes #3146.

## Why

The frontend used to derive subtask card state by string-matching the
leading text of the `task` tool's result. That contract surface was
fragile — `#3107` BUG-007 and the `#3131` review both surfaced cases
where new backend wording (`Task cancelled by user.`,
`Task polling timed out after N minutes`, `ToolErrorHandlingMiddleware`
exception wrappers) silently broke the card lifecycle. The frontend
fallback kept growing more prefixes; any future rewording would break
it again.

## Design

1. **Backend → frontend contract**: `ToolMessage.additional_kwargs`
   carries `subagent_status` (one of `completed | failed | cancelled |
   timed_out | polling_timed_out`) and an optional `subagent_error`
   blob. The frontend prefers it over parsing `content`.

2. **Centralised stamping, not 8 sprinkled stamps**: rather than have
   each of `task_tool.py`'s 5 normal-return + 3 pre-execution `Error:`
   paths remember to set `additional_kwargs`, `ToolErrorHandlingMiddleware`
   stamps the field after every task-tool call. Adding a new return
   path in `task_tool.py` cannot now skip the stamp.

3. **Cross-language contract fixture**: the prefix→status mapping is
   the one piece both sides must agree on. The shared fixture at
   `contracts/subagent_status_contract.json` lists every backend return
   string, the expected status, and what the error substring should
   contain. Backend test (`backend/tests/test_subagent_status_contract.py`)
   and frontend test (`frontend/tests/unit/core/tasks/subtask-result.test.ts`)
   both load that fixture and assert the same cases. A wording drift on
   either side fails the matching language's test.

4. **Round-trip serialisation pinned**: the round-trip test asserts
   `ToolMessage.model_dump_json()` → `model_validate_json()` preserves
   `additional_kwargs.subagent_status`. Catches the case where a future
   LangChain or Pydantic upgrade silently strips unknown kwargs.

5. **Frontend status collapse documented**: the backend has five status
   values, the frontend card has three (`completed | failed |
   in_progress`). `cancelled` / `timed_out` / `polling_timed_out` all
   collapse to `failed` with the original status preserved in `error`.
   `parseSubtaskResult` returns `in_progress` for unknown values so a
   backend that ships a new enum variant before the frontend upgrades
   degrades to the legacy prefix fallback instead of getting pinned.

## Changes

Backend:
- `deerflow.subagents.status_contract` — new module exporting
  `SUBAGENT_STATUS_KEY`, `SUBAGENT_ERROR_KEY`,
  `SUBAGENT_STATUS_VALUES`, `extract_subagent_status(content)`, and
  `make_subagent_additional_kwargs(status, error)`.
- `ToolErrorHandlingMiddleware`: new `_stamp_task_subagent_status`
  helper centralises the stamp; `wrap_tool_call` / `awrap_tool_call`
  stamp on the success path; `_build_error_message` stamps on the
  wrapper path (carrying `ExcClass: detail` into `subagent_error`).
  Non-task tools are untouched.
- New tests: `test_subagent_status_contract.py` (19 cases from the
  shared fixture + status-enum / blank-error / unknown-status
  rejection) and `test_tool_error_handling_subagent_stamp.py`
  (middleware integration: terminal-content stamps, non-terminal
  doesn't, non-task tools untouched, async path mirrors sync,
  existing additional_kwargs survive, JSON round-trip preserved).

Frontend:
- `parseSubtaskResult(text, additionalKwargs?)` — prefers the
  structured stamp; falls back to the legacy prefix matcher for
  historical threads / unknown future status values.
- `STRUCTURED_STATUS_TO_SUBTASK` documents the five→three collapse.
- `message-list.tsx` passes `message.additional_kwargs` through.
- `subtask-result.test.ts` adds a structured-status block + a
  fixture-driven contract block; legacy prefix tests stay green for
  the fallback path.

Contract:
- `contracts/subagent_status_contract.json` — single source of truth
  both languages load. Whitespace variants, varied N for polling
  timeouts, the 3 pre-execution `Error:` returns task_tool produces,
  and the middleware wrapper shape are all in there.

## Test plan
- `make lint` clean (backend + frontend).
- `pytest tests/test_subagent_status_contract.py
   tests/test_tool_error_handling_subagent_stamp.py` → 37 passed.
- `pnpm test --run` → 103 passed (was 76, +27 new).

## Migration / fallback retirement

The text-prefix fallback stays in place until backend telemetry shows
the frontend never hits it for newly produced messages. At that point
a follow-up PR can drop the prefix branches and keep only the
structured-status branch.

Refs: bytedance/deer-flow#3138 (split summary), #3107 (origin), #3131
(prior prefix-only fix), #3146 (this issue).

* fix(subtask): back-fill result/error from text when structured status present

Three follow-ups on the PR #3154 review:

1. `readStructuredStatus` no longer short-circuits the prefix parse.
   The backend currently stamps only the `subagent_status` enum value;
   the human-facing `result` body and wrapped-error message still live
   in `ToolMessage.content`. Dropping the text parse meant successful
   tasks rendered empty completed pills and wrapped failures lost their
   diagnostic. Now both shapes get composed: structured status wins,
   `result`/`error` come from text when both sides agree, and a lying
   success body under a `failed` stamp is dropped instead of leaking.

2. Replace the ESM-incompatible `__dirname` fixture lookup in
   subtask-result.test.ts with `fileURLToPath(new URL(..., import.meta.url))`.
   The frontend package is `"type": "module"`, so the previous path
   would have thrown at runtime if anything ever changed under the
   contract directory.

3. Drop the `$schema` reference from contracts/subagent_status_contract.json
   pointing at a file that doesn't exist in the tree.

Three new tests cover the structured + text composition: completed
back-fills the success body, failed back-fills the wrapper text, and
unrecognised content under a `failed` stamp stays empty rather than
echoing noise.

backend/packages/harness/deerflow/subagents/status_contract.py

@@ -0,0 +1,102 @@
+"""Backend↔frontend contract for the structured subagent status.
+
+Bytedance/deer-flow issue #3146: the frontend used to derive the
+subtask card state by string-matching the leading text of the
+``task`` tool's result. That contract was fragile — any rewording on
+the backend silently broke the card lifecycle, and the issue history
+of #3107 BUG-007 / #3131 review showed it repeatedly.
+
+This module replaces the text-shaped contract with a small structured
+one carried inside ``ToolMessage.additional_kwargs``:
+
+- ``subagent_status``: one of ``SUBAGENT_STATUS_VALUES``.
+- ``subagent_error`` (optional): the human-readable error blob the
+  backend recorded.
+
+The mapping from "task tool result text" to status is the one piece
+the backend stamper (``ToolErrorHandlingMiddleware``) and the
+frontend fallback parser must agree on. The shared fixture at
+``contracts/subagent_status_contract.json`` is the single source of
+truth — both sides' tests load it and assert behaviour.
+"""
+
+from __future__ import annotations
+
+from typing import Literal
+
+SUBAGENT_STATUS_KEY = "subagent_status"
+SUBAGENT_ERROR_KEY = "subagent_error"
+
+SubagentStatusValue = Literal[
+    "completed",
+    "failed",
+    "cancelled",
+    "timed_out",
+    "polling_timed_out",
+]
+
+#: Enumeration of every value ``subagent_status`` may take. Mirrors the
+#: ``valid_status_values`` array in the shared fixture; the contract test
+#: pins them against each other.
+SUBAGENT_STATUS_VALUES: tuple[SubagentStatusValue, ...] = (
+    "completed",
+    "failed",
+    "cancelled",
+    "timed_out",
+    "polling_timed_out",
+)
+
+# Prefix table — ordered most-specific-first because some prefixes are
+# substrings of others ("Task timed out" vs "Task polling timed out", "Task
+# failed" vs "Task failed. Error: ..."). The "Task " prefixes come from
+# ``task_tool.py``'s 5 normal-return strings; the bare ``Error:`` prefix
+# catches both the 3 ``Error:`` pre-execution returns and the wrapper
+# produced by ``ToolErrorHandlingMiddleware`` for any task tool exception.
+_PREFIX_TO_STATUS: tuple[tuple[str, SubagentStatusValue], ...] = (
+    ("Task Succeeded. Result:", "completed"),
+    ("Task polling timed out", "polling_timed_out"),
+    ("Task timed out", "timed_out"),
+    ("Task cancelled by user", "cancelled"),
+    ("Task failed.", "failed"),
+    ("Error", "failed"),
+)
+
+
+def extract_subagent_status(content: str) -> SubagentStatusValue | None:
+    """Infer the structured status for a ``task`` tool result string.
+
+    Returns ``None`` when the content does not match any known terminal
+    prefix. Non-terminal streaming chunks fall into this branch by
+    design — the middleware then leaves ``subagent_status`` unset so
+    the frontend keeps the card on its in-progress placeholder until
+    the real terminal frame arrives.
+    """
+    trimmed = content.strip()
+    for prefix, status in _PREFIX_TO_STATUS:
+        if trimmed.startswith(prefix):
+            return status
+    return None
+
+
+def make_subagent_additional_kwargs(
+    status: SubagentStatusValue,
+    *,
+    error: str | None = None,
+) -> dict[str, str]:
+    """Build the ``additional_kwargs`` payload the middleware stamps.
+
+    Drops the error field when blank so the JSON wire format never carries
+    a misleading empty ``subagent_error: ""``.
+
+    Raises:
+        ValueError: when ``status`` is not in :data:`SUBAGENT_STATUS_VALUES`.
+            We do not accept arbitrary strings: a typo would silently leak
+            through to the frontend and degrade to the legacy prefix
+            fallback rather than failing loudly.
+    """
+    if status not in SUBAGENT_STATUS_VALUES:
+        raise ValueError(f"invalid subagent status {status!r}; expected one of {SUBAGENT_STATUS_VALUES}")
+    payload: dict[str, str] = {SUBAGENT_STATUS_KEY: status}
+    if error and error.strip():
+        payload[SUBAGENT_ERROR_KEY] = error.strip()
+    return payload