diff --git a/.github/workflows/wework-e2e.yml b/.github/workflows/wework-e2e.yml index 9bd06ce402..8a1ba14991 100644 --- a/.github/workflows/wework-e2e.yml +++ b/.github/workflows/wework-e2e.yml @@ -116,6 +116,7 @@ jobs: libayatana-appindicator3-dev \ libssl-dev \ libwebkit2gtk-4.1-dev \ + imagemagick \ librsvg2-dev \ xvfb diff --git a/executor/src/agents/codex.rs b/executor/src/agents/codex.rs index 8886d1ef84..832bbd2386 100644 --- a/executor/src/agents/codex.rs +++ b/executor/src/agents/codex.rs @@ -187,6 +187,87 @@ pub struct CodexAppServerTurn { pub type CodexRequestUserInputReceiver = mpsc::Receiver; +#[derive(Default)] +struct InteractionAnswerRouterState { + pending: HashMap>>, + buffered: HashMap, + closed: bool, +} + +struct InteractionAnswerRouter { + state: Arc>, +} + +impl InteractionAnswerRouter { + fn new(mut receiver: CodexRequestUserInputReceiver) -> Arc { + let router = Arc::new(Self { + state: Arc::new(Mutex::new(InteractionAnswerRouterState::default())), + }); + let state = router.state.clone(); + tokio::spawn(async move { + while let Some(answer) = receiver.recv().await { + let mut state = state.lock().await; + let key = interaction_answer_key(&answer).or_else(|| { + (state.pending.len() == 1) + .then(|| state.pending.keys().next().cloned()) + .flatten() + }); + let Some(key) = key else { + continue; + }; + if let Some(sender) = state.pending.remove(&key) { + let _ = sender.send(Ok(answer)); + } else { + state.buffered.insert(key, answer); + } + } + let mut state = state.lock().await; + state.closed = true; + for (_, sender) in state.pending.drain() { + let _ = sender.send(Err("request_user_input response channel closed".to_owned())); + } + }); + router + } + + async fn receive(&self, key: String) -> Result { + let receiver = { + let mut state = self.state.lock().await; + if let Some(answer) = state.buffered.remove(&key) { + return Ok(answer); + } + if state.closed { + return Err("request_user_input response router closed".to_owned()); + } + let (sender, receiver) = oneshot::channel(); + if state.pending.insert(key, sender).is_some() { + return Err("duplicate pending interaction correlation key".to_owned()); + } + receiver + }; + receiver + .await + .map_err(|_| "request_user_input response router closed".to_owned())? + } +} + +fn interaction_answer_key(answer: &Value) -> Option { + answer + .get("requestId") + .or_else(|| answer.get("request_id")) + .or_else(|| answer.get("itemId")) + .or_else(|| answer.get("item_id")) + .and_then(interaction_value_key) +} + +fn interaction_value_key(value: &Value) -> Option { + match value { + Value::String(value) if !value.trim().is_empty() => Some(value.clone()), + Value::Number(value) => Some(value.to_string()), + _ => None, + } +} + #[derive(Debug, Clone)] pub struct CodexAppServerEngine { binary: String, @@ -1302,20 +1383,29 @@ async fn read_shared_turn_notifications( ) -> Result { let mut last_outcome: Option = None; let mut waiting_for_initial_progress = true; + let request_user_input_answers = options + .request_user_input_answers + .take() + .map(InteractionAnswerRouter::new); + let (response_error_tx, mut response_error_rx) = mpsc::unbounded_channel(); loop { let receive_notification = async { if let Some(cancel_rx) = options.cancellation.as_mut() { tokio::select! { - _ = cancel_rx => None, - message = notification_rx.recv() => Some(message), + _ = cancel_rx => Ok(None), + error = response_error_rx.recv() => Err(error.unwrap_or_else(|| "request_user_input response task closed".to_owned())), + message = notification_rx.recv() => Ok(Some(message)), } } else { - Some(notification_rx.recv().await) + tokio::select! { + error = response_error_rx.recv() => Err(error.unwrap_or_else(|| "request_user_input response task closed".to_owned())), + message = notification_rx.recv() => Ok(Some(message)), + } } }; let received = if waiting_for_initial_progress { match timeout_at(startup_deadline, receive_notification).await { - Ok(received) => received, + Ok(received) => received?, Err(_) => { return Err(recover_stalled_shared_turn( client, @@ -1326,7 +1416,7 @@ async fn read_shared_turn_notifications( } } } else { - receive_notification.await + receive_notification.await? }; let Some(received) = received else { options.cancellation = None; @@ -1388,12 +1478,12 @@ async fn read_shared_turn_notifications( .and_then(Value::as_str) .is_some_and(|method| method == "item/tool/requestUserInput") { - answer_shared_request_user_input( + spawn_shared_request_user_input_response( client, &message, - &mut options.request_user_input_answers, - ) - .await?; + request_user_input_answers.clone(), + response_error_tx.clone(), + )?; continue; } @@ -1402,12 +1492,12 @@ async fn read_shared_turn_notifications( .and_then(Value::as_str) .is_some_and(|method| method == "mcpServer/elicitation/request") { - answer_shared_mcp_server_elicitation( + spawn_shared_mcp_server_elicitation_response( client, &message, - &mut options.request_user_input_answers, - ) - .await?; + request_user_input_answers.clone(), + response_error_tx.clone(), + )?; continue; } @@ -1518,35 +1608,62 @@ async fn interrupt_shared_turn( .map(|_| ()) } -async fn answer_shared_request_user_input( +fn spawn_shared_request_user_input_response( client: &CodexAppServerClient, message: &Value, - request_user_input_answers: &mut Option, + request_user_input_answers: Option>, + response_error_tx: mpsc::UnboundedSender, ) -> Result<(), String> { let request_id = json_rpc_request_id(message) .ok_or_else(|| "request_user_input message is missing JSON-RPC id".to_owned())?; let Some(receiver) = request_user_input_answers else { return Err("request_user_input requires a runtime response channel".to_owned()); }; - let response = receiver - .recv() - .await - .ok_or_else(|| "request_user_input response channel closed".to_owned())?; - client - .send_response(request_id, request_user_input_result(response)) - .await + let correlation_key = interaction_value_key(&request_id) + .ok_or_else(|| "request_user_input message has invalid JSON-RPC id".to_owned())?; + let client = client.clone(); + tokio::spawn(async move { + let result = async { + let response = receiver.receive(correlation_key).await?; + client + .send_response(request_id, request_user_input_result(response)) + .await + } + .await; + if let Err(error) = result { + let _ = response_error_tx.send(error); + } + }); + Ok(()) } -async fn answer_shared_mcp_server_elicitation( +fn spawn_shared_mcp_server_elicitation_response( client: &CodexAppServerClient, message: &Value, - request_user_input_answers: &mut Option, + request_user_input_answers: Option>, + response_error_tx: mpsc::UnboundedSender, ) -> Result<(), String> { let request_id = json_rpc_request_id(message) .ok_or_else(|| "mcpServer/elicitation/request is missing JSON-RPC id".to_owned())?; - let result = - receive_mcp_server_elicitation_response(message, request_user_input_answers).await?; - client.send_response(request_id, result).await + let correlation_key = interaction_value_key(&request_id) + .ok_or_else(|| "mcpServer/elicitation/request has invalid JSON-RPC id".to_owned())?; + let client = client.clone(); + let message = message.clone(); + tokio::spawn(async move { + let result = async { + let response = match request_user_input_answers { + Some(receiver) => Some(receiver.receive(correlation_key).await?), + None => None, + }; + let result = mcp_server_elicitation_response(&message, response.as_ref())?; + client.send_response(request_id, result).await + } + .await; + if let Err(error) = result { + let _ = response_error_tx.send(error); + } + }); + Ok(()) } async fn notification_belongs_to_thread( @@ -2066,7 +2183,8 @@ impl JsonRpcConnection { let request_id = json_rpc_request_id(message) .ok_or_else(|| "mcpServer/elicitation/request is missing JSON-RPC id".to_owned())?; let result = - receive_mcp_server_elicitation_response(message, request_user_input_answers).await?; + receive_mcp_server_elicitation_response(message, request_user_input_answers.as_mut()) + .await?; self.write_message(json!({ "id": request_id, "result": result, @@ -4446,7 +4564,24 @@ const MCP_ELICITATION_DECLINE: &str = "Decline"; async fn receive_mcp_server_elicitation_response( message: &Value, - request_user_input_answers: &mut Option, + request_user_input_answers: Option<&mut CodexRequestUserInputReceiver>, +) -> Result { + if mcp_server_elicitation_request_user_input_params(message_params(message)).is_none() { + return mcp_server_elicitation_response(message, None); + } + let Some(receiver) = request_user_input_answers else { + return mcp_server_elicitation_response(message, None); + }; + let response = receiver + .recv() + .await + .ok_or_else(|| "mcp elicitation response channel closed".to_owned())?; + mcp_server_elicitation_response(message, Some(&response)) +} + +fn mcp_server_elicitation_response( + message: &Value, + response: Option<&Value>, ) -> Result { let params = message_params(message); let mode = params.get("mode").and_then(Value::as_str).unwrap_or(""); @@ -4489,14 +4624,10 @@ async fn receive_mcp_server_elicitation_response( return Ok(mcp_server_elicitation_decline_result()); } - let Some(receiver) = request_user_input_answers else { + let Some(response) = response else { return Ok(mcp_server_elicitation_cancel_result()); }; - let response = receiver - .recv() - .await - .ok_or_else(|| "mcp elicitation response channel closed".to_owned())?; - let result = mcp_server_elicitation_result(params, &response); + let result = mcp_server_elicitation_result(params, response); log_executor_event( "codex mcp elicitation response", &[ @@ -4822,6 +4953,67 @@ mod tests { use super::*; + #[tokio::test] + async fn interaction_answer_router_matches_reverse_order_answers() { + let (sender, receiver) = mpsc::channel(2); + let router = InteractionAnswerRouter::new(receiver); + let first = { + let router = router.clone(); + tokio::spawn(async move { router.receive("41".to_owned()).await }) + }; + let second = { + let router = router.clone(); + tokio::spawn(async move { router.receive("42".to_owned()).await }) + }; + + sender + .send(json!({"requestId": 42, "answers": {"choice": "second"}})) + .await + .expect("second answer should be sent"); + sender + .send(json!({"requestId": 41, "answers": {"choice": "first"}})) + .await + .expect("first answer should be sent"); + + assert_eq!( + first.await.expect("first waiter should join").unwrap()["answers"]["choice"], + "first" + ); + assert_eq!( + second.await.expect("second waiter should join").unwrap()["answers"]["choice"], + "second" + ); + } + + #[tokio::test] + async fn interaction_answer_router_rejects_waiters_after_channel_closes() { + let (sender, receiver) = mpsc::channel(1); + let router = InteractionAnswerRouter::new(receiver); + drop(sender); + tokio::time::timeout(Duration::from_secs(1), async { + loop { + if router.state.lock().await.closed { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("router should observe the closed response channel"); + + let result = tokio::time::timeout( + Duration::from_secs(1), + router.receive("closed-request".to_owned()), + ) + .await + .expect("closed router should not leave the waiter pending"); + + assert_eq!( + result.unwrap_err(), + "request_user_input response router closed" + ); + } + #[test] fn normalize_reasoning_effort_preserves_supported_codex_levels() { for effort in ["low", "medium", "high", "xhigh", "max", "ultra"] { diff --git a/executor/tests/app_runtime_work_send_contract.rs b/executor/tests/app_runtime_work_send_contract.rs index f0e5e20881..c3bc7b6c83 100644 --- a/executor/tests/app_runtime_work_send_contract.rs +++ b/executor/tests/app_runtime_work_send_contract.rs @@ -1152,6 +1152,13 @@ async fn runtime_tasks_send_answers_pending_request_user_input_while_running() { block_event["payload"]["data"]["block"]["render_payload"]["questions"][0]["id"], "goal" ); + recv_events_until(&mut events, |runtime_events| { + find_runtime_event(runtime_events, "response.block.created", |event| { + event["payload"]["data"]["block"]["id"] == "flood-sentinel" + }) + .is_some() + }) + .await; let sent = handler .handle_runtime_rpc(json!({ @@ -1178,10 +1185,10 @@ async fn runtime_tasks_send_answers_pending_request_user_input_while_running() { assert_eq!(sent["accepted"], true); wait_until_task_idle(&handler, "local-task-input").await; - let calls = read_json_lines(&log_path); - assert!(calls.iter().any(|call| { + wait_for_json_call(&log_path, |call| { call["id"] == 99 && call["result"]["answers"]["goal"]["answers"][0] == "Work goal" - })); + }) + .await; } #[tokio::test] @@ -2170,6 +2177,16 @@ while IFS= read -r line; do elif printf '%s\n' "$line" | grep -q '"method":"turn/start"'; then printf '%s\n' '{{"id":'"$request_id"',"result":{{"turn":{{"id":"turn-input","status":"inProgress"}}}}}}' printf '%s\n' '{{"id":99,"method":"item/tool/requestUserInput","params":{{"threadId":"thread-input","turnId":"turn-input","itemId":"item-input","questions":[{{"id":"goal","header":"工作目标","question":"你希望我接下来问你哪些问题？","options":[{{"label":"Work goal","description":"Focus on one concrete task."}}]}}],"autoResolutionMs":null}}}}' + sleep 0.1 + notification_index=0 + while [ "$notification_index" -lt 2200 ]; do + printf '%s\n' '{{"method":"thread/name/updated","params":{{"threadId":"thread-noise","name":"waiting"}}}}' + notification_index=$((notification_index + 1)) + if [ $((notification_index % 50)) -eq 0 ]; then + sleep 0.002 + fi + done + printf '%s\n' '{{"method":"item/started","params":{{"threadId":"thread-input","turnId":"turn-input","item":{{"id":"flood-sentinel","type":"commandExecution","status":"inProgress","command":"sentinel"}}}}}}' elif printf '%s\n' "$line" | grep -q '"id":99' && printf '%s\n' "$line" | grep -q '"result"'; then printf '%s\n' '{{"method":"item/agentMessage/delta","params":{{"delta":"answered","phase":"finalAnswer"}}}}' printf '%s\n' '{{"method":"turn/completed","params":{{"turn":{{"id":"turn-input","status":"completed"}}}}}}' @@ -2628,6 +2645,19 @@ async fn wait_for_turn_count(log_path: &Path, expected_turns: usize) { panic!("expected at least {expected_turns} turn/start calls"); } +async fn wait_for_json_call(log_path: &Path, matches: F) +where + F: Fn(&Value) -> bool, +{ + for _ in 0..50 { + if read_json_lines(log_path).iter().any(&matches) { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + panic!("expected JSON-RPC call was not recorded"); +} + fn drain_events(events: &mut broadcast::Receiver) { while events.try_recv().is_ok() {} } diff --git a/wework/e2e/desktop/task-flow.e2e.mjs b/wework/e2e/desktop/task-flow.e2e.mjs index ea0808e6b2..4245464cd1 100644 --- a/wework/e2e/desktop/task-flow.e2e.mjs +++ b/wework/e2e/desktop/task-flow.e2e.mjs @@ -17,6 +17,10 @@ const TASK_PROMPT = 'WEWORK_DESKTOP_E2E_TASK: create the requested verification const COMPLETION_TEXT = 'WEWORK_DESKTOP_E2E_COMPLETE' const FOLLOW_UP_PROMPT = 'WEWORK_DESKTOP_E2E_FOLLOW_UP: confirm the completed task.' const FOLLOW_UP_COMPLETION_TEXT = 'WEWORK_DESKTOP_E2E_FOLLOW_UP_COMPLETE' +const REQUEST_USER_INPUT_PROMPT = + 'WEWORK_DESKTOP_E2E_REQUEST_INPUT: ask which implementation direction to use.' +const REQUEST_USER_INPUT_QUESTION = 'Which implementation direction should be used?' +const REQUEST_USER_INPUT_COMPLETION_TEXT = 'WEWORK_DESKTOP_E2E_REQUEST_INPUT_COMPLETE' const CANCELLATION_PROMPT = 'WEWORK_DESKTOP_E2E_CANCEL: wait until the response is cancelled.' const CANCELLATION_COMPLETION_TEXT = 'WEWORK_DESKTOP_E2E_CANCEL_COMPLETE' const RETRY_PROMPT = 'WEWORK_DESKTOP_E2E_RETRY: fail once and then succeed after retry.' @@ -38,6 +42,7 @@ const FRESH_CHAT_COMPLETION_TEXT = 'WEWORK_DESKTOP_E2E_FRESH_CHAT_COMPLETE' const ACTIVE_WORKBENCH_SELECTOR = '[data-testid="desktop-workbench-main"]' const ACTIVE_COMPOSER_SELECTOR = `${ACTIVE_WORKBENCH_SELECTOR} [data-testid="chat-message-input"][contenteditable="true"]` const ACTIVE_SEND_BUTTON_SELECTOR = `${ACTIVE_WORKBENCH_SELECTOR} [data-testid="send-message-button"]` +const REQUEST_INPUT_ONLY = process.env.WEWORK_DESKTOP_E2E_REQUEST_INPUT_ONLY === '1' const scriptDir = dirname(fileURLToPath(import.meta.url)) const weworkDir = resolve(scriptDir, '..', '..') @@ -157,6 +162,19 @@ async function waitForSnapshot(control, predicate, message, timeoutMs = UI_TIMEO throw new Error(message) } +async function captureVerificationScreenshot(control, name) { + const screenshotPath = join(resultDir, name) + if (process.platform === 'linux') { + await runChecked('import', ['-window', 'root', screenshotPath]) + return screenshotPath + } + const dataUrl = await control.command('capture', 'body') + const prefix = 'data:image/png;base64,' + assert.ok(dataUrl.startsWith(prefix), 'Desktop screenshot did not return PNG data') + await writeFile(screenshotPath, Buffer.from(dataUrl.slice(prefix.length), 'base64')) + return screenshotPath +} + async function triggerModelReloadUntilCloudFailure(control) { const failedCloudModelRequest = control.awaitFailedCloudModelRequest() for (let attempt = 0; attempt < 10 && control.failedCloudModelRequests === 0; attempt += 1) { @@ -174,33 +192,29 @@ async function triggerModelReloadUntilCloudFailure(control) { } async function sendPromptUntilScenarioRequest(control, selector, prompt, scenario) { - let lastError - for (let attempt = 0; attempt < 5; attempt += 1) { - await sendPrompt(control, selector, prompt) - try { - return await withTimeout( - control.awaitScenarioRequest(scenario), - 2_000, - `The model service did not receive the ${scenario} request` - ) - } catch (error) { - lastError = error - await new Promise(resolvePromise => setTimeout(resolvePromise, 250)) - } - } - throw lastError + const scenarioRequest = control.awaitScenarioRequest(scenario) + await sendPrompt(control, selector, prompt) + return withTimeout( + scenarioRequest, + UI_TIMEOUT_MS, + `The model service did not receive the ${scenario} request` + ) } async function selectE2EModel(control, modelId = MODEL_ID, modelLabel = MODEL_LABEL) { - await control.command('waitFor', '[data-testid="model-selector-button"]', { - timeoutMs: WORKBENCH_READY_TIMEOUT_MS, - }) + const selectedModelText = await control.command( + 'waitFor', + '[data-testid="model-selector-button"]', + { + timeoutMs: WORKBENCH_READY_TIMEOUT_MS, + } + ) + if (selectedModelText.includes(modelLabel)) return await control.command('clickWhenEnabled', '[data-testid="model-selector-button"]', { stableMs: COMPOSER_READY_STABILITY_MS, timeoutMs: UI_TIMEOUT_MS, }) - await control.command('clickWhenEnabled', '[data-testid="model-control-menu-model"]', { - stableMs: COMPOSER_READY_STABILITY_MS, + await control.command('hover', '[data-testid="model-control-menu-model"]', { timeoutMs: UI_TIMEOUT_MS, }) await control.command('waitFor', `[data-testid="model-option-${modelId}"]`, { @@ -382,6 +396,12 @@ class DesktopE2EServer { this.retryCompletionRelease = new Promise(resolvePromise => { this.releaseRetryCompletion = resolvePromise }) + this.requestUserInputRelease = new Promise(resolvePromise => { + this.releaseRequestUserInput = resolvePromise + }) + this.requestUserInputResponseWritten = new Promise(resolvePromise => { + this.resolveRequestUserInputResponseWritten = resolvePromise + }) this.scenarioRequests = new Map() this.scenarioWaiters = new Map() } @@ -459,7 +479,14 @@ class DesktopE2EServer { setScenario(scenario) { assert.ok( - ['initial', 'follow_up', 'cancellation', 'retry', 'fresh_chat'].includes(scenario), + [ + 'initial', + 'follow_up', + 'request_user_input', + 'cancellation', + 'retry', + 'fresh_chat', + ].includes(scenario), `Unknown desktop E2E scenario: ${scenario}` ) this.scenario = scenario @@ -492,6 +519,11 @@ class DesktopE2EServer { this.releaseRetryCompletion() } + releaseRequestUserInputResponse() { + this.releaseRequestUserInput() + return this.requestUserInputResponseWritten + } + async command(action, selector, options = {}) { const id = randomUUID() const command = { id, action, selector, ...options } @@ -688,6 +720,43 @@ class DesktopE2EServer { return } + if (this.scenario === 'request_user_input') { + this.recordScenarioRequest('request_user_input', modelRequest) + if (JSON.stringify(body.input).includes('wework-e2e-request-user-input')) { + this.writeSse(response, [ + responseCreated(responseId), + assistantMessage(REQUEST_USER_INPUT_COMPLETION_TEXT), + responseCompleted(responseId), + ]) + return + } + assert.ok( + JSON.stringify(body).includes(REQUEST_USER_INPUT_PROMPT), + 'The real Codex request did not contain the request-user-input prompt' + ) + const tool = selectTool(body, 'request_user_input', { + questions: [ + { + header: 'Direction', + id: 'direction', + question: REQUEST_USER_INPUT_QUESTION, + options: [ + { label: 'Minimal', description: 'Make the smallest focused change.' }, + { label: 'Complete', description: 'Cover the full interaction flow.' }, + ], + }, + ], + }) + await this.requestUserInputRelease + this.writeSse(response, [ + responseCreated(responseId), + functionCall('wework-e2e-request-user-input', tool.name, tool.arguments), + responseCompleted(responseId), + ]) + this.resolveRequestUserInputResponseWritten() + return + } + if (this.scenario === 'fresh_chat') { this.recordScenarioRequest('fresh_chat', modelRequest) assert.ok(JSON.stringify(body).includes(FRESH_CHAT_PROMPT), 'The fresh chat prompt was lost') @@ -898,6 +967,9 @@ async function main() { WEGENT_EXECUTOR_LOG_DIR: resultDir, WEGENT_EXECUTOR_LOG_FILE: 'executor.log', DEVICE_ID: `wework-e2e-device-${process.pid}`, + DEVICE_SESSION_GATEWAY_HOST: '127.0.0.1', + DEVICE_SESSION_GATEWAY_PORT: '0', + VITE_WEWORK_E2E: 'true', WEWORK_E2E_MODEL_API_KEY: MODEL_API_KEY, WEWORK_EMBEDDED_BROWSER_BRIDGE_ADDR: '127.0.0.1:0', WEWORK_EXECUTOR_SIDECAR: executorBinary, @@ -929,7 +1001,6 @@ async function main() { await control.command('waitFor', '[data-testid="projects-create-button"]', { timeoutMs: WORKBENCH_READY_TIMEOUT_MS, }) - await selectE2EModel(control) control.failBlockedCloudModels() await triggerModelReloadUntilCloudFailure(control) @@ -1114,33 +1185,81 @@ async function main() { testId.startsWith('runtime-local-task-row-') ) assert.ok(taskRowTestId, 'The completed task row was not found') + if (!REQUEST_INPUT_ONLY) { + await control.command('click', '[data-testid="new-chat-button"]') + await control.command('waitFor', composerSelector, { + timeoutMs: WORKBENCH_READY_TIMEOUT_MS, + }) + await selectE2EModel(control, DEFAULT_MODEL_ID, DEFAULT_MODEL_LABEL) + await control.command('click', `[data-testid="${taskRowTestId}"]`) + await control.command('waitFor', '[data-testid="model-selector-button"]', { + text: MODEL_LABEL, + timeoutMs: UI_TIMEOUT_MS, + }) + + phase = 'follow-up' + control.setScenario('follow_up') + const followUpRequest = await sendPromptUntilScenarioRequest( + control, + composerSelector, + FOLLOW_UP_PROMPT, + 'follow_up' + ) + await control.command('waitFor', '[data-testid="message-assistant"]', { + text: FOLLOW_UP_COMPLETION_TEXT, + timeoutMs: UI_TIMEOUT_MS, + }) + assert.ok( + JSON.stringify(followUpRequest.body).includes(FOLLOW_UP_PROMPT), + 'The follow-up request did not preserve the user prompt' + ) + } + + phase = 'background-request-user-input' + control.setScenario('request_user_input') + await control.command('click', '[data-testid="add-context-button"]') + await control.command('click', '[data-testid="set-plan-mode-button"]') + await control.command('waitFor', '[data-testid="plan-mode-pill"]', { + timeoutMs: UI_TIMEOUT_MS, + }) + await sendPromptUntilScenarioRequest( + control, + composerSelector, + REQUEST_USER_INPUT_PROMPT, + 'request_user_input' + ) await control.command('click', '[data-testid="new-chat-button"]') await control.command('waitFor', composerSelector, { timeoutMs: WORKBENCH_READY_TIMEOUT_MS, }) - await selectE2EModel(control, DEFAULT_MODEL_ID, DEFAULT_MODEL_LABEL) + await control.command('click', composerSelector) + await control.command('press', 'body', { key: 'Escape' }) + await captureVerificationScreenshot(control, '01-request-running-in-background.png') + await withTimeout( + control.releaseRequestUserInputResponse(), + UI_TIMEOUT_MS, + 'Timed out waiting for the request-user-input SSE response' + ) + await control.command('press', 'body', { key: 'Escape' }) await control.command('click', `[data-testid="${taskRowTestId}"]`) - await control.command('waitFor', '[data-testid="model-selector-button"]', { - text: MODEL_LABEL, + await control.command('waitFor', '[data-testid="request-user-input-card"]', { + text: REQUEST_USER_INPUT_QUESTION, + visible: true, + stableMs: COMPOSER_READY_STABILITY_MS, timeoutMs: UI_TIMEOUT_MS, }) - - phase = 'follow-up' - control.setScenario('follow_up') - await sendPrompt(control, composerSelector, FOLLOW_UP_PROMPT) + await captureVerificationScreenshot(control, '02-background-request-user-input-visible.png') + await new Promise(resolvePromise => setTimeout(resolvePromise, 3_000)) + await control.command('click', '[data-testid="request-user-input-option-direction-1"]') await control.command('waitFor', '[data-testid="message-assistant"]', { - text: FOLLOW_UP_COMPLETION_TEXT, + text: REQUEST_USER_INPUT_COMPLETION_TEXT, + visible: true, + stableMs: COMPOSER_READY_STABILITY_MS, timeoutMs: UI_TIMEOUT_MS, }) - const followUpRequest = await withTimeout( - control.awaitScenarioRequest('follow_up'), - UI_TIMEOUT_MS, - 'The model service did not receive the follow-up request' - ) - assert.ok( - JSON.stringify(followUpRequest.body).includes(FOLLOW_UP_PROMPT), - 'The follow-up request did not preserve the user prompt' - ) + await captureVerificationScreenshot(control, '03-delayed-answer-completed.png') + await control.command('click', '[data-testid="cancel-plan-mode-button"]') + if (REQUEST_INPUT_ONLY) return phase = 'cancellation' control.setScenario('cancellation') diff --git a/wework/scripts/ai-verify-environment.mjs b/wework/scripts/ai-verify-environment.mjs index 71c365b7e7..62ff91ddb9 100644 --- a/wework/scripts/ai-verify-environment.mjs +++ b/wework/scripts/ai-verify-environment.mjs @@ -12,6 +12,8 @@ export function buildAiVerifyEnvironment( CODEX_HOME: codexHome, WEGENT_CODEX_HOME: codexHome, DEVICE_ID: deviceId, + DEVICE_SESSION_GATEWAY_HOST: '127.0.0.1', + DEVICE_SESSION_GATEWAY_PORT: '0', WEGENT_EXECUTOR_APP_IPC_SOCKET: socketPath, WEGENT_EXECUTOR_HOME: executorHome, WEGENT_EXECUTOR_PROJECTS_DIR: join(executorHome, 'workspace', 'projects'), diff --git a/wework/src/api/hybrid/hybridServices.ts b/wework/src/api/hybrid/hybridServices.ts index ff5c217345..14fb6c3281 100644 --- a/wework/src/api/hybrid/hybridServices.ts +++ b/wework/src/api/hybrid/hybridServices.ts @@ -2,7 +2,7 @@ import { createBackendWorkbenchServices } from '@/api/backend/backendServices' import { createCloudRuntimeIpcClient } from '@/api/backend/runtimeIpc' import { createExecutorClientFromApis } from '@/api/executorAccess' import { createLocalAppServices, createRuntimeWorkApiFromIpc } from '@/api/local/localServices' -import { createLocalChatStream } from '@/api/local/localChatStream' +import { createRuntimeChatStream } from '@/api/runtime/runtimeChatStream' import type { WorkbenchServices } from '@/features/workbench/workbenchServices' import { notifyWorkbenchCloudArchivesChanged, @@ -870,16 +870,17 @@ export function createHybridWorkbenchServices( }, } + const cloudRuntimeChatStream = createRuntimeChatStream({ + request: (method, params) => { + const deviceId = cloudDeviceIdFromData(params) + return cloudRuntimeIpc.request(method, params, deviceId) + }, + subscribe: cloudRuntimeIpc.subscribe, + }) const hybridChatStream: WorkbenchServices['chatStream'] = { subscribe(handlers) { const cleanupLocal = localServices.chatStream.subscribe(handlers) - const cleanupCloudRuntime = createLocalChatStream({ - request: (method, params) => { - const deviceId = cloudDeviceIdFromData(params) - return cloudRuntimeIpc.request(method, params, deviceId) - }, - subscribe: cloudRuntimeIpc.subscribe, - }).subscribe(handlers) + const cleanupCloudRuntime = cloudRuntimeChatStream.subscribe(handlers) const cleanupCloudDeviceEvents = cloudServices.chatStream.subscribe({ onDeviceOnline: handlers.onDeviceOnline, onDeviceOffline: handlers.onDeviceOffline, diff --git a/wework/src/api/local/localServices.ts b/wework/src/api/local/localServices.ts index 19ff1fd249..84389d4051 100644 --- a/wework/src/api/local/localServices.ts +++ b/wework/src/api/local/localServices.ts @@ -100,7 +100,7 @@ import { type LocalModelConfig, } from '@/features/model-settings/localModelSettings' import { getLocalProxyUrl } from '@/features/model-settings/localProxySettings' -import { createLocalChatStream } from './localChatStream' +import { createRuntimeChatStream } from '../runtime/runtimeChatStream' import { createLocalAttachmentApi } from './localAttachments' import { LOCAL_USER, saveLocalUserPreferences } from './localSession' import type { KeybindingOverride } from '@/lib/keybindings' @@ -2212,6 +2212,6 @@ export function createLocalAppServices(deps: LocalAppServicesDeps = {}): Workben uploadRuntimeAuthJson: () => cloudConnectionRequired('uploadRuntimeAuthJson'), importRuntimeAuthJson: () => cloudConnectionRequired('importRuntimeAuthJson'), }, - chatStream: createLocalChatStream({ subscribe, request }), + chatStream: createRuntimeChatStream({ subscribe, request }), } as unknown as WorkbenchServices } diff --git a/wework/src/api/local/localChatStream.test.ts b/wework/src/api/runtime/runtimeChatStream.test.ts similarity index 84% rename from wework/src/api/local/localChatStream.test.ts rename to wework/src/api/runtime/runtimeChatStream.test.ts index 69c9a5106e..fb46a2db8f 100644 --- a/wework/src/api/local/localChatStream.test.ts +++ b/wework/src/api/runtime/runtimeChatStream.test.ts @@ -1,13 +1,14 @@ import { beforeEach, describe, expect, test, vi } from 'vitest' import type { LocalExecutorEvent } from '@/tauri/localExecutor' -import { createLocalChatStream, setLocalChatStreamDebugEnabled } from './localChatStream' +import { createRuntimeChatStream, setRuntimeChatStreamDebugEnabled } from './runtimeChatStream' -describe('createLocalChatStream', () => { +describe('createRuntimeChatStream', () => { const subscribe = vi.fn() const request = vi.fn() beforeEach(() => { subscribe.mockReset() + subscribe.mockResolvedValue(vi.fn()) request.mockReset() localStorage.clear() }) @@ -19,7 +20,7 @@ describe('createLocalChatStream', () => { return vi.fn() }) const onChatChunk = vi.fn() - const stream = createLocalChatStream({ subscribe, request }) + const stream = createRuntimeChatStream({ subscribe, request }) stream.subscribe({ onChatChunk }) await Promise.resolve() @@ -50,7 +51,7 @@ describe('createLocalChatStream', () => { return vi.fn() }) const onRuntimePlanUpdated = vi.fn() - const stream = createLocalChatStream({ subscribe, request }) + const stream = createRuntimeChatStream({ subscribe, request }) stream.subscribe({ scope: { deviceId: 'local-device', taskId: 'previous-task' }, @@ -87,7 +88,7 @@ describe('createLocalChatStream', () => { return vi.fn() }) const consoleDebug = vi.spyOn(console, 'debug').mockImplementation(() => undefined) - const stream = createLocalChatStream({ subscribe, request }) + const stream = createRuntimeChatStream({ subscribe, request }) stream.subscribe({ onChatChunk: vi.fn() }) await Promise.resolve() @@ -102,7 +103,7 @@ describe('createLocalChatStream', () => { }) expect( - consoleDebug.mock.calls.some(call => call[0] === '[Wework] Local chat stream event') + consoleDebug.mock.calls.some(call => call[0] === '[Wework] Runtime chat stream event') ).toBe(false) consoleDebug.mockRestore() @@ -115,7 +116,7 @@ describe('createLocalChatStream', () => { return vi.fn() }) const consoleDebug = vi.spyOn(console, 'debug').mockImplementation(() => undefined) - const stream = createLocalChatStream({ subscribe, request }) + const stream = createRuntimeChatStream({ subscribe, request }) stream.subscribe({ onBlockUpdated: vi.fn() }) await Promise.resolve() @@ -130,7 +131,7 @@ describe('createLocalChatStream', () => { }) expect( - consoleDebug.mock.calls.some(call => call[0] === '[Wework] Local chat stream event') + consoleDebug.mock.calls.some(call => call[0] === '[Wework] Runtime chat stream event') ).toBe(false) consoleDebug.mockRestore() @@ -139,21 +140,21 @@ describe('createLocalChatStream', () => { test('logs local stream lifecycle only when stream debug is enabled', async () => { subscribe.mockImplementation(async () => vi.fn()) const consoleDebug = vi.spyOn(console, 'debug').mockImplementation(() => undefined) - const stream = createLocalChatStream({ subscribe, request }) + const stream = createRuntimeChatStream({ subscribe, request }) const cleanupWithoutDebug = stream.subscribe({ onChatChunk: vi.fn() }) cleanupWithoutDebug() expect(consoleDebug).not.toHaveBeenCalledWith( - '[Wework] Local chat stream subscription', + '[Wework] Runtime chat stream subscription', expect.anything() ) - setLocalChatStreamDebugEnabled(true) + setRuntimeChatStreamDebugEnabled(true) const cleanupWithDebug = stream.subscribe({ onChatChunk: vi.fn() }) cleanupWithDebug() expect(consoleDebug).toHaveBeenCalledWith( - '[Wework] Local chat stream subscription', + '[Wework] Runtime chat stream subscription', expect.objectContaining({ action: 'subscribed' }) ) @@ -168,7 +169,7 @@ describe('createLocalChatStream', () => { }) const onChatDone = vi.fn() const consoleInfo = vi.spyOn(console, 'info').mockImplementation(() => undefined) - const stream = createLocalChatStream({ subscribe, request }) + const stream = createRuntimeChatStream({ subscribe, request }) stream.subscribe({ onChatDone }) await Promise.resolve() @@ -189,7 +190,7 @@ describe('createLocalChatStream', () => { result: { value: 'complete' }, }) expect(consoleInfo).toHaveBeenCalledWith( - '[Wework] Local chat stream terminal event received', + '[Wework] Runtime chat stream terminal event received', expect.objectContaining({ event: 'response.completed', taskId: 'task-1', @@ -199,14 +200,14 @@ describe('createLocalChatStream', () => { consoleInfo.mockRestore() }) - test('does not open a local executor listener for device-only subscriptions', () => { - const stream = createLocalChatStream({ subscribe, request }) + test('opens the local executor listener before a task pane subscribes', () => { + const stream = createRuntimeChatStream({ subscribe, request }) const cleanup = stream.subscribe({ onDeviceStatus: vi.fn() }) - expect(subscribe).not.toHaveBeenCalled() + expect(subscribe).toHaveBeenCalledTimes(1) cleanup() - expect(subscribe).not.toHaveBeenCalled() + expect(subscribe).toHaveBeenCalledTimes(1) }) test('shares one native event listener across multiple stream subscribers', async () => { @@ -218,7 +219,7 @@ describe('createLocalChatStream', () => { }) const firstChunk = vi.fn() const secondChunk = vi.fn() - const stream = createLocalChatStream({ subscribe, request }) + const stream = createRuntimeChatStream({ subscribe, request }) const cleanupFirst = stream.subscribe({ onChatChunk: firstChunk }) const cleanupSecond = stream.subscribe({ onChatChunk: secondChunk }) @@ -241,7 +242,7 @@ describe('createLocalChatStream', () => { cleanupFirst() expect(unlisten).not.toHaveBeenCalled() cleanupSecond() - expect(unlisten).toHaveBeenCalledTimes(1) + expect(unlisten).not.toHaveBeenCalled() }) test('routes scoped events only to matching stream subscribers', async () => { @@ -252,7 +253,7 @@ describe('createLocalChatStream', () => { }) const firstChunk = vi.fn() const secondChunk = vi.fn() - const stream = createLocalChatStream({ subscribe, request }) + const stream = createRuntimeChatStream({ subscribe, request }) stream.subscribe({ scope: { deviceId: 'local-device', taskId: 'task-1' }, @@ -278,7 +279,7 @@ describe('createLocalChatStream', () => { expect(secondChunk).not.toHaveBeenCalled() }) - test('cleans up a late native listener when all subscribers release before it is ready', async () => { + test('keeps a late native listener active when no pane is subscribed', async () => { let resolveSubscribe!: (unlisten: () => void) => void const unlisten = vi.fn() subscribe.mockImplementation( @@ -287,20 +288,20 @@ describe('createLocalChatStream', () => { resolveSubscribe = resolve }) ) - const stream = createLocalChatStream({ subscribe, request }) + const stream = createRuntimeChatStream({ subscribe, request }) const cleanup = stream.subscribe({ onChatChunk: vi.fn() }) cleanup() resolveSubscribe(unlisten) await Promise.resolve() - expect(unlisten).toHaveBeenCalledTimes(1) + expect(unlisten).not.toHaveBeenCalled() }) test('routes guidance and cancel requests through app ipc', async () => { request.mockResolvedValueOnce({ success: true, guidance_id: 'guide-1' }) request.mockResolvedValueOnce({ success: true }) - const stream = createLocalChatStream({ subscribe, request }) + const stream = createRuntimeChatStream({ subscribe, request }) await expect( stream.sendGuidance({ diff --git a/wework/src/api/local/localChatStream.ts b/wework/src/api/runtime/runtimeChatStream.ts similarity index 71% rename from wework/src/api/local/localChatStream.ts rename to wework/src/api/runtime/runtimeChatStream.ts index c381a7f4c5..6c241164b3 100644 --- a/wework/src/api/local/localChatStream.ts +++ b/wework/src/api/runtime/runtimeChatStream.ts @@ -7,28 +7,28 @@ import { RESPONSE_API_STREAM_EVENTS, } from '@/stream/responseApiStream' -interface LocalChatStreamDeps { +interface RuntimeChatStreamDeps { subscribe: (handler: (event: LocalExecutorEvent) => void) => Promise<() => void> request: (method: string, params?: Record) => Promise } -let nextLocalChatStreamSubscriptionId = 1 -let activeLocalChatStreamSubscriptions = 0 -const LOCAL_CHAT_STREAM_DEBUG_STORAGE_KEY = 'wework:debug-local-chat-stream' +let nextRuntimeChatStreamSubscriptionId = 1 +let activeRuntimeChatStreamSubscriptions = 0 +const RUNTIME_CHAT_STREAM_DEBUG_STORAGE_KEY = 'wework:debug-runtime-chat-stream' -export function isLocalChatStreamDebugEnabled(): boolean { - return globalThis.localStorage?.getItem(LOCAL_CHAT_STREAM_DEBUG_STORAGE_KEY) === '1' +export function isRuntimeChatStreamDebugEnabled(): boolean { + return globalThis.localStorage?.getItem(RUNTIME_CHAT_STREAM_DEBUG_STORAGE_KEY) === '1' } -export function setLocalChatStreamDebugEnabled(enabled: boolean): void { +export function setRuntimeChatStreamDebugEnabled(enabled: boolean): void { if (enabled) { - globalThis.localStorage?.setItem(LOCAL_CHAT_STREAM_DEBUG_STORAGE_KEY, '1') + globalThis.localStorage?.setItem(RUNTIME_CHAT_STREAM_DEBUG_STORAGE_KEY, '1') return } - globalThis.localStorage?.removeItem(LOCAL_CHAT_STREAM_DEBUG_STORAGE_KEY) + globalThis.localStorage?.removeItem(RUNTIME_CHAT_STREAM_DEBUG_STORAGE_KEY) } -export function createLocalChatStream(deps: LocalChatStreamDeps) { +export function createRuntimeChatStream(deps: RuntimeChatStreamDeps) { const subscriptions = new Map< number, { @@ -40,23 +40,22 @@ export function createLocalChatStream(deps: LocalChatStreamDeps) { >() let nativeCleanup: (() => void) | null = null let nativeSubscribePromise: Promise | null = null - function ensureNativeListener(): void { if (nativeCleanup || nativeSubscribePromise) return - nativeSubscribePromise = deps - .subscribe(event => { + nativeSubscribePromise = Promise.resolve( + deps.subscribe(event => { if (import.meta.env.DEV && event.event === 'runtime.plan.updated') { - console.warn('[Wework] Local runtime task plan event received', { + console.warn('[Wework] Runtime task plan event received', { taskId: stringField(asRecord(event.payload), 'taskId') ?? null, deviceId: stringField(asRecord(event.payload), 'deviceId') ?? null, }) } const subscriptionEntries = Array.from(subscriptions) - if (shouldLogLocalChatStreamEvent(event.event)) { + if (shouldLogRuntimeChatStreamEvent(event.event)) { const matchedSubscriptionCount = subscriptionEntries.filter(([, subscription]) => isLocalExecutorEventInScope(event, subscription.handlers.scope) ).length - logLocalChatTerminalEvent(event, matchedSubscriptionCount, subscriptionEntries.length) + logRuntimeChatTerminalEvent(event, matchedSubscriptionCount, subscriptionEntries.length) } for (const [subscriptionId, subscription] of subscriptionEntries) { if (event.event === 'executor.runtime_replaced') { @@ -72,18 +71,15 @@ export function createLocalChatStream(deps: LocalChatStreamDeps) { continue } if (!inScope && import.meta.env.DEV) { - console.warn( - '[Wework] Local runtime task plan event cached outside subscription scope', - { - subscriptionId, - } - ) + console.warn('[Wework] Runtime task plan event forwarded outside subscription scope', { + subscriptionId, + }) } subscription.eventCount += 1 if (event.event === 'response.output_text.delta') { subscription.textDeltaCount += 1 } - logLocalChatStreamEvent( + logRuntimeChatStreamEvent( subscriptionId, event, subscription.eventCount, @@ -100,35 +96,27 @@ export function createLocalChatStream(deps: LocalChatStreamDeps) { globalThis.dispatchEvent(new Event('wework-runtime-plan-updated')) } }) + ) .then(unlisten => { nativeSubscribePromise = null - if (subscriptions.size === 0) { - logLocalChatStreamNativeSubscription('late-native-listener-cleanup') - unlisten() - return - } nativeCleanup = unlisten - logLocalChatStreamNativeSubscription('native-listener-ready') + logRuntimeChatStreamNativeSubscription('native-listener-ready') }) .catch(error => { nativeSubscribePromise = null - console.error('[Wework] Local chat stream native listener failed', { + console.error('[Wework] Runtime chat stream native listener failed', { error: error instanceof Error ? error.message : String(error), - activeSubscriptions: activeLocalChatStreamSubscriptions, + activeSubscriptions: activeRuntimeChatStreamSubscriptions, }) - logLocalChatStreamNativeSubscription('native-listener-failed', { + logRuntimeChatStreamNativeSubscription('native-listener-failed', { error: error instanceof Error ? error.message : String(error), }) }) } - function releaseNativeListenerIfIdle(): void { - if (subscriptions.size > 0) return - if (!nativeCleanup) return - nativeCleanup() - nativeCleanup = null - logLocalChatStreamNativeSubscription('native-listener-released') - } + // Start listening before a task pane exists. Local task creation and the + // first tool event can otherwise race the pane's asynchronous subscription. + ensureNativeListener() return { sendGuidance(payload: ChatGuidePayload): Promise { @@ -147,7 +135,7 @@ export function createLocalChatStream(deps: LocalChatStreamDeps) { if (!hasLocalExecutorResponseHandlers(handlers)) { return () => undefined } - const subscriptionId = nextLocalChatStreamSubscriptionId++ + const subscriptionId = nextRuntimeChatStreamSubscriptionId++ let released = false subscriptions.set(subscriptionId, { handlers, @@ -155,11 +143,12 @@ export function createLocalChatStream(deps: LocalChatStreamDeps) { eventCount: 0, textDeltaCount: 0, }) - activeLocalChatStreamSubscriptions += 1 - logLocalChatStreamSubscription('subscribed', subscriptionId, { - activeSubscriptions: activeLocalChatStreamSubscriptions, + activeRuntimeChatStreamSubscriptions += 1 + logRuntimeChatStreamSubscription('subscribed', subscriptionId, { + activeSubscriptions: activeRuntimeChatStreamSubscriptions, ...streamScopeDebug(handlers.scope), }) + // Retry setup if the eager native subscription failed. ensureNativeListener() return () => { @@ -167,26 +156,25 @@ export function createLocalChatStream(deps: LocalChatStreamDeps) { released = true const subscription = subscriptions.get(subscriptionId) subscriptions.delete(subscriptionId) - activeLocalChatStreamSubscriptions = Math.max(0, activeLocalChatStreamSubscriptions - 1) - logLocalChatStreamSubscription('unsubscribed', subscriptionId, { - activeSubscriptions: activeLocalChatStreamSubscriptions, + activeRuntimeChatStreamSubscriptions = Math.max(0, activeRuntimeChatStreamSubscriptions - 1) + logRuntimeChatStreamSubscription('unsubscribed', subscriptionId, { + activeSubscriptions: activeRuntimeChatStreamSubscriptions, eventCount: subscription?.eventCount ?? 0, textDeltaCount: subscription?.textDeltaCount ?? 0, ...streamScopeDebug(subscription?.handlers.scope), }) - releaseNativeListenerIfIdle() } }, } } -function logLocalChatTerminalEvent( +function logRuntimeChatTerminalEvent( event: LocalExecutorEvent, matchedSubscriptionCount: number, subscriptionCount: number ): void { const payload = asRecord(event.payload) - console.info('[Wework] Local chat stream terminal event received', { + console.info('[Wework] Runtime chat stream terminal event received', { event: event.event, taskId: stringField(payload, 'taskId') ?? null, subtaskId: stringField(payload, 'subtaskId') ?? null, @@ -230,44 +218,44 @@ function streamScopeDebug(scope: ChatStreamHandlers['scope']): Record ): void { - if (!isLocalChatStreamDebugEnabled()) return - console.debug('[Wework] Local chat stream subscription', { + if (!isRuntimeChatStreamDebugEnabled()) return + console.debug('[Wework] Runtime chat stream subscription', { action, subscriptionId, ...details, }) } -function logLocalChatStreamNativeSubscription( +function logRuntimeChatStreamNativeSubscription( action: string, details: Record = {} ): void { - if (!isLocalChatStreamDebugEnabled()) return - console.debug('[Wework] Local chat stream native subscription', { + if (!isRuntimeChatStreamDebugEnabled()) return + console.debug('[Wework] Runtime chat stream native subscription', { action, - activeSubscriptions: activeLocalChatStreamSubscriptions, + activeSubscriptions: activeRuntimeChatStreamSubscriptions, ...details, }) } -function logLocalChatStreamEvent( +function logRuntimeChatStreamEvent( subscriptionId: number, event: LocalExecutorEvent, eventCount: number, textDeltaCount: number ): void { - if (!isLocalChatStreamDebugEnabled()) return - if (!shouldLogLocalChatStreamEvent(event.event)) return + if (!isRuntimeChatStreamDebugEnabled()) return + if (!shouldLogRuntimeChatStreamEvent(event.event)) return const payload = asRecord(event.payload) const data = asRecord(payload.data) - console.debug('[Wework] Local chat stream event', { + console.debug('[Wework] Runtime chat stream event', { subscriptionId, - activeSubscriptions: activeLocalChatStreamSubscriptions, + activeSubscriptions: activeRuntimeChatStreamSubscriptions, event: event.event, mapped: isMappedResponseApiEvent(event.event), eventCount, @@ -293,7 +281,7 @@ function isLocalExecutorEventInScope( return true } -function shouldLogLocalChatStreamEvent(eventName: string): boolean { +function shouldLogRuntimeChatStreamEvent(eventName: string): boolean { return ( eventName === 'response.completed' || eventName === 'response.failed' || diff --git a/wework/src/components/layout/DesktopWorkbenchMain.tsx b/wework/src/components/layout/DesktopWorkbenchMain.tsx index ce1d5b8b8f..fcb1720355 100644 --- a/wework/src/components/layout/DesktopWorkbenchMain.tsx +++ b/wework/src/components/layout/DesktopWorkbenchMain.tsx @@ -77,7 +77,7 @@ import { import { pendingRequestUserInputPayload } from './requestUserInputOverlay' import { CachedWorkbenchPaneStack, - getRunningRuntimeWorkbenchPaneKeys, + getRuntimeWorkbenchPaneKeys, getWorkbenchPaneKey, useWorkbenchPaneActive, WorkbenchPaneActiveOnly, @@ -100,7 +100,7 @@ import { useRuntimeTaskContinueInIm } from './useRuntimeTaskContinueInIm' import { requestOpenCloudDeviceSettings } from './workbenchShellEvents' import { SubagentStatusIndicator } from './SubagentStatusIndicator' import { WEWORK_OPEN_TERMINAL_EVENT } from '@/lib/keybindings' -import type { RuntimeTaskAddress, RuntimeWorkListResponse } from '@/types/api' +import type { RuntimeTaskAddress } from '@/types/api' import type { WorkbenchMessage } from '@/types/workbench' import { BufferedChatInput } from './BufferedChatInput' import { DesktopEmptyTaskLauncher } from './DesktopEmptyTaskLauncher' @@ -177,29 +177,6 @@ function consumeLatestBlankBrowserMigration(): PendingBlankBrowserMigration | nu return migration } -function getRuntimeWorkbenchPaneKeys( - runtimeWork: RuntimeWorkListResponse | null | undefined -): string[] { - if (!runtimeWork) return [] - - const workspaces = [ - ...runtimeWork.chats, - ...runtimeWork.projects.flatMap(project => project.deviceWorkspaces), - ] - - return workspaces.flatMap(workspace => - workspace.tasks.map(task => - getWorkbenchPaneKey({ - currentRuntimeTask: { - deviceId: workspace.deviceId, - taskId: task.taskId, - }, - currentProject: null, - }) - ) - ) -} - function createBottomPanelWorkspaceKey({ currentRuntimeTask, workspaceProjectId, @@ -289,22 +266,11 @@ export function DesktopWorkbenchMain(props: DesktopWorkbenchMainProps) { [state.runtimeWork] ) const validRuntimePaneKeySet = useMemo(() => new Set(runtimePaneKeys), [runtimePaneKeys]) - const runningPaneKeys = useMemo( - () => getRunningRuntimeWorkbenchPaneKeys(state.runtimeWork), - [state.runtimeWork] - ) - const validTerminalPinnedPaneKeys = useMemo( - () => terminalPinnedPaneKeys.filter(key => validRuntimePaneKeySet.has(key)), - [terminalPinnedPaneKeys, validRuntimePaneKeySet] - ) const prunedPaneKeys = useMemo( () => terminalPinnedPaneKeys.filter(key => !validRuntimePaneKeySet.has(key)), [terminalPinnedPaneKeys, validRuntimePaneKeySet] ) - const pinnedPaneKeys = useMemo( - () => Array.from(new Set([...runningPaneKeys, ...validTerminalPinnedPaneKeys])), - [runningPaneKeys, validTerminalPinnedPaneKeys] - ) + const pinnedPaneKeys = runtimePaneKeys const pinTerminalPane = useCallback((paneKey: string) => { setTerminalPinnedPaneKeys(current => current.includes(paneKey) ? current : [...current, paneKey] diff --git a/wework/src/components/layout/MobileWorkbenchLayout.tsx b/wework/src/components/layout/MobileWorkbenchLayout.tsx index fc3978b8c0..0651828b17 100644 --- a/wework/src/components/layout/MobileWorkbenchLayout.tsx +++ b/wework/src/components/layout/MobileWorkbenchLayout.tsx @@ -34,7 +34,7 @@ import { import { TaskForkDialog } from './TaskForkDialog' import { CachedWorkbenchPaneStack, - getRunningRuntimeWorkbenchPaneKeys, + getRuntimeWorkbenchPaneKeys, type WorkbenchPaneIdentity, } from './workbenchPaneStack' import { useWorkbenchPaneSession } from './useWorkbenchPaneSession' @@ -64,7 +64,7 @@ export function MobileWorkbenchLayout() { standaloneChatKey: state.standaloneChatKey, } const pinnedPaneKeys = useMemo( - () => getRunningRuntimeWorkbenchPaneKeys(state.runtimeWork), + () => getRuntimeWorkbenchPaneKeys(state.runtimeWork), [state.runtimeWork] ) diff --git a/wework/src/components/layout/workbenchPaneStack.test.tsx b/wework/src/components/layout/workbenchPaneStack.test.tsx index 9bbfb11f5c..703cf3bba6 100644 --- a/wework/src/components/layout/workbenchPaneStack.test.tsx +++ b/wework/src/components/layout/workbenchPaneStack.test.tsx @@ -4,7 +4,7 @@ import { useState } from 'react' import { CachedWorkbenchPaneStack, WorkbenchPaneActiveOnly, - getRunningRuntimeWorkbenchPaneKeys, + getRuntimeWorkbenchPaneKeys, getWorkbenchPaneKey, type WorkbenchPaneIdentity, } from './workbenchPaneStack' @@ -101,13 +101,13 @@ describe('workbenchPaneStack', () => { ) }) - test('keeps pinned runtime panes mounted beyond the normal cache limit', async () => { + test('keeps background runtime panes mounted beyond the normal cache limit', async () => { const panes: WorkbenchPaneIdentity[] = [ { currentRuntimeTask: { deviceId: 'device-1', taskId: 101 }, currentProject: null }, { currentRuntimeTask: { deviceId: 'device-1', taskId: 102 }, currentProject: null }, { currentRuntimeTask: { deviceId: 'device-1', taskId: 103 }, currentProject: null }, ] - const pinnedKeys = [getWorkbenchPaneKey(panes[0])] + const pinnedKeys = panes.map(getWorkbenchPaneKey) function RuntimePaneStackProbe() { const [index, setIndex] = useState(0) @@ -135,8 +135,8 @@ describe('workbenchPaneStack', () => { await userEvent.click(screen.getByText('open third')) expect(screen.getByTestId('runtime-pane-101')).toBeInTheDocument() + expect(screen.getByTestId('runtime-pane-102')).toBeInTheDocument() expect(screen.getByTestId('runtime-pane-103')).toBeInTheDocument() - expect(screen.queryByTestId('runtime-pane-102')).not.toBeInTheDocument() }) test('hides cached runtime pane content after switching back to standalone chat', async () => { @@ -248,9 +248,9 @@ describe('workbenchPaneStack', () => { expect(screen.getByTestId('standalone-local-message')).toHaveTextContent('empty') }) - test('derives running runtime pane keys from task ids', () => { + test('derives every runtime pane key so background subscriptions stay mounted', () => { expect( - getRunningRuntimeWorkbenchPaneKeys({ + getRuntimeWorkbenchPaneKeys({ projects: [ { project: { id: 7, name: 'Wegent' }, @@ -284,7 +284,7 @@ describe('workbenchPaneStack', () => { chats: [], totalTasks: 2, }) - ).toEqual(['runtime:device-1:101']) + ).toEqual(['runtime:device-1:101', 'runtime:device-1:102']) }) }) diff --git a/wework/src/components/layout/workbenchPaneStack.tsx b/wework/src/components/layout/workbenchPaneStack.tsx index 6d53dad37c..58cb3a801d 100644 --- a/wework/src/components/layout/workbenchPaneStack.tsx +++ b/wework/src/components/layout/workbenchPaneStack.tsx @@ -30,30 +30,22 @@ export function getWorkbenchPaneKey({ return `blank:${blankPaneKey}` } -export function getRunningRuntimeWorkbenchPaneKeys( +export function getRuntimeWorkbenchPaneKeys( runtimeWork: RuntimeWorkListResponse | null | undefined ): string[] { if (!runtimeWork) return [] - const keys: string[] = [] const workspaces = [ ...runtimeWork.chats, ...runtimeWork.projects.flatMap(project => project.deviceWorkspaces), ] - workspaces.forEach(workspace => { - workspace.tasks.forEach(task => { - if (task.running !== true) return - keys.push( - getWorkbenchPaneKey({ - currentRuntimeTask: { - deviceId: workspace.deviceId, - taskId: task.taskId, - }, - currentProject: null, - }) - ) - }) - }) - return keys + return workspaces.flatMap(workspace => + workspace.tasks.map(task => + getWorkbenchPaneKey({ + currentRuntimeTask: { deviceId: workspace.deviceId, taskId: task.taskId }, + currentProject: null, + }) + ) + ) } interface CachedWorkbenchPaneStackProps { diff --git a/wework/src/e2e/automation.ts b/wework/src/e2e/automation.ts index f6cb8e62aa..997fae8b24 100644 --- a/wework/src/e2e/automation.ts +++ b/wework/src/e2e/automation.ts @@ -46,6 +46,7 @@ interface DesktopControlCommand { text?: string timeoutMs?: number enabled?: boolean + visible?: boolean stableMs?: number key?: string } @@ -300,6 +301,21 @@ function desktopControlElementEnabled(element: HTMLElement): boolean { return !('disabled' in element) || !(element as HTMLButtonElement).disabled } +function desktopControlElementVisible(element: HTMLElement): boolean { + const style = window.getComputedStyle(element) + const rect = element.getBoundingClientRect() + return ( + style.display !== 'none' && + style.visibility !== 'hidden' && + rect.width > 0 && + rect.height > 0 && + rect.bottom > 0 && + rect.right > 0 && + rect.top < window.innerHeight && + rect.left < window.innerWidth + ) +} + function desktopControlEventOptions(element: HTMLElement): MouseEventInit & PointerEventInit { const rect = element.getBoundingClientRect() const clientX = Math.max(0, Math.floor(rect.left + rect.width / 2)) @@ -356,10 +372,13 @@ async function waitForDesktopControlElement(command: DesktopControlCommand): Pro while (Date.now() - startedAt < timeoutMs) { const elements = findDesktopControlElements(command.selector) - const text = elements.map(element => element.textContent?.trim() ?? '').join('\n') + const matchingElements = command.visible + ? elements.filter(desktopControlElementVisible) + : elements + const text = matchingElements.map(element => element.textContent?.trim() ?? '').join('\n') const hasExpectedText = !command.text || text.includes(command.text) - const isEnabled = !command.enabled || elements.some(desktopControlElementEnabled) - if (elements.length > 0 && hasExpectedText && isEnabled) { + const isEnabled = !command.enabled || matchingElements.some(desktopControlElementEnabled) + if (matchingElements.length > 0 && hasExpectedText && isEnabled) { matchedAt ??= Date.now() if (Date.now() - matchedAt >= (command.stableMs ?? 0)) { return text diff --git a/wework/src/lib/developerCommandMenu.test.ts b/wework/src/lib/developerCommandMenu.test.ts index 582b3506fe..24e381c1b5 100644 --- a/wework/src/lib/developerCommandMenu.test.ts +++ b/wework/src/lib/developerCommandMenu.test.ts @@ -72,7 +72,20 @@ describe('developerCommandMenu', () => { expect(requestLocalExecutorMock).toHaveBeenCalledWith('runtime.codex.stream_debug.set', { enabled: true, }) - expect(localStorage.getItem('wework:debug-local-chat-stream')).toBe('1') + expect(localStorage.getItem('wework:debug-runtime-chat-stream')).toBe('1') + }) + + test('restores the frontend stream flag when the executor toggle fails', async () => { + requestLocalExecutorMock + .mockResolvedValueOnce({ enabled: true }) + .mockRejectedValueOnce(new Error('executor unavailable')) + dispatchDeveloperShortcut() + await flushPromises() + + screenCommand('toggle-stream-logs').click() + await flushPromises() + + expect(localStorage.getItem('wework:debug-runtime-chat-stream')).toBeNull() }) test('dispatches an event to simulate an app update', () => { diff --git a/wework/src/lib/developerCommandMenu.ts b/wework/src/lib/developerCommandMenu.ts index 32b59430c9..12330b7194 100644 --- a/wework/src/lib/developerCommandMenu.ts +++ b/wework/src/lib/developerCommandMenu.ts @@ -1,9 +1,9 @@ import { invoke } from '@tauri-apps/api/core' import { requestLocalExecutor } from '@/tauri/localExecutor' import { - isLocalChatStreamDebugEnabled, - setLocalChatStreamDebugEnabled, -} from '@/api/local/localChatStream' + isRuntimeChatStreamDebugEnabled, + setRuntimeChatStreamDebugEnabled, +} from '@/api/runtime/runtimeChatStream' import { clearWorkbenchDebugLogs, getWorkbenchDebugSnapshot, @@ -146,7 +146,7 @@ function renderDeveloperCommandList(list: HTMLElement) { function getDeveloperCommands(): DeveloperCommand[] { const diagnosticsEnabled = isPerformanceDiagnosticsEnabled() const streamDebugKnown = codexStreamDebugEnabled !== null - const streamLogsEnabled = isLocalChatStreamDebugEnabled() || (codexStreamDebugEnabled ?? false) + const streamLogsEnabled = isRuntimeChatStreamDebugEnabled() || (codexStreamDebugEnabled ?? false) return [ { id: 'open-debug-panel', @@ -256,9 +256,9 @@ async function refreshCodexStreamDebugStatus(onLoaded?: () => void) { async function toggleCodexStreamDebug() { if (!isTauriRuntime()) { - setLocalChatStreamDebugEnabled(!isLocalChatStreamDebugEnabled()) + setRuntimeChatStreamDebugEnabled(!isRuntimeChatStreamDebugEnabled()) console.info( - `[Wework dev] Frontend stream logs ${isLocalChatStreamDebugEnabled() ? 'enabled' : 'disabled'}.` + `[Wework dev] Frontend stream logs ${isRuntimeChatStreamDebugEnabled() ? 'enabled' : 'disabled'}.` ) return } @@ -266,8 +266,9 @@ async function toggleCodexStreamDebug() { if (codexStreamDebugEnabled === null) { await refreshCodexStreamDebugStatus() } - const enabled = !(isLocalChatStreamDebugEnabled() || (codexStreamDebugEnabled ?? false)) - setLocalChatStreamDebugEnabled(enabled) + const frontendWasEnabled = isRuntimeChatStreamDebugEnabled() + const enabled = !(frontendWasEnabled || (codexStreamDebugEnabled ?? false)) + setRuntimeChatStreamDebugEnabled(enabled) try { const state = await requestLocalExecutor(CODEX_STREAM_DEBUG_SET_METHOD, { enabled, @@ -275,7 +276,7 @@ async function toggleCodexStreamDebug() { codexStreamDebugEnabled = state.enabled console.info(`[Wework dev] Stream logs ${state.enabled ? 'enabled' : 'disabled'}.`) } catch (error) { - setLocalChatStreamDebugEnabled(!enabled) + setRuntimeChatStreamDebugEnabled(frontendWasEnabled) console.error('[Wework dev] Failed to toggle Codex stream logs', error) } }