Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-agents/src/lexigram/ai/agents/strategies/guard_hook.py: 25%

1"""Shared guard hook for tool observations entering the LLM context (D3). 

2 

3Called by every strategy at the OBSERVE boundary: the point a tool result 

4becomes a message the model will read. Content is checked with the input 

5side of the guard pipeline; a block aborts the run (fail-closed), redaction 

6is applied, and guard infrastructure errors are raised, never swallowed. 

7 

8The executor passes its effectively-resolved pipeline to ``strategy.execute`` 

9as a ``guard_pipeline`` kwarg; strategies hand it to :func:`guard_observation`. 

10""" 

11 

12from __future__ import annotations 

13 

14from typing import TYPE_CHECKING, Any 

15 

16from lexigram.ai.agents.exceptions import AgentExecutionError 

17 

18if TYPE_CHECKING: 

19 from lexigram.contracts.ai.guards import GuardPipelineProtocol 

20 

21 

22class ToolObservationBlockedError(AgentExecutionError): 

23 """A guard blocked tool output from entering the model context.""" 

24 

25 

26class ToolObservationGuardError(AgentExecutionError): 

27 """The guard pipeline failed while checking tool output.""" 

28 

29 

30async def guard_observation( 

31 pipeline: GuardPipelineProtocol | None, 

32 content: str, 

33 *, 

34 tool_name: str, 

35 metadata: dict[str, Any] | None = None, 

36) -> str: 

37 """Check *content* (a tool observation) against the pipeline's input guards. 

38 

39 Args: 

40 pipeline: The effective guard pipeline (may be ``None`` for no-op). 

41 content: The raw tool result text about to enter the model context. 

42 tool_name: Name of the tool that produced the result (for metadata). 

43 metadata: Extra metadata merged under ``source=tool_observation``. 

44 

45 Returns: 

46 The (possibly redacted) content to write into context. 

47 

48 Raises: 

49 ToolObservationBlockedError: If a guard blocks the content. 

50 ToolObservationGuardError: If the pipeline itself fails — fail-closed. 

51 

52 Example: 

53 ```python 

54 obs = await guard_observation(pipeline, obs_text, tool_name="web_fetch") 

55 messages.append(ChatMessage(role=Role.USER, content=_OBSERVATION_TEMPLATE.format(observation=obs))) 

56 ``` 

57 """ 

58 if pipeline is None: 

59 return content 

60 

61 scope = {"source": "tool_observation", "tool_name": tool_name} 

62 if metadata: 

63 scope.update(metadata) 

64 

65 try: 

66 result = await pipeline.check_input(content=content, metadata=scope) 

67 except (RuntimeError, OSError) as exc: 

68 raise ToolObservationGuardError( 

69 f"Guard evaluation failed on tool observation: {exc}" 

70 ) from exc 

71 

72 if result.is_err(): 

73 raise ToolObservationGuardError( 

74 f"Guard pipeline error on tool observation: {result.unwrap_err()}" 

75 ) 

76 

77 agg = result.unwrap() 

78 if bool(getattr(agg, "blocked", False)): 

79 blocking = getattr(agg, "blocking_result", None) 

80 reason = getattr(blocking, "reason", None) or "Tool output blocked by guards" 

81 raise ToolObservationBlockedError(reason) 

82 

83 final = getattr(agg, "final_content", content) 

84 return str(final if final is not None else content)