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)