fix(guard): split on & and unwrap subshells so real daemon kills classify (Closes #787)

`runtime_recovery_guard._SEGMENT_SPLIT_RE` claimed to split a compound command
line "on shell separators", but omitted the background-job separator `&`, and no
caller stripped a subshell wrapper before tokenising. `_analyse_kill_segment`
only inspects the first command token of a segment, so in both forms the kill
verb never reached the classifier.

Measured against the previous implementation at 35e94e10:

    sleep 1 & pkill -f mcp_server.py   ->  process_kill False, contamination False
    (pkill -f mcp_server.py)           ->  process_kill False, contamination False

Changes:

- Add `&` to `_SEGMENT_SPLIT_RE` as `(?:\|\||&&|[|;&\n])`; the two-character
  logical forms stay first so `&&` and `||` are consumed whole.
- Add `_strip_subshell()` and apply it in `_split_segments()`, removing leading
  `(` and trailing `)` (including nested wrappers) before tokenising.

Both forms now classify as contamination with reason class `manual_daemon_kill`,
and the record tool writes a durable marker for each. Direct-command behaviour is
unchanged. The contamination model, gated-task set, marker lifecycle, and
reconciler-only clearing path are untouched.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

runtime_recovery_guard.py

@@ -86,8 +86,12 @@ REMEDIATION = (
 
 # Split a compound command line into simple commands on shell separators so
 # ``ps aux | grep mcp_server`` is analysed segment by segment and its harmless
-# inspection half never reaches the kill classifier.
+# inspection half never reaches the kill classifier. The background separator
-_SEGMENT_SPLIT_RE = re.compile(r"(?:\|\||&&|\||;|\n)")
+# ``&`` is a separator too: without it ``sleep 1 & pkill -f mcp_server.py`` was
+# a single segment whose command position held ``sleep``, so the kill was never
+# classified (#787). The two-character logical forms are listed first so ``&&``
+# and ``||`` are consumed whole rather than split into single characters.
+_SEGMENT_SPLIT_RE = re.compile(r"(?:\|\||&&|[|;&\n])")
 
 _KILL_VERBS = frozenset({"kill", "pkill", "killall"})
 
@@ -133,8 +137,24 @@ def _clean(value: str | None) -> str:
     return (value or "").strip()
 
 
+def _strip_subshell(segment: str) -> str:
+    """Remove subshell wrappers so ``(pkill -f mcp_server.py)`` is classified.
+
+    The parentheses are shell syntax, not part of the simple command, so a
+    wrapped kill otherwise put ``(pkill`` in command position and never
+    reached the kill classifier (#787). Nested wrappers are unwrapped too.
+    """
+    stripped = segment.strip()
+    while stripped.startswith("("):
+        stripped = stripped[1:].strip()
+    while stripped.endswith(")"):
+        stripped = stripped[:-1].strip()
+    return stripped
+
+
 def _split_segments(command: str) -> list[str]:
-    return [seg for seg in _SEGMENT_SPLIT_RE.split(command) if seg.strip()]
+    segments = (_strip_subshell(seg) for seg in _SEGMENT_SPLIT_RE.split(command))
+    return [seg for seg in segments if seg]
 
 
 def is_sanctioned_recovery(text: str | None) -> bool:

tests/test_issue_630_runtime_recovery_guard.py

@@ -91,6 +91,44 @@ def test_compound_command_detects_the_kill_half():
     assert result["contamination"] is True
 
 
+# ── #787: background separator and subshell forms reach the classifier ───────
+
+def test_background_separator_kill_is_contamination():
+    result = guard.classify_recovery_command("sleep 1 & pkill -f mcp_server.py")
+    assert result["process_kill"] is True
+    assert result["contamination"] is True
+    assert result["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL
+    assert result["ambiguous"] is False
+
+
+def test_subshell_wrapped_kill_is_contamination():
+    result = guard.classify_recovery_command("(pkill -f mcp_server.py)")
+    assert result["process_kill"] is True
+    assert result["contamination"] is True
+    assert result["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL
+    assert result["ambiguous"] is False
+
+
+def test_further_background_and_subshell_forms_are_contamination():
+    for command in (
+        "pkill -f mcp_server.py &",
+        "( sudo pkill -f mcp_server.py )",
+        "((pkill -f gitea_mcp_server))",
+        "sleep 1 & killall mcp_server",
+        "(ps aux | grep mcp_server) & pkill -f mcp_server.py",
+    ):
+        result = guard.classify_recovery_command(command)
+        assert result["contamination"] is True, command
+        assert result["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL, command
+
+
+def test_logical_operators_are_not_split_into_single_characters():
+    # ``&&``/``||`` must still be consumed whole by the widened alternation.
+    assert guard._split_segments("a && b || c") == ["a", "b", "c"]
+    assert guard._split_segments("a & b") == ["a", "b"]
+    assert guard._split_segments("(a)") == ["a"]
+
+
 # ── no false positives ───────────────────────────────────────────────────────
 
 def test_read_only_inspection_is_not_a_kill():
@@ -112,6 +150,22 @@ def test_unrelated_pkill_target_is_not_contamination():
     assert result["ambiguous"] is False
 
 
+def test_user_scoped_pkill_of_unrelated_app_is_not_contamination():
+    # ``-u`` consumes ``mcpuser``; the surviving operand names no daemon (#787).
+    result = guard.classify_recovery_command("pkill -u mcpuser -f myapp")
+    assert result["process_kill"] is True
+    assert result["contamination"] is False
+    assert result["ambiguous"] is False
+
+
+def test_commit_message_quoting_the_kill_string_is_not_a_kill():
+    result = guard.classify_recovery_command(
+        'git commit -m "block pkill -f mcp_server.py as workflow recovery"'
+    )
+    assert result["process_kill"] is False
+    assert result["contamination"] is False
+
+
 def test_bare_kill_of_unknown_pid_is_ambiguous_not_contamination():
     result = guard.classify_recovery_command("kill 31337")
     assert result["contamination"] is False
@@ -333,6 +387,32 @@ def test_record_tool_marks_manual_daemon_kill():
     assert "mcp_server.py" in loaded["command_summary"]
 
 
+def test_record_tool_marks_background_separator_kill():
+    _clear_marker()
+    res = srv.gitea_record_daemon_process_kill_attempt(
+        command="sleep 1 & pkill -f mcp_server.py", remote="prgs"
+    )
+    assert res["contaminated"] is True
+    assert res["marked"] is True
+    assert res["marker"]["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL
+    loaded = srv._load_runtime_recovery_marker("prgs")
+    assert loaded is not None
+    assert "mcp_server.py" in loaded["command_summary"]
+
+
+def test_record_tool_marks_subshell_wrapped_kill():
+    _clear_marker()
+    res = srv.gitea_record_daemon_process_kill_attempt(
+        command="(pkill -f mcp_server.py)", remote="prgs"
+    )
+    assert res["contaminated"] is True
+    assert res["marked"] is True
+    assert res["marker"]["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL
+    loaded = srv._load_runtime_recovery_marker("prgs")
+    assert loaded is not None
+    assert "mcp_server.py" in loaded["command_summary"]
+
+
 def test_record_tool_marks_broad_sweep():
     _clear_marker()
     res = srv.gitea_record_daemon_process_kill_attempt(