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
10 changes: 10 additions & 0 deletions executor/src/agents/codex.rs
Original file line number Diff line number Diff line change
Expand Up @@ -340,6 +340,16 @@ pub async fn run_codex_app_server_turn(
.to_owned();
thread_fields.push(("thread_id", thread_id.clone()));
log_executor_event("codex thread request finished", &thread_fields);
if let Some(sender) = &notifications {
let _ = sender.send(json!({
"method": "thread/started",
"params": {
"thread": {
"id": thread_id.clone()
}
}
}));
}

let turn_input = turn_input(&request.prompt);
let mut turn_fields = task_fields(request.task_id, request.subtask_id);
Expand Down
28 changes: 24 additions & 4 deletions executor/src/runtime_work/handler.rs
Original file line number Diff line number Diff line change
Expand Up @@ -599,10 +599,6 @@ impl RuntimeWorkRpcHandler {
"code": "bad_request",
}));
}
let thread_id = existing_link
.as_ref()
.and_then(|link| link.thread_id.clone())
.unwrap_or_else(|| local_task_id.clone());
let workspace_path = workspace_path(&payload)
.or_else(|| {
existing_link
Expand All @@ -620,6 +616,19 @@ impl RuntimeWorkRpcHandler {
if request.project_workspace_path.is_none() && !workspace_path.is_empty() {
request.project_workspace_path = Some(workspace_path.clone());
}
let Some(thread_id) = runtime_session_id_from_payload(&payload).or_else(|| {
existing_link
.as_ref()
.and_then(runtime_session_id_from_link)
}) else {
return Ok(json!({
"success": false,
"error": "runtime task session is not ready",
"code": "missing_runtime_session",
"localTaskId": local_task_id,
"runtime": "codex",
}));
};

let mut fields = task_fields(request.task_id, request.subtask_id);
fields.push(("local_task_id", local_task_id.clone()));
Expand Down Expand Up @@ -1869,12 +1878,23 @@ fn runtime_session_id_from_link(link: &RuntimeTaskLink) -> Option<String> {
}

fn runtime_session_id_from_payload(payload: &Value) -> Option<String> {
let address = payload.get("address");
payload
.get("runtimeHandle")
.or_else(|| payload.get("runtime_handle"))
.and_then(runtime_session_id_from_handle)
.or_else(|| {
address.and_then(|address| {
address
.get("runtimeHandle")
.or_else(|| address.get("runtime_handle"))
.and_then(runtime_session_id_from_handle)
})
})
.or_else(|| string_field(payload, "providerSessionId"))
.or_else(|| string_field(payload, "provider_session_id"))
.or_else(|| address.and_then(|address| string_field(address, "providerSessionId")))
.or_else(|| address.and_then(|address| string_field(address, "provider_session_id")))
}

fn runtime_session_id_from_handle(handle: &Value) -> Option<String> {
Expand Down
18 changes: 18 additions & 0 deletions executor/src/runtime_work/runtime_handle_messages.rs
Original file line number Diff line number Diff line change
Expand Up @@ -144,6 +144,7 @@ impl CodexNotificationCacheMapper {
}
self.agent_message_phases.forget_item(params);
}
"thread/started" => cache_runtime_thread_id(store, local_task_id, params),
"turn/completed" => complete_runtime_assistant_message(store, local_task_id, request),
_ => debug_ignored_codex_notification(message, &notification.method, params),
}
Expand Down Expand Up @@ -227,6 +228,23 @@ fn truncate_log_text(text: &str, max_chars: usize) -> String {
result
}

fn cache_runtime_thread_id(store: &RuntimeWorkStore, local_task_id: &str, params: &Value) {
let Some(thread_id) = params
.get("thread")
.and_then(|thread| string_field(thread, "id"))
.or_else(|| string_field(params, "threadId"))
.or_else(|| string_field(params, "thread_id"))
.filter(|value| !value.is_empty())
else {
return;
};

store.update_task(local_task_id, |link| {
link.thread_id = Some(thread_id);
link.updated_at = now_ms();
});
}

pub(crate) fn retain_runtime_handle_user_messages(runtime_handle: &mut Value) {
let messages = runtime_handle
.get("messages")
Expand Down
198 changes: 198 additions & 0 deletions executor/tests/app_runtime_work_send_contract.rs
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,7 @@ async fn runtime_tasks_send_accepts_address_content_source_and_attachments() {
.expect("create should be accepted");
assert_eq!(created["accepted"], true);
wait_for_thread_mapping(&handler, "local-task-1", "thread-1").await;
wait_until_task_idle(&handler, "local-task-1").await;
drain_events(&mut events);

let source = json!({
Expand Down Expand Up @@ -350,6 +351,7 @@ async fn runtime_tasks_send_includes_local_text_attachment_content() {
.await
.expect("create should be accepted");
wait_for_thread_mapping(&handler, "local-task-text", "thread-1").await;
wait_until_task_idle(&handler, "local-task-text").await;
drain_events(&mut events);

let attachment_path = temp_path("runtime-send-pasted-text", "txt");
Expand Down Expand Up @@ -504,6 +506,122 @@ async fn runtime_tasks_send_rejects_running_local_task_until_cancelled() {
assert_eq!(cancelled["accepted"], true);
}

#[tokio::test]
async fn runtime_tasks_send_after_cancel_resumes_started_thread_not_local_task_id() {
let _lock = env_lock().await;
let _home = EnvGuard::set(
"WEGENT_EXECUTOR_HOME",
&temp_path("runtime-send-guide-home", "dir")
.display()
.to_string(),
);
let _codex_home = EnvGuard::set(
"CODEX_HOME",
&temp_path("runtime-send-guide-codex-home", "dir")
.display()
.to_string(),
);
let log_path = temp_path("runtime-send-guide-log", "jsonl");
let fake_codex = write_fake_codex_slow_first_turn(&log_path);
let handler = RuntimeWorkRpcHandler::new("device-1", fake_codex.display().to_string());

handler
.handle_runtime_rpc(json!({
"method": "runtime.tasks.create",
"payload": {
"localTaskId": "local-visible-task",
"workspacePath": "/tmp/project",
"message": "first turn",
"modelId": "gpt-5.5"
}
}))
.await
.expect("create should be accepted");
wait_for_thread_mapping(&handler, "local-visible-task", "thread-1").await;

handler
.handle_runtime_rpc(json!({
"method": "runtime.tasks.cancel",
"payload": {
"workspacePath": "/tmp/project",
"localTaskId": "local-visible-task"
}
}))
.await
.expect("cancel should be accepted");

let sent = handler
.handle_runtime_rpc(json!({
"method": "runtime.tasks.send",
"payload": {
"workspacePath": "/tmp/project",
"localTaskId": "local-visible-task",
"message": "second turn",
"modelId": "gpt-5.5"
}
}))
.await
.expect("send should be accepted");
assert_eq!(sent["accepted"], true);

wait_for_method_count(&log_path, "thread/resume", 1).await;
let calls = read_json_lines(&log_path);
let resume = calls
.iter()
.rev()
.find(|call| call["method"] == "thread/resume")
.expect("send should resume a provider thread");
assert_eq!(resume["params"]["threadId"], "thread-1");
}

#[tokio::test]
async fn runtime_tasks_send_uses_nested_address_runtime_handle_without_local_index() {
let _lock = env_lock().await;
let _home = EnvGuard::set(
"WEGENT_EXECUTOR_HOME",
&temp_path("runtime-send-address-handle-home", "dir")
.display()
.to_string(),
);
let _codex_home = EnvGuard::set(
"CODEX_HOME",
&temp_path("runtime-send-address-handle-codex-home", "dir")
.display()
.to_string(),
);
let log_path = temp_path("runtime-send-address-handle-log", "jsonl");
let fake_codex = write_fake_codex(&log_path);
let handler = RuntimeWorkRpcHandler::new("device-1", fake_codex.display().to_string());

let sent = handler
.handle_runtime_rpc(json!({
"method": "runtime.tasks.send",
"payload": {
"address": {
"deviceId": "device-1",
"workspacePath": "/tmp/project",
"localTaskId": "local-visible-task",
"runtimeHandle": {
"threadId": "thread-1"
}
},
"content": "continue from address handle",
"modelId": "gpt-4.1"
}
}))
.await
.expect("send should be accepted");
assert_eq!(sent["accepted"], true);

wait_for_method_count(&log_path, "thread/resume", 1).await;
let calls = read_json_lines(&log_path);
let resume = calls
.iter()
.find(|call| call["method"] == "thread/resume")
.expect("send should resume the nested runtime handle");
assert_eq!(resume["params"]["threadId"], "thread-1");
}

fn write_fake_codex(log_path: &Path) -> PathBuf {
let path = temp_path("fake-codex-send", "sh");
let _ = fs::remove_file(log_path);
Expand Down Expand Up @@ -555,6 +673,51 @@ done
path
}

fn write_fake_codex_slow_first_turn(log_path: &Path) -> PathBuf {
let path = temp_path("fake-codex-send-guide", "sh");
let _ = fs::remove_file(log_path);
let content = format!(
r#"#!/bin/sh
LOG_PATH='{}'
while IFS= read -r line; do
printf '%s\n' "$line" >> "$LOG_PATH"
request_id=$(printf '%s\n' "$line" | sed -n 's/.*"id":\([0-9][0-9]*\).*/\1/p')
case "$line" in
*'"method":"initialize"'*)
printf '%s\n' '{{"id":'"$request_id"',"result":{{"protocolVersion":1}}}}'
;;
*'"method":"initialized"'*)
;;
*'"method":"thread/list"'*)
printf '%s\n' '{{"id":'"$request_id"',"result":{{"data":[{{"id":"thread-1","cwd":"/tmp/project","name":"Runtime task","preview":"runtime","path":"/tmp/codex/thread-1.jsonl","createdAt":1780000000,"updatedAt":1780000060,"status":"inProgress","turns":[]}}],"nextCursor":null,"backwardsCursor":null}}}}'
;;
*'"method":"thread/start"'*)
printf '%s\n' '{{"id":'"$request_id"',"result":{{"thread":{{"id":"thread-1"}}}}}}'
;;
*'"method":"thread/resume"'*)
printf '%s\n' '{{"id":'"$request_id"',"result":{{"thread":{{"id":"thread-1"}}}}}}'
;;
*'"method":"turn/start"'*)
printf '%s\n' '{{"id":'"$request_id"',"result":{{"turn":{{"id":"turn-1","status":"inProgress"}}}}}}'
case "$line" in
*'second turn'*)
printf '%s\n' '{{"method":"turn/completed","params":{{"turn":{{"id":"turn-2","status":"completed"}}}}}}'
exit 0
;;
esac
sleep 2
printf '%s\n' '{{"method":"turn/completed","params":{{"turn":{{"id":"turn-1","status":"completed"}}}}}}'
exit 0
;;
esac
done
"#,
log_path.display()
);
write_executable(&path, &content);
path
}

fn write_fake_codex_hanging_turn(log_path: &Path) -> PathBuf {
let path = temp_path("fake-codex-send-hang", "sh");
let _ = fs::remove_file(log_path);
Expand Down Expand Up @@ -718,6 +881,18 @@ async fn wait_for_thread_mapping(
panic!("runtime task mapping was not persisted");
}

async fn wait_for_method_count(log_path: &Path, method: &str, expected: usize) {
for _ in 0..200 {
let calls = read_json_lines(log_path);
let count = calls.iter().filter(|call| call["method"] == method).count();
if count >= expected {
return;
}
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
}
panic!("expected at least {expected} {method} calls");
}

async fn wait_until_task_running(handler: &RuntimeWorkRpcHandler, local_task_id: &str) {
for _ in 0..50 {
let listed = handler
Expand All @@ -741,6 +916,29 @@ async fn wait_until_task_running(handler: &RuntimeWorkRpcHandler, local_task_id:
panic!("runtime task did not become running");
}

async fn wait_until_task_idle(handler: &RuntimeWorkRpcHandler, local_task_id: &str) {
for _ in 0..50 {
let listed = handler
.handle_runtime_rpc(json!({
"method": "runtime.tasks.list",
"payload": {}
}))
.await
.expect("list should succeed");
let idle = listed["workspaces"]
.as_array()
.into_iter()
.flatten()
.flat_map(|workspace| workspace["localTasks"].as_array().into_iter().flatten())
.any(|task| task["localTaskId"] == local_task_id && task["running"] == false);
if idle {
return;
}
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
}
panic!("runtime task did not become idle");
}

async fn wait_for_turn_count(log_path: &Path, expected_turns: usize) {
for _ in 0..50 {
let count = read_json_lines(log_path)
Expand Down
45 changes: 45 additions & 0 deletions wework/src/api/local/localServices.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,51 @@ describe('createLocalAppServices', () => {
expect(request).toHaveBeenCalledWith('runtime.tasks.list', {})
})

test('normalizes runtime handles returned by local executor task lists', async () => {
const request = vi.fn().mockResolvedValue({
workspaces: [
{
workspace_path: '/Users/me/project',
local_tasks: [
{
local_task_id: 'local-visible-task',
workspace_path: '/Users/me/project',
title: 'Fix guidance',
runtime: 'codex',
runtime_handle: {
threadId: '019ee7f6-456a-78a1-96b1-66451afc310e',
},
},
],
},
],
})
const services = createLocalAppServices({
ensure: vi.fn().mockResolvedValue({ running: true, ready: true, deviceId: 'device-uuid' }),
request,
subscribe: vi.fn(),
})

await expect(services.runtimeWorkApi?.listRuntimeWork()).resolves.toMatchObject({
projects: [
{
deviceWorkspaces: [
{
localTasks: [
{
localTaskId: 'local-visible-task',
runtimeHandle: {
threadId: '019ee7f6-456a-78a1-96b1-66451afc310e',
},
},
],
},
],
},
],
})
})

test('keeps local device visible when executor startup fails', async () => {
const services = createLocalAppServices({
ensure: vi.fn().mockRejectedValue(new Error('sidecar missing')),
Expand Down
Loading
Loading