fix: emit Lingma tool approve/invoke roundtrip

Forward tool/call/sync and tool/invoke events to Lingma with auto-approve and invokeResult so tool calls can complete end-to-end.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
GitHub Actions
2026-04-19 21:35:05 +08:00
parent 3498b81fa2
commit e208025f35
2 changed files with 147 additions and 1 deletions

View File

@@ -798,7 +798,6 @@ class SessionCacheToolFingerprintTests(unittest.TestCase):
"result": {"hits": 3},
}
)
self.assertEqual(
event,
{
@@ -808,3 +807,97 @@ class SessionCacheToolFingerprintTests(unittest.TestCase):
"result": {"hits": 3},
},
)
def test_tool_sync_triggers_approve_and_invoke_result_requests(self) -> None:
from app.lingma_client import LspWsRpcClient
class _WsStub:
def __init__(self) -> None:
self.frames: list[bytes] = []
async def send(self, data: bytes) -> None:
self.frames.append(data)
def _decode(frame: bytes) -> dict:
body = frame.split(b"\r\n\r\n", 1)[1]
return json.loads(body.decode("utf-8"))
ws = _WsStub()
rpc = LspWsRpcClient(ws)
async def run() -> None:
rpc.create_stream("req-1")
await rpc._handle_server_message(
{
"jsonrpc": "2.0",
"method": "tool/call/sync",
"params": {
"sessionId": "sess-1",
"requestId": "req-1",
"toolCallId": "call-1",
"name": "run_in_terminal",
"parameters": {"command": "pwd"},
},
}
)
decoded = [_decode(frame) for frame in ws.frames]
methods = [item.get("method") for item in decoded]
self.assertIn("tool/call/approve", methods)
self.assertIn("tool/invokeResult", methods)
approve = next(item for item in decoded if item.get("method") == "tool/call/approve")
self.assertEqual(
approve["params"],
{
"type": "tool_call",
"sessionId": "sess-1",
"requestId": "req-1",
"toolCallId": "call-1",
"approval": True,
},
)
invoke_result = next(item for item in decoded if item.get("method") == "tool/invokeResult")
self.assertEqual(invoke_result["params"]["toolCallId"], "call-1")
self.assertEqual(invoke_result["params"]["name"], "run_in_terminal")
self.assertTrue(invoke_result["params"]["success"])
self.assertEqual(invoke_result["params"]["errorMessage"], "")
import asyncio
asyncio.run(run())
def test_tool_sync_does_not_emit_roundtrip_without_request_id(self) -> None:
from app.lingma_client import LspWsRpcClient
class _WsStub:
def __init__(self) -> None:
self.frames: list[bytes] = []
async def send(self, data: bytes) -> None:
self.frames.append(data)
ws = _WsStub()
rpc = LspWsRpcClient(ws)
async def run() -> None:
rpc.create_stream("req-1")
await rpc._handle_server_message(
{
"jsonrpc": "2.0",
"method": "tool/call/sync",
"params": {
"sessionId": "sess-1",
"toolCallId": "call-1",
"name": "run_in_terminal",
"parameters": {"command": "pwd"},
},
}
)
self.assertEqual(ws.frames, [])
import asyncio
asyncio.run(run())