fix(studio): judge migration batches on a dynamic-tool app-server turn

Evaluation judging no longer parses the last `codex exec` agent message: the
runner writes a durable `judge/request.json` for the batch it is judging and
Studio answers it with one Codex app-server turn that returns the verdict
through the `reportEvaluation` dynamic tool. The runner keeps owning batching,
the thread record, the batch cache and the report, and still validates the
verdict before caching it, so a rejected batch falls back to `codex exec`
inside the same batch budget. Setting `AGENTKIT_MIGRATION_JUDGE_APP_SERVER=0`
pins that scripted judge for the whole run instead of writing requests.

The request declares how long the runner will listen, and the turn is sized to
answer inside that window, so a batch that is too slow comes back as an
answered failure instead of a verdict nobody reads. Two defects came out of the
first real runs and are fixed here: the turn is now bounded by wall-clock time
instead of the app-server's inactivity window, and a transport failure is
answered with an error envelope instead of leaving the runner waiting. The
`codex exec` fallback also uses the budget left after the channel rather than
the budget the batch started with.

`run_tool_turn` carries the shared one-turn contract for both the route
analysis and the judge, and the analysis now uses it too.

Verified with a real migration-evaluation run (Dify workflow source, Dev
Sandbox): the channel delivered the verdict in 285.9s, the runner accepted it
without falling back, and the report matches the legacy shape field for field.

Author: Garming

frontend/README.md

@@ -600,9 +600,22 @@ See [deployment and operation](service/studio_release_notifier/README.md).
   the framework, entry point, and open questions. Structured frameworks run the
   preinstalled `ak migrate`; Dify and Any projects run
   `ak migrate --execution in-place` with Codex in the same Session. Evaluation
-  deploys a temporary Runtime, checkpoints per-case execution as JSONL, judges
-  batches in one fresh resumable Codex thread, and always reconciles Runtime
-  cleanup before completing or cancelling. Reports show 0–100 display scores,
+  deploys a temporary Runtime, checkpoints per-case execution as JSONL, and
+  judges batches in one fresh resumable Codex thread. Analysis and judging both
+  deliver their contract through an app-server dynamic tool instead of parsing
+  free text, and the analysis turn runs on a Studio background worker so a long
+  analysis never waits inside the upload request. Each judged batch is a durable
+  request the runner writes into the Session: Studio answers it with one
+  app-server turn carrying the verdict, and the runner validates and caches that
+  verdict before the next batch. A request stays on disk until it is answered, so
+  a Studio restart replays the same batch; a request that cannot be answered
+  falls back to `codex exec` inside the same batch budget. The request declares
+  how long the runner will listen, and the turn is sized to answer inside that
+  window, so even a judge batch that is too slow comes back as an answered
+  failure rather than a timeout. Setting
+  `AGENTKIT_MIGRATION_JUDGE_APP_SERVER=0` pins that scripted judge for the
+  whole run instead of writing judge requests at all. Cleanup is always
+  reconciled before completing or cancelling. Reports show 0–100 display scores,
   execution success, evidence coverage, N/A counts, low-scoring and failed
   cases, versions, evidence severity, and cleanup status without a pass/fail
   verdict. Each raw judge score is rounded half up to a 0–100 integer before

frontend/server/migration/app_server.py

@@ -26,12 +26,9 @@
 from collections.abc import Callable
 import os
 
-from veadk.cli.codex_app_server import (
-    CodexAppServerError,
-    CodexAppServerSession,
-    CodexDynamicToolResult,
-)
+from veadk.cli.codex_app_server import CodexDynamicToolResult
 
+from .codex_tool_turn import ToolTurnUnavailable, run_tool_turn
 from .contracts import MigrationContractError, validate_analysis_result
 
 ROUTE_TOOL_NAME = "reportRoute"
@@ -104,36 +101,22 @@ async def run_route_analysis(
 ) -> dict[str, object] | None:
     """Run one analysis turn and return the validated route contract, if any."""
     recorder = RouteRecorder(attempt=attempt, input_sha256=input_sha256)
-    session = CodexAppServerSession(endpoint)
-    session.cwd = cwd
-    if model:
-        session.model = model
-    session.register_dynamic_tool(
-        ROUTE_TOOL_NAME,
-        ROUTE_TOOL_DESCRIPTION,
-        schema,
-        recorder.submit,
-    )
-    try:
-        await session.connect()
-    except CodexAppServerError as error:
-        raise MigrationAnalysisUnavailable(str(error)) from error
     try:
-        async for event in session.stream_turn(
-            prompt,
+        await run_tool_turn(
+            endpoint=endpoint,
+            prompt=prompt,
+            cwd=cwd,
+            tool_name=ROUTE_TOOL_NAME,
+            tool_description=ROUTE_TOOL_DESCRIPTION,
+            tool_schema=schema,
+            handler=recorder.submit,
+            has_result=lambda: recorder.result is not None,
+            model=model,
             timeout_seconds=timeout_seconds,
-        ):
-            if event_sink is not None:
-                event_sink(event)
-            if recorder.result is not None:
-                # 结果已经到手：终止本轮，避免继续消耗 token 和沙箱时间。
-                await session.interrupt()
-                break
-    except CodexAppServerError as error:
-        if recorder.result is None:
-            raise MigrationAnalysisUnavailable(str(error)) from error
-    finally:
-        await session.close()
+            event_sink=event_sink,
+        )
+    except ToolTurnUnavailable as error:
+        raise MigrationAnalysisUnavailable(str(error)) from error
     return recorder.result
 
 

frontend/server/migration/codex_tool_turn.py

@@ -0,0 +1,129 @@
+# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""One Codex app-server turn that reports its result through a dynamic tool.
+
+Studio drives every structured Codex turn the same way: a dynamic tool registered on
+``thread/start`` carries the result as typed JSON-RPC arguments, a rejected call returns
+``success: false`` so the same turn can correct itself, and the turn ends as soon as the
+contract has been delivered.  Callers own the contract: they pass the tool spec, the
+validator, and the check that says the result has arrived.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+
+from veadk.cli.codex_app_server import (
+    CodexAppServerError,
+    CodexAppServerSession,
+    CodexDynamicToolResult,
+)
+
+ToolHandler = Callable[[dict[str, object]], CodexDynamicToolResult]
+
+__all__ = [
+    "ToolHandler",
+    "ToolTurnDeadlineExceeded",
+    "ToolTurnUnavailable",
+    "run_tool_turn",
+]
+
+
+class ToolTurnUnavailable(RuntimeError):
+    """The Sandbox app-server could not start or finish the tool turn."""
+
+
+class ToolTurnDeadlineExceeded(ToolTurnUnavailable):
+    """The turn kept making progress but ran past the caller's wall-clock window.
+
+    Callers treat this differently from a transport failure: the app-server works,
+    the batch is simply too slow for the window that was granted.
+    """
+
+
+async def run_tool_turn(
+    *,
+    endpoint: str,
+    prompt: str,
+    cwd: str,
+    tool_name: str,
+    tool_description: str,
+    tool_schema: dict[str, object],
+    handler: ToolHandler,
+    has_result: Callable[[], bool],
+    thread_id: str = "",
+    model: str = "",
+    timeout_seconds: float,
+    event_sink: Callable[[object], None] | None = None,
+) -> str:
+    """Run one turn and return the thread id that carried it.
+
+    ``thread_id`` resumes a durable thread instead of starting a new one; the
+    app-server restores that thread's dynamic tools, so the same contract keeps
+    arriving.  ``timeout_seconds`` is a wall-clock budget, not just the app-server's
+    inactivity window: callers that must answer inside a caller-owned window cannot
+    rely on a turn that keeps making progress, so this interrupts it at the deadline.
+    Any protocol or transport failure becomes ``ToolTurnUnavailable`` so the caller can
+    fall back, unless the result already arrived.
+    """
+    session = CodexAppServerSession(endpoint)
+    session.cwd = cwd
+    if model:
+        session.model = model
+    session.register_dynamic_tool(
+        tool_name,
+        tool_description,
+        tool_schema,
+        handler,
+    )
+    used_thread = thread_id
+    try:
+        try:
+            if thread_id:
+                await session.attach_thread(thread_id)
+            else:
+                await session.connect()
+        except CodexAppServerError as error:
+            raise ToolTurnUnavailable(str(error)) from error
+        deadline = asyncio.get_running_loop().time() + timeout_seconds
+        try:
+            async for event in session.stream_turn(
+                prompt,
+                timeout_seconds=timeout_seconds,
+            ):
+                if event_sink is not None:
+                    event_sink(event)
+                if has_result():
+                    # 结果已经到手：终止本轮，避免继续消耗 token 和沙箱时间。
+                    await session.interrupt()
+                    break
+                if asyncio.get_running_loop().time() >= deadline:
+                    # 回合一直在产生进度，但已经超出调用方的窗口：主动收尾。
+                    await session.interrupt()
+                    raise ToolTurnDeadlineExceeded("Codex 回合超出时间预算。")
+        except TimeoutError as error:
+            # app-server 客户端自己的空闲超时：同样是「没在窗口内交付」。
+            if not has_result():
+                raise ToolTurnDeadlineExceeded(
+                    str(error) or "Codex 回合超时。"
+                ) from error
+        except CodexAppServerError as error:
+            if not has_result():
+                raise ToolTurnUnavailable(str(error)) from error
+        used_thread = session.thread_id or thread_id
+    finally:
+        await session.close()
+    return used_thread