Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions backend/app/schemas/runtime_work.py
Original file line number Diff line number Diff line change
Expand Up @@ -466,6 +466,7 @@ class RuntimeTaskCreateRequest(BaseModel):
)
device_id: Optional[str] = Field(default=None, alias="deviceId")
workspace_path: Optional[str] = Field(default=None, alias="workspacePath")
local_task_id: Optional[str] = Field(default=None, alias="localTaskId")
team_id: int = Field(..., alias="teamId", ge=1)
runtime: RuntimeName
message: str = Field(..., min_length=1)
Expand Down
2 changes: 2 additions & 0 deletions backend/app/services/runtime_work_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -763,6 +763,8 @@ async def create_runtime_task(
"title": _runtime_task_title(request),
"executionRequest": execution_request.to_dict(),
}
if request.local_task_id:
payload["localTaskId"] = request.local_task_id
try:
result = await runtime_rpc_service.call(
user_id=user_id,
Expand Down
2 changes: 2 additions & 0 deletions backend/tests/services/test_runtime_work_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -1325,6 +1325,7 @@ async def test_create_runtime_task_dispatches_to_project_device_without_task_row
user_id=test_user.id,
request=RuntimeTaskCreateRequest(
projectId=project.id,
localTaskId="runtime-client-1",
teamId=3,
runtime="claude_code",
message="create runtime task",
Expand All @@ -1341,6 +1342,7 @@ async def test_create_runtime_task_dispatches_to_project_device_without_task_row
method="runtime.tasks.create",
payload={
"runtime": "claude_code",
"localTaskId": "runtime-client-1",
"workspacePath": "/repo/Wegent",
"message": "create runtime task",
"title": "Create runtime task",
Expand Down
4 changes: 4 additions & 0 deletions docs/en/developer-guide/runtime-local-work.md
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,8 @@ After an image attachment uploads successfully, Wework keeps a frontend-local `l

When rendering a message that already has persisted image attachments, Wework prefers those attachment previews and ignores local image file mentions embedded in the Codex prompt. This avoids showing both the uploaded attachment and a temporary local path. Codex local image mentions are used only as a same-device preview fallback when no attachment record exists. If the current environment cannot convert the local path through Tauri `convertFileSrc`, or the converted image fails to load, the frontend does not display that local path.

When the executor discovers a user message from a native Codex session, it writes local image paths from `local_images`, `localImages`, or `images` into the user-visible text so refreshes still show which files the user mentioned. If those paths are readable on the current device, have an image MIME type, and are no larger than 5 MB, the executor also creates ready attachments used only for transcript rendering and stores `local_preview_url` as a data URL. This preview attachment is not a persisted Backend attachment, and it is not uploaded or synced to the central database.

Native Codex tasks have one additional rule: transcript refreshes trust only Codex's own session transcript. `runtimeHandle.messages` from a fork package or the executor JSON index is only an import-time snapshot and must not be used as a fallback for native Codex transcripts; otherwise Wework can show stale messages or lose follow-up turns after refresh. Non-SDK native tasks may still use the executor JSON index as their local transcript source.

Assistant messages in a runtime transcript may include a `fileChanges` summary. During native Codex creation and continuation, the executor attaches a `NativeTurnFileChangeTracker` to the Codex SDK `turn/diff/updated` events and records the latest cumulative diff for the current turn. When the response completes, the tracker returns `file_changes` through Responses completion fields, and `runtime.tasks.create`, `runtime.tasks.send`, and `runtime.tasks.transcript` must normalize it onto the message as `fileChanges`. This lets the frontend show the file changes card under the current assistant message without waiting for the next list refresh.
Expand All @@ -117,6 +119,8 @@ POST /api/runtime-work/create

Backend resolves the target device and directory from either a Project mapping or a standalone device workspace, builds a transient execution request, and calls device RPC `runtime.tasks.create`. This flow does not `db.add()` any `TaskResource` or `Subtask`.

Before calling create, Wework generates a client-side `localTaskId` and sends it to Backend as `localTaskId`. Backend only forwards that value to the target device; it does not write it to the central database. The frontend immediately opens the runtime URL from `deviceId + localTaskId`, renders the user message, and shows the waiting state. If the device returns a different `localTaskId`, the frontend switches to the device-confirmed address. This lets a newly created task appear before the Backend RPC completes or the next list refresh runs, and queued sends wait until the current waiting state becomes a real assistant turn before continuing.

The runtime owns persistence for newly created tasks:

- Claude Code creates an executor JSON LocalTask and stores the transcript and runtime handle in that index.
Expand Down
4 changes: 4 additions & 0 deletions docs/zh/developer-guide/runtime-local-work.md
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,8 @@ Backend 将 `deviceId + localTaskId` 转发为 `runtime.tasks.cancel`。executor

消息渲染时,如果消息已经带有持久化图片附件,Wework 优先展示附件预览,并忽略 Codex prompt 中的本地图片文件提及,避免同时展示上传附件和临时本机路径。只有没有附件记录时,才把 Codex 本地图片提及作为本机预览兜底;如果当前环境不能通过 Tauri `convertFileSrc` 转换本机路径,或转换后的图片加载失败,前端不展示该本机路径。

executor 从原生 Codex session 发现用户消息时,会把 `local_images`、`localImages` 或 `images` 中的本机图片路径写入用户可见文本,保持刷新后仍能看到“用户提到了哪些文件”。如果这些路径在当前设备上可读、是图片 MIME 类型且不超过 5 MB,executor 会额外生成只用于 transcript 渲染的 ready 附件,并把 `local_preview_url` 写成 data URL。这个预览附件不代表 Backend 持久化附件,也不会上传或同步到中心库。

原生 Codex 任务有一个额外约束:刷新 transcript 时只信任 Codex 本身的会话记录。fork 包或 executor JSON 索引中携带的 `runtimeHandle.messages` 只是导入瞬间的快照,不能作为原生 Codex transcript 的回退来源,否则 Wework 刷新后会显示旧消息或丢失用户追问。非 SDK 原生任务仍可以使用 executor JSON 索引中的本地 transcript。

runtime transcript 中的 assistant 消息可以携带 `fileChanges` 摘要。原生 Codex 创建和继续任务时,executor 会把 `NativeTurnFileChangeTracker` 接到 Codex SDK 的 `turn/diff/updated` 事件上,记录最新的本轮累计 diff;回复完成时 tracker 通过 Responses completion fields 返回 `file_changes`,`runtime.tasks.create`、`runtime.tasks.send` 和 `runtime.tasks.transcript` 都必须把它规范化为消息上的 `fileChanges`。这样前端无需等待下一次列表刷新,就能在当前 assistant 消息下显示本轮文件变更卡片。
Expand All @@ -117,6 +119,8 @@ POST /api/runtime-work/create

Backend 根据请求中的项目映射或独立设备工作区解析目标设备和目录,构造一次临时 execution request,然后调用设备 RPC `runtime.tasks.create`。这个流程不会 `db.add()` 任何 `TaskResource` 或 `Subtask`。

Wework 在调用 create 前先生成客户端侧 `localTaskId`,并在请求体中作为 `localTaskId` 传给 Backend。Backend 只把这个值转发给目标设备,不把它写入中心数据库。前端会立即用 `deviceId + localTaskId` 打开运行时 URL、展示用户消息和等待态;如果设备返回了不同的 `localTaskId`,前端再切换到设备确认的地址。这样新建任务不需要等待 Backend RPC 完成或下一次列表刷新,队列发送也会等当前等待态进入真实 assistant turn 后再继续。

运行时创建的持久化位置由具体 runtime 决定:

- Claude Code 创建 executor JSON LocalTask,并在该索引中保存 transcript 和 runtime handle。
Expand Down
151 changes: 131 additions & 20 deletions executor/runtime_work/codex_discovery.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,12 @@

"""Discover Codex sessions as device-local runtime work items."""

import base64
import contextlib
import gzip
import hashlib
import json
import mimetypes
import os
import re
import time
Expand All @@ -16,6 +18,7 @@
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Iterable, Optional
from urllib.parse import unquote, urlparse

from executor.agents.codex.config_builder import _resolve_codex_binary
from executor.config import config
Expand All @@ -36,6 +39,7 @@
CODEX_TRANSCRIPT_MAX_LIMIT = 200
CODEX_TRANSCRIPT_INITIAL_WINDOW_BYTES = 1024 * 1024
CODEX_TRANSCRIPT_MAX_WINDOW_BYTES = 64 * 1024 * 1024
CODEX_LOCAL_IMAGE_PREVIEW_MAX_BYTES = 5 * 1024 * 1024
CODEX_TERMINAL_EVENT_TYPES = {"task_complete", "turn_aborted"}
CODEX_CONVERSATION_DATE_PATTERN = re.compile(r"^\d{4}-\d{2}-\d{2}$")
CODEX_TRANSCRIPT_CURSOR_PATTERN = re.compile(r"^offset:(\d+)$")
Expand Down Expand Up @@ -832,6 +836,113 @@ def _first_text(record: dict[str, Any], *keys: str) -> Optional[str]:
return None


def _codex_user_message_content(payload: dict[str, Any]) -> Optional[str]:
message = _first_text(payload, "message") or ""
local_image_paths = _local_image_paths_from_payload(payload)
if not local_image_paths:
return message or None
return _build_files_mentioned_text(message, local_image_paths).lstrip()


def _local_image_paths_from_payload(payload: dict[str, Any]) -> list[str]:
paths: list[str] = []
seen: set[str] = set()
for key in ("local_images", "localImages", "images"):
value = payload.get(key)
if not isinstance(value, list):
continue
for item in value:
path = _local_image_path_from_item(item)
if path is None or path in seen:
continue
seen.add(path)
paths.append(path)
return paths


def _local_image_path_from_item(item: Any) -> Optional[str]:
if isinstance(item, str):
return _normalize_local_image_path(item)
if not isinstance(item, dict):
return None

for key in ("path", "local_path", "localPath", "file_path", "filePath"):
path = _normalize_local_image_path(item.get(key))
if path is not None:
return path
return None


def _normalize_local_image_path(value: Any) -> Optional[str]:
if not isinstance(value, str):
return None
path = value.strip()
if not path:
return None
parsed = urlparse(path)
if parsed.scheme and parsed.scheme != "file":
return None
return path


def _set_local_image_attachments(
message: dict[str, Any],
payload: dict[str, Any],
created_at: str,
) -> None:
attachments = [
attachment
for path in _local_image_paths_from_payload(payload)
if (attachment := _local_image_attachment(path, created_at)) is not None
]
if attachments:
message["attachments"] = attachments


def _local_image_attachment(path: str, created_at: str) -> Optional[dict[str, Any]]:
filesystem_path = _local_image_filesystem_path(path)
file_path = Path(filesystem_path)
try:
stat = file_path.stat()
except OSError:
return None
if not file_path.is_file() or stat.st_size > CODEX_LOCAL_IMAGE_PREVIEW_MAX_BYTES:
return None

mime_type = mimetypes.guess_type(file_path.name)[0] or "application/octet-stream"
if not mime_type.startswith("image/"):
return None

try:
encoded = base64.b64encode(file_path.read_bytes()).decode("ascii")
except OSError:
return None

return {
"id": _local_image_attachment_id(path),
"filename": file_path.name,
"file_size": stat.st_size,
"mime_type": mime_type,
"status": "ready",
"file_extension": file_path.suffix,
"created_at": created_at,
"local_preview_url": f"data:{mime_type};base64,{encoded}",
}


def _local_image_filesystem_path(path: str) -> str:
if not path.startswith("file://"):
return path
parsed = urlparse(path)
pathname = unquote(parsed.path)
return pathname[1:] if re.match(r"^/[a-zA-Z]:/", pathname) else pathname


def _local_image_attachment_id(path: str) -> int:
digest = hashlib.sha256(path.encode("utf-8")).hexdigest()
return int(digest[:12], 16)


def _parse_datetime(value: Any) -> Optional[datetime]:
if not isinstance(value, str) or not value:
return None
Expand Down Expand Up @@ -918,18 +1029,18 @@ def _read_session_transcript(path: Path, thread_id: str) -> list[dict[str, Any]]
processing_blocks = []
tool_blocks_by_call_id = {}
turn_started_at = timestamp
message = _first_text(payload, "message")
message = _codex_user_message_content(payload)
if message:
messages.append(
_transcript_message(
thread_id=thread_id,
turn_counter=turn_counter,
role="user",
content=message,
created_at=timestamp,
status="done",
)
user_message = _transcript_message(
thread_id=thread_id,
turn_counter=turn_counter,
role="user",
content=message,
created_at=timestamp,
status="done",
)
_set_local_image_attachments(user_message, payload, timestamp)
messages.append(user_message)
elif event_type == "agent_message":
message = _first_text(payload, "message")
if message:
Expand Down Expand Up @@ -1139,18 +1250,18 @@ def _read_session_transcript_window(
cwd = payload.get("cwd")
if isinstance(cwd, str) and cwd.strip():
turn_workspace_path = cwd.strip()
message = _first_text(payload, "message")
message = _codex_user_message_content(payload)
if message:
messages.append(
_transcript_message_from_offset(
thread_id=thread_id,
offset=line_start,
role="user",
content=message,
created_at=timestamp,
status="done",
)
user_message = _transcript_message_from_offset(
thread_id=thread_id,
offset=line_start,
role="user",
content=message,
created_at=timestamp,
status="done",
)
_set_local_image_attachments(user_message, payload, timestamp)
messages.append(user_message)
elif event_type == "agent_message":
message = _first_text(payload, "message")
if message:
Expand Down
88 changes: 88 additions & 0 deletions executor/tests/runtime_work/test_local_task_store.py
Original file line number Diff line number Diff line change
Expand Up @@ -2086,6 +2086,94 @@ def test_codex_discovery_reads_user_visible_transcript(tmp_path):
assert transcript[1]["status"] == "done"


def test_codex_discovery_preserves_local_images_in_user_message(tmp_path):
from executor.runtime_work.codex_discovery import CodexSessionDiscovery

codex_home = tmp_path / "codex-home"
session_dir = codex_home / "sessions" / "2026" / "06" / "20"
session_dir.mkdir(parents=True)
thread_id = "018f2d6b-8c7a-7abc-9def-0123456789b0"
session_path = session_dir / f"rollout-2026-06-20T13-52-19-{thread_id}.jsonl"
image_path = "/var/folders/tmp/codex-clipboard.png"
_write_codex_session(
session_path,
{
"timestamp": "2026-06-20T05:52:21Z",
"type": "event_msg",
"payload": {
"type": "user_message",
"message": "分析下这个图片",
"images": [],
"local_images": [image_path],
},
},
{
"timestamp": "2026-06-20T05:52:23Z",
"type": "event_msg",
"payload": {
"type": "task_complete",
"last_agent_message": "看到了",
},
},
)

page = CodexSessionDiscovery(codex_home=codex_home).read_transcript_page(
thread_id,
str(session_path),
)

assert [message["role"] for message in page.messages] == ["user", "assistant"]
assert page.messages[0]["content"] == "\n".join(
[
"# Files mentioned by the user:",
"",
f"## codex-clipboard.png: {image_path}",
"",
"## My request for Codex:",
"分析下这个图片",
"",
]
)
assert page.messages[1]["content"] == "看到了"


def test_codex_discovery_embeds_readable_local_image_previews(tmp_path):
from executor.runtime_work.codex_discovery import CodexSessionDiscovery

codex_home = tmp_path / "codex-home"
session_dir = codex_home / "sessions" / "2026" / "06" / "20"
session_dir.mkdir(parents=True)
thread_id = "018f2d6b-8c7a-7abc-9def-0123456789b1"
session_path = session_dir / f"rollout-2026-06-20T13-52-19-{thread_id}.jsonl"
image_path = tmp_path / "codex-clipboard.png"
image_path.write_bytes(b"\x89PNG\r\n\x1a\n")
_write_codex_session(
session_path,
{
"timestamp": "2026-06-20T05:52:21Z",
"type": "event_msg",
"payload": {
"type": "user_message",
"message": "分析下这个图片",
"local_images": [str(image_path)],
},
},
)

page = CodexSessionDiscovery(codex_home=codex_home).read_transcript_page(
thread_id,
str(session_path),
)

attachment = page.messages[0]["attachments"][0]
assert attachment["filename"] == "codex-clipboard.png"
assert attachment["mime_type"] == "image/png"
assert attachment["file_extension"] == ".png"
assert attachment["file_size"] == 8
assert attachment["status"] == "ready"
assert attachment["local_preview_url"] == "data:image/png;base64,iVBORw0KGgo="


def test_codex_discovery_keeps_commentary_when_task_complete_is_empty(tmp_path):
from executor.runtime_work.codex_discovery import CodexSessionDiscovery

Expand Down
Loading
Loading