fix(installer): preserve healthy Python runtime and Codex connections

.gitea/workflows/ci.yml

@@ -35,6 +35,8 @@ jobs:
           "$RUNNER_TEMP/ci-venv/bin/python" -m pytest tests/tools/test_protected_instruction_unattended.py tests/tools/test_file_write_safety.py tests/tools/test_computer_use_approval_isolation.py tests/cli/test_cli_approval_ui.py tests/cli/test_single_query_session_finalize.py -q -p no:cacheprovider -rf
       - name: Codex bridge regression tests
         run: '"$RUNNER_TEMP/ci-venv/bin/python" -m unittest discover -s apps/desktop/assets/codex-bridge'
+      - name: Managed Python update preservation
+        run: '"$RUNNER_TEMP/ci-venv/bin/python" -m pytest tests/test_install_ps1_reuse_healthy_venv.py tests/test_install_ps1_python_fallback_venv.py -q -p no:cacheprovider -rf'
       - name: Project scoped and bounded search regressions
         run: '"$RUNNER_TEMP/ci-venv/bin/python" -m pytest tests/tools/test_search_project_scope.py tests/tools/test_search_files_cpu_windows.py tests/tools/test_search_zero_match_and_multipath.py tests/tools/test_search_giant_line_containment.py tests/tools/test_search_budget_truncation.py -q -p no:cacheprovider -rf'
       - name: Native watch and speech mux pytest regressions

apps/desktop/assets/codex-bridge/bridge.py

@@ -32,9 +32,10 @@ import time
 import urllib.error
 import urllib.request
 import gecis_context
+import terminal_stream
 
 HERE = os.path.dirname(os.path.abspath(__file__))
-SURUM = "17"         # AITURK IDE'nin bekledigi kopru surumu (codex-link-config.ts CODEX_BRIDGE_VERSION)
+SURUM = "18"         # AITURK IDE'nin bekledigi kopru surumu (codex-link-config.ts CODEX_BRIDGE_VERSION)
 try:
     with open(os.path.abspath(__file__), "rb") as _f:
         KOD = hashlib.sha256(_f.read()).hexdigest()[:8]
@@ -348,6 +349,36 @@ class H(http.server.BaseHTTPRequestHandler):
             self.end_headers()
             self.wfile.write(hata)
             return True
+        observer = None
+        first_chunks = []
+        failure_key = getattr(self, "_failure_key", None)
+        if (hedef == KOPRU and failure_key is not None and durum == 200
+                and "text/event-stream" in yanit.headers.get("Content-Type", "")
+                and yanit.headers.get("Content-Encoding", "identity") == "identity"):
+            observer = terminal_stream.Observer()
+            if not started:
+                # Inspect only until the first SSE frame/comment. Keep normal
+                # streaming and heartbeats; an error-only response can still
+                # return a truthful HTTP failure before headers are committed.
+                try:
+                    while True:
+                        first = yanit.read1(4096)
+                        if not first:
+                            break
+                        first_chunks.append(first)
+                        observer.feed(first)
+                        if observer.error and not observer.payload_seen:
+                            yanit.close()
+                            self._json(400, terminal_stream.error_body(observer.error))
+                            return True
+                        if observer.comment_seen or observer.payload_seen or sum(map(len, first_chunks)) >= observer.LIMIT:
+                            break
+                except (OSError, http.client.HTTPException):
+                    # A transport interruption without an explicit provider
+                    # failure keeps the client's existing recovery behavior.
+                    yanit.close()
+                    self.close_connection = True
+                    return True
         if not started:
             self.send_response(durum)
             for k, v in yanit.headers.items():
@@ -359,6 +390,10 @@ class H(http.server.BaseHTTPRequestHandler):
         event_tail = b""
         completed = False
         try:
+            for first in first_chunks:
+                toplam += len(first)
+                self.wfile.write(b"%x\r\n" % len(first) + first + b"\r\n")
+                self.wfile.flush()
             while True:
                 try:
                     parca = yanit.read1(4096)      # gelen kadarini hemen verir; bekletmez
@@ -370,6 +405,8 @@ class H(http.server.BaseHTTPRequestHandler):
                 if not parca:
                     break
                 toplam += len(parca)
+                if observer is not None:
+                    observer.feed(parca)
                 if getattr(self, "_native_return_scope", None):
                     event_tail += parca
                     while b"\n" in event_tail:
@@ -382,6 +419,9 @@ class H(http.server.BaseHTTPRequestHandler):
                                 pass
                 self.wfile.write(b"%x\r\n" % len(parca) + parca + b"\r\n")
                 self.wfile.flush()
+            if observer is not None and observer.error and not observer.completed:
+                self.server.failure_replays.remember(failure_key, observer.error)
+                log(f"{tag} terminal worker error; identical automatic replay will return HTTP 400")
             self.wfile.write(b"0\r\n\r\n")
             self.wfile.flush()
         except (ConnectionAbortedError, BrokenPipeError, ConnectionResetError):
@@ -453,6 +493,7 @@ class H(http.server.BaseHTTPRequestHandler):
 
     def _isle(self):
         self._native_return_scope = None
+        self._failure_key = None
         with kilit:
             sira[0] += 1
             n = sira[0]
@@ -479,6 +520,12 @@ class H(http.server.BaseHTTPRequestHandler):
                     req = {}
             except Exception:
                 req = {}
+            if external:
+                self._failure_key = terminal_stream.request_key(req or {"raw": govde.hex()}, self.headers)
+                previous_failure = self.server.failure_replays.take(self._failure_key)
+                if previous_failure:
+                    log(f"[{n} {model}] terminal worker error replay stopped before upstream request")
+                    return self._json(400, terminal_stream.error_body(previous_failure))
             thread = str(req.get("prompt_cache_key") or "")
             scope = hashlib.sha256((str(self.headers.get("chatgpt-account-id") or "") + "|" + thread).encode()).hexdigest()
             source = TRANSFERS.get(scope) if thread else None
@@ -541,6 +588,10 @@ class H(http.server.BaseHTTPRequestHandler):
 
 
 class Sunucu(http.server.ThreadingHTTPServer):
+    def __init__(self, *args, **kwargs):
+        self.failure_replays = terminal_stream.FailureReplays()
+        super().__init__(*args, **kwargs)
+
     daemon_threads = True
     # Shipped Python 3.11 defaults to only five pending connections. Model
     # discovery, voice and simultaneous chats can arrive in the same burst.

apps/desktop/assets/codex-bridge/kopru.py

@@ -40,7 +40,7 @@ from bekci_watch import MetadataFilter
 # Installed only by owned native launcher, never an HTTP control endpoint.
 WATCH_OWNER = None
 
-SURUM = "17"
+SURUM = "18"
 HERE = os.path.dirname(os.path.abspath(__file__))
 try:
     with open(os.path.abspath(__file__), "rb") as _f:

apps/desktop/assets/codex-bridge/terminal_stream.py

@@ -0,0 +1,119 @@
+"""Transport-only handling of a worker's explicit terminal failure.
+
+Never selects a model, evaluates content, or changes worker/Jev decisions.
+If HTTP headers have already gone out, remember the failure for exactly one
+identical retry. Codex then receives an HTTP error instead of re-running the
+whole exhausted provider chain. Ordinary network interruptions remain retries.
+"""
+import hashlib
+import json
+import threading
+import time
+from collections import OrderedDict
+
+
+def request_key(request, headers):
+    principal = "\0".join(headers.get(name, "") for name in
+                          ("Authorization", "ChatGPT-Account-Id", "session_id"))
+    body = json.dumps(request, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
+    return hashlib.sha256((principal + "\0" + body).encode()).digest()
+
+
+class FailureReplays:
+    def __init__(self, ttl=120, capacity=128, clock=time.monotonic):
+        self.ttl, self.capacity, self.clock = ttl, capacity, clock
+        self.items = OrderedDict()
+        self.lock = threading.Lock()
+
+    def remember(self, key, error):
+        with self.lock:
+            self.items[key] = (self.clock() + self.ttl, error)
+            self.items.move_to_end(key)
+            while len(self.items) > self.capacity:
+                self.items.popitem(last=False)
+
+    def take(self, key):
+        with self.lock:
+            entry = self.items.pop(key, None)
+        return entry[1] if entry and entry[0] > self.clock() else None
+
+
+def error_body(error):
+    # Preserve the actual provider failure; don't relabel it invalid_prompt,
+    # fabricate response.completed, or expose a rejected model response.
+    return {"error": {"type": "upstream_terminal_error", "code": error["code"],
+                      "message": "Model yanıtı hata ile sonlandı. " + error["message"]}}
+
+
+class Observer:
+    """Bounded SSE observer. Forwarded stream bytes are never rewritten."""
+    LIMIT = 65536
+
+    def __init__(self):
+        self.tail = b""
+        self.data = []
+        self.size = 0
+        self.discard = False
+        self.discard_line = False
+        self.comment_seen = False
+        self.payload_seen = False
+        self.completed = False
+        self.error = None
+
+    def feed(self, chunk):
+        # read1() supplies small chunks, but also bound arbitrary caller input.
+        for offset in range(0, len(chunk), 4096):
+            self.tail += chunk[offset:offset + 4096]
+            while b"\n" in self.tail:
+                line, self.tail = self.tail.split(b"\n", 1)
+                if self.discard_line:
+                    self.discard_line = False
+                    continue
+                self._line(line.rstrip(b"\r"))
+            if len(self.tail) > self.LIMIT:
+                self.tail = b""
+                self.discard = self.discard_line = True
+
+    def _line(self, line):
+        if line.startswith(b":"):
+            self.comment_seen = True
+        elif line.startswith(b"data:"):
+            self.size += len(line)
+            if self.size > self.LIMIT:
+                self.discard = True
+                self.data.clear()
+            elif not self.discard:
+                self.data.append(line[5:].lstrip())
+        elif not line:
+            if self.data and not self.discard:
+                self._event(b"\n".join(self.data))
+            elif self.discard:
+                self.payload_seen = True
+            self.data, self.size, self.discard = [], 0, False
+
+    def _event(self, data):
+        try:
+            event = json.loads(data)
+        except (ValueError, UnicodeError):
+            self.payload_seen = True
+            return
+        if not isinstance(event, dict):
+            self.payload_seen = True
+            return
+        kind = event.get("type")
+        response = event.get("response")
+        error = event.get("error") if kind == "error" else response.get("error") if isinstance(response, dict) else None
+        # Known worker terminal errors only. Preserve context/rate-limit and
+        # other native error classifications rather than guessing semantics.
+        is_failure = (kind == "error" and isinstance(error, dict)
+                      and error.get("type") == "upstream_error")
+        is_failure |= (kind == "response.failed" and isinstance(error, dict)
+                       and error.get("code") == "jev_no_fallback")
+        if is_failure:
+            self.error = {"code": str(error.get("code") or "upstream_error")[:100],
+                          "message": str(error.get("message") or "Sağlayıcı yanıtı tamamlanamadı.")[:2000]}
+        else:
+            self.payload_seen = True
+        if kind == "response.completed":
+            self.completed = True
+            self.error = None

apps/desktop/assets/codex-bridge/test_terminal_stream.py

@@ -0,0 +1,143 @@
+"""Real-socket regressions for terminal worker errors (no provider calls)."""
+import json
+import unittest
+
+import test_ayirici as fixtures
+import terminal_stream
+
+
+ERROR = b'event: error\ndata: {"type":"error","error":{"type":"upstream_error","message":"upstream 503"}}\n\n'
+COMPLETE = b'event: response.completed\ndata: {"type":"response.completed","response":{"id":"resp_ok","status":"completed","output":[]}}\n\n'
+
+
+class TerminalStream(unittest.TestCase):
+    setUp = fixtures.Ayirici.setUp
+    tearDown = fixtures.Ayirici.tearDown
+    iste = fixtures.Ayirici.iste
+
+    def request(self, token="a", text="check"):
+        body = json.dumps({"model": "ide/test", "input": [{"role": "user", "content": text}]})
+        return self.iste("POST", "/backend-api/codex/responses", body,
+                         {"Authorization": "Bearer " + token})
+
+    def worker(self, payload):
+        def respond(h, body):
+            if h.path.endswith("/prepare-transfer"):
+                return fixtures.duz(200, json.dumps({"request": json.loads(body)["request"]}).encode())(h, body)
+            return fixtures.duz(200, payload, "text/event-stream")(h, body)
+        self.kopru.cevap = respond
+
+    def inference_count(self):
+        return sum(x["yol"].endswith("/responses") for x in self.kopru.gelen)
+
+    def test_error_only_stream_is_an_honest_http_failure(self):
+        self.worker(ERROR)
+        status, body, _ = self.request()
+        self.assertEqual(status, 400)
+        self.assertIn("upstream 503", json.loads(body)["error"]["message"])
+        self.assertNotIn(b"response.completed", body)
+        self.assertEqual(self.inference_count(), 1)
+
+    def test_error_after_heartbeat_stops_automatic_replay_once(self):
+        def respond(h, body):
+            if h.path.endswith("/prepare-transfer"):
+                return fixtures.duz(200, json.dumps({"request": json.loads(body)["request"]}).encode())(h, body)
+            h.send_response(200)
+            h.send_header("Content-Type", "text/event-stream")
+            h.send_header("Transfer-Encoding", "chunked")
+            h.end_headers()
+            for chunk in (b": keepalive\n\n", ERROR):
+                h.wfile.write(b"%x\r\n" % len(chunk) + chunk + b"\r\n")
+                h.wfile.flush()
+                fixtures.time.sleep(.08)
+            h.wfile.write(b"0\r\n\r\n")
+        self.kopru.cevap = respond
+        status, body, _ = self.request()
+        self.assertEqual((status, body), (200, b": keepalive\n\n" + ERROR))
+        status, body, _ = self.request()
+        self.assertEqual(status, 400)
+        self.assertIn(b"upstream 503", body)
+        self.assertEqual(self.inference_count(), 1)
+        # A later explicit retry is allowed; the one-shot guard is consumed.
+        self.worker(COMPLETE)
+        self.assertEqual(self.request()[:2], (200, COMPLETE))
+        self.assertEqual(self.inference_count(), 2)
+
+    def test_native_error_stream_is_not_intercepted(self):
+        self.ozgun.cevap = fixtures.duz(200, ERROR, "text/event-stream")
+        body = json.dumps({"model": "gpt-6-astra", "input": []})
+        self.assertEqual(self.iste("POST", "/backend-api/codex/responses", body)[:2], (200, ERROR))
+
+    def test_success_stream_is_byte_identical(self):
+        self.worker(COMPLETE)
+        self.assertEqual(self.request()[:2], (200, COMPLETE))
+        self.assertEqual(self.request()[:2], (200, COMPLETE))
+        self.assertEqual(self.inference_count(), 2)
+
+    def test_eof_without_upstream_error_keeps_network_retry(self):
+        self.worker(b": keepalive\n\n")
+        self.assertEqual(self.request()[0], 200)
+        self.assertEqual(self.request()[0], 200)
+        self.assertEqual(self.inference_count(), 2)
+
+    def test_worker_exhausted_failure_is_terminal_without_changing_its_reason(self):
+        self.worker(b'event: response.failed\ndata: {"type":"response.failed","response":{"status":"failed","error":{"code":"jev_no_fallback","message":"No accepted response"}}}\n\n')
+        status, body, _ = self.request()
+        self.assertEqual(status, 400)
+        self.assertEqual(json.loads(body)["error"]["code"], "jev_no_fallback")
+
+    def test_existing_context_error_classification_is_preserved(self):
+        data = b'event: response.failed\ndata: {"type":"response.failed","response":{"error":{"code":"context_length_exceeded","message":"Context full"}}}\n\n'
+        self.worker(data)
+        self.assertEqual(self.request()[:2], (200, data))
+
+
+class ObserverTests(unittest.TestCase):
+    def test_fragmented_crlf_and_multiline_json(self):
+        data = b'event: error\r\ndata: {"type":"error",\r\ndata: "error":{"type":"upstream_error","message":"503"}}\r\n\r\n'
+        observer = terminal_stream.Observer()
+        for byte in data:
+            observer.feed(bytes([byte]))
+        self.assertEqual(observer.error['message'], '503')
+
+    def test_large_and_malformed_frames_are_bounded_and_do_not_crash(self):
+        observer = terminal_stream.Observer()
+        observer.feed(b'data: ' + b'x'*200000)
+        self.assertLessEqual(len(observer.tail), observer.LIMIT)
+        observer.feed(b'\n\ndata: {"response":[]}\n\n' + ERROR)
+        self.assertEqual(observer.error['message'], 'upstream 503')
+
+    def test_completed_stream_does_not_poison_next_request(self):
+        observer = terminal_stream.Observer()
+        observer.feed(ERROR + COMPLETE)
+        self.assertIsNone(observer.error)
+        self.assertTrue(observer.completed)
+
+    def test_one_shot_expiry_and_capacity(self):
+        now = [10]
+        cache = terminal_stream.FailureReplays(ttl=5, capacity=2, clock=lambda: now[0])
+        cache.remember(b'a', {'code': 'a'})
+        self.assertEqual(cache.take(b'a'), {'code': 'a'})
+        self.assertIsNone(cache.take(b'a'))
+        cache.remember(b'a', {})
+        now[0] = 16
+        self.assertIsNone(cache.take(b'a'))
+        for key in (b'a', b'b', b'c'):
+            cache.remember(key, {'code': 'error'})
+        self.assertIsNone(cache.take(b'a'))
+        self.assertEqual(len(cache.items), 2)
+
+    def test_request_identity_separates_accounts_sessions_models_and_inputs(self):
+        base = {'model': 'ide/a', 'input': []}
+        headers = {'Authorization': 'Bearer a', 'session_id': 'one'}
+        key = terminal_stream.request_key(base, headers)
+        self.assertEqual(key, terminal_stream.request_key(dict(reversed(list(base.items()))), headers))
+        for request, other in [({'model': 'ide/b', 'input': []}, headers),
+                               ({'model': 'ide/a', 'input': ['next']}, headers),
+                               (base, dict(headers, Authorization='Bearer b')),
+                               (base, dict(headers, session_id='two'))]:
+            self.assertNotEqual(key, terminal_stream.request_key(request, other))
+
+
+if __name__ == "__main__":
+    unittest.main()

apps/desktop/electron/codex-link-config.ts

@@ -27,7 +27,7 @@ export const CODEX_LINK_REFRESH_URL = `${CODEX_LINK_ORIGIN}/aiturk/codex-baglama
 export const CODEX_LINK_MARKER = '# AITURK IDE Codex Baglama'
 // Must equal SURUM in assets/codex-bridge/bridge.py; a running bridge that reports
 // another version is restarted so an updated IDE never keeps an old bridge alive.
-export const CODEX_BRIDGE_VERSION = '17'
+export const CODEX_BRIDGE_VERSION = '18'
 export const CODEX_LINK_LINE = `openai_base_url = "${CODEX_LINK_ORIGIN}/backend-api/codex"  ${CODEX_LINK_MARKER}`
 
 export const CODEX_LINK_RUN_KEY = 'HKCU\\Software\\Microsoft\\Windows\\CurrentVersion\\Run'

apps/desktop/package.json

@@ -2,7 +2,7 @@
   "name": "aiturk-ide",
   "productName": "AITURK IDE",
   "private": true,
-  "version": "1.0.0-beta.36",
+  "version": "1.0.0-beta.37",
   "description": "TurkServis resmi yapay zekâ geliştirme ortamı. Hermes Agent tabanlıdır.",
   "author": "AITURK / TurkServis",
   "repository": {

package-lock.json

@@ -65,7 +65,7 @@
     },
     "apps/desktop": {
       "name": "aiturk-ide",
-      "version": "1.0.0-beta.36",
+      "version": "1.0.0-beta.37",
       "license": "MIT",
       "dependencies": {
         "@assistant-ui/core": "0.2.23",

scripts/install.ps1

@@ -2550,6 +2550,39 @@ function Install-Repository {
     Write-Success "Repository ready"
 }
 
+function Test-ReusableManagedVenv {
+    param([string]$VenvPath, [string]$BasePython, [string]$Version)
+    $pythonExe = Join-Path $VenvPath "Scripts\python.exe"
+    if (-not (Test-Path -LiteralPath $pythonExe -PathType Leaf)) { return $false }
+    $process = New-Object System.Diagnostics.Process
+    try {
+        $info = New-Object System.Diagnostics.ProcessStartInfo
+        $info.FileName = $pythonExe
+        $info.Arguments = '-I -c "import json,sys,ssl,ctypes,venv,encodings; print(json.dumps(dict(version=str(sys.version_info.major)+''.''+str(sys.version_info.minor),prefix=sys.prefix,base=sys._base_executable)))"'
+        $info.UseShellExecute = $false
+        $info.CreateNoWindow = $true
+        $info.RedirectStandardOutput = $true
+        $info.RedirectStandardError = $true
+        $process.StartInfo = $info
+        if (-not $process.Start()) { return $false }
+        $stdout = $process.StandardOutput.ReadToEndAsync()
+        $stderr = $process.StandardError.ReadToEndAsync()
+        if (-not $process.WaitForExit(15000)) {
+            try { $process.Kill() } catch { }
+            return $false
+        }
+        if ($process.ExitCode -ne 0) { return $false }
+        $health = $stdout.Result | ConvertFrom-Json
+        return ($health.version -eq $Version -and
+            [string]::Equals([IO.Path]::GetFullPath($health.prefix), [IO.Path]::GetFullPath($VenvPath), [StringComparison]::OrdinalIgnoreCase) -and
+            [string]::Equals([IO.Path]::GetFullPath($health.base), [IO.Path]::GetFullPath($BasePython), [StringComparison]::OrdinalIgnoreCase))
+    } catch {
+        return $false
+    } finally {
+        $process.Dispose()
+    }
+}
+
 function Install-Venv {
     if ($NoVenv) {
         Write-Info "Skipping virtual environment (-NoVenv)"
@@ -2567,6 +2600,19 @@ function Install-Venv {
         throw "Hermes-managed Python is unavailable. Run install.ps1 -Stage python first."
     }
 
+    # Routine source updates do not require replacing a healthy interpreter.
+    # Recreating it kills detached Codex bridges sharing this venv and leaves
+    # every subscription model disconnected until bootstrap finishes. Verify
+    # version, provenance and stdlib first; the dependency stage still performs
+    # the locked sync and import checks. Broken/mismatched runtimes retain the
+    # transactional repair path below.
+    $existingVenvPath = Join-Path $InstallDir "venv"
+    if (Test-ReusableManagedVenv -VenvPath $existingVenvPath -BasePython $resolvedPython.Path -Version $resolvedPython.Version) {
+        $env:UV_PYTHON = Join-Path $existingVenvPath "Scripts\python.exe"
+        Write-Success "Reusing healthy managed virtual environment; running Codex bridges preserved"
+        return
+    }
+
     Write-Info "Creating virtual environment with Python $($resolvedPython.Version)..."
     
     Push-Location $InstallDir

tests/test_install_ps1_reuse_healthy_venv.py

@@ -0,0 +1,77 @@
+"""Run the actual installer functions against a real disposable Python venv."""
+import json
+import os
+import shutil
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+SOURCE = Path(__file__).resolve().parents[1] / 'scripts/install.ps1'
+PS = shutil.which('powershell') or shutil.which('pwsh')
+pytestmark = pytest.mark.skipif(os.name != 'nt' or not PS, reason='Windows installer runtime test')
+
+
+def function(source, name):
+    start = source.index('function ' + name + ' {')
+    opening = source.index('{', start)
+    depth = 0
+    for index in range(opening, len(source)):
+        if source[index] == '{': depth += 1
+        elif source[index] == '}': depth -= 1
+        if depth == 0: return source[start:index + 1]
+    raise AssertionError(name)
+
+
+@pytest.fixture(scope='module')
+def runtime(tmp_path_factory):
+    root = tmp_path_factory.mktemp('installer-reuse')
+    subprocess.run([sys.executable, '-m', 'venv', '--without-pip', str(root / 'venv')], check=True)
+    return root
+
+
+def run(runtime, body):
+    text = SOURCE.read_text(encoding='utf-8-sig')
+    script = runtime / 'probe.ps1'
+    prefix = "$ErrorActionPreference = 'Stop'\n"
+    prefix += function(text, 'Test-ReusableManagedVenv') + '\n'
+    prefix += function(text, 'Install-Venv') + '\n'
+    prefix += "$InstallDir = '" + str(runtime).replace("'", "''") + "'\n"
+    prefix += "$BasePython = '" + sys._base_executable.replace("'", "''") + "'\n"
+    prefix += "$Version = '" + str(sys.version_info.major) + '.' + str(sys.version_info.minor) + "'\n"
+    script.write_text(prefix + body, encoding='utf-8-sig')
+    result = subprocess.run([PS, '-NoProfile', '-File', str(script)], capture_output=True, text=True, timeout=40)
+    assert result.returncode == 0, result.stdout + result.stderr
+    return result.stdout
+
+
+def test_healthy_update_keeps_live_python_and_never_reaches_kill_sweep(runtime):
+    python = runtime / 'venv/Scripts/python.exe'
+    live = subprocess.Popen([str(python), '-c', 'import time;time.sleep(90)'])
+    try:
+        output = run(runtime, r'''
+function Resolve-AvailablePythonVersion { return @{Path=$BasePython;Version=$Version} }
+function Write-Success { param($Message); Write-Output $Message }
+function Write-Info { throw 'Unexpected venv recreation/kill path' }
+function taskkill { throw 'Live process must not be killed' }
+Install-Venv
+''')
+        assert 'Reusing healthy' in output
+        assert live.poll() is None
+        assert python.is_file()
+    finally:
+        live.terminate();live.wait(timeout=10)
+
+
+def test_wrong_version_or_foreign_interpreter_does_not_reuse(runtime):
+    output = run(runtime, r'''
+$venvPath = Join-Path $InstallDir 'venv'
+$results = @(
+  (Test-ReusableManagedVenv $venvPath $BasePython '0.0'),
+  (Test-ReusableManagedVenv $venvPath (Join-Path $InstallDir 'other-python.exe') $Version),
+  (Test-ReusableManagedVenv (Join-Path $InstallDir 'missing') $BasePython $Version)
+)
+ConvertTo-Json -Compress -InputObject $results
+''')
+    assert json.loads(output) == [False, False, False]