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
21 changes: 21 additions & 0 deletions backend/app/api/endpoints/runtime_work.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@
RuntimeFileChangesRevertRequest,
RuntimeFileChangesRevertResponse,
RuntimeGlobalIMNotificationUpdateRequest,
RuntimeGuidanceRequest,
RuntimeGuidanceResponse,
RuntimeIMNotificationSettingsResponse,
RuntimeSendRequest,
RuntimeSendResponse,
Expand Down Expand Up @@ -224,6 +226,25 @@ async def send_runtime_message_endpoint(
)


@router.post(
"/guidance",
response_model=RuntimeGuidanceResponse,
response_model_by_alias=True,
)
async def send_runtime_guidance_endpoint(
request: RuntimeGuidanceRequest,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Steer an active native runtime turn through the owning local executor."""

return await runtime_work_service.send_runtime_guidance(
db=db,
user_id=current_user.id,
request=request,
)


@router.post(
"/im-sessions",
response_model=BindRuntimeTaskIMSessionsResponse,
Expand Down
28 changes: 28 additions & 0 deletions backend/app/schemas/runtime_work.py
Original file line number Diff line number Diff line change
Expand Up @@ -434,6 +434,34 @@ class RuntimeSendResponse(BaseModel):
error: Optional[str] = None


class RuntimeGuidanceRequest(BaseModel):
"""Request to steer an active native runtime turn."""

model_config = ConfigDict(populate_by_name=True)

address: RuntimeTaskAddress
message: str = Field(..., min_length=1)
client_guidance_id: Optional[str] = Field(
default=None,
alias="clientGuidanceId",
validation_alias=AliasChoices("clientGuidanceId", "client_guidance_id"),
)


class RuntimeGuidanceResponse(BaseModel):
"""Acknowledgement from the runtime guidance RPC."""

model_config = ConfigDict(populate_by_name=True)

accepted: bool
success: bool = True
local_task_id: str = Field(..., alias="taskId")
guidance_id: Optional[str] = Field(default=None, alias="guidanceId")
turn_id: Optional[str] = Field(default=None, alias="turnId")
error: Optional[str] = None
code: Optional[str] = None


class RuntimeWorkspaceOpenRequest(BaseModel):
"""Request to register/open a device-local runtime workspace."""

Expand Down
52 changes: 52 additions & 0 deletions backend/app/services/runtime_work_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,8 @@
RuntimeFileChangesRevertRequest,
RuntimeFileChangesRevertResponse,
RuntimeGlobalIMNotificationUpdateRequest,
RuntimeGuidanceRequest,
RuntimeGuidanceResponse,
RuntimeIMNotificationSession,
RuntimeIMNotificationSettingsResponse,
RuntimeProjectRef,
Expand Down Expand Up @@ -582,6 +584,39 @@ async def send_runtime_message(
return _runtime_send_response(result, address.local_task_id)


async def send_runtime_guidance(
*,
db: Session,
user_id: int,
request: RuntimeGuidanceRequest,
) -> RuntimeGuidanceResponse:
"""Steer an active LocalTask turn through the owning local executor."""

address = _normalized_address(request.address)
_ensure_owned_device(db, user_id, address.device_id)
_touch_workspace_mapping(db, user_id, address)
payload = {
**_runtime_task_address_payload(address),
"message": request.message,
}
if request.client_guidance_id:
payload["clientGuidanceId"] = request.client_guidance_id
try:
result = await runtime_rpc_service.call(
user_id=user_id,
device_id=address.device_id,
method="runtime.tasks.guidance",
payload=payload,
timeout_seconds=RUNTIME_SEND_TIMEOUT_SECONDS,
)
except RuntimeRpcError as exc:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=str(exc),
) from exc
return _runtime_guidance_response(result, address.local_task_id)


async def bind_runtime_task_to_im_sessions(
*,
db: Session,
Expand Down Expand Up @@ -1817,6 +1852,23 @@ def _runtime_send_response(
)


def _runtime_guidance_response(
result: dict[str, Any],
local_task_id: str,
) -> RuntimeGuidanceResponse:
success = result.get("success")
accepted = bool(result.get("accepted", success is not False))
return RuntimeGuidanceResponse(
accepted=accepted,
success=success is not False,
taskId=str(result.get("taskId") or local_task_id),
guidanceId=result.get("guidanceId") or result.get("guidance_id"),
turnId=result.get("turnId") or result.get("turn_id"),
error=result.get("error"),
code=result.get("code"),
)


def _runtime_archive_response(
result: dict[str, Any],
address: RuntimeTaskAddress,
Expand Down
43 changes: 43 additions & 0 deletions backend/tests/api/endpoints/test_runtime_work_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -350,6 +350,49 @@ def test_runtime_cancel_endpoint_dispatches_address(
assert service_mock.await_args.kwargs["address"].local_task_id == "codex-1"


def test_runtime_guidance_endpoint_dispatches_request(
test_client,
test_token,
monkeypatch,
):
from app.api.endpoints import runtime_work

service_mock = AsyncMock(
return_value={
"accepted": True,
"success": True,
"taskId": "codex-1",
"guidanceId": "guide-1",
"turnId": "turn-1",
"error": None,
"code": None,
}
)
monkeypatch.setattr(
runtime_work.runtime_work_service, "send_runtime_guidance", service_mock
)

response = test_client.post(
"/api/runtime-work/guidance",
headers=_auth_headers(test_token),
json={
"address": {
"deviceId": "device-1",
"workspacePath": "/repo/Wegent",
"taskId": "codex-1",
},
"message": "use this context",
"clientGuidanceId": "guide-1",
},
)

assert response.status_code == 200
assert response.json()["accepted"] is True
request = service_mock.await_args.kwargs["request"]
assert request.address.local_task_id == "codex-1"
assert request.client_guidance_id == "guide-1"


def test_archived_conversations_list_endpoint_dispatches_filters(
test_client,
test_token,
Expand Down
108 changes: 108 additions & 0 deletions backend/tests/services/test_runtime_work_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -1674,6 +1674,114 @@ async def test_send_runtime_message_normalizes_runtime_rpc_failure_without_task_
assert test_db.query(TaskResource).count() == 0


@pytest.mark.asyncio
async def test_send_runtime_guidance_dispatches_to_owned_device_without_task_rows(
test_db,
test_user,
monkeypatch,
):
from app.schemas.runtime_work import (
RuntimeGuidanceRequest,
RuntimeTaskAddress,
)
from app.services import runtime_work_service

monkeypatch.setattr(
runtime_work_service.device_service,
"get_device_by_device_id",
lambda db, user_id, device_id: object(),
)
rpc = AsyncMock(
return_value={
"success": True,
"accepted": True,
"taskId": "codex-1",
"guidanceId": "guide-1",
"turnId": "turn-1",
}
)
monkeypatch.setattr(runtime_work_service.runtime_rpc_service, "call", rpc)

response = await runtime_work_service.send_runtime_guidance(
db=test_db,
user_id=test_user.id,
request=RuntimeGuidanceRequest(
address=RuntimeTaskAddress(
deviceId="device-1",
localTaskId="codex-1",
),
message="use this context",
clientGuidanceId="guide-1",
),
)

assert response.accepted is True
assert response.local_task_id == "codex-1"
assert response.guidance_id == "guide-1"
assert response.turn_id == "turn-1"
rpc.assert_awaited_once_with(
user_id=test_user.id,
device_id="device-1",
method="runtime.tasks.guidance",
payload={
"deviceId": "device-1",
"taskId": "codex-1",
"message": "use this context",
"clientGuidanceId": "guide-1",
},
timeout_seconds=600,
)
assert test_db.query(TaskResource).count() == 0


@pytest.mark.asyncio
async def test_send_runtime_guidance_normalizes_runtime_rpc_failure_without_task_rows(
test_db,
test_user,
monkeypatch,
):
from app.schemas.runtime_work import (
RuntimeGuidanceRequest,
RuntimeTaskAddress,
)
from app.services import runtime_work_service

monkeypatch.setattr(
runtime_work_service.device_service,
"get_device_by_device_id",
lambda db, user_id, device_id: object(),
)
rpc = AsyncMock(
return_value={
"success": False,
"accepted": False,
"taskId": "codex-1",
"error": "no active turn to guide",
"code": "no_active_turn",
}
)
monkeypatch.setattr(runtime_work_service.runtime_rpc_service, "call", rpc)

response = await runtime_work_service.send_runtime_guidance(
db=test_db,
user_id=test_user.id,
request=RuntimeGuidanceRequest(
address=RuntimeTaskAddress(
deviceId="device-1",
localTaskId="codex-1",
),
message="use this context",
),
)

assert response.accepted is False
assert response.success is False
assert response.local_task_id == "codex-1"
assert response.error == "no active turn to guide"
assert response.code == "no_active_turn"
assert test_db.query(TaskResource).count() == 0


@pytest.mark.asyncio
async def test_send_runtime_message_forwards_ready_attachments_without_task_rows(
test_db,
Expand Down
16 changes: 13 additions & 3 deletions docs/en/developer-guide/runtime-local-work.md
Original file line number Diff line number Diff line change
Expand Up @@ -97,9 +97,9 @@ cargo test --test manual_runtime_perf -- --ignored --nocapture

Current local measurements:

| Sample | File size | List | First open | Loaded switch | Append refresh |
| --- | --- | ---: | ---: | ---: | ---: |
| "Fix running task tool calls not shown" | about 61 MB | 13 ms | 2.09 s | 33 ms | 53 ms |
| Sample | File size | List | First open | Loaded switch | Append refresh |
| --------------------------------------- | ----------- | ----: | ---------: | ------------: | -------------: |
| "Fix running task tool calls not shown" | about 61 MB | 13 ms | 2.09 s | 33 ms | 53 ms |

The current target is therefore met: list under 1 second, first open under 3 seconds, and loaded switch plus fresh-data refresh under 500 ms. The first cold parse for even larger extreme histories is still bounded by JSONL size, but loaded switching and append refresh no longer grow with total history length.

Expand All @@ -121,6 +121,16 @@ POST /api/runtime-work/cancel

Backend forwards `deviceId + localTaskId` as `runtime.tasks.cancel`. The Rust executor Codex app-server path currently does not interrupt an in-flight turn across processes, so app mode returns `accepted: false`; the frontend should keep the queued state or wait for the current turn to finish. If `turn/interrupt` support is added later, it must still identify the task only by `deviceId + localTaskId`; `workspacePath` remains device-directory context.

When the current Codex LocalTask is still replying, Wework can also send a queued user message as native guidance:

```text
POST /api/runtime-work/guidance
```

Backend only validates the user, device, and LocalTask ownership, then forwards `deviceId + localTaskId`, the user text, and the frontend-generated `clientGuidanceId` as device RPC `runtime.tasks.guidance`. The executor must find the running Codex turn and append the user text through Codex app-server's native guidance capability. If there is no active turn to guide, it should return `no_active_turn`; the frontend then sends the same message as a regular follow-up. `runtime.tasks.guidance` does not create central Task or Subtask rows, and it must not use `workspacePath` as task identity.

The frontend must insert the local guidance user message at the current streaming assistant position immediately when guidance sending starts, not after `runtime.tasks.guidance` returns. That insertion splits the active assistant into a frozen "before guidance" message and a continuing "after guidance" message. The continuing assistant keeps the original `subtaskId` so later stream events land after the guidance message. Later `chat:chunk` or `chat:done` events may still carry full text, so the frontend trims the text prefix recorded at split time. This keeps live streaming order consistent with the refreshed transcript order.

Continuing a LocalTask may include already uploaded attachment ids that are in the ready state. Backend verifies those attachments belong to the current user and converts them into executor attachment metadata. The executor downloads and converts the files on the target device before passing them to the runtime. The frontend never sends local attachment paths directly to Backend or executor.

## Archived Conversations
Expand Down
Loading
Loading