fix(approval): destructive-tool registry (#56)

* refactor(approval): make destructive-tool classification a registry seam

* feat(approval): add DeliberationScope contextvar primitive

* fix(approval): scope deliberation one-shot to context + generation

* feat(approval): bind deliberation scope around per-step tool execution

* fix(approval): propagate deliberation scope into tool futures

* fix(approval): preserve newer deliberation generations

* docs(changelog): note auto-deliberation scoping

Author: elkaix

CHANGELOG.md

@@ -15,6 +15,10 @@ GitHub Releases page; `0.8.0` is the new starting line.
 
 ## Unreleased
 
+- **Auto-mode destructive actions deliberate per turn and context.** Auto-deliberation now
+  scopes destructive-command one-shots to the active execution context and LLM generation,
+  so duplicate destructive calls in one response keep bouncing while later deliberate retries
+  and isolated subagent calls are handled independently.
 - **Release packaging keeps SDK/core pins in lockstep.** The SDK's `pythinker-core`
   dependency is now updated by release automation and checked by CI/release validation,
   preventing no-sources binary builds from resolving against a stale core pin.

docs/en/release-notes/changelog.md

@@ -17,6 +17,10 @@ GitHub Releases page; `0.8.0` is the new starting line.
 
 ## Unreleased
 
+- **Auto-mode destructive actions deliberate per turn and context.** Auto-deliberation now
+  scopes destructive-command one-shots to the active execution context and LLM generation,
+  so duplicate destructive calls in one response keep bouncing while later deliberate retries
+  and isolated subagent calls are handled independently.
 - **Release packaging keeps SDK/core pins in lockstep.** The SDK's `pythinker-core`
   dependency is now updated by release automation and checked by CI/release validation,
   preventing no-sources binary builds from resolving against a stale core pin.

src/pythinker_code/soul/approval.py

@@ -2,7 +2,10 @@
 
 import json
 import uuid
-from collections.abc import Callable
+from collections.abc import Callable, Generator
+from contextlib import contextmanager
+from contextvars import ContextVar
+from dataclasses import dataclass
 from typing import Literal
 
 from pythinker_core.utils.typing import JsonType
@@ -30,6 +33,33 @@
 )
 
 
+@dataclass(frozen=True)
+class DeliberationScope:
+    """Execution context + LLM generation a deliberation decision is scoped to.
+
+    ``context_id`` separates the main agent from each subagent (approval state is shared
+    via ``Approval.share()``); ``generation`` is the step number within that context.
+    """
+
+    context_id: str
+    generation: int
+
+
+_current_deliberation_scope: ContextVar[DeliberationScope | None] = ContextVar(
+    "deliberation_scope", default=None
+)
+
+
+@contextmanager
+def deliberation_scope(context_id: str, generation: int) -> Generator[None, None, None]:
+    """Bind the active deliberation scope for the duration of one step's tool execution."""
+    token = _current_deliberation_scope.set(DeliberationScope(context_id, generation))
+    try:
+        yield
+    finally:
+        _current_deliberation_scope.reset(token)
+
+
 class ApprovalResult:
     """Result of an approval request. Behaves as bool for backward compatibility."""
 
@@ -107,8 +137,9 @@ def __init__(
         """Set of action names that should automatically be approved."""
         self.approved_orchestration_fingerprints: set[str] = set()
         """RunAgents orchestration shapes approved for this in-memory session."""
-        self.deliberated_fingerprints: set[str] = set()
-        """Destructive (tool, command) shapes already bounced once; the re-issue runs."""
+        self.deliberated_fingerprints: dict[str, int] = {}
+        """Maps a context-namespaced destructive fingerprint to the generation it was last
+        bounced at; a re-issue in a later generation of the same context consumes it once."""
         self._on_change = on_change
 
     def notify_change(self) -> None:
@@ -211,10 +242,12 @@ def _tool_arguments(tool_call: ToolCall) -> dict[str, JsonType] | None:
         return args if isinstance(args, dict) else None
 
     @staticmethod
-    def _deliberation_fingerprint(tool_name: str, arguments: dict[str, JsonType]) -> str:
-        """Stable, tool-agnostic identity for a destructive call (name + sorted args)."""
+    def _deliberation_fingerprint(
+        context_id: str, tool_name: str, arguments: dict[str, JsonType]
+    ) -> str:
+        """Context-namespaced identity for a destructive call (context + name + sorted args)."""
         encoded = json.dumps(arguments, sort_keys=True, separators=(",", ":"))
-        return f"{tool_name}::{encoded}"
+        return f"{context_id}::{tool_name}::{encoded}"
 
     def deliberation_gate(self, tool_call: ToolCall) -> str | None:
         """Reason a destructive auto-approved action must deliberate once, else ``None``.
@@ -223,9 +256,11 @@ def deliberation_gate(self, tool_call: ToolCall) -> str | None:
         auto-approved (auto *or* yolo — so it gates ahead of the yolo bypass), and the
         tool call is destructive per the tool-agnostic classifier in ``permission``
         (today only ``Shell``; other destructive tools register their classifier there).
-        One-shot: the first occurrence is bounced for the agent to weigh alternatives;
-        the identical re-issue is let through once, so a deliberated ``rm -rf`` runs
-        without being permanently whitelisted.
+        One-shot, scoped to (execution context, generation): the first sighting and any
+        same-generation duplicate are bounced; only a re-issue in a later generation of the
+        same context is let through once, so a deliberated ``rm -rf`` runs without being
+        permanently whitelisted, while two identical calls in one model response both
+        deliberate and a subagent cannot consume the main agent's one-shot.
         """
         if not self._state.auto_deliberate:
             return None
@@ -239,20 +274,22 @@ def deliberation_gate(self, tool_call: ToolCall) -> str | None:
         reason = tool_destructive_reason(tool_call.function.name, arguments)
         if reason is None:
             return None
-        fingerprint = self._deliberation_fingerprint(tool_call.function.name, arguments)
-        # NOTE (known limitation, tracked): the one-shot is keyed only by the
-        # (tool, command) fingerprint, not by an assistant-turn boundary. If a
-        # model emits two byte-identical destructive calls within the SAME
-        # response (no intervening deliberation turn), the second consumes the
-        # one-shot and runs. Distinguishing that from a genuine re-issue needs a
-        # turn/generation signal not plumbed into the approval layer; spec §6 #2
-        # treats the one-shot as an open decision. Only reachable when a user has
-        # opted into the auto_deliberate policy (not a default), and requires the
-        # model to emit identical destructive calls in one response.
-        if fingerprint in self._state.deliberated_fingerprints:
-            self._state.deliberated_fingerprints.discard(fingerprint)  # consume one-shot
-            return None
-        self._state.deliberated_fingerprints.add(fingerprint)
+        scope = _current_deliberation_scope.get()
+        context_id = scope.context_id if scope is not None else "unscoped"
+        generation = scope.generation if scope is not None else 0
+        fingerprint = self._deliberation_fingerprint(context_id, tool_call.function.name, arguments)
+        # One-shot keyed by (execution context, generation): the first sighting and any
+        # same-generation duplicate are bounced; only a re-issue in a strictly LATER
+        # generation of the same context is let through once. The context_id prefix prevents
+        # a subagent's identical call from consuming the main agent's one-shot (state is
+        # shared via Approval.share()).
+        prior_generation = self._state.deliberated_fingerprints.get(fingerprint)
+        if prior_generation is not None:
+            if prior_generation < generation:
+                del self._state.deliberated_fingerprints[fingerprint]
+                return None
+            return reason
+        self._state.deliberated_fingerprints[fingerprint] = generation
         return reason
 
     async def request(

src/pythinker_code/soul/permission.py

@@ -2,6 +2,7 @@
 
 import re
 import shlex
+from collections.abc import Callable
 from contextvars import ContextVar, Token
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, Literal
@@ -497,18 +498,27 @@ def shell_destructive_reason(command: str) -> str | None:
     return None
 
 
+def _shell_args_destructive_reason(arguments: dict[str, Any]) -> str | None:
+    command = arguments.get("command")
+    return shell_destructive_reason(command) if isinstance(command, str) else None
+
+
+# The single, auditable place where a tool opts into auto-deliberation. Today only the
+# irreversible surface (Shell, including background shell — same tool name) is classified;
+# reversible file tools (WriteFile/StrReplaceFile: restore-point + VCS backed) are
+# intentionally excluded. A future destructive tool adds one entry here.
+_DESTRUCTIVE_CLASSIFIERS: dict[str, Callable[[dict[str, Any]], str | None]] = {
+    "Shell": _shell_args_destructive_reason,
+}
+
+
 def tool_destructive_reason(tool_name: str, arguments: dict[str, Any]) -> str | None:
     """Reason a tool call is irreversibly destructive (warrants deliberation), else ``None``.
 
-    Tool-agnostic dispatch point for the auto-deliberation gate. Today only ``Shell``
-    is classified; a future destructive tool registers its own argument classifier
-    here instead of the gate hard-coding a single tool name.
+    Declarative dispatch: classification lives in ``_DESTRUCTIVE_CLASSIFIERS``.
     """
-    if tool_name == "Shell":
-        command = arguments.get("command")
-        if isinstance(command, str):
-            return shell_destructive_reason(command)
-    return None
+    classifier = _DESTRUCTIVE_CLASSIFIERS.get(tool_name)
+    return classifier(arguments) if classifier is not None else None
 
 
 def _segment_destructive_reason(tokens: list[str]) -> str | None:

src/pythinker_code/soul/pythinkersoul.py

@@ -51,6 +51,7 @@
     wire_send,
 )
 from pythinker_code.soul.agent import Agent, Runtime
+from pythinker_code.soul.approval import deliberation_scope
 from pythinker_code.soul.compaction import (
     CompactionResult,
     SimpleCompaction,
@@ -311,6 +312,8 @@ def __init__(
         self._approval = agent.runtime.approval
         self._context = context
         self._loop_control = agent.runtime.config.loop_control
+        self._current_step_no = 0
+        self._deliberation_generation = 0
         self._sleep_inhibitor = SleepInhibitor(enabled=agent.runtime.config.prevent_idle_sleep)
         self._compaction = SimpleCompaction()  # TODO: maybe configurable and composable
 
@@ -1329,6 +1332,19 @@ async def _step(self) -> StepOutcome | None:
         # already checked in `run`
         assert self._runtime.llm is not None
         chat_provider = self._runtime.llm.chat_provider
+        self._deliberation_generation += 1
+        deliberation_generation = self._deliberation_generation
+        approval_source = get_current_approval_source_or_none()
+        if approval_source is not None:
+            deliberation_context_id = f"{approval_source.kind}:{approval_source.id}"
+            if approval_source.agent_id is not None:
+                deliberation_context_id = f"{deliberation_context_id}:{approval_source.agent_id}"
+        elif self._runtime.subagent_id is not None:
+            deliberation_context_id = self._runtime.subagent_id
+        elif self._runtime.role == "root":
+            deliberation_context_id = "root"
+        else:
+            deliberation_context_id = f"subagent:{self._runtime.session.id}"
 
         if self._runtime.role == "root":
 
@@ -1394,14 +1410,15 @@ async def _run_step_once() -> StepResult:
                         permission_profile_for_runtime(self._runtime)
                     )
                     try:
-                        step_result = await pythinker_core.step(
-                            chat_provider,
-                            self._agent.system_prompt,
-                            self._agent.toolset,
-                            effective_history,
-                            on_message_part=wire_send,
-                            on_tool_result=wire_send,
-                        )
+                        with deliberation_scope(deliberation_context_id, deliberation_generation):
+                            step_result = await pythinker_core.step(
+                                chat_provider,
+                                self._agent.system_prompt,
+                                self._agent.toolset,
+                                effective_history,
+                                on_message_part=wire_send,
+                                on_tool_result=wire_send,
+                            )
                     finally:
                         reset_step_permission_profile(profile_token)
                 except Exception as exc:
@@ -1491,7 +1508,12 @@ async def _pythinker_core_step_with_retry() -> StepResult:
 
         # wait for all tool results (may be interrupted)
         plan_mode_before_tools = self._plan_mode
-        results = await result.tool_results()
+        # Scope the deliberation one-shot to this context + step. Tool futures normally
+        # inherit this ContextVar when created during pythinker_core.step above; keeping
+        # it bound here also covers any future implementation that starts work lazily in
+        # tool_results().
+        with deliberation_scope(deliberation_context_id, deliberation_generation):
+            results = await result.tool_results()
         logger.debug("Got tool results: {results}", results=results)
 
         # If a tool (EnterPlanMode/ExitPlanMode) changed plan mode during execution,

tests/core/test_approval_auto.py

@@ -4,7 +4,7 @@
 
 import json
 
-from pythinker_code.soul.approval import Approval, ApprovalState
+from pythinker_code.soul.approval import Approval, ApprovalState, deliberation_scope
 from pythinker_code.wire.types import ToolCall
 
 
@@ -15,6 +15,39 @@ def _shell_call(cmd: str) -> ToolCall:
     )
 
 
+def test_tool_destructive_reason_gates_background_shell() -> None:
+    from pythinker_code.soul.permission import tool_destructive_reason
+
+    # Background shell is the same "Shell" tool (run_in_background=true); a destructive
+    # background command must still be classified as destructive.
+    reason = tool_destructive_reason(
+        "Shell", {"command": "rm -rf build", "run_in_background": True}
+    )
+    assert reason is not None
+
+
+def test_tool_destructive_reason_ignores_unregistered_tool() -> None:
+    from pythinker_code.soul.permission import tool_destructive_reason
+
+    assert (
+        tool_destructive_reason("WriteFile", {"path": "x", "content": "y", "mode": "overwrite"})
+        is None
+    )
+
+
+def test_deliberation_scope_sets_and_restores_contextvar() -> None:
+    from pythinker_code.soul.approval import (
+        DeliberationScope,
+        _current_deliberation_scope,
+        deliberation_scope,
+    )
+
+    assert _current_deliberation_scope.get() is None
+    with deliberation_scope("root", 3):
+        assert _current_deliberation_scope.get() == DeliberationScope("root", 3)
+    assert _current_deliberation_scope.get() is None
+
+
 def test_yolo_only() -> None:
     approval = Approval(yolo=True)
     assert approval.is_yolo() is True
@@ -117,19 +150,43 @@ def test_set_auto_false_clears_runtime_auto() -> None:
 
 
 def test_destructive_action_deliberates_once_then_proceeds_under_auto() -> None:
-    """auto + auto_deliberate: a destructive Shell command deliberates the first
-    time, the identical re-issue runs once (one-shot retry), and a third issue
-    deliberates again — so deliberation never permanently whitelists ``rm -rf``."""
+    """auto + auto_deliberate: a destructive command deliberates the first time, the
+    re-issue in a LATER generation runs once, and a fresh issue later deliberates again."""
     approval = Approval(state=ApprovalState(auto=True, auto_deliberate=True))
+    with deliberation_scope("root", 1):
+        assert approval.deliberation_gate(_shell_call("rm -rf build")) is not None
+    with deliberation_scope("root", 2):
+        assert approval.deliberation_gate(_shell_call("rm -rf build")) is None
+    with deliberation_scope("root", 3):
+        assert approval.deliberation_gate(_shell_call("rm -rf build")) is not None
 
-    first = approval.deliberation_gate(_shell_call("rm -rf build"))
-    assert first is not None, "first destructive issue should deliberate"
 
-    second = approval.deliberation_gate(_shell_call("rm -rf build"))
-    assert second is None, "identical re-issue is the one-shot retry: allowed through"
+def test_same_generation_duplicate_destructive_calls_both_bounce() -> None:
+    # Property (a): two byte-identical destructive calls in ONE generation both deliberate.
+    approval = Approval(state=ApprovalState(auto=True, auto_deliberate=True))
+    with deliberation_scope("root", 1):
+        assert approval.deliberation_gate(_shell_call("rm -rf build")) is not None
+        assert approval.deliberation_gate(_shell_call("rm -rf build")) is not None
+
 
-    third = approval.deliberation_gate(_shell_call("rm -rf build"))
-    assert third is not None, "one-shot consumed; a fresh issue deliberates again"
+def test_subagent_identical_call_does_not_consume_main_one_shot() -> None:
+    # Property (c): a subagent's identical call must not ride on the main agent's bounce.
+    approval = Approval(state=ApprovalState(auto=True, auto_deliberate=True))
+    with deliberation_scope("root", 1):
+        assert approval.deliberation_gate(_shell_call("rm -rf build")) is not None
+    with deliberation_scope("sub-1", 1):
+        assert approval.deliberation_gate(_shell_call("rm -rf build")) is not None
+
+
+def test_older_generation_duplicate_destructive_call_still_bounces() -> None:
+    # Defensive guard: only a strictly later generation can consume a prior bounce.
+    approval = Approval(state=ApprovalState(auto=True, auto_deliberate=True))
+    with deliberation_scope("root", 2):
+        assert approval.deliberation_gate(_shell_call("rm -rf build")) is not None
+    with deliberation_scope("root", 1):
+        assert approval.deliberation_gate(_shell_call("rm -rf build")) is not None
+    with deliberation_scope("root", 2):
+        assert approval.deliberation_gate(_shell_call("rm -rf build")) is not None
 
 
 def test_deliberation_gate_conditions() -> None:
@@ -164,14 +221,16 @@ async def test_request_bounces_destructive_then_approves_retry() -> None:
 
     approval = Approval(state=ApprovalState(auto=True, auto_deliberate=True))
     with tool_call_context("Shell", arguments={"command": "rm -rf build"}):
-        first = await approval.request("Shell", "run command", "Run command `rm -rf build`")
+        with deliberation_scope("root", 1):
+            first = await approval.request("Shell", "run command", "Run command `rm -rf build`")
         assert not first, "destructive action is bounced for deliberation"
         assert first.deliberation is True
         assert "irreversible" in first.feedback
         assert "rejected by the user" not in first.rejection_error().message
 
-        second = await approval.request("Shell", "run command", "Run command `rm -rf build`")
-        assert second, "one-shot consumed: the deliberated retry runs"
+        with deliberation_scope("root", 2):
+            second = await approval.request("Shell", "run command", "Run command `rm -rf build`")
+        assert second, "one-shot consumed in a later generation: the deliberated retry runs"
 
 
 def test_approval_state_honors_auto_deliberate_flag() -> None: