test: verify restore response loss and owned runtime restart

tests/e2e/README.md

@@ -22,7 +22,7 @@
 
 所有crateの`test-support`依存はテスト側の非default `fixture-support` featureだけで有効にする。製品binary providerは別Cargo invocationと専用`target/e2e-products`へ`yoi`・`yoi-server`・`yoi-runtime`を追加featureなしでビルドする。support test自身をこのtargetへビルドしてはならない。`YOI_E2E_BIN`等の明示binary overrideを使用する場合も、呼出し側が同じ分離を保証する。既存TUI用`e2e-test` featureは別の`target/e2e-tui-products`へビルドし、restore製品binaryへ混入しない。
 
-固定資産の正常restore・並行隔離E2E:
+固定資産の正常restore・並行隔離・応答喪失回復E2E:
 
 ```sh
 cargo test -p yoi-e2e --features e2e,fixture-support --test worker_restore
@@ -33,6 +33,12 @@ YOI_E2E_RESTORE_FIXTURE=target/fixture-candidate-v1 \
 
 標準資産は`tests/e2e/fixtures/worker-codex-v1`の`manifest.json`・`worker-template.json`・`expected.json`・`workspace-config.dcdl`。manifestの各asset hash/size、schemaと固定ID、生成元commit、modelを検証する。Workspace configのimport参照を黙って最新へ置換せず、Gitで生成元commitに対する`resources`と`crates/builtin-source`の一致を確認する（Git履歴がない場合も失敗）。移植したSessionの正規公開projectionを供給された会話行と照合し、その全snapshotを実executionのWebSocketと比較する。欠落fixture・hash不一致・未実行は成功扱いしない。
 
+応答喪失ケースは同じ`WorkerE2eEnvironment`内の単発loopback relayで公開restore要求をそのまま転送し、完全なupstream応答を受け取った後に呼出元へ0byteで切断する。呼出元のtransport失敗とrelay側のAccepted観測は別の事実として記録する。回収は元のrequest ID・観測tokenをそのまま使い、fresh intentへの古いtokenはHTTP409を要求する。
+
+process終了ケースはそのAccepted receipt確認後に環境が所有するRuntime childだけをkill/reapし、同じroot・Runtime identity・trustから再起動する。seedの再実行・tokenの差替えはしない。再起動前後のcanonical owner全体のhash・operation ID・owner数が一致し、公開APIの保存結果とfull Session snapshotを回収できることを確認する。公開Worker投影の観測tokenは再起動で変わるため、そのfieldだけはcurrent detailとの一致を別に確認する。canonical診断はboundedな読取りのみで、要求・receiptの内容や認証情報はartifactへ保存しない。
+
+これは**commit済みかつterminal receipt保存後**のprocess終了であり、未commit候補やcommit直後・receipt前の終了証明ではない。固定Codex資産へSubmit/Notifyを送らないため、通常入力・Paused admissionの製品E2Eの証明でもない。検証jobはこの明示targetを選択する必要があり、通常suiteだけでは代替しない。
+
 通常のharness回帰試験（製品processは起動しない）:
 
 ```sh

tests/e2e/src/worker_environment.rs

@@ -1,3 +1,6 @@
+#[path = "worker_environment/response_loss.rs"]
+mod response_loss;
+
 use std::fs::{self, File, OpenOptions};
 use std::io::{Read, Write};
 use std::path::{Path, PathBuf};
@@ -571,12 +574,7 @@ impl WorkerE2eEnvironment {
         if self.runtime_public_identity.is_none() {
             self.initialize_runtime_identity(self.runtime_id.clone())?;
         }
-        let runtime_identity = self
-            .runtime_public_identity
-            .clone()
-            .expect("initialized identity");
         let bundle_path = self.root.join("workspace-public-identity.json");
-        let server_url = self.required_server_url()?.to_string();
         let workspace_id = self.required_workspace_id()?.to_string();
         let runtime_binary = self.binaries.binary("yoi-runtime")?;
         self.run_command(
@@ -616,6 +614,20 @@ impl WorkerE2eEnvironment {
 
         seed(self)?;
         self.write_metadata("runtime-seeded")?;
+        self.start_runtime_process()
+    }
+
+    fn start_runtime_process(&mut self) -> Result<()> {
+        if self.runtime.is_some() {
+            return Err(HarnessError::Protocol("Runtime is already running".into()));
+        }
+        let runtime_binary = self.binaries.binary("yoi-runtime")?;
+        let runtime_identity = self
+            .runtime_public_identity
+            .clone()
+            .ok_or_else(|| HarnessError::Protocol("Runtime identity is not initialized".into()))?;
+        let server_url = self.required_server_url()?.to_string();
+        let workspace_id = self.required_workspace_id()?.to_string();
         let mut command = self.isolated_command(&runtime_binary);
         command.args([
             "--bind",
@@ -669,6 +681,36 @@ impl WorkerE2eEnvironment {
         Ok(())
     }
 
+    /// Kill/reap only the Runtime child owned by this environment, retaining roots
+    /// and identities. Restart neither seeds storage nor rotates restore requests.
+    pub fn crash_and_restart_runtime(&mut self) -> Result<()> {
+        let runtime = self
+            .runtime
+            .as_mut()
+            .ok_or_else(|| HarnessError::Protocol("Runtime is not running".into()))?;
+        let pid = runtime.id();
+        let status = runtime.kill_and_reap(EXIT_WAIT)?;
+        runtime.join_readers_until(Instant::now() + EXIT_WAIT)?;
+        fs::write(
+            self.artifacts_dir.join("runtime-crash.json"),
+            serde_json::to_vec_pretty(&serde_json::json!({
+                "owned_pid": pid, "reaped": true, "exit_status": status.to_string(),
+                "same_runtime_id": self.runtime_id,
+            }))?,
+        )?;
+        self.runtime.take();
+        for stream in ["stdout", "stderr"] {
+            fs::rename(
+                self.artifacts_dir.join(format!("runtime.{stream}.log")),
+                self.artifacts_dir
+                    .join(format!("runtime-before-crash.{stream}.log")),
+            )?;
+        }
+        self.runtime_url = None;
+        self.runtime_shutdown = WorkerProcessShutdownOutcome::NotStarted;
+        self.start_runtime_process()
+    }
+
     pub fn assert_connected(&self) -> Result<()> {
         let server_url = self.required_server_url()?;
         let workspace_id = self.required_workspace_id()?;
@@ -752,6 +794,38 @@ impl WorkerE2eEnvironment {
         )
     }
 
+    /// Forward one unchanged public restore request through an owned loopback
+    /// relay, consume its complete upstream response, and close without delivering
+    /// any response bytes. The returned JSON is relay evidence, not caller success.
+    pub fn restore_worker_losing_response(
+        &self,
+        worker_id: &str,
+        request_id: &str,
+        expected_observation_token: &str,
+    ) -> Result<Value> {
+        let token = self.access_token.as_deref().ok_or_else(|| {
+            HarnessError::Protocol("Backend access token is not initialized".into())
+        })?;
+        let request = self
+            .client
+            .post(self.worker_url(worker_id, "restore")?)
+            .bearer_auth(token)
+            .header(reqwest::header::CONNECTION, "close")
+            .timeout(STARTUP_WAIT + EXIT_WAIT)
+            .json(&serde_json::json!({
+                "request_id": request_id,
+                "expected_observation_token": expected_observation_token,
+            }))
+            .build()
+            .map_err(|_| HarnessError::Protocol("build response-loss request".into()))?;
+        let evidence = response_loss::drop_response(&self.client, request)?;
+        self.append_operation(
+            "restore response consumed by relay, caller transport failed",
+            0,
+        )?;
+        Ok(evidence)
+    }
+
     pub fn stop_worker(&self, worker_id: &str) -> Result<Value> {
         self.request_json(
             self.client

tests/e2e/src/worker_environment/response_loss.rs

@@ -0,0 +1,198 @@
+//! A single bounded response-loss injection in the shared product harness.
+use std::io::{Read, Write};
+use std::net::{IpAddr, TcpListener, TcpStream};
+use std::time::Instant;
+
+use reqwest::blocking::{Client, Request};
+use serde_json::Value;
+
+use super::{EXIT_WAIT, HTTP_WAIT_STEP, STARTUP_WAIT};
+use crate::{HarnessError, Result};
+
+const MAX_HEADER: usize = 16 * 1024;
+const MAX_MESSAGE: usize = 2 * 1024 * 1024;
+
+fn failure(message: &str) -> HarnessError {
+    HarnessError::Protocol(message.into())
+}
+
+pub(super) fn drop_response(client: &Client, mut request: Request) -> Result<Value> {
+    let host: IpAddr = request
+        .url()
+        .host_str()
+        .ok_or_else(|| failure("missing relay host"))?
+        .parse()
+        .map_err(|_| failure("relay target must be a literal loopback address"))?;
+    if !host.is_loopback() || request.url().scheme() != "http" {
+        return Err(failure(
+            "response-loss relay only permits owned loopback HTTP",
+        ));
+    }
+    let port = request
+        .url()
+        .port()
+        .ok_or_else(|| failure("missing relay port"))?;
+    let listener = TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0))?;
+    listener.set_nonblocking(true)?;
+    let relay_address = listener.local_addr()?;
+    request
+        .url_mut()
+        .set_port(Some(relay_address.port()))
+        .map_err(|_| failure("set relay port"))?;
+    request
+        .url_mut()
+        .set_host(Some("127.0.0.1"))
+        .map_err(|_| failure("set relay host"))?;
+    std::thread::scope(|scope| {
+        let relay = scope.spawn(move || -> Result<Value> {
+            let deadline = Instant::now() + STARTUP_WAIT;
+            let mut downstream = loop {
+                match listener.accept() {
+                    Ok((stream, peer)) if peer.ip().is_loopback() => break stream,
+                    Ok(_) => return Err(failure("non-loopback relay peer")),
+                    Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
+                        if Instant::now() >= deadline {
+                            return Err(failure("response-loss relay accept deadline exceeded"));
+                        }
+                        std::thread::sleep(HTTP_WAIT_STEP);
+                    }
+                    Err(error) => return Err(error.into()),
+                }
+            };
+            let request_bytes = read_message(&mut DeadlineStream {
+                stream: &mut downstream,
+                deadline,
+            })?;
+            let mut upstream = TcpStream::connect_timeout(&(host, port).into(), EXIT_WAIT)?;
+            upstream.set_write_timeout(Some(EXIT_WAIT))?;
+            // The body, request identity and authorization bytes are forwarded
+            // unchanged. Neither request nor response headers enter diagnostics.
+            upstream.write_all(&request_bytes)?;
+            let response = read_message(&mut DeadlineStream {
+                stream: &mut upstream,
+                deadline,
+            })?;
+            let boundary =
+                header_end(&response).ok_or_else(|| failure("missing upstream header"))?;
+            let status = std::str::from_utf8(&response[..boundary])
+                .ok()
+                .and_then(|header| header.lines().next())
+                .and_then(|line| line.split_whitespace().nth(1));
+            if status != Some("200") {
+                return Err(failure("restore relay upstream did not return HTTP 200"));
+            }
+            let value = serde_json::from_slice(&response[boundary..])?;
+            // Intentionally send zero bytes back; the caller must observe loss.
+            drop(downstream);
+            Ok(value)
+        });
+        let caller_failed = client.execute(request).is_err();
+        let evidence = relay
+            .join()
+            .map_err(|_| failure("response-loss relay panicked"))??;
+        if !caller_failed {
+            return Err(failure("caller unexpectedly received a response"));
+        }
+        Ok(evidence)
+    })
+}
+
+struct DeadlineStream<'a> {
+    stream: &'a mut TcpStream,
+    deadline: Instant,
+}
+
+impl Read for DeadlineStream<'_> {
+    fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
+        let remaining = self.deadline.saturating_duration_since(Instant::now());
+        if remaining.is_zero() {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::TimedOut,
+                "relay deadline exceeded",
+            ));
+        }
+        self.stream.set_read_timeout(Some(remaining))?;
+        self.stream.read(buffer)
+    }
+}
+
+fn header_end(bytes: &[u8]) -> Option<usize> {
+    bytes
+        .windows(4)
+        .position(|window| window == b"\r\n\r\n")
+        .map(|i| i + 4)
+}
+
+fn read_message(stream: &mut impl Read) -> Result<Vec<u8>> {
+    let mut bytes = Vec::new();
+    let mut buffer = [0; 4096];
+    let mut expected = None;
+    loop {
+        let count = stream.read(&mut buffer)?;
+        if count == 0 {
+            return Err(failure("relay peer closed before complete HTTP message"));
+        }
+        bytes.extend_from_slice(&buffer[..count]);
+        if bytes.len() > MAX_MESSAGE {
+            return Err(failure("relay HTTP byte budget exceeded"));
+        }
+        if expected.is_none() {
+            if let Some(boundary) = header_end(&bytes) {
+                if boundary > MAX_HEADER {
+                    return Err(failure("relay HTTP header budget exceeded"));
+                }
+                let header = std::str::from_utf8(&bytes[..boundary])
+                    .map_err(|_| failure("invalid relay HTTP header"))?;
+                let mut lengths = header
+                    .lines()
+                    .filter_map(|line| line.split_once(':'))
+                    .filter(|(name, _)| name.eq_ignore_ascii_case("content-length"))
+                    .map(|(_, value)| value.trim().parse::<usize>());
+                let length = lengths
+                    .next()
+                    .ok_or_else(|| failure("relay requires Content-Length"))?
+                    .map_err(|_| failure("invalid relay Content-Length"))?;
+                if lengths.next().is_some()
+                    || length > MAX_MESSAGE - boundary
+                    || header
+                        .lines()
+                        .filter_map(|line| line.split_once(':'))
+                        .any(|(name, _)| name.eq_ignore_ascii_case("transfer-encoding"))
+                {
+                    return Err(failure("unsupported relay HTTP framing"));
+                }
+                expected = Some(boundary + length);
+            } else if bytes.len() > MAX_HEADER {
+                return Err(failure("relay HTTP header budget exceeded"));
+            }
+        }
+        if let Some(expected) = expected {
+            if bytes.len() == expected {
+                return Ok(bytes);
+            }
+            if bytes.len() > expected {
+                return Err(failure("unexpected relay trailing bytes"));
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn framing_rejects_incomplete_ambiguous_or_unbounded_messages() {
+        for bytes in [
+            b"HTTP/1.1 200 OK\r\n\r\n".as_slice(),
+            b"HTTP/1.1 200 OK\r\nContent-Length: 8\r\n\r\nshort",
+            b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nContent-Length: 0\r\n\r\n",
+            b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nTransfer-Encoding: chunked\r\n\r\n",
+            b"HTTP/1.1 200 OK\r\nContent-Length: 999999999\r\n\r\n",
+        ] {
+            assert!(read_message(&mut &bytes[..]).is_err());
+        }
+        let valid = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n{}";
+        assert_eq!(read_message(&mut &valid[..]).unwrap(), valid);
+    }
+}

tests/e2e/tests/worker_restore.rs

@@ -143,6 +143,186 @@ fn concurrent_fixed_id_restores_keep_sessions_storage_and_cleanup_isolated() ->
     result
 }
 
+#[test]
+fn lost_http_response_recovers_exact_request_and_rejects_fresh_stale_intent() -> Result<()> {
+    response_loss_recovery(false)
+}
+
+#[test]
+fn committed_response_loss_survives_owned_runtime_process_kill_and_restart() -> Result<()> {
+    response_loss_recovery(true)
+}
+
+fn response_loss_recovery(crash_runtime: bool) -> Result<()> {
+    let fixture = Fixture::load()?;
+    let result = (|| {
+        let mut installed = seed::start(&fixture)?;
+        let stopped = assert_single_state(&installed.environment, &fixture, "stopped")?;
+        let token = stopped
+            .restore_observation_token
+            .as_deref()
+            .filter(|token| !token.is_empty())
+            .ok_or_else(|| failure("missing stopped observation token"))?;
+        let request_id = "fixed-fixture-response-loss";
+        phase(
+            &installed.environment,
+            "drop-http-response-after-complete-upstream",
+        )?;
+        let observed = installed.environment.restore_worker_losing_response(
+            &fixture.manifest.worker_id,
+            request_id,
+            token,
+        )?;
+        let receipt: WorkerRestoreResponse = serde_json::from_value(observed.clone())?;
+        if receipt.result.state != WorkerRestoreState::Accepted {
+            return Err(failure(
+                "upstream did not reach Accepted before response loss",
+            ));
+        }
+        fs::write(
+            installed
+                .environment
+                .artifacts_dir
+                .join("response-loss.json"),
+            serde_json::to_vec_pretty(&serde_json::json!({
+                "caller_received_response": false,
+                "relay_upstream_state": receipt.result.state,
+                "request_id": request_id,
+                "crash_after_committed_terminal_receipt": crash_runtime,
+            }))?,
+        )?;
+        let owner_before = restore_owner_evidence(&installed.environment, &fixture, request_id)?;
+        if crash_runtime {
+            phase(
+                &installed.environment,
+                "kill-owned-runtime-after-committed-receipt",
+            )?;
+            installed.environment.crash_and_restart_runtime()?;
+        }
+        phase(
+            &installed.environment,
+            "same-request-original-token-recovery",
+        )?;
+        let recovered =
+            installed
+                .environment
+                .restore_worker(&fixture.manifest.worker_id, request_id, token)?;
+        if crash_runtime {
+            // The public response projects the current observation fence. Runtime
+            // startup invalidates prior active observations; this is not a change
+            // to the original request token or its saved outcome.
+            let current = installed
+                .environment
+                .worker_detail(&fixture.manifest.worker_id)?;
+            assert_eq!(
+                recovered.pointer("/result/worker/restore_observation_token"),
+                current.get("restore_observation_token")
+            );
+            let mut expected = observed.clone();
+            expected["result"]["worker"]["restore_observation_token"] =
+                recovered["result"]["worker"]["restore_observation_token"].clone();
+            assert_eq!(
+                recovered, expected,
+                "restart may change only the public observation fence, not the saved outcome"
+            );
+        } else {
+            assert_eq!(
+                recovered, observed,
+                "exact request must recover the saved receipt"
+            );
+        }
+        assert_single_state(&installed.environment, &fixture, "idle")?;
+        let mut subscription = WorkerSessionSubscription::connect(
+            &installed.environment,
+            &fixture.manifest.worker_id,
+            &installed.snapshot,
+        )?;
+        subscription.assert_alive()?;
+        let stale = installed.environment.restore_worker(
+            &fixture.manifest.worker_id,
+            "fixed-fixture-fresh-stale-intent",
+            token,
+        );
+        match stale {
+            Err(error) if error.to_string().contains("HTTP 409") => {}
+            _ => {
+                return Err(failure(
+                    "fresh intent with old observation was not HTTP 409",
+                ));
+            }
+        }
+        let owner_after = restore_owner_evidence(&installed.environment, &fixture, request_id)?;
+        assert_eq!(
+            owner_after, owner_before,
+            "saved owner/operation/receipt must not change"
+        );
+        fs::write(
+            installed
+                .environment
+                .artifacts_dir
+                .join("restore-owner-evidence.json"),
+            serde_json::to_vec_pretty(&serde_json::json!({
+                "before": owner_before, "after": owner_after,
+                "request_token_replaced": false,
+            }))?,
+        )?;
+        installed.environment.assert_connected()?;
+        subscription.assert_alive()?;
+        stop_and_cleanup(installed, vec![subscription], &fixture)
+    })();
+    fixture.source.assert_unchanged()?;
+    result
+}
+
+// Read-only diagnostic of the product-written aggregate; never seeds or edits it.
+// Persist only a digest, operation ID and owner count, not prompts or credentials.
+fn restore_owner_evidence(
+    environment: &WorkerE2eEnvironment,
+    fixture: &Fixture,
+    request_id: &str,
+) -> Result<serde_json::Value> {
+    use sha2::{Digest, Sha256};
+    use std::io::Read;
+    let path = environment
+        .runtime_data_dir
+        .join("runtime/workers")
+        .join(&fixture.manifest.worker_id)
+        .join("worker.json");
+    let mut bytes = Vec::new();
+    fs::File::open(path)?
+        .take(8 * 1024 * 1024 + 1)
+        .read_to_end(&mut bytes)?;
+    if bytes.len() > 8 * 1024 * 1024 {
+        return Err(failure("Worker aggregate byte budget exceeded"));
+    }
+    let aggregate: serde_json::Value = serde_json::from_slice(&bytes)?;
+    let owners = aggregate
+        .pointer("/restore_guard/owners")
+        .and_then(serde_json::Value::as_object)
+        .ok_or_else(|| failure("canonical Worker omitted restore owners"))?;
+    let owner = owners
+        .get(request_id)
+        .ok_or_else(|| failure("canonical owner missing"))?;
+    if owner
+        .pointer("/receipt/state")
+        .and_then(serde_json::Value::as_str)
+        != Some("accepted")
+    {
+        return Err(failure("canonical owner has no Accepted receipt"));
+    }
+    let operation = owner
+        .get("operation_id")
+        .and_then(serde_json::Value::as_str)
+        .ok_or_else(|| failure("canonical owner omitted operation ID"))?;
+    let digest: String = Sha256::digest(serde_json::to_vec(owner)?)
+        .iter()
+        .map(|byte| format!("{byte:02x}"))
+        .collect();
+    Ok(
+        serde_json::json!({"operation_id": operation, "owner_sha256": digest, "owner_count": owners.len()}),
+    )
+}
+
 fn restore(
     installed: &Installed,
     fixture: &Fixture,