Replace killpg-on-self with thread-scoped subprocess kill on quit

Removes the shell's "Killed" message and lets fill3's terminal-restore
codes flush via normal interpreter exit. Worker.kill() now SIGKILLs only
the worker thread's subprocess descendants (via /proc/PID/task/TID/children,
recursing through shell=True grandchildren), so editor and xdg-open spawns
from the main thread survive quit. A _killed flag skips persisting result
pickles for aborted tool runs. start_new_session=True is no longer needed
on editor/xdg-open launches since the kill is worker-thread-scoped.
This commit is contained in:
Andrew Hamilton 2026-08-04 08:31:28 +10:00
parent b932ebc6b8
commit 980ee7071c
3 changed files with 80 additions and 8 deletions

View file

@ -34,6 +34,47 @@ class WorkerTestCase(unittest.TestCase):
result_path = os.path.join(tools.CACHE_PATH, "foo-metadata")
self.assertTrue(os.path.exists(result_path))
def test_kill_terminates_in_flight_subprocess(self):
import subprocess
import threading
import time
loop = asyncio.new_event_loop()
self.worker_ = worker.Worker("none")
process_holder = {}
original_run_tool_no_error = tools.run_tool_no_error
def slow_tool(path, tool):
process = subprocess.Popen(["sleep", "60"])
process_holder["process"] = process
process.wait()
return tools.Status.ok, ""
tools.run_tool_no_error = slow_tool
try:
results = []
def run_future():
results.append(loop.run_until_complete(self.worker_.run_tool("foo", tools.metadata)))
future_thread = threading.Thread(target=run_future)
future_thread.start()
deadline = time.time() + 5
while "process" not in process_holder and time.time() < deadline:
time.sleep(0.01)
self.assertIn("process", process_holder)
start = time.time()
self.worker_.kill()
future_thread.join(timeout=5)
self.assertLess(time.time() - start, 5)
self.assertFalse(future_thread.is_alive())
self.assertEqual(len(results), 1)
self.assertIsNotNone(process_holder["process"].poll())
result_path = os.path.join(tools.CACHE_PATH, "foo-metadata")
self.assertFalse(os.path.exists(result_path))
finally:
tools.run_tool_no_error = original_run_tool_no_error
if __name__ == "__main__":
unittest.main()