fix(pipeline): avoid duplicate handoff confirmations

Author: guima-why

src/iac_code/a2a/pipeline_executor.py

@@ -412,6 +412,7 @@ def __init__(
         backup_service: Any | None = None,
         aliyun_delegated_executor_factory: Any | None = None,
         pipeline_name: str | None = None,
+        permission_context_getter: Callable[[], Any] | None = None,
     ) -> None:
         self._task_store = task_store
         self._model = model
@@ -438,6 +439,7 @@ def __init__(
         self._backup_service = backup_service or SessionBackupService()
         self._aliyun_delegated_executor_factory = aliyun_delegated_executor_factory
         self._pipeline_name_override = pipeline_name or None
+        self._permission_context_getter = permission_context_getter
 
     def _resolve_pipeline_name(self) -> str:
         """Pipeline this executor must run.
@@ -1678,7 +1680,9 @@ def _create_pipeline(
             services = getattr(runtime, "aliyun_services", None)
             delegated_factory = getattr(services, "delegated_executor_factory", None)
         agent_loop = getattr(runtime, "agent_loop", None)
-        permission_context_getter = getattr(agent_loop, "_permission_context_getter", None)
+        permission_context_getter = self._permission_context_getter
+        if not callable(permission_context_getter):
+            permission_context_getter = getattr(agent_loop, "_permission_context_getter", None)
         if not callable(permission_context_getter) and agent_loop is not None:
 
             def permission_context_getter() -> Any:

src/iac_code/pipeline/engine/handoff.py

@@ -24,6 +24,7 @@ def build_handoff_summary(
     outcome: TerminalOutcome,
     context_snapshot: dict,
     include_fields: list[str],
+    release_tools_any_allowed: bool | None = None,
 ) -> str:
     """Build deterministic text for continuing in normal chat after a pipeline."""
     included = {
@@ -43,18 +44,28 @@ def build_handoff_summary(
     if missing:
         lines.extend(["", "Missing context fields:"])
         lines.extend(f"- {field_name}" for field_name in missing)
-    lines.extend(
-        [
-            "",
-            "Safety requirements for normal chat:",
-            (
-                "- Before performing any operation that releases, deletes, or otherwise destroys a resource, "
-                "obtain a fresh, explicit confirmation from the user in normal chat. Any confirmation given "
-                "during the pipeline does not count."
-            ),
-            ("- Exception: pipeline-managed automatic cleanup may proceed without this additional confirmation."),
-            "",
-            "Use this context when answering follow-up questions after the pipeline handoff.",
-        ]
-    )
+    lines.extend(["", "Safety requirements for normal chat:"])
+    if release_tools_any_allowed is False:
+        lines.extend(
+            [
+                (
+                    "- For operations that release, delete, or otherwise destroy a resource, rely on the tool "
+                    "permission confirmation as the sole confirmation. Do not ask for a separate confirmation "
+                    "in normal chat, and do not proceed unless the permission request is approved."
+                ),
+                ("- Exception: pipeline-managed automatic cleanup may proceed without an additional confirmation."),
+            ]
+        )
+    else:
+        lines.extend(
+            [
+                (
+                    "- Before performing any operation that releases, deletes, or otherwise destroys a resource, "
+                    "obtain a fresh, explicit confirmation from the user in normal chat. Any confirmation given "
+                    "during the pipeline does not count."
+                ),
+                ("- Exception: pipeline-managed automatic cleanup may proceed without this additional confirmation."),
+            ]
+        )
+    lines.extend(["", "Use this context when answering follow-up questions after the pipeline handoff."])
     return "\n".join(lines)

src/iac_code/pipeline/engine/pipeline_runner.py

@@ -1093,8 +1093,50 @@ def build_normal_handoff_summary(self, completed_event_data: dict) -> str:
             outcome=outcome,
             context_snapshot=self.context.snapshot(),
             include_fields=include_fields,
+            release_tools_any_allowed=self._normal_chat_release_tools_any_allowed(),
         )
 
+    def _normal_chat_release_tools_any_allowed(self) -> bool | None:
+        permission_context = self._normal_chat_permission_context()
+        if permission_context is None:
+            return None
+        mode = getattr(permission_context, "mode", None)
+        mode_value = getattr(mode, "value", mode)
+        mode_name = mode_value if isinstance(mode_value, str) else None
+        allow_rules = getattr(permission_context, "allow_rules", {})
+        ask_rules = getattr(permission_context, "ask_rules", {})
+        deny_rules = getattr(permission_context, "deny_rules", {})
+        if not all(isinstance(rules, dict) for rules in (allow_rules, ask_rules, deny_rules)):
+            return None
+
+        for tool_name in ("aliyun_api", "ros_stack"):
+            if self._permission_rules_contain_tool(deny_rules, tool_name):
+                continue
+            if self._permission_rules_contain_tool(ask_rules, tool_name):
+                continue
+            if self._permission_rules_contain_tool(allow_rules, tool_name) or mode_name == "bypass_permissions":
+                return True
+        return False
+
+    @staticmethod
+    def _permission_rules_contain_tool(rules_by_source: dict[str, list[str]], tool_name: str) -> bool:
+        return any(
+            isinstance(rule, str) and rule.split("(", 1)[0].strip() == tool_name
+            for rules in rules_by_source.values()
+            if isinstance(rules, list)
+            for rule in rules
+        )
+
+    def _normal_chat_permission_context(self) -> Any | None:
+        getter = self._permission_context_getter
+        if not callable(getter):
+            return None
+        try:
+            return getter()
+        except Exception:
+            logger.debug("Failed to read permission context for normal chat handoff", exc_info=True)
+            return None
+
     def mark_normal_handoff(self, status: str, failed_reason: str | None = None) -> None:
         """Record terminal pipeline-to-normal handoff metadata without deleting the sidecar."""
         if not self.session:

src/iac_code/web/app.py

@@ -298,8 +298,8 @@ async def state_transaction(function: Callable[..., Any], /, *args: Any, **kwarg
         )
     )
     make_shell_runner: WebShellRunnerFactory = shell_runner_factory or (lambda: WebShellEscapeRunner(manager))
-    make_pipeline_action_runner: WebPipelineActionRunnerFactory = (
-        pipeline_action_runner_factory or create_pipeline_action_runner
+    make_pipeline_action_runner: WebPipelineActionRunnerFactory = pipeline_action_runner_factory or (
+        lambda: create_pipeline_action_runner(manager.ensure_permission_context)
     )
     command_dispatcher = WebCommandDispatcher(manager)
     shell_runner = make_shell_runner()

src/iac_code/web/pipeline_actions.py

@@ -96,7 +96,11 @@ class PipelineActionResult:
 class A2APipelineActionRunner:
     """Route Web pipeline actions through the existing A2A pipeline executor."""
 
-    def __init__(self, runtime_owner: Any | None = None) -> None:
+    def __init__(
+        self,
+        runtime_owner: Any | None = None,
+        permission_context_getter: Callable[[Any], Any] | None = None,
+    ) -> None:
         from iac_code.a2a.exposure import A2AExposureType
         from iac_code.a2a.metrics import NoOpA2AMetrics
         from iac_code.a2a.persistence import A2APersistenceStore
@@ -126,6 +130,7 @@ def __init__(self, runtime_owner: Any | None = None) -> None:
         self._owner = owner
         self._task_store = owner.task_store
         self._uses_web_global_defaults = uses_web_global_defaults
+        self._permission_context_getter = permission_context_getter
 
     async def startup(self) -> None:
         """Start background maintenance for the fallback store owned by Web."""
@@ -361,6 +366,11 @@ def _executor_for_session(
             # without this the executor would always fall back to the process-wide
             # IAC_CODE_PIPELINE_NAME default and silently run `selling`.
             pipeline_name=_session_pipeline_name(session),
+            permission_context_getter=(
+                (lambda: self._permission_context_getter(session))
+                if callable(getattr(self, "_permission_context_getter", None))
+                else None
+            ),
         )
 
     async def rebuild_permission_audit_event(
@@ -565,8 +575,10 @@ class _WebA2AContext:
     metadata: dict[str, Any] = {}
 
 
-def create_pipeline_action_runner() -> PipelineActionRunner:
-    return A2APipelineActionRunner()
+def create_pipeline_action_runner(
+    permission_context_getter: Callable[[Any], Any] | None = None,
+) -> PipelineActionRunner:
+    return A2APipelineActionRunner(permission_context_getter=permission_context_getter)
 
 
 async def load_pipeline_snapshot(*, context_id: str | None, task_id: str | None) -> dict[str, Any] | None:

tests/a2a/test_pipeline_executor.py

@@ -9,6 +9,7 @@
 import threading
 from pathlib import Path
 from types import SimpleNamespace
+from typing import Any
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
@@ -230,6 +231,38 @@ def _pipeline_executor(*, aliyun_delegated_executor_factory=None, candidate_pres
     )
 
 
+def test_create_pipeline_prefers_explicit_permission_context_getter(monkeypatch: pytest.MonkeyPatch) -> None:
+    import iac_code.a2a.pipeline_executor as pipeline_executor_module
+
+    explicit_permission_context = object()
+    runtime_permission_context = object()
+    captured: dict[str, Any] = {}
+
+    def fake_create_pipeline(*_args: Any, **kwargs: Any) -> object:
+        captured.update(kwargs)
+        return object()
+
+    monkeypatch.setattr(pipeline_executor_module, "create_pipeline", fake_create_pipeline)
+    executor = _pipeline_executor()
+    executor._permission_context_getter = lambda: explicit_permission_context
+
+    executor._create_pipeline(
+        session_id="session-1",
+        cwd="/workspace",
+        runtime=SimpleNamespace(
+            provider_manager=object(),
+            tool_registry=object(),
+            agent_loop=SimpleNamespace(_permission_context_getter=lambda: runtime_permission_context),
+            aliyun_services=None,
+        ),
+        session_storage=MagicMock(),
+        prerequisite_metadata={},
+        pipeline_name="selling",
+    )
+
+    assert captured["permission_context_getter"]() is explicit_permission_context
+
+
 @pytest.mark.asyncio
 async def test_pipeline_permission_backup_records_staged_generation_before_publication() -> None:
     task_store = MagicMock()

tests/pipeline/engine/test_pipeline_handoff.py

@@ -1,5 +1,6 @@
 from pathlib import Path
 from textwrap import dedent
+from types import SimpleNamespace
 from unittest.mock import MagicMock
 
 import yaml
@@ -8,7 +9,11 @@
 from iac_code.pipeline.engine.pipeline_runner import PipelineRunner
 
 
-def _make_runner(tmp_path: Path, on_complete: dict | None = None) -> PipelineRunner:
+def _make_runner(
+    tmp_path: Path,
+    on_complete: dict | None = None,
+    permission_mode: str | None = None,
+) -> PipelineRunner:
     body = {
         "name": "test",
         "context_dependencies": {
@@ -31,13 +36,15 @@ def _make_runner(tmp_path: Path, on_complete: dict | None = None) -> PipelineRun
     (tmp_path / "prompts").mkdir()
     (tmp_path / "prompts" / "step.md").write_text("step", encoding="utf-8")
 
+    permission_context_getter = (lambda: SimpleNamespace(mode=permission_mode)) if permission_mode is not None else None
     return PipelineRunner(
         pipeline_dir=tmp_path,
         provider_manager=MagicMock(),
         base_tool_registry=MagicMock(),
         session_storage=MagicMock(),
         session_id="session",
         cwd=str(tmp_path),
+        permission_context_getter=permission_context_getter,
     )
 
 
@@ -133,6 +140,33 @@ def test_build_handoff_summary_requires_new_confirmation_for_resource_release_ex
     assert "pipeline-managed automatic cleanup may proceed without this additional confirmation" in summary
 
 
+def test_build_handoff_summary_uses_permission_confirmation_when_release_tools_are_not_allowed():
+    summary = build_handoff_summary(
+        pipeline_name="selling",
+        outcome="completed",
+        context_snapshot={},
+        include_fields=[],
+        release_tools_any_allowed=False,
+    )
+
+    assert "tool permission confirmation as the sole confirmation" in summary
+    assert "Do not ask for a separate confirmation in normal chat" in summary
+    assert "obtain a fresh, explicit confirmation" not in summary
+
+
+def test_build_handoff_summary_keeps_chat_confirmation_when_a_release_tool_is_allowed():
+    summary = build_handoff_summary(
+        pipeline_name="selling",
+        outcome="completed",
+        context_snapshot={},
+        include_fields=[],
+        release_tools_any_allowed=True,
+    )
+
+    assert "obtain a fresh, explicit confirmation" in summary
+    assert "tool permission confirmation as the sole confirmation" not in summary
+
+
 def test_runner_should_switch_to_normal_for_completed_policy(tmp_path):
     runner = _make_runner(tmp_path, _switch_policy("completed"))
 
@@ -181,3 +215,36 @@ def test_runner_build_normal_handoff_summary_uses_configured_context_values(tmp_
     assert "Outcome: completed" in summary
     assert '"summary": "deploy nginx"' in summary
     assert "Missing context fields:\n- architecture" in summary
+
+
+def test_runner_build_normal_handoff_summary_uses_current_permission_rules(tmp_path):
+    runner = _make_runner(tmp_path, _switch_policy("completed"), permission_mode="default")
+    runner._permission_context_getter = lambda: SimpleNamespace(
+        mode="default",
+        allow_rules={
+            "user_settings": [
+                "bash(**)",
+                "read_file",
+            ]
+        },
+        ask_rules={"user_settings": ["ros_stack", "aliyun_api"]},
+        deny_rules={},
+    )
+
+    summary = runner.build_normal_handoff_summary({"total_steps": 1})
+
+    assert "tool permission confirmation as the sole confirmation" in summary
+
+
+def test_runner_build_normal_handoff_summary_uses_chat_confirmation_when_a_release_tool_is_allowed(tmp_path):
+    runner = _make_runner(tmp_path, _switch_policy("completed"), permission_mode="default")
+    runner._permission_context_getter = lambda: SimpleNamespace(
+        mode="default",
+        allow_rules={"user_settings": ["aliyun_api"]},
+        ask_rules={"user_settings": ["ros_stack"]},
+        deny_rules={},
+    )
+
+    summary = runner.build_normal_handoff_summary({"total_steps": 1})
+
+    assert "obtain a fresh, explicit confirmation" in summary

tests/web/test_pipeline_actions.py

@@ -108,7 +108,11 @@ async def get_task_record(self, _task_id: str) -> Any:
         return SimpleNamespace(state="working")
 
 
-async def _executor_kwargs_for_session(monkeypatch: pytest.MonkeyPatch, session: Any) -> dict[str, Any]:
+async def _executor_kwargs_for_session(
+    monkeypatch: pytest.MonkeyPatch,
+    session: Any,
+    permission_context_getter: Any = None,
+) -> dict[str, Any]:
     import iac_code.a2a.pipeline_executor as pipeline_executor_module
     from iac_code.web import pipeline_actions
 
@@ -122,6 +126,7 @@ async def no_snapshot(**_kwargs: Any) -> None:
     runner = A2APipelineActionRunner.__new__(A2APipelineActionRunner)
     runner._task_store = _StubTaskStore()
     runner._uses_web_global_defaults = False
+    runner._permission_context_getter = permission_context_getter
     runner._owner = SimpleNamespace(
         model="qwen3.6-plus",
         metrics=None,
@@ -175,6 +180,22 @@ async def test_execute_leaves_the_process_default_when_the_session_has_no_select
     assert kwargs["pipeline_name"] is None
 
 
+@pytest.mark.asyncio
+async def test_execute_passes_the_web_session_permission_context_to_pipeline(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    permission_context = object()
+    session = _pipeline_session(permission_context=permission_context)
+
+    kwargs = await _executor_kwargs_for_session(
+        monkeypatch,
+        session,
+        permission_context_getter=lambda current_session: current_session.permission_context,
+    )
+
+    assert kwargs["permission_context_getter"]() is permission_context
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("stored_name", ["no_such_pipeline", "   "])
 async def test_execute_falls_back_when_the_session_pipeline_is_unknown(