diff --git a/backend/app/api/endpoints/subtasks.py b/backend/app/api/endpoints/subtasks.py
index 63e3ebd5a8..6c88dcd73d 100644
--- a/backend/app/api/endpoints/subtasks.py
+++ b/backend/app/api/endpoints/subtasks.py
@@ -23,8 +23,13 @@
SubtaskListResponse,
SubtaskUpdate,
)
+from app.schemas.turn_file_changes import (
+ TurnFileChangesDiffResponse,
+ TurnFileChangesRevertResponse,
+)
from app.services.chat.storage import session_manager
from app.services.subtask import subtask_service
+from app.services.turn_file_changes import turn_file_changes_service
router = APIRouter()
@@ -103,6 +108,40 @@ def get_subtask(
)
+@router.get(
+ "/{subtask_id}/file-changes/diff",
+ response_model=TurnFileChangesDiffResponse,
+)
+async def get_turn_file_changes_diff(
+ subtask_id: int,
+ current_user: User = Depends(security.get_current_user),
+ db: Session = Depends(get_db),
+) -> TurnFileChangesDiffResponse:
+ """Load one assistant turn's validated diff from its execution device."""
+ return await turn_file_changes_service.get_diff(
+ db=db,
+ user_id=current_user.id,
+ subtask_id=subtask_id,
+ )
+
+
+@router.post(
+ "/{subtask_id}/file-changes/revert",
+ response_model=TurnFileChangesRevertResponse,
+)
+async def revert_turn_file_changes(
+ subtask_id: int,
+ current_user: User = Depends(security.get_current_user),
+ db: Session = Depends(get_db),
+) -> TurnFileChangesRevertResponse:
+ """Reverse one assistant turn without overwriting later workspace changes."""
+ return await turn_file_changes_service.revert(
+ db=db,
+ user_id=current_user.id,
+ subtask_id=subtask_id,
+ )
+
+
@router.put("/{subtask_id}", response_model=SubtaskInDB)
def update_subtask(
subtask_id: int,
diff --git a/backend/app/schemas/turn_file_changes.py b/backend/app/schemas/turn_file_changes.py
new file mode 100644
index 0000000000..1b4b8e39c1
--- /dev/null
+++ b/backend/app/schemas/turn_file_changes.py
@@ -0,0 +1,54 @@
+# SPDX-FileCopyrightText: 2025 Weibo, Inc.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Schemas for per-turn coding file changes."""
+
+from datetime import datetime
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class TurnFileChangeItem(BaseModel):
+ """One file changed during an agent turn."""
+
+ model_config = ConfigDict(extra="forbid")
+
+ old_path: str | None = None
+ path: str = Field(min_length=1)
+ change_type: Literal["created", "modified", "deleted", "renamed"]
+ additions: int = Field(ge=0)
+ deletions: int = Field(ge=0)
+ binary: bool = False
+
+
+class TurnFileChangesSummary(BaseModel):
+ """Persisted lightweight summary for a turn patch artifact."""
+
+ model_config = ConfigDict(extra="forbid")
+
+ version: Literal[1]
+ status: Literal["active", "reverted", "conflicted", "artifact_missing"]
+ artifact_id: str = Field(min_length=1)
+ device_id: str = Field(min_length=1)
+ workspace_path: str = Field(min_length=1)
+ file_count: int = Field(ge=0)
+ additions: int = Field(ge=0)
+ deletions: int = Field(ge=0)
+ files: list[TurnFileChangeItem]
+ reverted_at: datetime | None = None
+
+
+class TurnFileChangesDiffResponse(BaseModel):
+ """Validated unified diff returned from the owning device."""
+
+ subtask_id: int
+ diff: str
+
+
+class TurnFileChangesRevertResponse(BaseModel):
+ """Updated summary returned after an idempotent revert."""
+
+ subtask_id: int
+ file_changes: TurnFileChangesSummary
diff --git a/backend/app/services/chat/trigger/unified.py b/backend/app/services/chat/trigger/unified.py
index b5c28ef7b2..efff66e351 100644
--- a/backend/app/services/chat/trigger/unified.py
+++ b/backend/app/services/chat/trigger/unified.py
@@ -386,6 +386,7 @@ async def build_execution_request(
previous_bot_id=previous_bot_id,
web_runtime_guidance=web_runtime_guidance,
)
+ request.device_id = device_id or request.device_id
# Merge reasoning config from API/model selection into model_config.
# Priority: explicit API reasoning_config > UI model_options > model think_config.
diff --git a/backend/app/services/device/command_registry.py b/backend/app/services/device/command_registry.py
index 5b53112347..311710a938 100644
--- a/backend/app/services/device/command_registry.py
+++ b/backend/app/services/device/command_registry.py
@@ -353,6 +353,156 @@ def fail(message, code=64):
f"python3 -c {shlex.quote(READ_RUNTIME_AUTH_FILE_SCRIPT)}"
)
+TURN_FILE_CHANGES_SCRIPT = """
+import gzip
+import hashlib
+import json
+import os
+import re
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+MAX_PATCH_BYTES = 20 * 1024 * 1024
+ARTIFACT_PATTERN = re.compile(
+ r"turn-file-changes/([1-9][0-9]*)/([1-9][0-9]*)"
+)
+
+
+def finish(payload, code=0):
+ print(json.dumps(payload, ensure_ascii=False))
+ sys.exit(code)
+
+
+def fail(message, code=64, status=None):
+ payload = {"success": False, "error": message}
+ if status:
+ payload["status"] = status
+ finish(payload, code)
+
+
+if len(sys.argv) != 3:
+ fail("mode and artifact id are required")
+
+mode = sys.argv[1]
+artifact_id = sys.argv[2]
+if mode not in {"review", "revert"}:
+ fail("invalid mode")
+
+match = ARTIFACT_PATTERN.fullmatch(artifact_id)
+if not match:
+ fail("invalid artifact id")
+
+task_id = int(match.group(1))
+subtask_id = int(match.group(2))
+executor_home = Path(
+ os.environ.get("WEGENT_EXECUTOR_HOME", "~/.wegent-executor")
+).expanduser()
+artifact_root = (executor_home / "artifacts").resolve()
+artifact_dir = (artifact_root / artifact_id).resolve()
+if artifact_root not in artifact_dir.parents:
+ fail("invalid artifact id")
+
+metadata_path = artifact_dir / "metadata.json"
+patch_path = artifact_dir / "changes.patch.gz"
+if not metadata_path.is_file() or not patch_path.is_file():
+ finish(
+ {
+ "success": False,
+ "status": "artifact_missing",
+ "error": "turn file changes artifact is missing",
+ }
+ )
+
+try:
+ metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
+except (OSError, json.JSONDecodeError) as exc:
+ fail(f"invalid artifact metadata: {exc}", code=65)
+
+if not isinstance(metadata, dict):
+ fail("invalid artifact metadata", code=65)
+if metadata.get("task_id") != task_id or metadata.get("subtask_id") != subtask_id:
+ fail("artifact metadata id mismatch", code=65)
+
+workspace = Path.cwd().resolve()
+try:
+ metadata_workspace = Path(str(metadata["workspace_path"])).resolve()
+except (KeyError, OSError):
+ fail("invalid artifact workspace", code=65)
+if metadata_workspace != workspace:
+ fail("artifact workspace mismatch", code=65)
+
+try:
+ with gzip.open(patch_path, "rb") as patch_file:
+ patch = patch_file.read(MAX_PATCH_BYTES + 1)
+except (OSError, gzip.BadGzipFile) as exc:
+ fail(f"failed to read artifact patch: {exc}", code=65)
+if len(patch) > MAX_PATCH_BYTES:
+ fail("artifact patch exceeds size limit", code=65)
+if hashlib.sha256(patch).hexdigest() != metadata.get("checksum"):
+ fail("artifact patch checksum mismatch", code=65)
+
+if mode == "review":
+ finish(
+ {
+ "success": True,
+ "diff": patch.decode("utf-8", errors="replace"),
+ }
+ )
+
+temp_path = None
+try:
+ with tempfile.NamedTemporaryFile(
+ prefix="wegent-validated-turn-",
+ suffix=".patch",
+ delete=False,
+ ) as temp_file:
+ temp_file.write(patch)
+ temp_path = Path(temp_file.name)
+
+ check = subprocess.run(
+ ["git", "apply", "--reverse", "--check", "--binary", str(temp_path)],
+ cwd=workspace,
+ capture_output=True,
+ text=True,
+ )
+ if check.returncode != 0:
+ finish(
+ {
+ "success": False,
+ "status": "conflicted",
+ "error": "patch does not apply",
+ }
+ )
+
+ apply_result = subprocess.run(
+ ["git", "apply", "--reverse", "--binary", str(temp_path)],
+ cwd=workspace,
+ capture_output=True,
+ text=True,
+ )
+ if apply_result.returncode != 0:
+ finish(
+ {
+ "success": False,
+ "status": "conflicted",
+ "error": "patch does not apply",
+ }
+ )
+ finish({"success": True, "status": "reverted"})
+finally:
+ if temp_path is not None:
+ temp_path.unlink(missing_ok=True)
+""".strip()
+
+TURN_FILE_CHANGES_REVIEW_COMMAND = (
+ f"python3 -c {shlex.quote(TURN_FILE_CHANGES_SCRIPT)} review"
+)
+TURN_FILE_CHANGES_REVERT_COMMAND = (
+ f"python3 -c {shlex.quote(TURN_FILE_CHANGES_SCRIPT)} revert"
+)
+
DEFAULT_LOCAL_DEVICE_COMMANDS: dict[str, LocalDeviceCommandDefinition] = {
"pwd": LocalDeviceCommandDefinition(command="pwd"),
@@ -438,6 +588,14 @@ def fail(message, code=64):
command=READ_RUNTIME_AUTH_FILE_COMMAND,
post_processor="json",
),
+ "turn_file_changes_review": LocalDeviceCommandDefinition(
+ command=TURN_FILE_CHANGES_REVIEW_COMMAND,
+ post_processor="json",
+ ),
+ "turn_file_changes_revert": LocalDeviceCommandDefinition(
+ command=TURN_FILE_CHANGES_REVERT_COMMAND,
+ post_processor="json",
+ ),
}
diff --git a/backend/app/services/execution/dispatcher.py b/backend/app/services/execution/dispatcher.py
index 68ecb8e357..722589a125 100644
--- a/backend/app/services/execution/dispatcher.py
+++ b/backend/app/services/execution/dispatcher.py
@@ -177,6 +177,7 @@ def extract_completed_result(response_data: dict) -> dict:
"standalone_chat_workspace_path": response_data.get(
"standalone_chat_workspace_path"
),
+ "file_changes": response_data.get("file_changes"),
"reasoning_content": reasoning_content,
}
diff --git a/backend/app/services/turn_file_changes.py b/backend/app/services/turn_file_changes.py
new file mode 100644
index 0000000000..d6a092b86d
--- /dev/null
+++ b/backend/app/services/turn_file_changes.py
@@ -0,0 +1,270 @@
+# SPDX-FileCopyrightText: 2025 Weibo, Inc.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Review and safely revert per-turn file change artifacts."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+
+from fastapi import HTTPException
+from pydantic import ValidationError
+from sqlalchemy.orm import Session
+from sqlalchemy.orm.attributes import flag_modified
+
+from app.models.subtask import Subtask, SubtaskRole
+from app.models.task import TaskResource
+from app.schemas.turn_file_changes import (
+ TurnFileChangesDiffResponse,
+ TurnFileChangesRevertResponse,
+ TurnFileChangesSummary,
+)
+from app.services.device.command_service import (
+ DeviceCommandError,
+ DeviceCommandNotFoundError,
+ execute_configured_device_command,
+)
+
+MAX_DIFF_OUTPUT_BYTES = 5 * 1024 * 1024
+
+
+def _error(status_code: int, code: str, message: str, **extra: Any) -> HTTPException:
+ return HTTPException(
+ status_code=status_code,
+ detail={"code": code, "message": message, **extra},
+ )
+
+
+class TurnFileChangesService:
+ """Load stored summaries and dispatch artifact commands to their device."""
+
+ async def get_diff(
+ self,
+ *,
+ db: Session,
+ user_id: int,
+ subtask_id: int,
+ ) -> TurnFileChangesDiffResponse:
+ subtask, summary = self._load_summary(db, user_id, subtask_id)
+ if summary.status == "artifact_missing":
+ raise self._artifact_missing_error()
+
+ payload = await self._execute(
+ db=db,
+ user_id=user_id,
+ summary=summary,
+ command_key="turn_file_changes_review",
+ )
+ self._raise_for_artifact_status(db, subtask, payload)
+ diff = payload.get("diff")
+ if payload.get("success") is not True or not isinstance(diff, str):
+ raise _error(
+ 422,
+ "TURN_FILE_CHANGES_ARTIFACT_INVALID",
+ str(payload.get("error") or "Device returned an invalid diff artifact"),
+ )
+ return TurnFileChangesDiffResponse(subtask_id=subtask.id, diff=diff)
+
+ async def revert(
+ self,
+ *,
+ db: Session,
+ user_id: int,
+ subtask_id: int,
+ ) -> TurnFileChangesRevertResponse:
+ subtask, summary = self._load_summary(db, user_id, subtask_id)
+ if summary.status == "reverted":
+ return TurnFileChangesRevertResponse(
+ subtask_id=subtask.id,
+ file_changes=summary,
+ )
+ if summary.status == "artifact_missing":
+ raise self._artifact_missing_error()
+
+ payload = await self._execute(
+ db=db,
+ user_id=user_id,
+ summary=summary,
+ command_key="turn_file_changes_revert",
+ )
+ self._raise_for_artifact_status(db, subtask, payload)
+ if payload.get("success") is True and payload.get("status") == "reverted":
+ updated = self._update_status(
+ db,
+ subtask,
+ status="reverted",
+ reverted_at=datetime.now(timezone.utc).isoformat(),
+ )
+ return TurnFileChangesRevertResponse(
+ subtask_id=subtask.id,
+ file_changes=updated,
+ )
+ if payload.get("status") == "conflicted":
+ updated = self._update_status(db, subtask, status="conflicted")
+ raise _error(
+ 409,
+ "TURN_FILE_CHANGES_CONFLICT",
+ str(payload.get("error") or "Patch does not apply"),
+ file_changes=updated.model_dump(mode="json"),
+ )
+ raise _error(
+ 422,
+ "TURN_FILE_CHANGES_ARTIFACT_INVALID",
+ str(payload.get("error") or "Device returned an invalid revert result"),
+ )
+
+ def _load_summary(
+ self,
+ db: Session,
+ user_id: int,
+ subtask_id: int,
+ ) -> tuple[Subtask, TurnFileChangesSummary]:
+ record = (
+ db.query(Subtask, TaskResource)
+ .join(TaskResource, TaskResource.id == Subtask.task_id)
+ .filter(
+ Subtask.id == subtask_id,
+ Subtask.role == SubtaskRole.ASSISTANT,
+ TaskResource.user_id == user_id,
+ TaskResource.kind == "Task",
+ TaskResource.is_active.in_(TaskResource.is_active_query()),
+ )
+ .first()
+ )
+ if record is None:
+ raise _error(
+ 404,
+ "TURN_FILE_CHANGES_NOT_FOUND",
+ "Assistant message file changes were not found",
+ )
+
+ subtask, task = record
+ result = subtask.result if isinstance(subtask.result, dict) else {}
+ raw_summary = result.get("file_changes")
+ try:
+ summary = TurnFileChangesSummary.model_validate(raw_summary)
+ except ValidationError as exc:
+ raise _error(
+ 409,
+ "TURN_FILE_CHANGES_INVALID_STATE",
+ "Stored file changes summary is invalid",
+ ) from exc
+
+ task_json = task.json if isinstance(task.json, dict) else {}
+ task_spec = task_json.get("spec")
+ expected_device_id = (
+ task_spec.get("device_id") if isinstance(task_spec, dict) else None
+ )
+ if expected_device_id and expected_device_id != summary.device_id:
+ raise _error(
+ 409,
+ "TURN_FILE_CHANGES_DEVICE_MISMATCH",
+ "Stored artifact device does not match the task device",
+ )
+ return subtask, summary
+
+ async def _execute(
+ self,
+ *,
+ db: Session,
+ user_id: int,
+ summary: TurnFileChangesSummary,
+ command_key: str,
+ ) -> dict[str, Any]:
+ try:
+ result = await execute_configured_device_command(
+ db=db,
+ user_id=user_id,
+ device_id=summary.device_id,
+ command_key=command_key,
+ path=summary.workspace_path,
+ args=[summary.artifact_id],
+ timeout_seconds=30,
+ max_output_bytes=MAX_DIFF_OUTPUT_BYTES,
+ )
+ except DeviceCommandNotFoundError as exc:
+ raise _error(
+ 409,
+ "TURN_FILE_CHANGES_DEVICE_UNAVAILABLE",
+ str(exc),
+ ) from exc
+ except DeviceCommandError as exc:
+ code = (
+ "TURN_FILE_CHANGES_DEVICE_OFFLINE"
+ if "offline" in str(exc).lower()
+ else "TURN_FILE_CHANGES_DEVICE_UNAVAILABLE"
+ )
+ raise _error(409, code, str(exc)) from exc
+
+ if not isinstance(result, dict):
+ raise _error(
+ 422,
+ "TURN_FILE_CHANGES_ARTIFACT_INVALID",
+ "Device returned an invalid command result",
+ )
+ if result.get("success") is not True:
+ raise _error(
+ 422,
+ "TURN_FILE_CHANGES_ARTIFACT_INVALID",
+ str(
+ result.get("error")
+ or result.get("stderr")
+ or "Artifact command failed"
+ ),
+ )
+ payload = result.get("stdout")
+ if not isinstance(payload, dict):
+ raise _error(
+ 422,
+ "TURN_FILE_CHANGES_ARTIFACT_INVALID",
+ "Device returned malformed artifact output",
+ )
+ return payload
+
+ def _raise_for_artifact_status(
+ self,
+ db: Session,
+ subtask: Subtask,
+ payload: dict[str, Any],
+ ) -> None:
+ if payload.get("status") != "artifact_missing":
+ return
+ updated = self._update_status(db, subtask, status="artifact_missing")
+ raise _error(
+ 410,
+ "TURN_FILE_CHANGES_ARTIFACT_MISSING",
+ str(payload.get("error") or "Turn file changes artifact is missing"),
+ file_changes=updated.model_dump(mode="json"),
+ )
+
+ def _update_status(
+ self,
+ db: Session,
+ subtask: Subtask,
+ *,
+ status: str,
+ reverted_at: str | None = None,
+ ) -> TurnFileChangesSummary:
+ updated_result = dict(subtask.result or {})
+ updated_file_changes = dict(updated_result["file_changes"])
+ updated_file_changes["status"] = status
+ if reverted_at is not None:
+ updated_file_changes["reverted_at"] = reverted_at
+ updated_result["file_changes"] = updated_file_changes
+ subtask.result = updated_result
+ flag_modified(subtask, "result")
+ db.commit()
+ return TurnFileChangesSummary.model_validate(updated_file_changes)
+
+ @staticmethod
+ def _artifact_missing_error() -> HTTPException:
+ return _error(
+ 410,
+ "TURN_FILE_CHANGES_ARTIFACT_MISSING",
+ "Turn file changes artifact is missing",
+ )
+
+
+turn_file_changes_service = TurnFileChangesService()
diff --git a/backend/tests/api/endpoints/test_subtask_file_changes.py b/backend/tests/api/endpoints/test_subtask_file_changes.py
new file mode 100644
index 0000000000..05a90e3d37
--- /dev/null
+++ b/backend/tests/api/endpoints/test_subtask_file_changes.py
@@ -0,0 +1,88 @@
+# SPDX-FileCopyrightText: 2025 Weibo, Inc.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+from unittest.mock import AsyncMock, patch
+
+from fastapi.testclient import TestClient
+
+from app.models.user import User
+
+
+def test_get_subtask_file_changes_diff_requires_authentication(
+ test_client: TestClient,
+):
+ response = test_client.get("/api/subtasks/123/file-changes/diff")
+
+ assert response.status_code == 401
+
+
+def test_get_subtask_file_changes_diff_delegates_to_service(
+ test_client: TestClient,
+ test_token: str,
+ test_user: User,
+):
+ with patch(
+ "app.api.endpoints.subtasks.turn_file_changes_service.get_diff",
+ AsyncMock(
+ return_value={
+ "subtask_id": 123,
+ "diff": "diff --git a/a.txt b/a.txt\n",
+ }
+ ),
+ ) as get_diff:
+ response = test_client.get(
+ "/api/subtasks/123/file-changes/diff",
+ headers={"Authorization": f"Bearer {test_token}"},
+ )
+
+ assert response.status_code == 200
+ assert response.json()["subtask_id"] == 123
+ get_diff.assert_awaited_once()
+ assert get_diff.await_args.kwargs["user_id"] == test_user.id
+
+
+def test_revert_subtask_file_changes_delegates_without_request_body(
+ test_client: TestClient,
+ test_token: str,
+ test_user: User,
+):
+ summary = {
+ "version": 1,
+ "status": "reverted",
+ "artifact_id": "turn-file-changes/10/20",
+ "device_id": "device-1",
+ "workspace_path": "/workspace/project",
+ "file_count": 1,
+ "additions": 1,
+ "deletions": 0,
+ "files": [
+ {
+ "old_path": None,
+ "path": "a.txt",
+ "change_type": "created",
+ "additions": 1,
+ "deletions": 0,
+ "binary": False,
+ }
+ ],
+ "reverted_at": "2026-06-11T00:00:00Z",
+ }
+ with patch(
+ "app.api.endpoints.subtasks.turn_file_changes_service.revert",
+ AsyncMock(
+ return_value={
+ "subtask_id": 123,
+ "file_changes": summary,
+ }
+ ),
+ ) as revert:
+ response = test_client.post(
+ "/api/subtasks/123/file-changes/revert",
+ headers={"Authorization": f"Bearer {test_token}"},
+ )
+
+ assert response.status_code == 200
+ assert response.json()["file_changes"]["status"] == "reverted"
+ revert.assert_awaited_once()
+ assert revert.await_args.kwargs["user_id"] == test_user.id
diff --git a/backend/tests/services/chat/test_trigger_unified.py b/backend/tests/services/chat/test_trigger_unified.py
index 2fed7b8afb..ffb3f1cec5 100644
--- a/backend/tests/services/chat/test_trigger_unified.py
+++ b/backend/tests/services/chat/test_trigger_unified.py
@@ -92,6 +92,35 @@ def test_apply_user_runtime_config_skips_non_codex_models(monkeypatch):
@pytest.mark.unit
class TestBuildExecutionRequestUserSubtaskId:
+ async def test_propagates_device_id_to_execution_request(self):
+ from app.services.chat.trigger import unified as trigger_unified
+
+ mock_db = MagicMock()
+ request_from_builder = ExecutionRequest(task_id=1, subtask_id=2)
+ mock_builder = MagicMock()
+ mock_builder.build.return_value = request_from_builder
+
+ with patch.object(trigger_unified, "SessionLocal", return_value=mock_db):
+ with patch(
+ "app.services.execution.TaskRequestBuilder",
+ return_value=mock_builder,
+ ):
+ task = MagicMock(id=1, json={})
+ assistant_subtask = MagicMock(id=2)
+ team = MagicMock()
+ user = MagicMock(id=7)
+
+ result = await trigger_unified.build_execution_request(
+ task=task,
+ assistant_subtask=assistant_subtask,
+ team=team,
+ user=user,
+ message="hello",
+ device_id="device-1",
+ )
+
+ assert result.device_id == "device-1"
+
async def test_propagates_user_subtask_id_to_execution_request(self):
"""Ensure user_subtask_id is always propagated for downstream RAG persistence."""
from app.services.chat.trigger import unified as trigger_unified
diff --git a/backend/tests/services/chat/trigger/test_lifecycle_completed_result.py b/backend/tests/services/chat/trigger/test_lifecycle_completed_result.py
index d56bf81c6e..f11bea3070 100644
--- a/backend/tests/services/chat/trigger/test_lifecycle_completed_result.py
+++ b/backend/tests/services/chat/trigger/test_lifecycle_completed_result.py
@@ -93,3 +93,44 @@ async def _empty_existing_result(_subtask_id: int) -> dict:
},
}
]
+
+
+@pytest.mark.asyncio
+async def test_collect_completed_result_preserves_file_changes_with_blocks(monkeypatch):
+ async def _empty_existing_result(_subtask_id: int) -> dict:
+ return {}
+
+ monkeypatch.setattr(
+ lifecycle,
+ "_get_existing_subtask_result",
+ _empty_existing_result,
+ )
+
+ import app.services.chat.storage as chat_storage
+
+ monkeypatch.setattr(chat_storage, "session_manager", _SessionManager())
+
+ result = await lifecycle.collect_completed_result(
+ 1234,
+ status="COMPLETED",
+ result={
+ "value": "done",
+ "file_changes": {
+ "version": 1,
+ "status": "active",
+ "artifact_id": "turn-file-changes/7/1234",
+ "device_id": "device-1",
+ "workspace_path": "/workspace/project",
+ "file_count": 1,
+ "additions": 4,
+ "deletions": 2,
+ "files": [],
+ "reverted_at": None,
+ },
+ },
+ )
+
+ assert result is not None
+ assert result["value"] == "done"
+ assert result["blocks"]
+ assert result["file_changes"]["file_count"] == 1
diff --git a/backend/tests/services/execution/test_extract_completed_result.py b/backend/tests/services/execution/test_extract_completed_result.py
index a5fbce78ab..33720e24ad 100644
--- a/backend/tests/services/execution/test_extract_completed_result.py
+++ b/backend/tests/services/execution/test_extract_completed_result.py
@@ -66,6 +66,26 @@ def test_preserves_standalone_chat_workspace_path(self):
result["standalone_chat_workspace_path"] == "/tmp/chats/2026-05-29/new-chat"
)
+ def test_preserves_file_changes(self):
+ file_changes = {
+ "version": 1,
+ "status": "active",
+ "file_count": 1,
+ "additions": 1,
+ "deletions": 0,
+ "files": [
+ {
+ "path": "data.txt",
+ "change_type": "created",
+ "additions": 1,
+ "deletions": 0,
+ "binary": False,
+ }
+ ],
+ }
+ result = extract_completed_result({"file_changes": file_changes})
+ assert result["file_changes"] == file_changes
+
def test_preserves_silent_exit_fields(self):
result = extract_completed_result(
{
diff --git a/backend/tests/services/test_local_device_command_service.py b/backend/tests/services/test_local_device_command_service.py
index bc2fbb1511..ef2ddcfc13 100644
--- a/backend/tests/services/test_local_device_command_service.py
+++ b/backend/tests/services/test_local_device_command_service.py
@@ -4,6 +4,8 @@
"""Tests for local device command RPC service."""
+import gzip
+import hashlib
import json
import os
import subprocess
@@ -13,6 +15,153 @@
import pytest
+def _run_git(repo, *args):
+ return subprocess.run(
+ ["git", "-C", str(repo), *args],
+ check=True,
+ capture_output=True,
+ ).stdout
+
+
+def _create_turn_file_changes_artifact(tmp_path):
+ repo = tmp_path / "repo"
+ repo.mkdir()
+ _run_git(repo, "init", "-q")
+ _run_git(repo, "config", "user.email", "tests@example.com")
+ _run_git(repo, "config", "user.name", "Tests")
+ changed_file = repo / "changed.txt"
+ changed_file.write_text("before\n", encoding="utf-8")
+ _run_git(repo, "add", "--all")
+ _run_git(repo, "commit", "-qm", "initial")
+ changed_file.write_text("after\n", encoding="utf-8")
+ patch = _run_git(repo, "diff", "--binary", "HEAD")
+ executor_home = tmp_path / "executor-home"
+ artifact_dir = executor_home / "artifacts" / "turn-file-changes" / "10" / "20"
+ artifact_dir.mkdir(parents=True)
+ (artifact_dir / "changes.patch.gz").write_bytes(gzip.compress(patch))
+ (artifact_dir / "metadata.json").write_text(
+ json.dumps(
+ {
+ "version": 1,
+ "task_id": 10,
+ "subtask_id": 20,
+ "workspace_path": str(repo.resolve()),
+ "checksum": hashlib.sha256(patch).hexdigest(),
+ }
+ ),
+ encoding="utf-8",
+ )
+ return repo, executor_home
+
+
+def test_turn_file_changes_commands_are_registered():
+ from app.services.device.command_registry import resolve_local_device_command
+
+ review = resolve_local_device_command("turn_file_changes_review", {})
+ revert = resolve_local_device_command("turn_file_changes_revert", {})
+
+ assert review is not None
+ assert revert is not None
+ assert review.post_processor == "json"
+ assert revert.post_processor == "json"
+
+
+@pytest.mark.parametrize(
+ "artifact_id",
+ [
+ "../../etc/passwd",
+ "turn-file-changes/1/2/../../../secret",
+ ],
+)
+def test_turn_file_changes_command_rejects_invalid_artifact_id(
+ tmp_path,
+ artifact_id,
+):
+ from app.services.device.command_registry import TURN_FILE_CHANGES_SCRIPT
+
+ result = subprocess.run(
+ ["python3", "-c", TURN_FILE_CHANGES_SCRIPT, "review", artifact_id],
+ cwd=tmp_path,
+ env={**os.environ, "WEGENT_EXECUTOR_HOME": str(tmp_path / "home")},
+ capture_output=True,
+ text=True,
+ )
+
+ assert result.returncode != 0
+ assert "invalid artifact id" in result.stdout
+
+
+def test_turn_file_changes_review_returns_validated_diff(tmp_path):
+ from app.services.device.command_registry import TURN_FILE_CHANGES_SCRIPT
+
+ repo, executor_home = _create_turn_file_changes_artifact(tmp_path)
+ result = subprocess.run(
+ [
+ "python3",
+ "-c",
+ TURN_FILE_CHANGES_SCRIPT,
+ "review",
+ "turn-file-changes/10/20",
+ ],
+ cwd=repo,
+ env={**os.environ, "WEGENT_EXECUTOR_HOME": str(executor_home)},
+ check=True,
+ capture_output=True,
+ text=True,
+ )
+
+ payload = json.loads(result.stdout)
+ assert payload["success"] is True
+ assert payload["diff"].startswith("diff --git a/changed.txt b/changed.txt")
+
+
+def test_turn_file_changes_revert_is_conflict_safe(tmp_path):
+ from app.services.device.command_registry import TURN_FILE_CHANGES_SCRIPT
+
+ repo, executor_home = _create_turn_file_changes_artifact(tmp_path)
+ (repo / "changed.txt").write_text("later change\n", encoding="utf-8")
+ result = subprocess.run(
+ [
+ "python3",
+ "-c",
+ TURN_FILE_CHANGES_SCRIPT,
+ "revert",
+ "turn-file-changes/10/20",
+ ],
+ cwd=repo,
+ env={**os.environ, "WEGENT_EXECUTOR_HOME": str(executor_home)},
+ check=True,
+ capture_output=True,
+ text=True,
+ )
+
+ assert json.loads(result.stdout)["status"] == "conflicted"
+ assert (repo / "changed.txt").read_text(encoding="utf-8") == "later change\n"
+
+
+def test_turn_file_changes_revert_applies_reverse_patch(tmp_path):
+ from app.services.device.command_registry import TURN_FILE_CHANGES_SCRIPT
+
+ repo, executor_home = _create_turn_file_changes_artifact(tmp_path)
+ result = subprocess.run(
+ [
+ "python3",
+ "-c",
+ TURN_FILE_CHANGES_SCRIPT,
+ "revert",
+ "turn-file-changes/10/20",
+ ],
+ cwd=repo,
+ env={**os.environ, "WEGENT_EXECUTOR_HOME": str(executor_home)},
+ check=True,
+ capture_output=True,
+ text=True,
+ )
+
+ assert json.loads(result.stdout) == {"success": True, "status": "reverted"}
+ assert (repo / "changed.txt").read_text(encoding="utf-8") == "before\n"
+
+
def test_local_device_command_registry_default_includes_diagnostic_commands():
"""Default command registry should include basic diagnostic commands."""
from app.core.config import Settings
diff --git a/backend/tests/services/test_turn_file_changes_service.py b/backend/tests/services/test_turn_file_changes_service.py
new file mode 100644
index 0000000000..21ac9034cd
--- /dev/null
+++ b/backend/tests/services/test_turn_file_changes_service.py
@@ -0,0 +1,263 @@
+# SPDX-FileCopyrightText: 2025 Weibo, Inc.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+from unittest.mock import AsyncMock
+
+import pytest
+from fastapi import HTTPException
+
+from app.models.subtask import Subtask, SubtaskRole, SubtaskStatus
+from app.models.task import TaskResource
+from app.services.device.command_service import DeviceCommandError
+from app.services.turn_file_changes import turn_file_changes_service
+
+
+def _summary(*, device_id="device-1", status="active"):
+ return {
+ "version": 1,
+ "status": status,
+ "artifact_id": "turn-file-changes/10/20",
+ "device_id": device_id,
+ "workspace_path": "/workspace/project",
+ "file_count": 1,
+ "additions": 3,
+ "deletions": 1,
+ "files": [
+ {
+ "old_path": None,
+ "path": "src/app.ts",
+ "change_type": "modified",
+ "additions": 3,
+ "deletions": 1,
+ "binary": False,
+ }
+ ],
+ "reverted_at": None,
+ }
+
+
+def _create_records(test_db, *, device_id="device-1", summary_device_id="device-1"):
+ task = TaskResource(
+ user_id=7,
+ kind="Task",
+ name="task-10",
+ namespace="default",
+ json={
+ "kind": "Task",
+ "spec": {"title": "Task", "device_id": device_id},
+ },
+ is_active=TaskResource.STATE_ACTIVE,
+ )
+ test_db.add(task)
+ test_db.flush()
+ subtask = Subtask(
+ user_id=7,
+ task_id=task.id,
+ team_id=1,
+ title="Assistant",
+ bot_ids=[],
+ role=SubtaskRole.ASSISTANT,
+ status=SubtaskStatus.COMPLETED,
+ result={
+ "value": "done",
+ "blocks": [{"id": "tool-1", "type": "tool"}],
+ "file_changes": _summary(device_id=summary_device_id),
+ },
+ )
+ test_db.add(subtask)
+ test_db.commit()
+ test_db.refresh(subtask)
+ return task, subtask
+
+
+def _command_result(payload):
+ return {
+ "success": True,
+ "exit_code": 0,
+ "stdout": payload,
+ "stderr": "",
+ "duration": 0.01,
+ }
+
+
+@pytest.mark.asyncio
+async def test_get_diff_dispatches_recorded_device_and_workspace(
+ test_db,
+ monkeypatch,
+):
+ _, subtask = _create_records(test_db)
+ execute = AsyncMock(
+ return_value=_command_result({"success": True, "diff": "diff --git\n"})
+ )
+ monkeypatch.setattr(
+ "app.services.turn_file_changes.execute_configured_device_command",
+ execute,
+ )
+
+ response = await turn_file_changes_service.get_diff(
+ db=test_db,
+ user_id=7,
+ subtask_id=subtask.id,
+ )
+
+ assert response.diff == "diff --git\n"
+ execute.assert_awaited_once_with(
+ db=test_db,
+ user_id=7,
+ device_id="device-1",
+ command_key="turn_file_changes_review",
+ path="/workspace/project",
+ args=["turn-file-changes/10/20"],
+ timeout_seconds=30,
+ max_output_bytes=5 * 1024 * 1024,
+ )
+
+
+@pytest.mark.asyncio
+async def test_get_diff_rejects_device_mismatch(test_db, monkeypatch):
+ _, subtask = _create_records(
+ test_db,
+ device_id="device-2",
+ summary_device_id="device-1",
+ )
+ execute = AsyncMock()
+ monkeypatch.setattr(
+ "app.services.turn_file_changes.execute_configured_device_command",
+ execute,
+ )
+
+ with pytest.raises(HTTPException) as exc_info:
+ await turn_file_changes_service.get_diff(
+ db=test_db,
+ user_id=7,
+ subtask_id=subtask.id,
+ )
+
+ assert exc_info.value.status_code == 409
+ assert exc_info.value.detail["code"] == "TURN_FILE_CHANGES_DEVICE_MISMATCH"
+ execute.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_get_diff_reports_offline_device(test_db, monkeypatch):
+ _, subtask = _create_records(test_db)
+ monkeypatch.setattr(
+ "app.services.turn_file_changes.execute_configured_device_command",
+ AsyncMock(side_effect=DeviceCommandError("Device 'device-1' is offline")),
+ )
+
+ with pytest.raises(HTTPException) as exc_info:
+ await turn_file_changes_service.get_diff(
+ db=test_db,
+ user_id=7,
+ subtask_id=subtask.id,
+ )
+
+ assert exc_info.value.status_code == 409
+ assert exc_info.value.detail["code"] == "TURN_FILE_CHANGES_DEVICE_OFFLINE"
+
+
+@pytest.mark.asyncio
+async def test_revert_updates_only_file_changes_status(test_db, monkeypatch):
+ _, subtask = _create_records(test_db)
+ monkeypatch.setattr(
+ "app.services.turn_file_changes.execute_configured_device_command",
+ AsyncMock(
+ return_value=_command_result({"success": True, "status": "reverted"})
+ ),
+ )
+
+ response = await turn_file_changes_service.revert(
+ db=test_db,
+ user_id=7,
+ subtask_id=subtask.id,
+ )
+ test_db.refresh(subtask)
+
+ assert response.file_changes.status == "reverted"
+ assert response.file_changes.reverted_at is not None
+ assert subtask.result["value"] == "done"
+ assert subtask.result["blocks"] == [{"id": "tool-1", "type": "tool"}]
+
+
+@pytest.mark.asyncio
+async def test_revert_conflict_keeps_existing_message_result(test_db, monkeypatch):
+ _, subtask = _create_records(test_db)
+ monkeypatch.setattr(
+ "app.services.turn_file_changes.execute_configured_device_command",
+ AsyncMock(
+ return_value=_command_result(
+ {
+ "success": False,
+ "status": "conflicted",
+ "error": "patch does not apply",
+ }
+ )
+ ),
+ )
+
+ with pytest.raises(HTTPException) as exc_info:
+ await turn_file_changes_service.revert(
+ db=test_db,
+ user_id=7,
+ subtask_id=subtask.id,
+ )
+ test_db.refresh(subtask)
+
+ assert exc_info.value.status_code == 409
+ assert exc_info.value.detail["code"] == "TURN_FILE_CHANGES_CONFLICT"
+ assert subtask.result["value"] == "done"
+ assert subtask.result["blocks"] == [{"id": "tool-1", "type": "tool"}]
+ assert subtask.result["file_changes"]["status"] == "conflicted"
+
+
+@pytest.mark.asyncio
+async def test_revert_is_idempotent_after_success(test_db, monkeypatch):
+ _, subtask = _create_records(test_db)
+ result = dict(subtask.result)
+ result["file_changes"] = _summary(status="reverted")
+ subtask.result = result
+ test_db.commit()
+ execute = AsyncMock()
+ monkeypatch.setattr(
+ "app.services.turn_file_changes.execute_configured_device_command",
+ execute,
+ )
+
+ response = await turn_file_changes_service.revert(
+ db=test_db,
+ user_id=7,
+ subtask_id=subtask.id,
+ )
+
+ assert response.file_changes.status == "reverted"
+ execute.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_missing_artifact_marks_artifact_missing(test_db, monkeypatch):
+ _, subtask = _create_records(test_db)
+ monkeypatch.setattr(
+ "app.services.turn_file_changes.execute_configured_device_command",
+ AsyncMock(
+ return_value=_command_result(
+ {
+ "success": False,
+ "status": "artifact_missing",
+ "error": "turn file changes artifact is missing",
+ }
+ )
+ ),
+ )
+
+ with pytest.raises(HTTPException) as exc_info:
+ await turn_file_changes_service.get_diff(
+ db=test_db,
+ user_id=7,
+ subtask_id=subtask.id,
+ )
+ test_db.refresh(subtask)
+
+ assert exc_info.value.status_code == 410
+ assert subtask.result["file_changes"]["status"] == "artifact_missing"
diff --git a/docs/en/user-guide/coding/turn-file-changes.md b/docs/en/user-guide/coding/turn-file-changes.md
new file mode 100644
index 0000000000..14f54c2a48
--- /dev/null
+++ b/docs/en/user-guide/coding/turn-file-changes.md
@@ -0,0 +1,34 @@
+---
+sidebar_position: 5
+---
+
+# Review and Revert Turn File Changes
+
+After a supported coding task completes, Wework displays a file changes card below the corresponding assistant message. The card covers only the changes produced during that user-and-assistant turn, not the accumulated changes for the conversation or workspace.
+
+## Review the Summary
+
+The card shows the number of changed files, added and deleted lines, and the file list. It displays the first three files by default and can expand to show the rest.
+
+Binary files do not have reliable text line counts, so the card displays only the file name and binary status.
+
+## Review the Full Diff
+
+Selecting **Review** loads the compressed patch from the original execution device and displays the unified diff grouped by file.
+
+Review is not available offline. The original device must be online and the device-side artifact must still exist. The summary remains visible while the device is offline, but Review and Revert are disabled.
+
+## Revert One Turn
+
+After selecting **Revert** and confirming, the device first checks whether the reverse patch can apply safely:
+
+- When the check succeeds, only the changes from that turn are reverted.
+- If later work changed the same content, the card reports a conflict and the workspace is not modified.
+- Revert never force-overwrites later changes.
+- A successful revert is idempotent, so repeated requests do not modify files again.
+
+## Supported Workspaces
+
+Turn file changes currently support Codex and Claude Code in Git project workspaces. Non-Git directories do not produce a file changes card.
+
+The complete patch is stored as a compressed artifact in the task directory on the execution device. Backend stores only the file count, line statistics, file list, status, device, and artifact identifier in the assistant message's `Subtask.result`.
diff --git a/docs/zh/user-guide/coding/turn-file-changes.md b/docs/zh/user-guide/coding/turn-file-changes.md
new file mode 100644
index 0000000000..3b67e9e6a1
--- /dev/null
+++ b/docs/zh/user-guide/coding/turn-file-changes.md
@@ -0,0 +1,34 @@
+---
+sidebar_position: 5
+---
+
+# 查看和撤销单轮文件变更
+
+Wework 会在支持的代码任务完成后,在对应的助手消息下方显示本轮文件变更卡片。卡片只统计这一问一答期间由智能体产生的变更,不代表整个会话或工作区的累计变更。
+
+## 查看变更摘要
+
+卡片显示本轮修改的文件数量、增加和删除行数,以及文件列表。默认显示前三个文件,可以展开查看其余文件。
+
+二进制文件没有可靠的文本行数,因此只显示文件名称和“二进制文件”状态。
+
+## 审核完整变更
+
+点击“审核”后,Wework 会从执行本轮任务的原设备读取压缩 patch,并按文件展示 unified diff。
+
+审核不是离线功能。原设备必须在线,并且设备侧 artifact 仍然存在。设备离线时,摘要仍可查看,但审核和撤销操作不可用。
+
+## 撤销本轮变更
+
+点击“撤销”并确认后,设备会先检查反向 patch 是否能够安全应用:
+
+- 检查通过时,只撤销这一轮产生的文件变更。
+- 如果后续操作修改了相同内容,状态会变为“存在后续冲突”,工作区不会被修改。
+- 撤销不会强制覆盖后续变更。
+- 已成功撤销的操作具有幂等性,重复请求不会再次修改文件。
+
+## 支持范围
+
+单轮文件变更目前适用于 Codex 和 Claude Code 的 Git 项目工作区。非 Git 目录不会生成变更卡片。
+
+完整 patch 以压缩 artifact 保存在执行设备的任务目录中;Backend 只在助手消息的 `Subtask.result` 中保存文件数量、增删行、文件列表、状态、设备和 artifact 标识。
diff --git a/executor/agents/base.py b/executor/agents/base.py
index 9fc2c77009..4f63d01a7d 100644
--- a/executor/agents/base.py
+++ b/executor/agents/base.py
@@ -10,9 +10,11 @@
import os
import threading
from collections import OrderedDict
+from pathlib import Path
from typing import Any, Dict, Optional, Tuple
from executor.config import config
+from executor.services.turn_file_changes import TurnFileChangeTracker
from shared.logger import setup_logger
from shared.models import EmitterBuilder, ResponsesAPIEmitter, TransportFactory
from shared.models.execution import ExecutionRequest
@@ -80,6 +82,7 @@ def __init__(
# Emitter is required and must be provided by caller
self.emitter: ResponsesAPIEmitter = emitter
+ self.turn_file_change_tracker: Optional[TurnFileChangeTracker] = None
self._progress_task_lock = threading.Lock()
self._inflight_progress_task: Optional[asyncio.Task] = None
@@ -106,6 +109,7 @@ def update_emitter(self, new_subtask_id: int) -> None:
old_subtask_id = self.subtask_id
self.subtask_id = new_subtask_id
self.task_data.subtask_id = new_subtask_id
+ self.turn_file_change_tracker = None
self.emitter = (
EmitterBuilder()
.with_task(self.task_id, new_subtask_id)
@@ -125,6 +129,61 @@ def update_emitter(self, new_subtask_id: int) -> None:
f"{old_subtask_id} -> {new_subtask_id}"
)
+ async def start_turn_file_change_tracking(self) -> None:
+ """Install per-turn Git tracking when the workspace supports it."""
+ device_id = getattr(self.task_data, "device_id", None)
+ if not self.project_path:
+ logger.info(
+ "Turn file change tracking skipped for task %s: project_path is empty",
+ self.task_id,
+ )
+ return
+ if not device_id:
+ logger.info(
+ "Turn file change tracking skipped for task %s: device_id is empty",
+ self.task_id,
+ )
+ return
+ tracker = TurnFileChangeTracker(
+ workspace=Path(self.project_path),
+ task_id=self.task_id,
+ subtask_id=self.subtask_id,
+ executor_home=Path(config.WEGENT_EXECUTOR_HOME),
+ device_id=device_id,
+ )
+ if await tracker.start():
+ self.turn_file_change_tracker = tracker
+ self.emitter.set_completion_fields_provider(tracker.finalize)
+ logger.info(
+ "Turn file change tracking started for task %s, subtask %s, workspace=%s",
+ self.task_id,
+ self.subtask_id,
+ self.project_path,
+ )
+ else:
+ logger.info(
+ "Turn file change tracking skipped for task %s: workspace is not a Git repository with HEAD, workspace=%s",
+ self.task_id,
+ self.project_path,
+ )
+
+ async def abort_turn_file_change_tracking(self) -> None:
+ """Discard the current turn tracker and release its workspace lock."""
+ tracker = self.turn_file_change_tracker
+ self.turn_file_change_tracker = None
+ self.emitter.set_completion_fields_provider(None)
+ if tracker is not None:
+ await tracker.abort()
+
+ async def collect_turn_file_change_completion_fields(self) -> dict[str, Any]:
+ """Finalize the current turn tracker and return response.completed fields."""
+ tracker = self.turn_file_change_tracker
+ self.turn_file_change_tracker = None
+ self.emitter.set_completion_fields_provider(None)
+ if tracker is None:
+ return {}
+ return await tracker.finalize()
+
async def handle(
self, pre_executed: Optional[TaskStatus] = None
) -> Tuple[TaskStatus, Optional[str]]:
diff --git a/executor/agents/claude_code/claude_code_agent.py b/executor/agents/claude_code/claude_code_agent.py
index da52aa501f..a1dd515a2f 100644
--- a/executor/agents/claude_code/claude_code_agent.py
+++ b/executor/agents/claude_code/claude_code_agent.py
@@ -863,6 +863,7 @@ async def _async_execute(self) -> TaskStatus:
interactive_form_answer
)
+ await self.start_turn_file_change_tracking()
if interactive_form_payload:
logger.info(
"Sending interactive form answer as tool_result for tool_use_id=%s",
@@ -910,6 +911,9 @@ async def _async_execute(self) -> TaskStatus:
session_id=self.session_id,
mcp_servers=self.options.get("mcp_servers")
or self.options.get("mcpServers"),
+ completion_fields_provider=(
+ self.collect_turn_file_change_completion_fields
+ ),
)
# Task completed or failed
@@ -919,6 +923,9 @@ async def _async_execute(self) -> TaskStatus:
logger.warning("No final result received from process_response")
result = TaskStatus.RUNNING
+ if result != TaskStatus.COMPLETED:
+ await self.abort_turn_file_change_tracking()
+
# Update task state based on result
if result == TaskStatus.COMPLETED:
self.task_state_manager.set_state(self.task_id, TaskState.COMPLETED)
@@ -935,6 +942,7 @@ async def _async_execute(self) -> TaskStatus:
return result
except Exception as e:
+ await self.abort_turn_file_change_tracking()
return self._handle_execution_error(e, "async execution")
async def _drain_answered_interactive_form_resume_result(
diff --git a/executor/agents/claude_code/response_processor.py b/executor/agents/claude_code/response_processor.py
index d5c8e57969..f679e94a71 100644
--- a/executor/agents/claude_code/response_processor.py
+++ b/executor/agents/claude_code/response_processor.py
@@ -3,7 +3,7 @@
import os
from dataclasses import asdict
from datetime import datetime
-from typing import Any, Dict, List, Optional, Tuple, Union
+from typing import Any, Awaitable, Callable, Dict, List, Optional, Tuple, Union
from claude_agent_sdk import ClaudeSDKClient
from claude_agent_sdk.types import (
@@ -46,6 +46,10 @@
logger = setup_logger("claude_response_processor")
+CompletionFieldsProvider = Callable[
+ [], Union[Dict[str, Any], Awaitable[Dict[str, Any]]]
+]
+
# Maximum retry count for API errors per session
MAX_API_ERROR_RETRIES = 3
@@ -77,6 +81,32 @@ def _standalone_workspace_result_fields(state_manager, content: str) -> dict[str
return {"standalone_chat_workspace_path": workspace_path}
+async def _collect_completion_fields(
+ provider: Optional[CompletionFieldsProvider],
+) -> Dict[str, Any]:
+ """Collect optional fields that must be attached to response.completed."""
+ if provider is None:
+ return {}
+
+ try:
+ fields = provider()
+ if hasattr(fields, "__await__"):
+ fields = await fields
+ if isinstance(fields, dict):
+ logger.info(
+ "Collected Claude response completion fields: keys=%s",
+ sorted(fields.keys()),
+ )
+ return fields
+ logger.info(
+ "Claude response completion fields provider returned non-dict: %s",
+ type(fields).__name__,
+ )
+ except Exception:
+ logger.exception("Failed to collect Claude response completion fields")
+ return {}
+
+
def _tool_result_completion(block: ToolResultBlock) -> tuple[str, str | None]:
"""Translate Claude tool result errors into emitter terminal status."""
if not block.is_error:
@@ -95,6 +125,7 @@ async def process_response(
task_state_manager=None,
session_id: str = None,
mcp_servers: Any = None,
+ completion_fields_provider: Optional[CompletionFieldsProvider] = None,
) -> Union[TaskStatus, str]:
"""
Process the response messages from Claude
@@ -106,6 +137,7 @@ async def process_response(
thinking_manager: Optional ThinkingStepManager instance for adding thinking steps
task_state_manager: Optional TaskStateManager instance for checking cancellation
session_id: Optional session ID for retry operations
+ completion_fields_provider: Optional provider for response.completed fields
Returns:
TaskStatus: Processing status
@@ -251,6 +283,7 @@ async def process_response(
mcp_servers=mcp_servers,
deferred_mcp_retry_count=deferred_mcp_retry_count,
max_deferred_mcp_retries=MAX_DEFERRED_MCP_RETRIES,
+ completion_fields_provider=completion_fields_provider,
)
if result_status == "RETRY":
@@ -874,6 +907,7 @@ async def _process_result_message(
mcp_servers: Any = None,
deferred_mcp_retry_count: int = 0,
max_deferred_mcp_retries: int = MAX_DEFERRED_MCP_RETRIES,
+ completion_fields_provider: Optional[CompletionFieldsProvider] = None,
) -> Union[TaskStatus, str, None]:
"""
Process a ResultMessage from Claude
@@ -890,6 +924,7 @@ async def _process_result_message(
propagated_silent_exit: Silent exit flag propagated from UserMessage tool results
propagated_silent_exit_reason: Silent exit reason propagated from UserMessage tool results
task_state_manager: Optional TaskStateManager for checking cancellation state
+ completion_fields_provider: Optional provider for response.completed fields
Returns:
TaskStatus: Processing status (COMPLETED if successful, CANCELLED, FAILED, or None)
@@ -1081,12 +1116,14 @@ async def _process_result_message(
if state_manager:
state_manager.set_task_status(TaskStatus.COMPLETED.value)
+ extra_fields = await _collect_completion_fields(completion_fields_provider)
await emitter.done(
content="",
usage=msg.usage,
stop_reason=stop_reason or "tool_deferred",
silent_exit=True,
silent_exit_reason=WAITING_FOR_USER_INPUT_REASON,
+ **extra_fields,
)
return TaskStatus.COMPLETED
@@ -1148,6 +1185,9 @@ async def _process_result_message(
extra_fields = _standalone_workspace_result_fields(
state_manager, content
)
+ extra_fields.update(
+ await _collect_completion_fields(completion_fields_provider)
+ )
await emitter.done(
content=content,
usage=msg.usage,
@@ -1180,6 +1220,9 @@ async def _process_result_message(
extra_fields = _standalone_workspace_result_fields(
state_manager, result_str
)
+ extra_fields.update(
+ await _collect_completion_fields(completion_fields_provider)
+ )
await emitter.done(
content=result_str,
usage=msg.usage,
diff --git a/executor/agents/codex/codex_agent.py b/executor/agents/codex/codex_agent.py
index 727ac00116..2c2b285bcf 100644
--- a/executor/agents/codex/codex_agent.py
+++ b/executor/agents/codex/codex_agent.py
@@ -143,6 +143,7 @@ async def execute_async(self) -> TaskStatus:
input_text = self._build_turn_input()
effort, summary = self._build_reasoning_params()
+ await self.start_turn_file_change_tracking()
self._turn = await self._thread.turn(
input_text,
cwd=self.project_path,
@@ -155,10 +156,14 @@ async def execute_async(self) -> TaskStatus:
async for event in self._turn.stream():
status = await mapper.handle(event)
if status is not None:
+ if status != TaskStatus.COMPLETED:
+ await self.abort_turn_file_change_tracking()
return status
+ await self.abort_turn_file_change_tracking()
return TaskStatus.FAILED
except Exception as exc:
logger.exception("CodeXAgent execution failed: %s", exc)
+ await self.abort_turn_file_change_tracking()
await self.emitter.error(str(exc), "execution_error")
return TaskStatus.FAILED
finally:
diff --git a/executor/agents/codex/event_mapper.py b/executor/agents/codex/event_mapper.py
index 066228263a..b4feabf4d0 100644
--- a/executor/agents/codex/event_mapper.py
+++ b/executor/agents/codex/event_mapper.py
@@ -33,6 +33,7 @@ def __init__(self, emitter: ResponsesAPIEmitter):
self._pending_agent_delta_text = ""
self._pending_agent_deltas: dict[str, str] = {}
self._tool_contexts: dict[str, dict[str, Any]] = {}
+ self.native_turn_diff = ""
async def handle(self, event: Any) -> Optional[TaskStatus]:
method = getattr(event, "method", "")
@@ -73,6 +74,14 @@ async def handle(self, event: Any) -> Optional[TaskStatus]:
self._handle_usage(payload)
return None
+ if method == "turn/diff/updated":
+ self.native_turn_diff = str(getattr(payload, "diff", "") or "")
+ logger.debug(
+ "Received native Codex turn diff: bytes=%s",
+ len(self.native_turn_diff.encode("utf-8")),
+ )
+ return None
+
if method == "turn/completed":
return await self._handle_turn_completed(payload)
@@ -362,8 +371,10 @@ def _extract_message_text(cls, item: Any) -> str:
@classmethod
def _call_id(cls, item: Any) -> str:
- value = cls._get(item, "call_id") or cls._get(item, "callId") or cls._get(
- item, "id"
+ value = (
+ cls._get(item, "call_id")
+ or cls._get(item, "callId")
+ or cls._get(item, "id")
)
return str(value or "")
diff --git a/executor/services/turn_file_changes.py b/executor/services/turn_file_changes.py
new file mode 100644
index 0000000000..2d08db2d4b
--- /dev/null
+++ b/executor/services/turn_file_changes.py
@@ -0,0 +1,420 @@
+# SPDX-FileCopyrightText: 2025 Weibo, Inc.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Capture Git file changes made during one agent turn."""
+
+from __future__ import annotations
+
+import gzip
+import hashlib
+import json
+import logging
+import os
+import socket
+import subprocess
+import tempfile
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Optional
+
+ARTIFACT_VERSION = 1
+LOCK_FILE_NAME = "wegent-turn-file-changes.lock"
+logger = logging.getLogger(__name__)
+
+
+class WorkspaceBusyError(RuntimeError):
+ """Raised when another tracked turn owns the workspace lock."""
+
+
+@dataclass(frozen=True)
+class GitTreeSnapshot:
+ """A Git tree representing the complete workspace state."""
+
+ tree_id: str
+
+
+@dataclass(frozen=True)
+class TurnFileArtifact:
+ """Paths and checksum for a persisted turn patch."""
+
+ artifact_id: str
+ patch_path: Path
+ metadata_path: Path
+ checksum: str
+
+
+@dataclass(frozen=True)
+class _ChangedPath:
+ path: str
+ change_type: str
+ old_path: Optional[str] = None
+
+
+def _run_git(
+ workspace: Path,
+ *args: str,
+ env: Optional[dict[str, str]] = None,
+ text: bool = True,
+) -> subprocess.CompletedProcess:
+ return subprocess.run(
+ ["git", "-C", str(workspace), *args],
+ check=True,
+ capture_output=True,
+ env={**os.environ, **(env or {})},
+ text=text,
+ )
+
+
+def _is_git_workspace(workspace: Path) -> bool:
+ result = subprocess.run(
+ ["git", "-C", str(workspace), "rev-parse", "--is-inside-work-tree"],
+ check=False,
+ capture_output=True,
+ text=True,
+ )
+ return result.returncode == 0 and result.stdout.strip() == "true"
+
+
+def _has_head(workspace: Path) -> bool:
+ result = subprocess.run(
+ ["git", "-C", str(workspace), "rev-parse", "--verify", "HEAD"],
+ check=False,
+ capture_output=True,
+ text=True,
+ )
+ return result.returncode == 0
+
+
+def _decode_path(value: bytes) -> str:
+ return value.decode("utf-8", errors="surrogateescape")
+
+
+class TurnFileChangeTracker:
+ """Create a per-turn Git diff without mutating the repository index."""
+
+ def __init__(
+ self,
+ workspace: Path,
+ task_id: int,
+ subtask_id: int,
+ executor_home: Path,
+ device_id: Optional[str] = None,
+ ) -> None:
+ self.workspace = workspace.resolve()
+ self.task_id = task_id
+ self.subtask_id = subtask_id
+ self.executor_home = executor_home.expanduser().resolve()
+ self.device_id = device_id
+ self._before: Optional[GitTreeSnapshot] = None
+ self._lock_path: Optional[Path] = None
+ self._active = False
+
+ async def start(self) -> bool:
+ """Capture the workspace state immediately before an agent turn."""
+ if not _is_git_workspace(self.workspace):
+ return False
+ self._acquire_lock()
+ try:
+ self._before = self._capture_tree()
+ self._active = True
+ logger.info(
+ "Captured turn file change baseline: task_id=%s subtask_id=%s workspace=%s tree=%s",
+ self.task_id,
+ self.subtask_id,
+ self.workspace,
+ self._before.tree_id,
+ )
+ return True
+ except Exception:
+ self._release_lock()
+ raise
+
+ async def finalize(self) -> dict[str, Any]:
+ """Persist and summarize changes made since ``start``."""
+ if not self._active or self._before is None:
+ return {}
+ try:
+ after = self._capture_tree()
+ patch = self._create_patch(self._before, after)
+ if not patch:
+ logger.info(
+ "No turn file changes detected: task_id=%s subtask_id=%s workspace=%s before=%s after=%s",
+ self.task_id,
+ self.subtask_id,
+ self.workspace,
+ self._before.tree_id,
+ after.tree_id,
+ )
+ return {}
+ files = self._create_file_summary(self._before, after)
+ artifact = self._persist_artifact(patch)
+ logger.info(
+ "Persisted turn file changes: task_id=%s subtask_id=%s workspace=%s artifact_id=%s files=%s additions=%s deletions=%s",
+ self.task_id,
+ self.subtask_id,
+ self.workspace,
+ artifact.artifact_id,
+ len(files),
+ sum(item["additions"] for item in files),
+ sum(item["deletions"] for item in files),
+ )
+ return {
+ "file_changes": {
+ "version": ARTIFACT_VERSION,
+ "status": "active",
+ "artifact_id": artifact.artifact_id,
+ "device_id": self.device_id,
+ "workspace_path": str(self.workspace),
+ "file_count": len(files),
+ "additions": sum(item["additions"] for item in files),
+ "deletions": sum(item["deletions"] for item in files),
+ "files": files,
+ "reverted_at": None,
+ }
+ }
+ finally:
+ self._active = False
+ self._release_lock()
+
+ async def abort(self) -> None:
+ """Stop tracking and release the workspace lock."""
+ self._active = False
+ self._before = None
+ self._release_lock()
+
+ def _capture_tree(self) -> GitTreeSnapshot:
+ with tempfile.TemporaryDirectory(prefix="wegent-turn-index-") as temp_dir:
+ index_path = Path(temp_dir) / "index"
+ env = {"GIT_INDEX_FILE": str(index_path)}
+ if _has_head(self.workspace):
+ _run_git(self.workspace, "read-tree", "HEAD", env=env)
+ _run_git(self.workspace, "add", "--all", "--", ".", env=env)
+ tree_id = _run_git(self.workspace, "write-tree", env=env).stdout.strip()
+ return GitTreeSnapshot(tree_id=tree_id)
+
+ def _create_patch(
+ self,
+ before: GitTreeSnapshot,
+ after: GitTreeSnapshot,
+ ) -> bytes:
+ return _run_git(
+ self.workspace,
+ "diff",
+ "--binary",
+ "--find-renames",
+ before.tree_id,
+ after.tree_id,
+ text=False,
+ ).stdout
+
+ def _create_file_summary(
+ self,
+ before: GitTreeSnapshot,
+ after: GitTreeSnapshot,
+ ) -> list[dict[str, Any]]:
+ changed_paths = self._changed_paths(before, after)
+ files = []
+ for changed in changed_paths:
+ additions, deletions, binary = self._line_stats(before, after, changed)
+ files.append(
+ {
+ "old_path": changed.old_path,
+ "path": changed.path,
+ "change_type": changed.change_type,
+ "additions": additions,
+ "deletions": deletions,
+ "binary": binary,
+ }
+ )
+ return sorted(files, key=lambda item: item["path"])
+
+ def _changed_paths(
+ self,
+ before: GitTreeSnapshot,
+ after: GitTreeSnapshot,
+ ) -> list[_ChangedPath]:
+ output = _run_git(
+ self.workspace,
+ "diff",
+ "--name-status",
+ "-z",
+ "--find-renames",
+ before.tree_id,
+ after.tree_id,
+ text=False,
+ ).stdout
+ tokens = output.split(b"\0")
+ changes: list[_ChangedPath] = []
+ index = 0
+ while index < len(tokens) and tokens[index]:
+ status = _decode_path(tokens[index])
+ index += 1
+ old_or_current = _decode_path(tokens[index])
+ index += 1
+ code = status[0]
+ if code == "R":
+ new_path = _decode_path(tokens[index])
+ index += 1
+ changes.append(
+ _ChangedPath(
+ old_path=old_or_current,
+ path=new_path,
+ change_type="renamed",
+ )
+ )
+ continue
+ change_type = {
+ "A": "created",
+ "D": "deleted",
+ }.get(code, "modified")
+ changes.append(_ChangedPath(path=old_or_current, change_type=change_type))
+ return changes
+
+ def _line_stats(
+ self,
+ before: GitTreeSnapshot,
+ after: GitTreeSnapshot,
+ changed: _ChangedPath,
+ ) -> tuple[int, int, bool]:
+ paths = [changed.path]
+ if changed.old_path is not None:
+ paths.insert(0, changed.old_path)
+ output = _run_git(
+ self.workspace,
+ "diff",
+ "--numstat",
+ "--find-renames",
+ before.tree_id,
+ after.tree_id,
+ "--",
+ *paths,
+ text=False,
+ ).stdout
+ additions = 0
+ deletions = 0
+ binary = False
+ for line in output.splitlines():
+ parts = line.split(b"\t", 2)
+ if len(parts) < 2:
+ continue
+ if parts[0] == b"-" or parts[1] == b"-":
+ binary = True
+ continue
+ additions += int(parts[0])
+ deletions += int(parts[1])
+ return additions, deletions, binary
+
+ def _persist_artifact(self, patch: bytes) -> TurnFileArtifact:
+ artifact_id = f"turn-file-changes/{self.task_id}/{self.subtask_id}"
+ artifact_dir = self.executor_home / "artifacts" / artifact_id
+ artifact_dir.mkdir(parents=True, exist_ok=True)
+ patch_path = artifact_dir / "changes.patch.gz"
+ metadata_path = artifact_dir / "metadata.json"
+ checksum = hashlib.sha256(patch).hexdigest()
+ metadata = {
+ "version": ARTIFACT_VERSION,
+ "task_id": self.task_id,
+ "subtask_id": self.subtask_id,
+ "workspace_path": str(self.workspace),
+ "checksum": checksum,
+ }
+ self._atomic_write(patch_path, gzip.compress(patch))
+ self._atomic_write(
+ metadata_path,
+ json.dumps(metadata, ensure_ascii=True, sort_keys=True).encode("utf-8"),
+ )
+ return TurnFileArtifact(
+ artifact_id=artifact_id,
+ patch_path=patch_path,
+ metadata_path=metadata_path,
+ checksum=checksum,
+ )
+
+ @staticmethod
+ def _atomic_write(path: Path, content: bytes) -> None:
+ with tempfile.NamedTemporaryFile(
+ dir=path.parent,
+ prefix=f".{path.name}.",
+ delete=False,
+ ) as temp_file:
+ temp_path = Path(temp_file.name)
+ temp_file.write(content)
+ temp_file.flush()
+ os.fsync(temp_file.fileno())
+ temp_path.replace(path)
+
+ def _acquire_lock(self) -> None:
+ common_dir = _run_git(
+ self.workspace,
+ "rev-parse",
+ "--git-common-dir",
+ ).stdout.strip()
+ common_path = Path(common_dir)
+ if not common_path.is_absolute():
+ common_path = self.workspace / common_path
+ self._lock_path = common_path.resolve() / LOCK_FILE_NAME
+ lock_metadata = {
+ "pid": os.getpid(),
+ "hostname": socket.gethostname(),
+ "created_at": datetime.now(timezone.utc).isoformat(),
+ "task_id": self.task_id,
+ "subtask_id": self.subtask_id,
+ }
+ for attempt in range(2):
+ try:
+ lock_fd = os.open(
+ self._lock_path,
+ os.O_WRONLY | os.O_CREAT | os.O_EXCL,
+ 0o600,
+ )
+ with os.fdopen(lock_fd, "w", encoding="utf-8") as lock_file:
+ json.dump(lock_metadata, lock_file)
+ return
+ except FileExistsError:
+ if attempt == 0 and self._remove_stale_lock():
+ continue
+ raise WorkspaceBusyError(
+ f"Workspace is busy with another tracked turn: {self.workspace}"
+ )
+
+ def _remove_stale_lock(self) -> bool:
+ if self._lock_path is None:
+ return False
+ try:
+ metadata = json.loads(self._lock_path.read_text(encoding="utf-8"))
+ if metadata.get("hostname") != socket.gethostname():
+ return False
+ pid = int(metadata["pid"])
+ except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError):
+ return False
+ if self._pid_exists(pid):
+ return False
+ try:
+ self._lock_path.unlink()
+ return True
+ except FileNotFoundError:
+ return True
+
+ @staticmethod
+ def _pid_exists(pid: int) -> bool:
+ if pid <= 0:
+ return False
+ try:
+ os.kill(pid, 0)
+ except ProcessLookupError:
+ return False
+ except PermissionError:
+ return True
+ return True
+
+ def _release_lock(self) -> None:
+ if self._lock_path is None:
+ return
+ try:
+ self._lock_path.unlink()
+ except FileNotFoundError:
+ pass
+ finally:
+ self._lock_path = None
diff --git a/executor/tests/agents/test_claude_response_processor.py b/executor/tests/agents/test_claude_response_processor.py
index c77c0e2f8e..60f845c434 100644
--- a/executor/tests/agents/test_claude_response_processor.py
+++ b/executor/tests/agents/test_claude_response_processor.py
@@ -15,6 +15,7 @@
_process_result_message,
)
from executor.tasks.task_state_manager import TaskState, TaskStateManager
+from shared.models import EmitterBuilder, GeneratorTransport
from shared.status import TaskStatus
@@ -172,3 +173,91 @@ async def test_cancelled_task_takes_precedence_over_error_result_message():
assert result == TaskStatus.CANCELLED
assert state_manager.statuses == [TaskStatus.CANCELLED.value]
+
+
+@pytest.mark.asyncio
+async def test_claude_success_finalizes_turn_file_changes():
+ transport = GeneratorTransport()
+ emitter = EmitterBuilder().with_task(1, 2).with_transport(transport).build()
+ completion_fields = AsyncMock(
+ return_value={"file_changes": {"version": 1, "file_count": 1}}
+ )
+ emitter.set_completion_fields_provider(completion_fields)
+ message = ResultMessage(
+ subtype="success",
+ duration_ms=1,
+ duration_api_ms=1,
+ is_error=False,
+ num_turns=1,
+ session_id="session-1",
+ result="done",
+ )
+
+ result = await _process_result_message(
+ msg=message,
+ emitter=emitter,
+ state_manager=DummyStateManager(),
+ )
+
+ completed = transport.get_events()[-1][1]["response"]
+ assert result == TaskStatus.COMPLETED
+ completion_fields.assert_awaited_once()
+ assert completed["file_changes"]["file_count"] == 1
+
+
+@pytest.mark.asyncio
+async def test_claude_success_uses_explicit_completion_fields_provider():
+ transport = GeneratorTransport()
+ emitter = EmitterBuilder().with_task(1, 2).with_transport(transport).build()
+ completion_fields = AsyncMock(
+ return_value={"file_changes": {"version": 1, "file_count": 1}}
+ )
+ message = ResultMessage(
+ subtype="success",
+ duration_ms=1,
+ duration_api_ms=1,
+ is_error=False,
+ num_turns=1,
+ session_id="session-1",
+ result="done",
+ )
+
+ result = await _process_result_message(
+ msg=message,
+ emitter=emitter,
+ state_manager=DummyStateManager(),
+ completion_fields_provider=completion_fields,
+ )
+
+ completed = transport.get_events()[-1][1]["response"]
+ assert result == TaskStatus.COMPLETED
+ completion_fields.assert_awaited_once()
+ assert completed["file_changes"]["file_count"] == 1
+
+
+@pytest.mark.asyncio
+async def test_claude_failure_does_not_finalize_turn_file_changes():
+ transport = GeneratorTransport()
+ emitter = EmitterBuilder().with_task(1, 2).with_transport(transport).build()
+ completion_fields = AsyncMock(
+ return_value={"file_changes": {"version": 1, "file_count": 1}}
+ )
+ emitter.set_completion_fields_provider(completion_fields)
+ message = ResultMessage(
+ subtype="error_during_execution",
+ duration_ms=1,
+ duration_api_ms=1,
+ is_error=True,
+ num_turns=1,
+ session_id="session-1",
+ result="failed",
+ )
+
+ result = await _process_result_message(
+ msg=message,
+ emitter=emitter,
+ state_manager=DummyStateManager(),
+ )
+
+ assert result == TaskStatus.FAILED
+ completion_fields.assert_not_awaited()
diff --git a/executor/tests/agents/test_codex_event_mapper.py b/executor/tests/agents/test_codex_event_mapper.py
index 9d322f09bd..a014d43b15 100644
--- a/executor/tests/agents/test_codex_event_mapper.py
+++ b/executor/tests/agents/test_codex_event_mapper.py
@@ -5,11 +5,14 @@
# SPDX-License-Identifier: Apache-2.0
from types import SimpleNamespace
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, MagicMock, patch
import pytest
+from executor.agents.base import Agent
from executor.agents.codex.event_mapper import CodeXEventMapper
+from shared.models import EmitterBuilder, GeneratorTransport
+from shared.models.execution import ExecutionRequest
from shared.status import TaskStatus
@@ -28,6 +31,70 @@ def assert_commentary_block(
return block
+@pytest.mark.asyncio
+async def test_agent_installs_turn_file_change_completion_provider():
+ emitter = MagicMock()
+ tracker = MagicMock()
+ tracker.start = AsyncMock(return_value=True)
+ tracker.finalize = AsyncMock(return_value={"file_changes": {"file_count": 1}})
+ task_data = ExecutionRequest(task_id=1, subtask_id=2, device_id="device-1")
+ agent = Agent(task_data, emitter)
+ agent.project_path = "/tmp/project"
+
+ with patch(
+ "executor.agents.base.TurnFileChangeTracker",
+ return_value=tracker,
+ ) as tracker_class:
+ await agent.start_turn_file_change_tracking()
+
+ tracker_class.assert_called_once()
+ tracker.start.assert_awaited_once()
+ emitter.set_completion_fields_provider.assert_called_once_with(tracker.finalize)
+ assert agent.turn_file_change_tracker is tracker
+
+
+@pytest.mark.asyncio
+async def test_codex_completion_includes_turn_file_changes():
+ transport = GeneratorTransport()
+ emitter = EmitterBuilder().with_task(1, 2).with_transport(transport).build()
+ completion_fields = AsyncMock(
+ return_value={"file_changes": {"version": 1, "file_count": 1}}
+ )
+ emitter.set_completion_fields_provider(completion_fields)
+ mapper = CodeXEventMapper(emitter)
+
+ status = await mapper.handle(
+ SimpleNamespace(
+ method="turn/completed",
+ payload=SimpleNamespace(
+ turn=SimpleNamespace(status=SimpleNamespace(value="completed"))
+ ),
+ )
+ )
+
+ completed = transport.get_events()[-1][1]["response"]
+ assert status == TaskStatus.COMPLETED
+ completion_fields.assert_awaited_once()
+ assert completed["file_changes"]["file_count"] == 1
+
+
+@pytest.mark.asyncio
+async def test_codex_records_native_turn_diff_for_diagnostics_only():
+ emitter = MagicMock()
+ mapper = CodeXEventMapper(emitter)
+
+ result = await mapper.handle(
+ SimpleNamespace(
+ method="turn/diff/updated",
+ payload=SimpleNamespace(diff="diff --git a/a.txt b/a.txt\n"),
+ )
+ )
+
+ assert result is None
+ assert mapper.native_turn_diff == "diff --git a/a.txt b/a.txt\n"
+ emitter.done.assert_not_called()
+
+
@pytest.mark.asyncio
async def test_codex_event_mapper_streams_final_answer_delta_and_completion():
emitter = SimpleNamespace(
diff --git a/executor/tests/services/test_turn_file_changes.py b/executor/tests/services/test_turn_file_changes.py
new file mode 100644
index 0000000000..89072de57f
--- /dev/null
+++ b/executor/tests/services/test_turn_file_changes.py
@@ -0,0 +1,291 @@
+# SPDX-FileCopyrightText: 2025 Weibo, Inc.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+import gzip
+import hashlib
+import json
+import os
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from executor.services.turn_file_changes import (
+ TurnFileChangeTracker,
+ WorkspaceBusyError,
+)
+
+
+def run_git(repo: Path, *args: str, input_bytes: bytes | None = None) -> bytes:
+ return subprocess.run(
+ ["git", "-C", str(repo), *args],
+ check=True,
+ capture_output=True,
+ input=input_bytes,
+ ).stdout
+
+
+def init_repo(path: Path, *, with_head: bool = True) -> Path:
+ repo = path / "repo"
+ repo.mkdir()
+ run_git(repo, "init", "-q")
+ run_git(repo, "config", "user.email", "tests@example.com")
+ run_git(repo, "config", "user.name", "Tests")
+ if with_head:
+ write(repo / ".gitkeep", "")
+ commit_all(repo, "initial")
+ return repo
+
+
+def write(path: Path, content: str | bytes) -> None:
+ path.parent.mkdir(parents=True, exist_ok=True)
+ if isinstance(content, bytes):
+ path.write_bytes(content)
+ else:
+ path.write_text(content, encoding="utf-8")
+
+
+def commit_all(repo: Path, message: str) -> None:
+ run_git(repo, "add", "--all")
+ run_git(repo, "commit", "-qm", message)
+
+
+def file_changes(summary: dict) -> dict:
+ return summary["file_changes"]
+
+
+@pytest.mark.asyncio
+async def test_tracker_excludes_changes_that_preexist_the_turn(tmp_path):
+ repo = init_repo(tmp_path)
+ write(repo / "existing.txt", "base\n")
+ commit_all(repo, "tracked")
+ write(repo / "existing.txt", "user dirty\n")
+ write(repo / "untracked.txt", "before\n")
+
+ tracker = TurnFileChangeTracker(
+ workspace=repo,
+ task_id=10,
+ subtask_id=20,
+ executor_home=tmp_path / "executor-home",
+ )
+ assert await tracker.start() is True
+
+ write(repo / "agent.txt", "created by agent\n")
+ summary = await tracker.finalize()
+
+ assert [item["path"] for item in file_changes(summary)["files"]] == ["agent.txt"]
+
+
+@pytest.mark.asyncio
+async def test_tracker_counts_modified_created_deleted_and_renamed_files(tmp_path):
+ repo = init_repo(tmp_path)
+ write(repo / "modified.txt", "old\n")
+ write(repo / "deleted.txt", "gone\n")
+ write(repo / "old-name.txt", "rename me\n")
+ commit_all(repo, "fixtures")
+ tracker = TurnFileChangeTracker(
+ workspace=repo,
+ task_id=1,
+ subtask_id=2,
+ executor_home=tmp_path / "home",
+ device_id="device-1",
+ )
+ await tracker.start()
+
+ write(repo / "modified.txt", "new\nextra\n")
+ write(repo / "created.txt", "created\n")
+ (repo / "deleted.txt").unlink()
+ (repo / "old-name.txt").rename(repo / "new-name.txt")
+ changes = file_changes(await tracker.finalize())
+
+ by_path = {item["path"]: item for item in changes["files"]}
+ assert changes["file_count"] == 4
+ assert changes["additions"] == 3
+ assert changes["deletions"] == 2
+ assert by_path["created.txt"]["change_type"] == "created"
+ assert by_path["deleted.txt"]["change_type"] == "deleted"
+ assert by_path["modified.txt"]["change_type"] == "modified"
+ assert by_path["new-name.txt"]["change_type"] == "renamed"
+ assert by_path["new-name.txt"]["old_path"] == "old-name.txt"
+ assert changes["device_id"] == "device-1"
+
+
+@pytest.mark.asyncio
+async def test_tracker_supports_repository_without_head(tmp_path):
+ repo = init_repo(tmp_path, with_head=False)
+ tracker = TurnFileChangeTracker(
+ workspace=repo,
+ task_id=1,
+ subtask_id=2,
+ executor_home=tmp_path / "home",
+ )
+ await tracker.start()
+
+ write(repo / "first.txt", "first\n")
+ changes = file_changes(await tracker.finalize())
+
+ assert changes["files"][0]["path"] == "first.txt"
+ assert changes["files"][0]["change_type"] == "created"
+
+
+@pytest.mark.asyncio
+async def test_tracker_marks_binary_files_without_line_counts(tmp_path):
+ repo = init_repo(tmp_path)
+ tracker = TurnFileChangeTracker(
+ workspace=repo,
+ task_id=1,
+ subtask_id=2,
+ executor_home=tmp_path / "home",
+ )
+ await tracker.start()
+
+ write(repo / "image.bin", b"\x00\x01\x02\x03")
+ changes = file_changes(await tracker.finalize())
+
+ assert changes["files"][0] == {
+ "old_path": None,
+ "path": "image.bin",
+ "change_type": "created",
+ "additions": 0,
+ "deletions": 0,
+ "binary": True,
+ }
+
+
+@pytest.mark.asyncio
+async def test_tracker_writes_gzip_patch_and_metadata_checksum(tmp_path):
+ repo = init_repo(tmp_path)
+ executor_home = tmp_path / "home"
+ tracker = TurnFileChangeTracker(
+ workspace=repo,
+ task_id=12,
+ subtask_id=34,
+ executor_home=executor_home,
+ )
+ await tracker.start()
+ write(repo / "hello.txt", "hello\n")
+
+ changes = file_changes(await tracker.finalize())
+ artifact_dir = executor_home / "artifacts" / "turn-file-changes" / "12" / "34"
+ patch = gzip.decompress((artifact_dir / "changes.patch.gz").read_bytes())
+ metadata = json.loads((artifact_dir / "metadata.json").read_text())
+
+ assert changes["artifact_id"] == "turn-file-changes/12/34"
+ assert patch.startswith(b"diff --git a/hello.txt b/hello.txt")
+ assert metadata["checksum"] == hashlib.sha256(patch).hexdigest()
+ assert metadata["workspace_path"] == str(repo.resolve())
+
+
+@pytest.mark.asyncio
+async def test_tracker_returns_empty_fields_when_workspace_is_not_git(tmp_path):
+ workspace = tmp_path / "plain"
+ workspace.mkdir()
+ tracker = TurnFileChangeTracker(
+ workspace=workspace,
+ task_id=1,
+ subtask_id=2,
+ executor_home=tmp_path / "home",
+ )
+
+ assert await tracker.start() is False
+ assert await tracker.finalize() == {}
+
+
+@pytest.mark.asyncio
+async def test_tracker_reverse_patch_restores_only_the_turn(tmp_path):
+ repo = init_repo(tmp_path)
+ write(repo / "existing.txt", "committed\n")
+ commit_all(repo, "tracked")
+ write(repo / "existing.txt", "dirty before\n")
+ tracker = TurnFileChangeTracker(
+ workspace=repo,
+ task_id=1,
+ subtask_id=2,
+ executor_home=tmp_path / "home",
+ )
+ await tracker.start()
+
+ write(repo / "agent.txt", "agent\n")
+ await tracker.finalize()
+ patch = gzip.decompress(
+ (
+ tmp_path / "home/artifacts/turn-file-changes/1/2/changes.patch.gz"
+ ).read_bytes()
+ )
+ run_git(repo, "apply", "--reverse", "--binary", "-", input_bytes=patch)
+
+ assert (repo / "existing.txt").read_text() == "dirty before\n"
+ assert not (repo / "agent.txt").exists()
+
+
+@pytest.mark.asyncio
+async def test_tracker_does_not_change_real_git_index(tmp_path):
+ repo = init_repo(tmp_path)
+ write(repo / "staged.txt", "base\n")
+ commit_all(repo, "tracked")
+ write(repo / "staged.txt", "staged\n")
+ run_git(repo, "add", "staged.txt")
+ before_index = run_git(repo, "ls-files", "--stage")
+ tracker = TurnFileChangeTracker(
+ workspace=repo,
+ task_id=1,
+ subtask_id=2,
+ executor_home=tmp_path / "home",
+ )
+
+ await tracker.start()
+ write(repo / "agent.txt", "agent\n")
+ await tracker.finalize()
+
+ assert run_git(repo, "ls-files", "--stage") == before_index
+
+
+@pytest.mark.asyncio
+async def test_tracker_rejects_concurrent_turns_in_same_workspace(tmp_path):
+ repo = init_repo(tmp_path)
+ first = TurnFileChangeTracker(
+ workspace=repo,
+ task_id=1,
+ subtask_id=2,
+ executor_home=tmp_path / "home",
+ )
+ second = TurnFileChangeTracker(
+ workspace=repo,
+ task_id=1,
+ subtask_id=3,
+ executor_home=tmp_path / "home",
+ )
+ await first.start()
+
+ with pytest.raises(WorkspaceBusyError):
+ await second.start()
+
+ await first.abort()
+
+
+@pytest.mark.asyncio
+async def test_tracker_removes_stale_same_host_lock(tmp_path):
+ repo = init_repo(tmp_path)
+ lock_path = repo / ".git" / "wegent-turn-file-changes.lock"
+ lock_path.write_text(
+ json.dumps(
+ {
+ "pid": 99999999,
+ "hostname": os.uname().nodename,
+ "task_id": 9,
+ "subtask_id": 9,
+ }
+ )
+ )
+ tracker = TurnFileChangeTracker(
+ workspace=repo,
+ task_id=1,
+ subtask_id=2,
+ executor_home=tmp_path / "home",
+ )
+
+ assert await tracker.start() is True
+
+ await tracker.abort()
+ assert not lock_path.exists()
diff --git a/shared/models/db/subtask.py b/shared/models/db/subtask.py
index d5baff6cab..ac1655be0e 100644
--- a/shared/models/db/subtask.py
+++ b/shared/models/db/subtask.py
@@ -4,6 +4,8 @@
"""Subtask database model."""
+from datetime import datetime
+
from sqlalchemy import JSON, Boolean, Column, DateTime
from sqlalchemy import Enum as SQLEnum
from sqlalchemy import Integer, String, Text
@@ -40,7 +42,11 @@ class Subtask(Base):
error_message = Column(Text)
created_at = Column(DateTime, default=func.now())
updated_at = Column(DateTime, default=func.now(), onupdate=func.now())
- completed_at = Column(DateTime, nullable=False, default="1970-01-01 00:00:00")
+ completed_at = Column(
+ DateTime,
+ nullable=False,
+ default=datetime(1970, 1, 1),
+ )
# Group chat fields
sender_type = Column(
diff --git a/shared/models/execution.py b/shared/models/execution.py
index 4f3810f8c1..3eb9d474c1 100644
--- a/shared/models/execution.py
+++ b/shared/models/execution.py
@@ -136,6 +136,7 @@ class ExecutionRequest:
workspace_source: Optional[str] = None
project_workspace_path: Optional[str] = None
execution_target_type: Optional[str] = None
+ device_id: Optional[str] = None
# === Git Configuration (from Task) ===
git_domain: Optional[str] = None
diff --git a/shared/models/responses_api_emitter.py b/shared/models/responses_api_emitter.py
index fda499b3d3..35c7f3fa79 100644
--- a/shared/models/responses_api_emitter.py
+++ b/shared/models/responses_api_emitter.py
@@ -25,17 +25,22 @@
await emitter.done(content="Hello world", usage={"input_tokens": 10})
"""
+import inspect
import json
import logging
import time
import uuid
from abc import ABC, abstractmethod
-from typing import Any, Callable, Optional, Union
+from typing import Any, Awaitable, Callable, Optional, Union
from .responses_api import ResponsesAPIEventBuilder, ResponsesAPIStreamEvents
logger = logging.getLogger(__name__)
SHELL_TOOL_NAMES = {"exec"}
+CompletionFieldsProvider = Callable[
+ [],
+ Union[dict[str, Any], Awaitable[dict[str, Any]]],
+]
__all__ = [
"ResponsesAPIEmitter",
@@ -87,6 +92,19 @@ def __init__(
self.executor_namespace = executor_namespace
self.builder = ResponsesAPIEventBuilder(subtask_id, model)
self._tool_contexts: dict[str, dict[str, Any]] = {}
+ self._completion_fields_provider: Optional[CompletionFieldsProvider] = None
+
+ def set_completion_fields_provider(
+ self,
+ provider: Optional[CompletionFieldsProvider],
+ ) -> None:
+ """Set a one-shot provider for fields added to the completion event."""
+ self._completion_fields_provider = provider
+ logger.info(
+ "Response completion fields provider %s for subtask_id=%s",
+ "set" if provider is not None else "cleared",
+ self.subtask_id,
+ )
# ============================================================
# Response Lifecycle Events
@@ -144,6 +162,35 @@ async def done(
# Flush any buffered events before sending done
await self.flush()
+ provider = self._completion_fields_provider
+ self._completion_fields_provider = None
+ logger.info(
+ "Preparing response.completed for subtask_id=%s completion_provider=%s",
+ self.subtask_id,
+ provider is not None,
+ )
+ if provider is not None:
+ try:
+ provider_fields = provider()
+ if inspect.isawaitable(provider_fields):
+ provider_fields = await provider_fields
+ if isinstance(provider_fields, dict):
+ logger.info(
+ "Collected response completion fields for subtask_id=%s keys=%s",
+ self.subtask_id,
+ sorted(provider_fields.keys()),
+ )
+ for key, value in provider_fields.items():
+ extra_fields.setdefault(key, value)
+ else:
+ logger.info(
+ "Response completion fields provider returned non-dict for subtask_id=%s: %s",
+ self.subtask_id,
+ type(provider_fields).__name__,
+ )
+ except Exception:
+ logger.exception("Failed to collect response completion fields")
+
data = self.builder.response_completed(
content=content,
usage=usage,
diff --git a/shared/tests/test_emitter_factory.py b/shared/tests/test_emitter_factory.py
index 21ba6e9631..932a2e54db 100644
--- a/shared/tests/test_emitter_factory.py
+++ b/shared/tests/test_emitter_factory.py
@@ -372,6 +372,69 @@ async def test_send_publishes_to_redis(self):
class TestResponsesAPIEmitter:
+ @pytest.mark.asyncio
+ async def test_done_merges_completion_provider_fields(self):
+ transport = GeneratorTransport()
+ emitter = EmitterBuilder().with_task(1, 2).with_transport(transport).build()
+
+ async def completion_fields():
+ return {"file_changes": {"version": 1, "file_count": 2}}
+
+ emitter.set_completion_fields_provider(completion_fields)
+
+ await emitter.done(content="done")
+
+ completed = transport.get_events()[-1][1]["response"]
+ assert completed["file_changes"] == {
+ "version": 1,
+ "file_count": 2,
+ }
+
+ @pytest.mark.asyncio
+ async def test_done_ignores_empty_completion_provider_fields(self):
+ transport = GeneratorTransport()
+ emitter = EmitterBuilder().with_task(1, 2).with_transport(transport).build()
+ emitter.set_completion_fields_provider(lambda: {})
+
+ await emitter.done(content="done")
+
+ completed = transport.get_events()[-1][1]["response"]
+ assert "file_changes" not in completed
+
+ @pytest.mark.asyncio
+ async def test_done_logs_provider_failure_and_still_emits_completion(self, caplog):
+ transport = GeneratorTransport()
+ emitter = EmitterBuilder().with_task(1, 2).with_transport(transport).build()
+
+ async def fail():
+ raise RuntimeError("snapshot failed")
+
+ emitter.set_completion_fields_provider(fail)
+
+ await emitter.done(content="done")
+
+ completed = transport.get_events()[-1][1]["response"]
+ assert completed["output"][0]["content"][0]["text"] == "done"
+ assert "Failed to collect response completion fields" in caplog.text
+
+ @pytest.mark.asyncio
+ async def test_done_consumes_completion_provider_once(self):
+ transport = GeneratorTransport()
+ emitter = EmitterBuilder().with_task(1, 2).with_transport(transport).build()
+ calls = 0
+
+ async def completion_fields():
+ nonlocal calls
+ calls += 1
+ return {"file_changes": {"version": 1}}
+
+ emitter.set_completion_fields_provider(completion_fields)
+
+ await emitter.done(content="first")
+ await emitter.done(content="second")
+
+ assert calls == 1
+
@pytest.mark.asyncio
async def test_reasoning_emits_summary_text_delta(self):
transport = GeneratorTransport()
diff --git a/wework/src/api/http.test.ts b/wework/src/api/http.test.ts
index d6576b4a20..b17514c0c6 100644
--- a/wework/src/api/http.test.ts
+++ b/wework/src/api/http.test.ts
@@ -97,6 +97,32 @@ describe('createHttpClient', () => {
})
})
+ test('preserves structured error details and codes', async () => {
+ fetchMock.mockResolvedValueOnce({
+ ok: false,
+ status: 409,
+ text: async () =>
+ JSON.stringify({
+ detail: {
+ code: 'TURN_FILE_CHANGES_CONFLICT',
+ message: 'Patch does not apply',
+ file_changes: { status: 'conflicted' },
+ },
+ }),
+ })
+
+ const client = createHttpClient({ baseUrl: '/api' })
+
+ await expect(client.post('/subtasks/9/file-changes/revert')).rejects.toMatchObject({
+ message: 'Patch does not apply',
+ status: 409,
+ errorCode: 'TURN_FILE_CHANGES_CONFLICT',
+ detail: {
+ file_changes: { status: 'conflicted' },
+ },
+ })
+ })
+
test('posts FormData without forcing a json content type', async () => {
localStorage.setItem('auth_token', 'token-1')
fetchMock.mockResolvedValueOnce({
diff --git a/wework/src/api/http.ts b/wework/src/api/http.ts
index 920aaae2b8..f2356f0467 100644
--- a/wework/src/api/http.ts
+++ b/wework/src/api/http.ts
@@ -4,12 +4,19 @@ import { redirectToLogin } from '@/features/auth/redirect'
export class ApiError extends Error {
status: number
errorCode?: string | number
-
- constructor(message: string, status: number, errorCode?: string | number) {
+ detail?: unknown
+
+ constructor(
+ message: string,
+ status: number,
+ errorCode?: string | number,
+ detail?: unknown,
+ ) {
super(message)
this.name = 'ApiError'
this.status = status
this.errorCode = errorCode
+ this.detail = detail
}
}
@@ -33,14 +40,20 @@ async function parseError(response: Response): Promise
+ {t('file_changes.edited_files', {
+ count: summary.file_count,
+ })}
+
+ +{summary.additions}
+ -{summary.deletions}
+
+ {t('file_changes.conflicted')}
+
+ {t('file_changes.artifact_missing')}
+
+ {t('file_changes.device_offline')}
+
+ {actionError}
+
+ {t('file_changes.confirm_revert_description')}
+
+ {t('file_changes.loading_diff')}
+
+ {error}
+
+ {t('file_changes.empty_diff')}
+ {message.error}
+ {t('file_changes.confirm_revert_title')}
+
+
+ {t('file_changes.review_title')}
+
+
+
+ {section.lines.map((line, lineIndex) => (
+
+ {line || ' '}
+
+ ))}
+
+ )}
+