test: connect shared Subject Body lifecycle and reproduce stale stop recovery

crates/workspace-server/src/server/tests/chat_web_connected.rs

@@ -1,8 +1,9 @@
 //! Explicit browser/component connection, not a separate product-process launcher.
 //! Reuses ConnectedChat's real server/auth/storage/Runtime/installed chat Feature.
 //! The browser uses the actual static Web build and Account API-token auth; only
-//! the external model and response-loss/Busy timing are controlled. T807 Body and
+//! the external model and response-loss/Busy timing are controlled. Shared Body
-//! T817 canonical navigation are integrated separately once their refs arrive.
+//! start/stop/restore and canonical/legacy Subject navigation stay real. Restore
+//! requires the runner-owned isolation checked before either server starts.
 //! Build `web/workspace` first, then explicitly select:
 //! `cargo test -p yoi-workspace-server --lib chat_web_connected -- --ignored --test-threads=1`.
 //! Uses the installed tools/web-ux Deno lock/Chromium; no browser download occurs.
@@ -11,9 +12,34 @@ use super::*;
 use std::process::Stdio;
 use tokio::io::AsyncWriteExt;
 
+#[derive(Clone)]
+struct BrowserModel {
+    model: ChatModel,
+    resumed: Arc<tokio::sync::Semaphore>,
+}
+
+#[async_trait::async_trait]
+impl agen::llm_client::LlmClient for BrowserModel {
+    fn clone_boxed(&self) -> Box<dyn agen::llm_client::LlmClient> {
+        Box::new(self.clone())
+    }
+
+    async fn stream(
+        &self,
+        request: agen::llm_client::Request,
+    ) -> std::result::Result<agen::llm_client::ResponseStream, agen::llm_client::ClientError> {
+        let stream = self.model.stream(request).await?;
+        if self.model.calls.load(std::sync::atomic::Ordering::SeqCst) == 6 {
+            self.resumed.add_permits(1);
+        }
+        Ok(stream)
+    }
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 6)]
 #[ignore = "explicit Web build, Deno and installed Playwright Chromium required"]
 async fn chat_web_connected_busy_unknown_replay_and_authorization_use_real_runtime_publications() {
+    crate::restore_test_environment::require_restore_test_environment();
     let root = Path::new(env!("CARGO_MANIFEST_DIR")).join("../..");
     let assets = root.join("web/workspace/build");
     assert!(
@@ -29,6 +55,11 @@ async fn chat_web_connected_busy_unknown_replay_and_authorization_use_real_runti
         after_busy_boundary: Arc::new(tokio::sync::Semaphore::new(1)),
     };
     let calls = model.calls.clone();
+    let resumed = Arc::new(tokio::sync::Semaphore::new(0));
+    let model = BrowserModel {
+        model,
+        resumed: resumed.clone(),
+    };
     let public_posts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
     let read_started = Arc::new(tokio::sync::Semaphore::new(0));
     let (release, wait) = tokio::sync::oneshot::channel::<()>();
@@ -37,6 +68,22 @@ async fn chat_web_connected_busy_unknown_replay_and_authorization_use_real_runti
     let f = ConnectedChat::with_web_assets(
         model,
         move |router| {
+            // Observe the substituted model only. An accepted Notify or an Idle
+            // snapshot alone cannot prove this new no-public-reply turn ran.
+            let router = router.route(
+                "/__fixture/model-resumed",
+                axum::routing::get(move || {
+                    let resumed = resumed.clone();
+                    async move {
+                        tokio::time::timeout(std::time::Duration::from_secs(30), resumed.acquire())
+                            .await
+                            .expect("restored input must reach the model")
+                            .unwrap()
+                            .forget();
+                        axum::Json(json!({"observed":true}))
+                    }
+                }),
+            );
             router.layer(axum::middleware::from_fn(
                 move |request: axum::extract::Request, next: axum::middleware::Next| {
                     let public_posts = public_posts.clone();
@@ -89,9 +136,15 @@ async fn chat_web_connected_busy_unknown_replay_and_authorization_use_real_runti
     let conversation = f.post(&conversations_path(), json!({"operation_id":format!("web-subject-conversation/v1/{subject}"),"subject_id":subject})).await;
     let c = conversation["id"].as_str().unwrap();
     *conversation_id.lock().unwrap() = c.into();
-    let created = f.post(&format!("/api/w/{TEST_WORKSPACE_ID}/workers"), json!({"runtime_id":"connected-chat","display_name":"Browser connected chat","profile":"builtin:async-chat","initial_submit":[],"workdir_attachments":[],"feature_connections":{"subjektiv":{"subject_id":subject}}})).await;
+    // Let the browser use normal Body admission, including its default profile.
-    let worker = created["worker_id"].as_str().unwrap();
+    // Remove only the unused fixture Runtime so the real signed HTTP path wins.
-    let mut events = f.runtime.subscribe_worker_observation().unwrap();
+    assert!(matches!(
+        f.api
+            .runtime
+            .unregister_if_idle(EMBEDDED_WORKER_RUNTIME_ID, 100)
+            .unwrap(),
+        crate::hosts::RuntimeRegistryUnregisterResult::Removed
+    ));
     f.lose_next_input_ack
         .store(true, std::sync::atomic::Ordering::SeqCst);
     let foreign = seed_test_api_token(f.api.store.as_ref(), "chat-connected-foreign");
@@ -111,7 +164,8 @@ async fn chat_web_connected_busy_unknown_replay_and_authorization_use_real_runti
         .kill_on_drop(true)
         .spawn()
         .expect("explicit browser test requires Deno");
-    let wire = serde_json::to_vec(&json!({"origin":f.url,"workspace_id":TEST_WORKSPACE_ID,"subject_id":subject,"subject_name":"Web connected companion","conversation_id":c,"owner_token":f.owner,"foreign_token":foreign})).unwrap();
+    let artifacts = root.join("target/t808-validation/body-connected");
+    let wire = serde_json::to_vec(&json!({"origin":f.url,"workspace_id":TEST_WORKSPACE_ID,"subject_id":subject,"subject_name":"Web connected companion","conversation_id":c,"owner_token":f.owner,"foreign_token":foreign,"artifact_dir":artifacts})).unwrap();
     let mut stdin = child.stdin.take().unwrap();
     stdin.write_all(&wire).await.unwrap();
     drop(stdin);
@@ -145,19 +199,28 @@ async fn chat_web_connected_busy_unknown_replay_and_authorization_use_real_runti
     for phase in 2..=4 {
         next_chat_input(&mut inputs, phase).await;
     }
-    chat_idle(&mut events, worker).await;
+    let resumed = next_chat_input(&mut inputs, 5).await;
+    assert!(resumed.contains("WEB_CONNECTED_AFTER_RESTART"));
+    assert!(resumed.contains("WEB_CONNECTED_FIRST_INPUT"));
     assert_eq!(
         calls.load(std::sync::atomic::Ordering::SeqCst),
-        5,
+        6,
-        "exact retry must not reactivate the model"
+        "only the new post-restore input may reactivate the model"
     );
     let (status, public) = account(&f.api, &f.owner, "GET", &messages_path(c), None).await;
     assert_eq!(status, StatusCode::OK);
-    assert_eq!(public["messages"].as_array().unwrap().len(), 4);
+    assert_eq!(public["messages"].as_array().unwrap().len(), 5);
     assert!(!public.to_string().contains("INTERNAL_ASSISTANT"));
-    f.post(
+    assert!(!public.to_string().contains("edited next draft"));
-        &format!("/api/w/{TEST_WORKSPACE_ID}/runtimes/connected-chat/workers/{worker}/stop"),
+    let (status, body) = account(
-        json!({}),
+        &f.api,
+        &f.owner,
+        "GET",
+        &format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/subjects/{subject}/body"),
+        None,
     )
     .await;
+    assert_eq!(status, StatusCode::OK);
+    assert_eq!(body["current"]["runtime_id"], "connected-chat");
+    assert_eq!(body["observation"]["worker"]["state"], "stopped");
 }

tools/web-ux/browser-tests/subject_conversation_connected.ts

@@ -1,8 +1,9 @@
-// Explicit connected test driver. Rust owns ephemeral Backend/Runtime/SQLite;
+// 明示選択の接続driver。隔離Backend/Runtime/SQLiteはRust fixtureが所有し、
-// this browser runs the production Web build. Only the external model and
+// 本番Webから共有Bodyと会話APIを操作する。外部model・ACK喪失・Busy時刻のみ制御する。
-// deliberate lost transport acknowledgement are substituted. No live instance.
 import { assert, assertEquals } from "@std/assert";
-import { chromium } from "playwright";
+import { chromium, type Page } from "playwright";
+import { sha256File } from "../src/artifacts.ts";
+import { assertNoPageWideOverflow } from "./memory_document_harness.ts";
 import {
   subjectChatHref,
   subjectHref,
@@ -12,37 +13,109 @@ const input = JSON.parse(await new Response(Deno.stdin.readable).text());
 const origin = new URL(input.origin);
 assert(origin.hostname === "127.0.0.1" && origin.protocol === "http:");
 const base = `/api/w/${encodeURIComponent(input.workspace_id)}/chat/conversations`;
+const bodyPath = `/api/w/${encodeURIComponent(input.workspace_id)}/subjektiv/subjects/${
+  encodeURIComponent(input.subject_id)
+}/body`;
 const detail = subjectHref(input.workspace_id, input.subject_id);
 const chat = subjectChatHref(input.workspace_id, input.subject_id);
 const legacy = `/w/${encodeURIComponent(input.workspace_id)}/memory/${
   encodeURIComponent(input.subject_id)
 }`;
 const suffix = "?from=legacy&encoded=%2Fkeep%20exact#subject-heading";
+const output = `${input.artifact_dir}/run-${Date.now()}`;
+await Deno.mkdir(output, { recursive: true });
+const git = async (args: string[]) => {
+  const result = await new Deno.Command("git", { args, stdout: "piped" }).output();
+  assert(result.success);
+  return new TextDecoder().decode(result.stdout).trim();
+};
+const pageErrors: string[] = [];
+const captures: unknown[] = [];
+const manifest = {
+  evidence: "actual-Web-Backend-SQLite-signed-Runtime; external model scripted",
+  source: {
+    commit: await git(["rev-parse", "HEAD"]),
+    status: await git(["status", "--porcelain"]),
+  },
+  driverSha256: await sha256File(new URL(import.meta.url).pathname),
+  buildIndexSha256: await sha256File("web/workspace/build/index.html"),
+  visualVerdict: "UNREVIEWED: implementation author must view images",
+  status: "running",
+  pageErrors,
+  captures,
+};
+const save = () =>
+  Deno.writeTextFile(`${output}/review-context.json`, JSON.stringify(manifest, null, 2));
+async function capture(page: Page, id: string, theme: "light" | "dark" = "light") {
+  await assertNoPageWideOverflow(page);
+  const image = `${output}/${id}.png`;
+  await page.screenshot({ path: image, fullPage: true, animations: "disabled" });
+  const accessibility = `${output}/${id}.aria.txt`;
+  await Deno.writeTextFile(accessibility, await page.locator("main").ariaSnapshot());
+  captures.push({
+    id,
+    route: new URL(page.url()).pathname,
+    viewport: page.viewportSize(),
+    theme,
+    image,
+    imageSha256: await sha256File(image),
+    accessibility,
+    visibleAlerts: await page.getByRole("alert").allTextContents(),
+    mainWidth: await page.locator("main").evaluate((el) => el.clientWidth),
+  });
+  await save();
+}
 const browser = await chromium.launch({ headless: true });
+let activePage: Page | null = null;
 try {
   const context = await browser.newContext({
+    viewport: { width: 1440, height: 1000 },
+    colorScheme: "light",
     extraHTTPHeaders: { Authorization: `Bearer ${input.owner_token}` },
   });
   const page = await context.newPage();
+  activePage = page;
   page.setDefaultTimeout(30000);
   const posts: Record<string, unknown>[] = [];
-  const pageErrors: string[] = [];
+  const bodyPosts: { path: string; input: Record<string, unknown> }[] = [];
   page.on("pageerror", (error) => pageErrors.push(error.message));
   page.on("request", (request) => {
-    if (
+    const path = new URL(request.url()).pathname;
-      request.method() === "POST" &&
+    if (request.method() !== "POST") return;
-      new URL(request.url()).pathname ===
+    if (path === `${base}/${encodeURIComponent(input.conversation_id)}/messages`) {
-        `${base}/${encodeURIComponent(input.conversation_id)}/messages`
+      posts.push(request.postDataJSON());
-    ) posts.push(request.postDataJSON());
+    }
+    if (path.startsWith(`${bodyPath}/`)) bodyPosts.push({ path, input: request.postDataJSON() });
   });
+  // 有限の状態観測で同期する。refreshやACKからRuntimeのidleを推論しない。
+  async function idle() {
+    await page.waitForFunction(async (path) => {
+      const response = await fetch(path, { cache: "no-store" });
+      if (!response.ok) return false;
+      const body = await response.json();
+      return body.observation.kind === "available" &&
+        body.observation.worker.worker_state?.state.kind === "idle";
+    }, bodyPath);
+    await page.getByRole("button", { name: "Refresh body state", exact: true }).click();
+    await page.getByText("idle", { exact: true }).first().waitFor();
+  }
   const response = await page.goto(`${origin.origin}${legacy}${suffix}`);
   assertEquals(response?.status(), 200);
   await page.waitForURL(`${origin.origin}${detail}${suffix}`);
+  await page.getByText("No current conversation body.", { exact: true }).waitFor();
+  await capture(page, "light-1440-absent-detail");
+  await page.getByRole("button", { name: "Start conversation body", exact: true }).click();
+  await idle();
+  assertEquals(bodyPosts.length, 1);
+  assertEquals(bodyPosts[0].path, `${bodyPath}/start`);
+  assertEquals(bodyPosts[0].input.expected_current, null);
+  const initialBody = await (await context.request.get(`${origin.origin}${bodyPath}`)).json();
   const open = page.getByRole("link", { name: "Open conversation" });
   assertEquals(await open.getAttribute("href"), chat);
   await open.click();
   await page.waitForURL(`${origin.origin}${chat}`);
   await page.getByText("No public messages yet.").waitFor();
+  await capture(page, "light-1440-empty-chat");
   for (const alias of ["conversation", "chat"]) {
     const response = await page.goto(`${origin.origin}${legacy}/${alias}${suffix}`);
     assertEquals(response?.status(), 200);
@@ -121,12 +194,90 @@ try {
   );
   assertEquals(await history.getByText("WEB_CONNECTED_BUSY_INPUT", { exact: true }).count(), 1);
   for (const entry of [chat, `${legacy}/conversation`, `${legacy}/chat`]) {
-    const response = await page.goto(`${origin.origin}${entry}${suffix}`);
+    const target = `${origin.origin}${entry}${suffix}`;
+    const response = page.url() === target ? await page.reload() : await page.goto(target);
     assertEquals(response?.status(), 200);
     await page.waitForURL(`${origin.origin}${chat}${suffix}`);
     await page.getByText("public reply 3", { exact: true }).waitFor();
     assertEquals(await history.getByRole("listitem").count(), 4);
   }
+  await idle();
+  // 共有Bodyの実停止後にACKだけ失う。reload後も完全なstop intentを維持し、
+  // read-only outcome照会は追加stopを送らないことを監視する。
+  let stopLost = false;
+  await page.route(`**${bodyPath}/stop`, async (route) => {
+    if (!stopLost) {
+      stopLost = true;
+      const actual = await route.fetch();
+      assertEquals(actual.status(), 200);
+      await route.abort("failed");
+    } else await route.continue();
+  });
+  await page.getByRole("button", { name: "Stop conversation body", exact: true }).click();
+  await page.getByRole("alert").filter({ hasText: /Request outcome unknown/ }).waitFor();
+  assertEquals(bodyPosts.length, 2);
+  const stopRequest = bodyPosts[1].input;
+  assertEquals(stopRequest.expected_current, initialBody.current);
+  await page.reload();
+  await page.getByRole("button", { name: "Check stop outcome", exact: true }).click();
+  await page.getByText("Historical stop confirmed.", { exact: true }).waitFor();
+  await page.getByText("Body stopped.", { exact: true }).waitFor();
+  assertEquals(bodyPosts.length, 3);
+  assertEquals(bodyPosts[2], { path: `${bodyPath}/stop/outcome`, input: stopRequest });
+  assertEquals(await history.getByRole("listitem").count(), 4);
+  await capture(page, "light-1440-stopped-retained-chat");
+  await page.getByRole("button", { name: "Restart conversation body", exact: true }).click();
+  await idle();
+  const restored = await (await context.request.get(`${origin.origin}${bodyPath}`)).json();
+  assertEquals(
+    restored.current,
+    initialBody.current,
+    "restore must keep the singleton Body, not create another",
+  );
+  await composer.fill("WEB_CONNECTED_AFTER_RESTART");
+  await page.getByRole("button", { name: "Send message", exact: true }).click();
+  await page.getByText("Runtime accepted this input. Processing and replies are not confirmed.")
+    .waitFor();
+  // ACKや初期idleは処理済みの証拠ではない。scripted modelの受信観測後に
+  // 本番Runtimeのidleを待ち、この返信なしturnを停止で潰さない。
+  const observed = await context.request.get(`${origin.origin}/__fixture/model-resumed`);
+  assertEquals(observed.status(), 200);
+  assertEquals(await observed.json(), { observed: true });
+  await idle();
+  await page.getByRole("button", { name: "Refresh history", exact: true }).click();
+  await history.getByText("WEB_CONNECTED_AFTER_RESTART", { exact: true }).waitFor();
+  assertEquals(
+    await history.getByRole("listitem").count(),
+    5,
+    "no reply turn must not expose internal assistant text",
+  );
+  for (const width of [1440, 768, 390]) {
+    for (const theme of ["light", "dark"] as const) {
+      await page.setViewportSize({ width, height: 1000 });
+      await page.emulateMedia({ colorScheme: theme });
+      await capture(page, `${theme}-${width}-restored-chat`, theme);
+    }
+  }
+  await page.getByRole("button", { name: "Stop conversation body", exact: true }).click();
+  const stopDisposition = await Promise.race([
+    page.getByText("Body stopped.", { exact: true }).waitFor().then(() => "stopped"),
+    page.getByText("Request rejected.", { exact: true }).waitFor().then(() => "rejected"),
+  ]);
+  if (stopDisposition === "rejected") {
+    // 実Backendの既知の未実行拒否だけが新観測・新依頼へ進める。
+    // unknownをここで清算したり、同IDの入力を差し替えたりしない。
+    await page.getByRole("button", { name: "Refresh body state", exact: true }).click();
+    await page.getByRole("button", { name: "Stop conversation body", exact: true }).click();
+    await page.getByText("Body stopped.", { exact: true }).waitFor();
+    const stops = bodyPosts.filter((post) => post.path === `${bodyPath}/stop`);
+    assert(stops.at(-1)!.input.request_id !== stops.at(-2)!.input.request_id);
+  }
+  assertEquals(await history.getByRole("listitem").count(), 5);
+  assertEquals(bodyPosts.filter((post) => post.path === `${bodyPath}/start`).length, 2);
+  assertEquals(
+    bodyPosts.filter((post) => post.path === `${bodyPath}/stop`).length,
+    stopDisposition === "rejected" ? 3 : 2,
+  );
   const deniedPath = `${origin.origin}${base}/${
     encodeURIComponent(input.conversation_id)
   }/messages`;
@@ -137,12 +288,24 @@ try {
   assertEquals(denied.status(), 403);
   await page.getByRole("button", { name: "Refresh history", exact: true }).click();
   await page.getByRole("alert").filter({ hasText: /do not have access/ }).waitFor();
+  await page.getByRole("button", { name: "Refresh body state", exact: true }).click();
+  await page.getByRole("alert").filter({ hasText: /Body state could not be confirmed/ }).waitFor();
   assertEquals(await page.getByRole("textbox").count(), 0);
   assertEquals(await history.getByRole("listitem").count(), 0);
+  await capture(page, "dark-390-permission-denied", "dark");
   assertEquals(pageErrors, []);
+  manifest.status = "passed";
   console.log(
-    "real Web -> Backend -> Runtime -> explicit public replies passed; scripted model, API-token browser transport; Body route not yet integrated",
+    "real Web -> Backend -> Runtime -> explicit public replies passed; scripted model; shared Body start/stop/restore and canonical/legacy routes included",
   );
+} catch (cause) {
+  manifest.status = "failed";
+  if (activePage) {
+    console.error("Visible alerts:", await activePage.getByRole("alert").allTextContents());
+    await capture(activePage, "failure").catch(() => {});
+  }
+  throw cause;
 } finally {
+  await save();
   await browser.close();
 }