Add Kill Job button to Running Tasks screen

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
shuki
2026-03-06 18:10:19 +02:00
parent 8a83812584
commit 2379c2fdeb
2 changed files with 38 additions and 1 deletions

View File

@@ -77,6 +77,16 @@ class JobManager:
app.post_message(JobFinished(job.id, rc))
return job.return_code if job.return_code is not None else 1
def kill_job(self, job_id: str) -> bool:
job = self._jobs.get(job_id)
if not job or job._proc is None:
return False
try:
job._proc.terminate()
return True
except ProcessLookupError:
return False
def kill_running(self) -> None:
for job in self._jobs.values():
if job._proc is not None:

View File

@@ -6,7 +6,7 @@ from textual.widgets import Header, Footer, Static, Button, DataTable
from textual.containers import Vertical, Horizontal
from tui.jobs import job_manager
from tui.widgets import OperationLog
from tui.widgets import ConfirmDialog, OperationLog
class RunningTasksScreen(Screen):
@@ -20,6 +20,7 @@ class RunningTasksScreen(Screen):
yield DataTable(id="rt-table")
with Horizontal(id="rt-buttons"):
yield Button("View Log", variant="primary", id="btn-rt-view")
yield Button("Kill Job", variant="error", id="btn-rt-kill")
yield Button("Clear Finished", variant="warning", id="btn-rt-clear")
yield Button("Back", id="btn-rt-back")
yield Footer()
@@ -66,6 +67,8 @@ class RunningTasksScreen(Screen):
elif event.button.id == "btn-rt-clear":
job_manager.remove_finished()
self._refresh_table()
elif event.button.id == "btn-rt-kill":
self._kill_selected()
elif event.button.id == "btn-rt-view":
table = self.query_one("#rt-table", DataTable)
if table.row_count == 0:
@@ -82,5 +85,29 @@ class RunningTasksScreen(Screen):
)
self.app.push_screen(log_screen)
def _kill_selected(self) -> None:
table = self.query_one("#rt-table", DataTable)
if table.row_count == 0:
self.notify("No jobs to kill", severity="warning")
return
row_key, _ = table.coordinate_to_cell_key(table.cursor_coordinate)
job_id = str(row_key)
job = job_manager.get_job(job_id)
if not job:
return
if job.status != "running":
self.notify("Job is not running", severity="warning")
return
self.app.push_screen(
ConfirmDialog(f"Kill job '{job.label}'?", "Confirm Kill"),
callback=lambda ok: self._do_kill(job_id) if ok else None,
)
def _do_kill(self, job_id: str) -> None:
if job_manager.kill_job(job_id):
self.notify("Job killed")
else:
self.notify("Could not kill job", severity="error")
def action_go_back(self) -> None:
self.app.pop_screen()