Merge current subagent naming stack into Team session UI

python/sdk/tests/test_approval.py

@@ -0,0 +1,55 @@
+"""Approval request decoding and explicit wire capability negotiation."""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+from pydantic import ValidationError
+
+from deepseek_harness import ApprovalRequestParams, ApprovalRequestResult, DeepSeekHarnessConfig, HarnessClient
+
+
+@pytest.mark.parametrize("enabled", [False, True])
+def test_approval_handshake_and_response(tmp_path: Path, enabled: bool) -> None:
+    peer = tmp_path / "approval_peer.py"
+    peer.write_text('''import json
+import sys
+for line in sys.stdin:
+    msg = json.loads(line)
+    if msg.get("method") == "initialize":
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"serverInfo": {"name": "test"}}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "method": "capability", "params": msg["params"]}), flush=True)
+        if msg["params"].get("approvalRequests"):
+            print(json.dumps({"jsonrpc": "2.0", "id": "ask", "method": "approval/request", "params": {"sessionId": "main", "toolName": "bash", "reason": "write"}}), flush=True)
+    elif msg.get("id") == "ask":
+        print(json.dumps({"jsonrpc": "2.0", "method": "decision", "params": msg["result"]}), flush=True)
+    elif msg.get("method") == "shutdown":
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {}}), flush=True)
+        break
+''')
+    with HarnessClient(_launch_args=(sys.executable, str(peer))) as client:
+        client.initialize(cwd=str(tmp_path), provider="mock", model="mock", approval_requests=enabled)
+        payload = client.next_notification().payload
+        assert payload.get("approvalRequests", False) is enabled
+        if not enabled:
+            assert "approvalRequests" not in payload
+            return
+        request = client.next_request()
+        assert request.method == "approval/request"
+        question = ApprovalRequestParams.model_validate(request.payload)
+        assert question.toolName == "bash"
+        assert question.callId is None
+        client.respond(request.id, ApprovalRequestResult(outcome="rejected").model_dump())
+        assert client.next_notification().payload == {"outcome": "rejected"}
+
+
+def test_approval_result_rejects_unrecognized_grants() -> None:
+    with pytest.raises(ValidationError):
+        ApprovalRequestResult(outcome="allow-always")
+
+
+def test_approval_opt_in_preserves_existing_config_positional_arguments() -> None:
+    config = DeepSeekHarnessConfig("mock", "model", None, None, "/workspace")
+    assert config.cwd == "/workspace"
+    assert config.approval_requests is False

python/sdk/tests/test_approval_runtime.py

@@ -0,0 +1,52 @@
+"""Python decisions control a guarded file operation through the built sdk profile."""
+from __future__ import annotations
+
+import json
+import shutil
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+
+import pytest
+from pydantic import BaseModel
+
+from deepseek_harness import ApprovalRequestParams, HarnessClient, HarnessConfig
+
+
+class EmptyResponse(BaseModel):
+    pass
+
+
+@pytest.mark.parametrize("outcome", ["allowed-once", "rejected"])
+def test_python_approval_controls_runtime_operation(tmp_path: Path, outcome: str) -> None:
+    repo = Path(__file__).resolve().parents[3]
+    node = shutil.which("node")
+    binary = repo / "apps/cli/lib/bin.js"
+    if node is None or not binary.exists():
+        pytest.skip("requires Node and built dsh CLI")
+    fixture = repo / "packages/sdk/server/tests/fixtures/approval.mjs"
+    marker = tmp_path / "approved.txt"
+    patch = tmp_path / "approval.patch.yml"
+    patch.write_text("- insert:\n    - id: approval-fixture\n      name: " + json.dumps(str(fixture))
+                     + "\n      config:\n        stopAfterApproval: true\n        path: " + json.dumps(str(marker)) + "\n")
+    # The client closes inside the executor lifetime, unblocking a reader even
+    # when a regression prevents the expected request from arriving.
+    with ThreadPoolExecutor(max_workers=1) as worker:
+        with HarnessClient(
+            HarnessConfig(cwd=str(repo), request_timeout_seconds=20, shutdown_timeout_seconds=20,
+                          env={"DSH_HOME": str(tmp_path / "home"), "DSH_PERMISSION_MODE": "workspace-write",
+                               "DSH_TELEMETRY_DISABLED": "1", "DEEPSEEK_API_KEY": "keyless-no-model-call"}),
+            _launch_args=(node, str(binary), "--profile", "sdk", "--patch", str(patch)),
+        ) as client:
+            client.initialize(cwd=str(tmp_path), provider="deepseek-official", model="deepseek-v4-flash",
+                              approval_requests=True)
+            client.session_prompt("main", [{"type": "text", "text": "Request the marker write."}])
+            request = worker.submit(client.next_request).result(timeout=20)
+            assert request.method == "approval/request"
+            question = ApprovalRequestParams.model_validate(request.payload)
+            assert question.sessionId == "main"
+            assert question.toolName == "fixture-write"
+            client.respond(request.id, {"outcome": outcome})
+            client.request("session/wait", {"sessionId": "main"}, response_model=EmptyResponse)
+            assert marker.exists() is (outcome == "allowed-once")
+            if marker.exists():
+                assert marker.read_text() == "approved\n"

python/sdk/tests/test_client.py

@@ -165,6 +165,7 @@ for line in sys.stdin:
     )
 
     with DeepSeekHarness(
+        approval_requests=True,
         model="deepseek-v4-flash",
         reasoning_effort="max",
         max_tokens=4096,
@@ -211,6 +212,7 @@ for line in sys.stdin:
     assert dumped_env["DSH_SESSION_ROOT"] is None
     assert dumped_env["DSH_CORDIS_CONFIG"] is None
     assert json.loads(init_dump.read_text()) == {
+        "approvalRequests": True,
         "cwd": str(tmp_path),
         "provider": "deepseek-official",
         "model": "deepseek-v4-flash",