fix(process): label background completion causes (#46659)

Track why a background process finished and include that source in notify-on-complete messages so SIGTERM from process.kill, kill_all, backend loss, and ordinary exits are distinguishable.

tests/tools/test_notify_on_complete.py

@@ -12,7 +12,7 @@ import json
 import os
 import time
 import pytest
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 from tools.process_registry import (
     ProcessRegistry,
@@ -99,6 +99,8 @@ class TestCompletionQueue:
         assert completion["session_id"] == s.id
         assert completion["command"] == "echo hello"
         assert completion["exit_code"] == 0
+        assert completion["completion_reason"] == "exited"
+        assert completion["termination_source"] == ""
         assert "build succeeded" in completion["output"]
 
     def test_move_to_finished_nonzero_exit(self, registry):
@@ -138,6 +140,35 @@ class TestCompletionQueue:
         completion = registry.completion_queue.get_nowait()
         assert completion["exit_code"] == -15  # from the first (kill) call
 
+    def test_kill_process_sets_completion_reason_and_source(self, registry):
+        s = _make_session(notify_on_complete=True, output="stopping")
+        s.process = MagicMock()
+        s.process.pid = 4242
+        registry._running[s.id] = s
+
+        class FakeProcess:
+            def __init__(self, pid):
+                self.pid = pid
+
+            def children(self, recursive=False):
+                return []
+
+            def terminate(self):
+                pass
+
+        import psutil as _psutil
+
+        with patch.object(_psutil, "Process", side_effect=lambda pid: FakeProcess(pid)), \
+             patch.object(registry, "_write_checkpoint"):
+            result = registry.kill_process(s.id)
+
+        assert result["status"] == "killed"
+        assert result["completion_reason"] == "killed"
+        assert result["termination_source"] == "process.kill"
+        completion = registry.completion_queue.get_nowait()
+        assert completion["completion_reason"] == "killed"
+        assert completion["termination_source"] == "process.kill"
+
     def test_output_truncated_to_2000(self, registry):
         """Long output is truncated to last 2000 chars."""
         long_output = "x" * 5000

tests/tools/test_process_registry.py

@@ -1012,12 +1012,41 @@ def test_format_completion_event():
         "output": "done",
     }
     result = format_process_notification(evt)
-    assert "[IMPORTANT: Background process proc_abc completed" in result
+    assert "[IMPORTANT: Background process proc_abc completed normally" in result
     assert "exit code 0" in result
     assert "Command: sleep 5" in result
     assert "Output:\ndone]" in result
 
 
+def test_format_killed_completion_event_names_source_and_signal():
+    evt = {
+        "type": "completion",
+        "session_id": "proc_killed",
+        "command": "sleep 5",
+        "exit_code": -15,
+        "completion_reason": "killed",
+        "termination_source": "process.kill",
+        "output": "",
+    }
+    result = format_process_notification(evt)
+    assert "proc_killed terminated by process.kill" in result
+    assert "exit code -15, SIGTERM" in result
+
+
+def test_format_external_sigterm_does_not_claim_agent_kill():
+    evt = {
+        "type": "completion",
+        "session_id": "proc_external",
+        "command": "sleep 5",
+        "exit_code": 143,
+        "output": "",
+    }
+    result = format_process_notification(evt)
+    assert "proc_external exited" in result
+    assert "terminated by" not in result
+    assert "exit code 143, SIGTERM" in result
+
+
 def test_format_watch_match_event():
     evt = {
         "type": "watch_match",
@@ -1087,7 +1116,7 @@ def test_drain_notifications_returns_pending_events():
         results = process_registry.drain_notifications()
         assert len(results) == 2
         assert results[0][0]["session_id"] == "proc_drain1"
-        assert "proc_drain1 completed" in results[0][1]
+        assert "proc_drain1 completed normally" in results[0][1]
         assert results[1][0]["session_id"] == "proc_drain2"
         assert "watch pattern" in results[1][1]
     finally: