From 1ea3d440467e8ccc747ef71e4ad0ca2f87abc892 Mon Sep 17 00:00:00 2001 From: Chisanan232 Date: Thu, 18 Jun 2026 14:50:46 +0800 Subject: [PATCH 1/9] =?UTF-8?q?=F0=9F=94=A7=20(mypy):=20Enable=20pydantic?= =?UTF-8?q?=20plugin=20so=20model=20defaults=20type-check?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pre-commit mypy hook was red on master: pydantic models with Field(default=...) defaults reported spurious 'Missing named argument' call-arg errors because the pydantic mypy plugin was not enabled. Enable it so defaulted fields are correctly treated as optional at construction. refs AAASM-3347 --- mypy.ini | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/mypy.ini b/mypy.ini index b57cd779..5a09919a 100644 --- a/mypy.ini +++ b/mypy.ini @@ -1,6 +1,10 @@ # Global options: [mypy] +# The pydantic plugin teaches mypy that BaseModel fields with Field(default=...) / +# Field(None) defaults are optional at construction; without it every model with +# defaulted fields reports spurious "Missing named argument" call-arg errors. +plugins = pydantic.mypy packages = agent_assembly,test exclude = (?x)( test/unit_test.{1,64}.py # Ignore the code of unit test because of the usage of mock From c6622ca07069e7e347e956c65b74b0ef5ba47b69 Mon Sep 17 00:00:00 2001 From: Chisanan232 Date: Thu, 18 Jun 2026 14:51:24 +0800 Subject: [PATCH 2/9] =?UTF-8?q?=E2=9C=85=20(test):=20Make=20runtime=20monk?= =?UTF-8?q?eypatch=20targets=20mypy-clean?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pre-commit mypy hook flagged runtime.socket / runtime.subprocess as non-exported attributes and a Popen-vs-str comparison-overlap. Patch via string targets (mypy does not check string monkeypatch paths) and widen the handle annotation to object. No behaviour change; tests still pass. refs AAASM-3347 --- test/unit/test_runtime.py | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/test/unit/test_runtime.py b/test/unit/test_runtime.py index 6062e6da..948890db 100644 --- a/test/unit/test_runtime.py +++ b/test/unit/test_runtime.py @@ -100,8 +100,7 @@ def __exit__(self, *_args: object) -> None: return None monkeypatch.setattr( - runtime.socket, - "create_connection", + "agent_assembly.runtime.socket.create_connection", lambda _address, timeout: _FakeConn(), # noqa: ARG005 — signature-matching stub ) @@ -114,14 +113,12 @@ def test_is_running_false_on_socket_error(monkeypatch: pytest.MonkeyPatch) -> No def refuse(address: object, timeout: object) -> object: raise ConnectionRefusedError("nothing listening") - monkeypatch.setattr(runtime.socket, "create_connection", refuse) + monkeypatch.setattr("agent_assembly.runtime.socket.create_connection", refuse) assert runtime.is_running(port=7878) is False -def test_start_runtime_spawns_detached_serve_process( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: +def test_start_runtime_spawns_detached_serve_process(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: """start_runtime opens the log file in the chosen dir and spawns the detached ``aasm serve --port`` subprocess with stdout/stderr redirected.""" captured: dict[str, object] = {} @@ -131,9 +128,9 @@ def fake_popen(cmd: list[str], **kwargs: object) -> str: captured["kwargs"] = kwargs return "popen-handle" - monkeypatch.setattr(runtime.subprocess, "Popen", fake_popen) + monkeypatch.setattr("agent_assembly.runtime.subprocess.Popen", fake_popen) - handle = runtime.start_runtime(Path("/bin/aasm"), port=9001, log_dir=tmp_path) + handle: object = runtime.start_runtime(Path("/bin/aasm"), port=9001, log_dir=tmp_path) assert handle == "popen-handle" assert captured["cmd"] == ["/bin/aasm", "serve", "--port", "9001"] From d60e73522771b12ee3268285886988bcdfc0d51c Mon Sep 17 00:00:00 2001 From: Chisanan232 Date: Thu, 18 Jun 2026 14:51:42 +0800 Subject: [PATCH 3/9] =?UTF-8?q?=F0=9F=8E=A8=20(mcp):=20Merge=20implicitly?= =?UTF-8?q?=20concatenated=20f-strings=20+=20narrow=20error=20type?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SonarCloud S5799 flagged two adjacent f-string literals in the blocked message; merge each into one. S112 flagged the generic Exception return of _build_blocked_error; narrow it to MCPToolBlockedError. No behaviour change. refs AAASM-3347 --- agent_assembly/adapters/mcp/patch.py | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/agent_assembly/adapters/mcp/patch.py b/agent_assembly/adapters/mcp/patch.py index 6ea4fb9f..67e8d4f9 100644 --- a/agent_assembly/adapters/mcp/patch.py +++ b/agent_assembly/adapters/mcp/patch.py @@ -6,7 +6,10 @@ import importlib.util import inspect from dataclasses import dataclass -from typing import Any, Literal, Mapping +from typing import TYPE_CHECKING, Any, Literal, Mapping + +if TYPE_CHECKING: + from agent_assembly.exceptions import MCPToolBlockedError from agent_assembly.adapters.crewai.patch import ( _get_pending_tool_approval_timeout_seconds as _resolve_pending_timeout_seconds, @@ -224,16 +227,14 @@ def _build_blocked_error( server_identifier: str, reason: str | None, is_pending_rejection: bool, -) -> Exception: +) -> MCPToolBlockedError: from agent_assembly.exceptions import MCPToolBlockedError reason_text = reason or "No reason provided." if is_pending_rejection: - message = f"MCP tool '{tool_name}' on server '{server_identifier}' " f"rejected during approval: {reason_text}" + message = f"MCP tool '{tool_name}' on server '{server_identifier}' rejected during approval: {reason_text}" else: - message = ( - f"MCP tool '{tool_name}' on server '{server_identifier}' " f"blocked by governance policy: {reason_text}" - ) + message = f"MCP tool '{tool_name}' on server '{server_identifier}' blocked by governance policy: {reason_text}" return MCPToolBlockedError( message, From 99ba61c40b5bb4fa6e649a0880d344b40d6eb054 Mon Sep 17 00:00:00 2001 From: Chisanan232 Date: Thu, 18 Jun 2026 14:51:55 +0800 Subject: [PATCH 4/9] =?UTF-8?q?=F0=9F=8E=A8=20(cli):=20Merge=20concat=20f-?= =?UTF-8?q?strings=20+=20drop=20empty=20TYPE=5FCHECKING=20block?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SonarCloud S5799 flagged three adjacent f-string literals in adapter validation messages; merge each into one. S108 flagged the empty `if TYPE_CHECKING: pass` block (no longer guarding any import); remove it and the now-unused TYPE_CHECKING import. No behaviour change. refs AAASM-3347 --- agent_assembly/cli/adapter_validator.py | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/agent_assembly/cli/adapter_validator.py b/agent_assembly/cli/adapter_validator.py index 0611a464..f2e20ff2 100644 --- a/agent_assembly/cli/adapter_validator.py +++ b/agent_assembly/cli/adapter_validator.py @@ -8,10 +8,6 @@ import tomllib from dataclasses import dataclass from pathlib import Path -from typing import TYPE_CHECKING - -if TYPE_CHECKING: - pass from agent_assembly.adapters.base import FrameworkAdapter, GovernanceInterceptor @@ -147,7 +143,7 @@ def _check_register_hooks_signature(cls: type) -> AdapterValidationResult: return AdapterValidationResult( check_name="register_hooks_signature", passed=False, - message=(f"register_hooks() first parameter annotated as {annotation}, " f"expected GovernanceInterceptor."), + message=f"register_hooks() first parameter annotated as {annotation}, expected GovernanceInterceptor.", ) @@ -162,7 +158,7 @@ def _check_unregister_hooks_idempotent( return AdapterValidationResult( check_name="unregister_hooks_idempotent", passed=False, - message=(f"unregister_hooks() is not idempotent: " f"second call raised {type(exc).__name__}: {exc}"), + message=f"unregister_hooks() is not idempotent: second call raised {type(exc).__name__}: {exc}", ) return AdapterValidationResult( check_name="unregister_hooks_idempotent", @@ -231,7 +227,7 @@ def _check_entry_point_metadata(cls: type, path_or_module: str) -> AdapterValida return AdapterValidationResult( check_name="entry_point_metadata", passed=False, - message=(f"No entry point references {class_qualname}. " f"Found: {entry_points}."), + message=f"No entry point references {class_qualname}. Found: {entry_points}.", ) From 86732006f0b1e09b227ad15b6fecccaccea802c4 Mon Sep 17 00:00:00 2001 From: Chisanan232 Date: Thu, 18 Jun 2026 14:52:09 +0800 Subject: [PATCH 5/9] =?UTF-8?q?=F0=9F=8F=B7=EF=B8=8F=20(pydantic=5Fai):=20?= =?UTF-8?q?Narrow=20error=20builders=20to=20PolicyViolationError=20(S112)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SonarCloud S112 flagged four `raise` sites whose builders returned the generic Exception type. Narrow _build_denied_error / _build_pending_rejected_error return annotations to PolicyViolationError (added under TYPE_CHECKING). No behaviour change. refs AAASM-3347 --- agent_assembly/adapters/pydantic_ai/patch.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/agent_assembly/adapters/pydantic_ai/patch.py b/agent_assembly/adapters/pydantic_ai/patch.py index 230cee2d..f5c0781f 100644 --- a/agent_assembly/adapters/pydantic_ai/patch.py +++ b/agent_assembly/adapters/pydantic_ai/patch.py @@ -7,7 +7,10 @@ from collections.abc import Mapping from dataclasses import dataclass from functools import wraps -from typing import Any, Literal +from typing import TYPE_CHECKING, Any, Literal + +if TYPE_CHECKING: + from agent_assembly.exceptions import PolicyViolationError from agent_assembly.adapters.crewai.patch import ( _get_pending_tool_approval_timeout_seconds as _resolve_pending_timeout_seconds, @@ -593,14 +596,14 @@ async def _record_async_tool_result( await recorded -def _build_denied_error(tool_name: str, reason: str | None) -> Exception: +def _build_denied_error(tool_name: str, reason: str | None) -> PolicyViolationError: from agent_assembly.exceptions import PolicyViolationError reason_text = reason or "No reason provided." return PolicyViolationError(f"Tool '{tool_name}' blocked by governance policy: {reason_text}") -def _build_pending_rejected_error(tool_name: str, reason: str | None) -> Exception: +def _build_pending_rejected_error(tool_name: str, reason: str | None) -> PolicyViolationError: from agent_assembly.exceptions import PolicyViolationError reason_text = reason or "No reason provided." From 8596a767287687437d5b88dbe666084ecacafbde Mon Sep 17 00:00:00 2001 From: Chisanan232 Date: Thu, 18 Jun 2026 14:52:12 +0800 Subject: [PATCH 6/9] =?UTF-8?q?=F0=9F=8F=B7=EF=B8=8F=20(google=5Fadk):=20N?= =?UTF-8?q?arrow=20error=20builders=20to=20PolicyViolationError=20(S112)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SonarCloud S112 flagged two `raise` sites whose builders returned the generic Exception type. Narrow _build_denied_error / _build_pending_rejected_error return annotations to PolicyViolationError (added under TYPE_CHECKING). No behaviour change. refs AAASM-3347 --- agent_assembly/adapters/google_adk/patch.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/agent_assembly/adapters/google_adk/patch.py b/agent_assembly/adapters/google_adk/patch.py index c20eb765..dfba1153 100644 --- a/agent_assembly/adapters/google_adk/patch.py +++ b/agent_assembly/adapters/google_adk/patch.py @@ -7,7 +7,10 @@ from collections.abc import Mapping from dataclasses import dataclass from functools import wraps -from typing import Any, Literal +from typing import TYPE_CHECKING, Any, Literal + +if TYPE_CHECKING: + from agent_assembly.exceptions import PolicyViolationError from agent_assembly.adapters.crewai.patch import ( _get_pending_tool_approval_timeout_seconds as _resolve_pending_timeout_seconds, @@ -381,14 +384,14 @@ async def _record_async_tool_result( await recorded -def _build_denied_error(tool_name: str, reason: str | None) -> Exception: +def _build_denied_error(tool_name: str, reason: str | None) -> PolicyViolationError: from agent_assembly.exceptions import PolicyViolationError reason_text = reason or "No reason provided." return PolicyViolationError(f"Tool '{tool_name}' blocked by governance policy: {reason_text}") -def _build_pending_rejected_error(tool_name: str, reason: str | None) -> Exception: +def _build_pending_rejected_error(tool_name: str, reason: str | None) -> PolicyViolationError: from agent_assembly.exceptions import PolicyViolationError reason_text = reason or "No reason provided." From d18af509f13e02616c575168ebcad04354b9a5df Mon Sep 17 00:00:00 2001 From: Chisanan232 Date: Thu, 18 Jun 2026 14:52:28 +0800 Subject: [PATCH 7/9] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20(langgraph):=20Drop=20?= =?UTF-8?q?unused=20node=5Fname=20params=20+=20redundant=20list()=20(S1172?= =?UTF-8?q?/S7504)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SonarCloud S1172 flagged the unused node_name parameter on _wrap_tool_node_subgraphs and _make_subgraph_spawn_wrapper; remove it and update all call sites (incl. tests). S7504 flagged the unnecessary list() around tools_by_name.items() — the loop mutates tool.func, not the dict, so iterate directly. No behaviour change. refs AAASM-3347 --- agent_assembly/adapters/langgraph/patch.py | 9 ++------- .../langchain/test_langgraph_spawn_patch.py | 16 +++++++--------- 2 files changed, 9 insertions(+), 16 deletions(-) diff --git a/agent_assembly/adapters/langgraph/patch.py b/agent_assembly/adapters/langgraph/patch.py index 06b85ea9..e6fc371b 100644 --- a/agent_assembly/adapters/langgraph/patch.py +++ b/agent_assembly/adapters/langgraph/patch.py @@ -244,7 +244,6 @@ def _is_tool_node(node_executor: Any) -> bool: def _wrap_tool_node_subgraphs( - node_name: str, tool_node: Any, process_agent_id: str | None, ) -> bool: @@ -253,11 +252,10 @@ def _wrap_tool_node_subgraphs( if not isinstance(tools_by_name, dict): return False wrapped_any = False - for tool_name, tool in list(tools_by_name.items()): + for tool_name, tool in tools_by_name.items(): tool_func = getattr(tool, "func", None) if tool_func is not None and _is_compiled_subgraph(tool_func): wrapper = _make_subgraph_spawn_wrapper( - str(tool_name), tool_func, process_agent_id, spawned_by_tool=str(tool_name), @@ -272,7 +270,6 @@ def _wrap_tool_node_subgraphs( def _make_subgraph_spawn_wrapper( - node_name: str, subgraph: Any, process_agent_id: str | None, *, @@ -325,7 +322,6 @@ def _wrap_subgraph_spawn_node(node_map: Any, node_name: Any, node_executor: Any, """Wrap a compiled-subgraph node (spawn point) for lineage. Return True when wrapped.""" node_delegation_reason = f"langgraph_node:{node_name}" sync_wrapper = _make_subgraph_spawn_wrapper( - str(node_name), node_executor, process_agent_id, spawned_by_tool=None, @@ -335,7 +331,6 @@ def _wrap_subgraph_spawn_node(node_map: Any, node_name: Any, node_executor: Any, node_map[node_name] = sync_wrapper if hasattr(node_executor, "ainvoke") and not getattr(node_executor, "_agent_assembly_ainvoke_spawned", False): async_wrapper = _make_subgraph_spawn_wrapper( - str(node_name), node_executor, process_agent_id, async_=True, @@ -369,7 +364,7 @@ def _wrap_node_entry( # ToolNode: intercept any compiled-subgraph tools it holds. # Must come before the callable() check since ToolNode is also callable. if _is_tool_node(node_executor): - return _wrap_tool_node_subgraphs(str(node_name), node_executor, process_agent_id) + return _wrap_tool_node_subgraphs(node_executor, process_agent_id) if callable(node_executor): return _wrap_callable_node_executor(node_map, node_name, node_executor, callback_handler) diff --git a/test/unit/adapters/langchain/test_langgraph_spawn_patch.py b/test/unit/adapters/langchain/test_langgraph_spawn_patch.py index ed437578..a486bc16 100644 --- a/test/unit/adapters/langchain/test_langgraph_spawn_patch.py +++ b/test/unit/adapters/langchain/test_langgraph_spawn_patch.py @@ -39,7 +39,7 @@ def test_sync_wrapper_sets_spawn_ctx_then_resets(self) -> None: subgraph = MagicMock() subgraph.invoke = original_invoke - wrapper = _make_subgraph_spawn_wrapper("subnode", subgraph, "parent-agent") + wrapper = _make_subgraph_spawn_wrapper(subgraph, "parent-agent") assert _SPAWN_CTX.get() is None result = wrapper({"input": "x"}) @@ -60,7 +60,6 @@ def test_delegation_reason_passed_through(self) -> None: subgraph.invoke = MagicMock(side_effect=lambda *_args, **_kwargs: captured.append(_SPAWN_CTX.get()) or "r") # type: ignore[func-returns-value] wrapper = _make_subgraph_spawn_wrapper( - "mynode", subgraph, "parent", delegation_reason="langgraph_node:mynode", @@ -78,7 +77,6 @@ def test_spawned_by_tool_passed_through_for_tool_node(self) -> None: subgraph.invoke = MagicMock(side_effect=lambda *_args, **_kwargs: captured.append(_SPAWN_CTX.get()) or "r") # type: ignore[func-returns-value] wrapper = _make_subgraph_spawn_wrapper( - "tool_x", subgraph, "parent", spawned_by_tool="tool_x", @@ -103,7 +101,7 @@ async def fake_ainvoke(*args: object, **kwargs: object) -> str: subgraph.ainvoke = fake_ainvoke subgraph.invoke = MagicMock() - wrapper = _make_subgraph_spawn_wrapper("asyncnode", subgraph, "parent-async", async_=True) + wrapper = _make_subgraph_spawn_wrapper(subgraph, "parent-async", async_=True) result = await wrapper({"input": "y"}) assert result == "async-result" @@ -122,7 +120,7 @@ def test_spawn_ctx_depth_increments_when_already_in_ctx(self) -> None: outer = SpawnContext(parent_agent_id="grandparent", depth=1) token = _SPAWN_CTX.set(outer) try: - wrapper = _make_subgraph_spawn_wrapper("child", subgraph, "parent-agent") + wrapper = _make_subgraph_spawn_wrapper(subgraph, "parent-agent") wrapper({}) finally: _SPAWN_CTX.reset(token) @@ -135,7 +133,7 @@ def test_wrapper_is_pass_through_on_exception(self) -> None: subgraph = MagicMock() subgraph.invoke = MagicMock(side_effect=RuntimeError("graph error")) - wrapper = _make_subgraph_spawn_wrapper("err_node", subgraph, "parent") + wrapper = _make_subgraph_spawn_wrapper(subgraph, "parent") with pytest.raises(RuntimeError, match="graph error"): wrapper({}) # Token must still be reset @@ -231,7 +229,7 @@ def test_wraps_compiled_subgraph_tool_inside_tool_node(self) -> None: tool_node = MagicMock(spec=["tools_by_name"]) tool_node.tools_by_name = {"search": tool} - result = _wrap_tool_node_subgraphs("tool_node", tool_node, "parent-agent") + result = _wrap_tool_node_subgraphs(tool_node, "parent-agent") assert result is True assert tool.func is not subgraph @@ -245,7 +243,7 @@ def test_returns_false_when_no_compiled_subgraph_tools(self) -> None: tool_node = MagicMock(spec=["tools_by_name"]) tool_node.tools_by_name = {"search": tool} - result = _wrap_tool_node_subgraphs("tool_node", tool_node, "parent-agent") + result = _wrap_tool_node_subgraphs(tool_node, "parent-agent") assert result is False @@ -260,7 +258,7 @@ def test_spawned_by_tool_and_delegation_reason_set_for_tool_node_path(self) -> N tool_node = MagicMock(spec=["tools_by_name"]) tool_node.tools_by_name = {"retriever": tool} - _wrap_tool_node_subgraphs("tool_node", tool_node, "parent-agent") + _wrap_tool_node_subgraphs(tool_node, "parent-agent") # Invoke the wrapped function to verify spawn context values tool.func({}) From acd8ea8dea1252ee81425cb87a6b29f21e4f91e6 Mon Sep 17 00:00:00 2001 From: Chisanan232 Date: Thu, 18 Jun 2026 14:52:40 +0800 Subject: [PATCH 8/9] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20(crewai):=20Reduce=20?= =?UTF-8?q?=5Fnormalize=5Fdecision=20cognitive=20complexity=20(S3776)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SonarCloud S3776 flagged _normalize_decision at complexity 18 (limit 15). Collapse the three per-status if-chains in both the str and Mapping branches into a single _coerce_known_status lookup against a status set. Behaviour is identical; unknown verdicts still fail closed under enforce. refs AAASM-3347 --- agent_assembly/adapters/crewai/patch.py | 32 +++++++++++++------------ 1 file changed, 17 insertions(+), 15 deletions(-) diff --git a/agent_assembly/adapters/crewai/patch.py b/agent_assembly/adapters/crewai/patch.py index 9efac099..6f270a48 100644 --- a/agent_assembly/adapters/crewai/patch.py +++ b/agent_assembly/adapters/crewai/patch.py @@ -7,7 +7,7 @@ from dataclasses import dataclass from functools import wraps from threading import local -from typing import Any, Literal +from typing import Any, Literal, cast from agent_assembly.core.spawn import _SPAWN_CTX, SpawnContext, spawn_context_scope @@ -227,31 +227,33 @@ def _unknown_decision(enforce: bool) -> tuple[Literal["allow", "deny", "pending" return "allow", None +_KNOWN_STATUSES: frozenset[str] = frozenset({"allow", "deny", "pending"}) + + +def _coerce_known_status(value: str) -> Literal["allow", "deny", "pending"] | None: + """Return the verdict literal for a recognized status string, else ``None``.""" + if value in _KNOWN_STATUSES: + return cast("Literal['allow', 'deny', 'pending']", value) + return None + + def _normalize_decision( decision: object, *, enforce: bool = False, ) -> tuple[Literal["allow", "deny", "pending"], str | None]: if isinstance(decision, str): - normalized = decision.strip().lower() - if normalized == "allow": - return "allow", None - if normalized == "deny": - return "deny", None - if normalized == "pending": - return "pending", None + status = _coerce_known_status(decision.strip().lower()) + if status is not None: + return status, None return _unknown_decision(enforce) if isinstance(decision, Mapping): - raw_status = str(decision.get("status", "")).strip().lower() reason_value = decision.get("reason") reason = str(reason_value) if reason_value is not None else None - if raw_status == "allow": - return "allow", reason - if raw_status == "deny": - return "deny", reason - if raw_status == "pending": - return "pending", reason + status = _coerce_known_status(str(decision.get("status", "")).strip().lower()) + if status is not None: + return status, reason unknown_status, unknown_reason = _unknown_decision(enforce) return unknown_status, reason if reason is not None else unknown_reason From 40d1443036dd80e195c7aac6406ea5c8b8c7b127 Mon Sep 17 00:00:00 2001 From: Chisanan232 Date: Thu, 18 Jun 2026 14:52:52 +0800 Subject: [PATCH 9/9] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20(langchain):=20Reduce?= =?UTF-8?q?=20=5Fnormalize=5Fdecision=20cognitive=20complexity=20(S3776)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SonarCloud S3776 flagged _normalize_decision at complexity 18 (limit 15). Collapse the three per-status if-chains in both the str and Mapping branches into a single _coerce_known_status lookup against a status set. Behaviour is identical; unknown verdicts still fail closed under enforce. refs AAASM-3347 --- .../adapters/langchain/callback_handler.py | 29 ++++++++++--------- 1 file changed, 15 insertions(+), 14 deletions(-) diff --git a/agent_assembly/adapters/langchain/callback_handler.py b/agent_assembly/adapters/langchain/callback_handler.py index d4251f5c..1bd50b72 100644 --- a/agent_assembly/adapters/langchain/callback_handler.py +++ b/agent_assembly/adapters/langchain/callback_handler.py @@ -9,6 +9,8 @@ from agent_assembly.exceptions import ToolExecutionBlockedError +_KNOWN_STATUSES: frozenset[str] = frozenset({"allow", "deny", "pending"}) + class _FallbackBaseCallbackHandler: """Fallback base type when langchain-core is not installed.""" @@ -60,30 +62,29 @@ def _unknown_decision(self) -> tuple[Literal["allow", "deny", "pending"], str | return "deny", self._UNKNOWN_DECISION_REASON return "allow", None + @staticmethod + def _coerce_known_status(value: str) -> Literal["allow", "deny", "pending"] | None: + """Return the verdict literal for a recognized status string, else ``None``.""" + if value in _KNOWN_STATUSES: + return cast("Literal['allow', 'deny', 'pending']", value) + return None + def _normalize_decision( self, decision: object, ) -> tuple[Literal["allow", "deny", "pending"], str | None]: if isinstance(decision, str): - normalized = decision.strip().lower() - if normalized == "allow": - return "allow", None - if normalized == "deny": - return "deny", None - if normalized == "pending": - return "pending", None + status = self._coerce_known_status(decision.strip().lower()) + if status is not None: + return status, None return self._unknown_decision() if isinstance(decision, Mapping): - raw_status = str(decision.get("status", "")).strip().lower() reason_value = decision.get("reason") reason = str(reason_value) if reason_value is not None else None - if raw_status == "allow": - return "allow", reason - if raw_status == "deny": - return "deny", reason - if raw_status == "pending": - return "pending", reason + status = self._coerce_known_status(str(decision.get("status", "")).strip().lower()) + if status is not None: + return status, reason unknown_status, unknown_reason = self._unknown_decision() return unknown_status, reason if reason is not None else unknown_reason