fix(runtime): fail released context handles instead of truncating them

crates/execution/assets/undici-shims/stream.js

@@ -45,6 +45,37 @@ const normalizeStreamError = (value) => {
 	return new Error(String(value));
 };
 
+// Web streams expose no way to observe close or error without locking the
+// stream. Node's eos waits on an internal promise that does not keep the event
+// loop alive, so watch web-streams-polyfill `_state` transitions. A polling
+// timer here held the guest process open while a fetch body stayed unread.
+const webStreamSettleListeners = new WeakMap();
+
+const watchWebStreamState = (stream) => {
+	const listeners = new Set();
+	let state = stream._state;
+	Object.defineProperty(stream, "_state", {
+		configurable: true,
+		enumerable: true,
+		get: () => state,
+		set: (next) => {
+			state = next;
+			if (next === "closed" || next === "errored") {
+				for (const listener of listeners) listener();
+			}
+		},
+	});
+	webStreamSettleListeners.set(stream, listeners);
+	return listeners;
+};
+
+const onWebStreamSettled = (stream, listener) => {
+	const listeners =
+		webStreamSettleListeners.get(stream) ?? watchWebStreamState(stream);
+	listeners.add(listener);
+	return () => listeners.delete(listener);
+};
+
 export const finished = (stream, options, callback) => {
 	let normalizedOptions = options;
 	let normalizedCallback = callback;
@@ -65,14 +96,12 @@ export const finished = (stream, options, callback) => {
 	const readableEnabled = normalizedOptions?.readable !== false;
 	const writableEnabled = normalizedOptions?.writable !== false;
 	let cancelled = false;
-	let timer = null;
+	let unsubscribe = null;
 
 	const cleanup = () => {
 		cancelled = true;
-		if (timer !== null) {
-			clearTimeout(timer);
-			timer = null;
-		}
+		unsubscribe?.();
+		unsubscribe = null;
 	};
 
 	const complete = (error = undefined) => {
@@ -83,27 +112,31 @@ export const finished = (stream, options, callback) => {
 		queueMicrotask(() => done(error));
 	};
 
-	const poll = () => {
+	const check = () => {
 		if (cancelled) {
-			return;
+			return true;
 		}
 		const state = stream?._state;
 		if (state === "errored") {
 			complete(normalizeStreamError(stream?._storedError));
-			return;
+			return true;
 		}
 		if (
 			state === "closed" ||
 			(isWebReadableStream(stream) && !readableEnabled) ||
 			(isWebWritableStream(stream) && !writableEnabled)
 		) {
 			complete();
-			return;
+			return true;
 		}
-		timer = setTimeout(poll, 0);
+		return false;
 	};
 
-	poll();
+	if (!check()) {
+		// The polyfill assigns `_storedError` after `_state`, so read it once the
+		// transition has finished.
+		unsubscribe = onWebStreamSettled(stream, () => queueMicrotask(check));
+	}
 	return cleanup;
 };
 

crates/native-sidecar/tests/language_execution.rs

@@ -1117,6 +1117,32 @@ fn execution_timeout_is_enforced_by_the_sidecar() {
     dispose_vm_and_close_session_wire(&mut sidecar, &connection_id, &session_id, &vm_id);
 }
 
+fn evaluate_in_context(
+    sidecar: &mut agentos_native_sidecar::NativeSidecar<support::RecordingBridge>,
+    (connection_id, session_id, vm_id): (&str, &str, &str),
+    context_id: &str,
+    format: wire::JavaScriptModuleFormat,
+    expression: &str,
+) -> wire::ExecutionCompletedResponse {
+    let mut process = context_process_options(context_id);
+    process.timeout_ms = Some(5_000);
+    let started = sidecar
+        .dispatch_wire_blocking(wire_request(
+            5,
+            wire_vm(connection_id, session_id, vm_id),
+            wire::RequestPayload::JavaScriptEvaluationRequest(wire::JavaScriptEvaluationRequest {
+                process,
+                expression: expression.to_owned(),
+                format: Some(format),
+                file_path: None,
+                inputs: None,
+            }),
+        ))
+        .expect("start context evaluation");
+    let execution_id = accepted_execution_id(started);
+    wait_for_execution(sidecar, connection_id, session_id, vm_id, &execution_id)
+}
+
 #[test]
 fn context_call_completes_when_its_result_settles_despite_leftover_handles() {
     let mut sidecar = new_sidecar("language-execution-leftover-handles");
@@ -1132,32 +1158,18 @@ fn context_call_completes_when_its_result_settles_despite_leftover_handles() {
         &cwd,
     );
 
-    let mut request_id = 4;
-    let mut evaluate =
+    let evaluate =
         |sidecar: &mut agentos_native_sidecar::NativeSidecar<support::RecordingBridge>,
          context_id: &str,
          format: wire::JavaScriptModuleFormat,
          expression: &str| {
-            let mut process = context_process_options(context_id);
-            process.timeout_ms = Some(5_000);
-            request_id += 1;
-            let started = sidecar
-                .dispatch_wire_blocking(wire_request(
-                    request_id,
-                    wire_vm(&connection_id, &session_id, &vm_id),
-                    wire::RequestPayload::JavaScriptEvaluationRequest(
-                        wire::JavaScriptEvaluationRequest {
-                            process,
-                            expression: expression.to_owned(),
-                            format: Some(format),
-                            file_path: None,
-                            inputs: None,
-                        },
-                    ),
-                ))
-                .expect("start context evaluation");
-            let execution_id = accepted_execution_id(started);
-            wait_for_execution(sidecar, &connection_id, &session_id, &vm_id, &execution_id)
+            evaluate_in_context(
+                sidecar,
+                (&connection_id, &session_id, &vm_id),
+                context_id,
+                format,
+                expression,
+            )
         };
 
     for (context_id, format) in [
@@ -1202,6 +1214,112 @@ fn context_call_completes_when_its_result_settles_despite_leftover_handles() {
     dispose_vm_and_close_session_wire(&mut sidecar, &connection_id, &session_id, &vm_id);
 }
 
+#[test]
+fn released_context_socket_fails_reads_in_later_calls() {
+    let mut sidecar = new_sidecar("language-execution-released-socket");
+    let connection_id = authenticate_wire(&mut sidecar, "released-socket-connection");
+    let session_id = open_session_wire(&mut sidecar, 2, &connection_id);
+    let cwd = temp_dir("language-execution-released-socket-cwd");
+    let (vm_id, _) = create_vm_wire(
+        &mut sidecar,
+        3,
+        &connection_id,
+        &session_id,
+        wire::GuestRuntimeKind::JavaScript,
+        &cwd,
+    );
+    create_context(
+        &mut sidecar,
+        &connection_id,
+        &session_id,
+        &vm_id,
+        "released",
+    );
+    let ids = (connection_id.as_str(), session_id.as_str(), vm_id.as_str());
+    let stderr = |result: &wire::ExecutionCompletedResponse| {
+        String::from_utf8_lossy(result.stderr.as_deref().unwrap_or_default()).into_owned()
+    };
+
+    // The body is far larger than one read, so most of it is still on the
+    // socket when the first call completes and releases that socket.
+    let stored = evaluate_in_context(
+        &mut sidecar,
+        ids,
+        "released",
+        wire::JavaScriptModuleFormat::Module,
+        r#"await (async () => {
+            const http = require("node:http");
+            const body = Buffer.alloc(8 * 1024 * 1024, 97);
+            const server = http.createServer((_request, response) => {
+                response.setHeader("content-length", body.length);
+                response.end(body);
+            });
+            await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
+            globalThis.response = await fetch(`http://127.0.0.1:${server.address().port}/`);
+            return response.status;
+        })()"#,
+    );
+    assert_eq!(
+        stored.outcome,
+        wire::ExecutionOutcome::Succeeded,
+        "{:?}",
+        stored.error
+    );
+    assert_eq!(stored.evaluation_value.as_deref(), Some("200"));
+
+    let read = evaluate_in_context(
+        &mut sidecar,
+        ids,
+        "released",
+        wire::JavaScriptModuleFormat::Module,
+        "await response.arrayBuffer().then((body) => ({ bytes: body.byteLength }), (error) => ({ code: error.cause?.code ?? error.code }))",
+    );
+    assert_eq!(
+        read.outcome,
+        wire::ExecutionOutcome::Succeeded,
+        "{:?}",
+        read.error
+    );
+    assert_eq!(
+        read.evaluation_value.as_deref(),
+        Some(r#"{"code":"ERR_AGENTOS_CONTEXT_HANDLE_RELEASED"}"#),
+        "a released response body must fail, not end early"
+    );
+
+    // A call that finishes its own I/O leaves nothing to release.
+    let complete = evaluate_in_context(
+        &mut sidecar,
+        ids,
+        "released",
+        wire::JavaScriptModuleFormat::Module,
+        r#"await (async () => {
+            const http = require("node:http");
+            const server = http.createServer((_request, response) => {
+                response.setHeader("connection", "close");
+                response.end("ok");
+            });
+            await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
+            const text = await (await fetch(`http://127.0.0.1:${server.address().port}/`)).text();
+            await new Promise((resolve) => server.close(resolve));
+            return text;
+        })()"#,
+    );
+    assert_eq!(
+        complete.outcome,
+        wire::ExecutionOutcome::Succeeded,
+        "{:?}",
+        complete.error
+    );
+    assert_eq!(complete.evaluation_value.as_deref(), Some(r#""ok""#));
+    assert!(
+        !stderr(&complete).contains("work still pending"),
+        "a call that finished its fetch reported leftover work: {}",
+        stderr(&complete)
+    );
+
+    dispose_vm_and_close_session_wire(&mut sidecar, &connection_id, &session_id, &vm_id);
+}
+
 #[test]
 fn evaluation_rejects_non_json_values_with_a_structured_result() {
     let mut sidecar = new_sidecar("language-execution-json-evaluation");

crates/v8-runtime/CLAUDE.md

@@ -8,7 +8,7 @@
 - In `src/session.rs`, top-level `SessionMessage` handling must never silently drop late `BridgeResponse`, `StreamEvent`, or `TerminateExecution` frames after an execution finishes. If there is no active event loop to consume them, emit a structured `RuntimeEvent::Log` warning with an explicit `ERR_LATE_*` code so host-side diagnostics can see the loss.
 - In `src/session.rs`, sync bridge waits need the same per-execution abort path for both CPU timeouts and explicit `TerminateExecution` requests. If a terminate signal only hits `isolate_handle.terminate_execution()` and does not also close the active bridge-wait receiver, a guest blocked inside a sync host call will hang until the host replies.
 - In `src/session.rs`, `run_event_loop()` must consult `_getPendingTimerCount()` directly in its exit checks, not only pending promises or `_waitForActiveHandles()`. Some callers enter the loop without first registering the keepalive promise, and otherwise a lone ref'd `setInterval()` can let the session exit between ticks.
-- In `src/session.rs`, `Execute` mode bit `2` (retained context) or bit `4` (semantic evaluation) means complete-on-settle: the loop exits when the entry evaluation settles (`EventLoopExit::EvaluationSettled`), then `_releaseLeftoverHandles()` cancels leftover timers and closes ref'd handles, then the loop runs until no async bridge call is in flight (`EventLoopExit::BridgeIdle`). Never apply `_waitForActiveHandles()` to these calls; the idle session loop does not run guest callbacks, so anything left would freeze or keep the call open until its timeout.
+- In `src/session.rs`, `Execute` mode bit `2` (retained context) or bit `4` (semantic evaluation) means complete-on-settle: the loop exits when the entry evaluation settles (`EventLoopExit::EvaluationSettled`), the result is finalized, then `_releaseLeftoverHandles()` cancels ref'd timers and closes ref'd handles with `ERR_AGENTOS_CONTEXT_HANDLE_RELEASED` (never a clean EOF), then the loop runs until no async bridge call is in flight (`EventLoopExit::BridgeIdle`). Failures and unhandled rejections after the result is finalized go to the call's stderr via `report_after_call_failure` and never change its outcome or leak into a later call. Never apply `_waitForActiveHandles()` to these calls, and never cancel unref'd timers: runtime-internal timers (undici's shared timer, stream pollers) must be unref'd so they survive across calls.
 - Session-quota regressions now live in `src/embedded_runtime.rs`, not a removed daemon `main.rs`: keep one `SessionManager` per `EmbeddedV8Runtime`, share it across every runtime handle/connection to that instance, and cover quota behavior in `tests/embedded_runtime_session.rs`.
 - In `tests/embedded_runtime_session.rs`, shared-quota assertions must saturate the runtime slots before registering any session that is supposed to stay queued. `SessionManager` acquires slots during `CreateSession`, so relying on creation order alone makes queued-session tests race with the sessions meant to hold the slots.
 - Guest `process.memoryUsage()`, `process.cpuUsage()`, `process.resourceUsage()`, and live `process.versions.v8` should be resolved locally on the V8 session thread in `src/bridge.rs`, using `v8::HeapStatistics`, `getrusage(RUSAGE_THREAD)`, and the bundled library version APIs. Do not reintroduce stale JS literals or route these per-isolate values back through sidecar RPC.

crates/v8-runtime/src/execution.rs

@@ -1144,6 +1144,51 @@ pub fn release_leftover_handles(scope: &mut v8::HandleScope) -> Result<(), Execu
     }
 }
 
+/// Report a failure that happened after a settle-complete call's result was
+/// fixed. It goes to the call's stderr (sidecar stderr when the bridge is
+/// absent) instead of changing the call's outcome.
+pub fn report_after_call_failure(scope: &mut v8::HandleScope, error: &ExecutionError) {
+    let detail = if error.stack.is_empty() {
+        format!("{}: {}", error.error_type, error.message)
+    } else {
+        error.stack.clone()
+    };
+    let message = format!("agentOS: after the context call completed: {detail}\n");
+    if !write_guest_stderr(scope, &message) {
+        eprint!("{message}");
+    }
+}
+
+fn write_guest_stderr(scope: &mut v8::HandleScope, message: &str) -> bool {
+    let tc = &mut v8::TryCatch::new(scope);
+    let global = tc.get_current_context().global(tc);
+    let (Some(error_key), Some(apply_key), Some(text)) = (
+        v8::String::new(tc, "_error"),
+        v8::String::new(tc, "applySync"),
+        v8::String::new(tc, message),
+    ) else {
+        return false;
+    };
+    let Some(target) = global
+        .get(tc, error_key.into())
+        .and_then(|value| v8::Local::<v8::Object>::try_from(value).ok())
+    else {
+        return false;
+    };
+    let Some(apply) = target
+        .get(tc, apply_key.into())
+        .and_then(|value| v8::Local::<v8::Function>::try_from(value).ok())
+    else {
+        return false;
+    };
+    let args = v8::Array::new_with_elements(tc, &[text.into()]);
+    let undefined = v8::undefined(tc).into();
+    apply
+        .call(tc, target.into(), &[undefined, args.into()])
+        .is_some()
+        && !tc.has_caught()
+}
+
 fn serialize_module_exports(
     scope: &mut v8::HandleScope,
     module: v8::Local<v8::Module>,