diff --git a/CLAUDE.md b/CLAUDE.md index 4a4c46b..1baaa4d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -671,9 +671,10 @@ functions); a test patches a name in the module that looks it up (`harness._MIN_ - **Toolkits** (`toolkits/`): what the agent can read, one `Toolkit` subclass per AWS service (`base.py` says what one gives). `LambdaToolkit` (`lambda_tools.py`) has seven tools (`find_functions`, `function_health`, `function_errors`, `function_logs`, `function_performance`, `function_details`, `function_code`), building blocks over `LambdaAnalyzer` - methods, not one per question: plain functions closed over the toolkit in its `tools()`. Their type hints and + methods, not one per question: methods of the toolkit that its `tools()` lists. Their type hints and docstrings are the schema and description the model sees (LangChain reads them), so write a docstring for the - model: what it returns, how to pass times (ISO UTC or '24h'), when to use it. Each returns a compact dict from the + model: what it returns, how to pass times (ISO UTC or '24h'), when to use it, and nothing about how it works + inside (that goes in a helper's docstring). Each returns a compact dict from the module's pure `summarize_*` (what the model reads: counts, causes, a few lines, secrets hidden by `results.py`'s `hide_secrets`) with its coverage (`window`, `regions`, `cut_short`, `covered_from`), plus `_summary` (the report's one line) and `_next` (View calls, made by `Toolkit._view_call`), which the runtime takes out before the model sees diff --git a/ruff.toml b/ruff.toml index e1e544d..9a2f4a5 100644 --- a/ruff.toml +++ b/ruff.toml @@ -29,13 +29,6 @@ select = ["E4", "E7", "E9", "F", "B006", "BLE001", "ERA001", "C901", "PLR0912", ".claude/skills/demo/shots.py" = ["C901", "PLR0912", "PLR0913", "PLR0915"] ".claude/skills/release/release.py" = ["BLE001", "C901", "PLR0912", "PLR0915"] "src/aws_analyzer/_bedrock.py" = ["C901", "PLR0912", "PLR0915"] -"src/aws_analyzer/agent/analyzer.py" = ["PLR0913"] -"src/aws_analyzer/agent/harness.py" = ["BLE001", "C901", "PLR0913"] -"src/aws_analyzer/agent/llm.py" = ["PLR0913"] -"src/aws_analyzer/agent/runtime.py" = ["BLE001", "C901"] -"src/aws_analyzer/agent/toolkits/base.py" = ["ERA001"] -"src/aws_analyzer/agent/toolkits/lambda_tools.py" = ["C901", "PLR0913", "PLR0915"] -"src/aws_analyzer/agent/toolkits/s3_tools.py" = ["C901", "PLR0912", "PLR0913", "PLR0915"] "src/aws_analyzer/bedrock_chat.py" = ["BLE001", "C901", "PLR0912", "PLR0913", "PLR0915"] "src/aws_analyzer/bedrock_kb.py" = ["C901", "PLR0912", "PLR0913", "PLR0915"] "src/aws_analyzer/dynamodb.py" = ["C901", "PLR0913"] diff --git a/src/aws_analyzer/agent/analyzer.py b/src/aws_analyzer/agent/analyzer.py index 283cab3..3614ec2 100644 --- a/src/aws_analyzer/agent/analyzer.py +++ b/src/aws_analyzer/agent/analyzer.py @@ -27,6 +27,27 @@ from ..s3 import S3Analyzer +_THREAD_ID_LENGTH = 12 +_WORKERS = 8 + + +def _new_thread() -> str: + """ + A new conversation's ID. + """ + return uuid.uuid4().hex[:_THREAD_ID_LENGTH] + + +def _checked_zone(timezone: str | tzinfo | None) -> str | tzinfo | None: + """ + timezone= as it's kept, after checking it names a zone, so a misspelled one is an error now rather than at the + first question. None is the zone reports show times in (set_timezone()). + """ + if timezone is not None: + _zone_from(timezone) + return timezone + + class AgentAnalyzer(_BaseAnalyzer): """ Answers questions about your AWS account with an agent: a model (an Azure OpenAI deployment) calling read-only @@ -47,11 +68,14 @@ class AgentAnalyzer(_BaseAnalyzer): AWS_ANALYZER_TZ's), the zone every report shows times in; `timezone` ('Asia/Kolkata', 'kolkata', '+05:30') sets one for the agent alone. `notes` are facts about the account the agent is told ("orders-etl runs at 02:00 IST"). `budget` caps each question (Budget, or a dict of its fields). + + It has no list prices: Azure OpenAI's depend on the deployment (see model_prices=). `thread` is the conversation + follow-up questions continue, and `history` every question asked, in order. """ - _LIST_PRICES: dict[str, Any] = {} # no list prices: Azure OpenAI's depend on the deployment; see model_prices= + _LIST_PRICES: dict[str, Any] = {} - def __init__( + def __init__( # noqa: PLR0913 self, session: Any = None, *, @@ -77,19 +101,17 @@ def __init__( ): super().__init__(session, region=region, profile=profile, prices=model_prices) self.deployment, self.endpoint, self.api_version = azure_settings(deployment, endpoint, api_version) - self._api_key = api_key # never shown; None reads AZURE_OPENAI_API_KEY when the model is made + self._api_key = api_key self._token_provider = azure_ad_token_provider self.chat_model = chat_model self.max_tokens = max_tokens - if timezone is not None: - _zone_from(timezone) # a misspelled time zone is an error now, not at the first question - self.timezone = timezone # None: the zone reports show times in (set_timezone()) + self.timezone = _checked_zone(timezone) self.regions = regions self.notes = notes self.budget = Budget.of(budget) self._clients: dict[Any, Any] = {} self._lock = threading.Lock() - self.max_workers = 8 + self.max_workers = _WORKERS self._guard = _Guard() self.toolkits: list[Toolkit] = [toolkit.bind(self) for toolkit in make_toolkits( toolkits, {"lambda": {"analyzer": lambda_core, "clients": lambda_clients, "prices": prices}, @@ -97,8 +119,8 @@ def __init__( self._graph: Any = None self._run: _Run | None = None self._tools: list[Callable[..., dict[str, Any]]] | None = None - self.thread = uuid.uuid4().hex[:12] # the conversation follow-up questions continue - self.history: list[AgentAnswer] = [] # every question asked, in order + self.thread = _new_thread() + self.history: list[AgentAnswer] = [] def toolkit(self, name: str) -> Toolkit: """ @@ -144,8 +166,6 @@ def reads(self) -> dict[str, tuple[str, ...]]: """ return {name: reads for toolkit in self.toolkits for name, reads in toolkit.reads.items()} - # ------------------------------------------------------------------ the model and the loop - @property def model(self) -> str: """ @@ -172,10 +192,10 @@ def instructions(self) -> str: def _harness(self) -> Any: """ The deepagents agent: the model, the tools, the instructions and a memory of the conversation. Made once; - new_conversation() starts a new thread in it. + new_conversation() starts a new thread in it. Before anything else is made, check_ready() says what's missing. """ if self._graph is None: - harness.check_ready() # says what's missing before anything else is made + harness.check_ready() prompt = self.instructions() self._graph = harness.build_graph(self._chat(), self.tools(), prompt) return self._graph @@ -184,11 +204,9 @@ def new_conversation(self) -> str: """ Starts a new conversation: the next question won't see the earlier ones. Returns its ID. """ - self.thread = uuid.uuid4().hex[:12] + self.thread = _new_thread() return self.thread - # ------------------------------------------------------------------ asking - def ask( self, question: str, @@ -234,8 +252,6 @@ def ask( self.history.append(answer) return answer - # ------------------------------------------------------------------ the tools - def tools(self) -> list[Callable[..., dict[str, Any]]]: """ The tools the agent can call, from every toolkit: plain Python functions (type hints and docstrings say how diff --git a/src/aws_analyzer/agent/answers.py b/src/aws_analyzer/agent/answers.py index e0e5a85..14ba3d5 100644 --- a/src/aws_analyzer/agent/answers.py +++ b/src/aws_analyzer/agent/answers.py @@ -20,17 +20,24 @@ class ToolCall: """ One call a tool made for a question: what it was asked, what it covered and found, or why it didn't run. + + summary one line: what it found + coverage its window, regions, whether it was cut short, ... + error why it failed: 'AccessDeniedException; needs logs:FilterLogEvents' + refused why it wasn't run: 'tool_calls', 'time', 'cost', 'log_window' or 'stopped' + next View calls that show more, and what they show + first the tool's first parameter when it has no default, which the call is written without the name of """ name: str args: dict[str, Any] = field(default_factory=dict) seconds: float = 0.0 - summary: str = "" # one line: what it found - coverage: dict[str, Any] = field(default_factory=dict) # window, regions, cut_short, ... - error: str | None = None # why it failed: 'AccessDeniedException; needs logs:FilterLogEvents' - refused: str | None = None # why it wasn't run: 'tool_calls', 'time', 'cost', 'log_window', 'stopped' - next: list[tuple[str, str]] = field(default_factory=list) # View calls that show more, and what they show - first: str | None = "function" # the tool's first parameter when it has no default: written without its name + summary: str = "" + coverage: dict[str, Any] = field(default_factory=dict) + error: str | None = None + refused: str | None = None + next: list[tuple[str, str]] = field(default_factory=list) + first: str | None = "function" @property def ok(self) -> bool: @@ -50,26 +57,38 @@ def call(self) -> str: class AgentAnswer: """ One question and its answer, with every tool call it made, the AWS reads, the tokens and what it cost. + + text the answer, in markdown + aws_calls the AWS reads it made: 'logs:FilterLogEvents' -> 12 + blocked AWS calls the read-only guard refused + metrics_read CloudWatch metrics requested (GetMetricData bills per 1,000) + aws_cost USD for those metrics ($0.01 per 1,000 at list price) and S3's requests; the rest are free + model the deployment or model that answered + cost estimated USD for the model's tokens; None when its prices aren't known + stopped None when it finished, else 'tool_calls', 'time', 'cost', 'steps', 'stopped' or 'error' + error what went wrong with the model or the agent loop, when something did + thread the conversation it belongs to + exception the error itself, to raise when debugging """ question: str - text: str = "" # the answer, in markdown + text: str = "" calls: list[ToolCall] = field(default_factory=list) - aws_calls: dict[str, int] = field(default_factory=dict) # 'logs:FilterLogEvents' -> 12 - blocked: list[str] = field(default_factory=list) # AWS calls the read-only guard refused - metrics_read: int = 0 # CloudWatch metrics requested (GetMetricData bills per 1,000) - aws_cost: float = 0.0 # USD for those metrics ($0.01 per 1,000 at list price) and S3's requests; the rest are free - model: str = "" # the deployment or model that answered + aws_calls: dict[str, int] = field(default_factory=dict) + blocked: list[str] = field(default_factory=list) + metrics_read: int = 0 + aws_cost: float = 0.0 + model: str = "" model_calls: int = 0 input_tokens: int = 0 output_tokens: int = 0 - cost: float | None = None # estimated USD for the model's tokens; None when its prices aren't known + cost: float | None = None seconds: float = 0.0 - stopped: str | None = None # None (it finished) | 'tool_calls' | 'time' | 'cost' | 'steps' | 'stopped' | 'error' - error: str | None = None # what went wrong with the model or the agent loop, when something did - thread: str = "" # the conversation it belongs to + stopped: str | None = None + error: str | None = None + thread: str = "" asked: datetime | None = None - exception: BaseException | None = field(default=None, repr=False) # the error itself, to raise when debugging + exception: BaseException | None = field(default=None, repr=False) def answer_findings(answer: AgentAnswer, budget: Budget | None = None) -> list[tuple[str, str]]: @@ -130,11 +149,11 @@ def answer_findings(answer: AgentAnswer, budget: Budget | None = None) -> list[t def answer_next(answer: AgentAnswer, n: int = 3) -> list[tuple[str, str]]: """ The commands worth running next: the View calls behind the tools the answer used, the most specific first, - without repeats. + without repeats: one per command and function (errors('etl') once, with the latest window). """ seen: dict[str, tuple[str, str]] = {} for call in reversed(answer.calls): - for command, what in call.next: # one per command and function: errors('etl') once, the latest window + for command, what in call.next: key = re.match(r"(?:\w+\(\)\.)?(\w+)\(('[^']*')?", command) seen.setdefault(key.group(0) if key else command, (command, what)) return list(seen.values())[:n] diff --git a/src/aws_analyzer/agent/budget.py b/src/aws_analyzer/agent/budget.py index 919d867..f72b401 100644 --- a/src/aws_analyzer/agent/budget.py +++ b/src/aws_analyzer/agent/budget.py @@ -1,6 +1,7 @@ """ How far one question may go (Budget), and the state of the question being answered (_Run): its tool calls, what -the model has cost so far, and the stop button. check_budget() is what every tool asks before it runs. +the model has cost so far, and the stop button. check_budget() is what every tool asks before it runs, and +_REFUSED is what the table of tool calls says about a call that wasn't run. """ from __future__ import annotations @@ -17,19 +18,29 @@ from .answers import ToolCall +_METRICS_PER_PRICE = 1000 + + @dataclass class Budget: """ How far one question may go. A tool that would go past a limit isn't run: it tells the model to answer with what it has, and the answer says what stopped it. + + tool_calls tool calls per question + seconds wall-clock time per question, model included + usd estimated model cost per question (only when the model's prices are known) + steps LangGraph's recursion limit: model calls, tool calls and the harness's own steps together + log_days the longest window a log read (errors, logs, performance) may cover + s3_keys the most keys one S3 listing may read (find_files, folder_summary, s3_triggers) """ - tool_calls: int = 25 # tool calls per question - seconds: float = 300.0 # wall-clock time per question, model included - usd: float | None = 1.00 # estimated model cost per question (only when the model's prices are known) - steps: int = 80 # LangGraph's recursion limit: model calls, tool calls and the harness's own steps together - log_days: float = 3.0 # the longest window a log read (errors, logs, performance) may cover - s3_keys: int = 100_000 # the most keys one S3 listing may read (find_files, folder_summary, s3_triggers) + tool_calls: int = 25 + seconds: float = 300.0 + usd: float | None = 1.00 + steps: int = 80 + log_days: float = 3.0 + s3_keys: int = 100_000 @classmethod def of(cls, value: Budget | dict[str, Any] | None) -> Budget: @@ -56,15 +67,19 @@ class _Run: """ One question's state while it's being answered: what the tools did and what's left of the budget. The tools of one turn run side by side, so everything they change is changed under `lock`. + + cost the model's estimated cost so far (USD) + metrics_read CloudWatch metrics the tools requested (GetMetricData bills per 1,000) + aws_cost USD for those metrics; every other read the tools make is free """ budget: Budget started: float = field(default_factory=time.monotonic) stop: threading.Event = field(default_factory=threading.Event) calls: list[ToolCall] = field(default_factory=list) - cost: float = 0.0 # the model's estimated cost so far (USD) - metrics_read: int = 0 # CloudWatch metrics the tools requested (GetMetricData bills per 1,000) - aws_cost: float = 0.0 # USD for those metrics; every other read the tools make is free + cost: float = 0.0 + metrics_read: int = 0 + aws_cost: float = 0.0 lock: threading.Lock = field(default_factory=threading.Lock) @property @@ -77,7 +92,7 @@ def read_metrics(self, count: int, price: float) -> None: """ with self.lock: self.metrics_read += count - self.aws_cost += count / 1000 * price + self.aws_cost += count / _METRICS_PER_PRICE * price class _Refused(Exception): @@ -90,7 +105,7 @@ def __init__(self, reason: str, message: str, **extra: Any): self.reason, self.message, self.extra = reason, message, extra -_REFUSED = { # what the table of tool calls says about a call that wasn't run +_REFUSED = { "log_window": "not run: the window is longer than the log limit", "tool_calls": "not run: the budget of tool calls was used up", "time": "not run: the time budget was used up", diff --git a/src/aws_analyzer/agent/guard.py b/src/aws_analyzer/agent/guard.py index 089a6da..021399f 100644 --- a/src/aws_analyzer/agent/guard.py +++ b/src/aws_analyzer/agent/guard.py @@ -2,6 +2,10 @@ The read-only guard: every AWS client a toolkit's analyzer makes for the agent is hooked, and any operation that isn't a read is refused before its request is built, whatever asks for it (the model, a tool, a bug). It also counts the reads it lets through, for the report's "AWS reads it made". + +READ_PREFIXES are the AWS operations the tools may call: reads only (FilterLogEvents reads a log group; HeadObject +and HeadBucket read an object's or a bucket's headers). The guard refuses every other one before its request is +built. """ from __future__ import annotations @@ -10,9 +14,7 @@ from collections import Counter from typing import Any -READ_PREFIXES = ("Get", "List", "Describe", "Head", "Filter") # the AWS operations the tools may call: reads only -# (FilterLogEvents reads a log group; HeadObject and HeadBucket read an object's or a bucket's headers). The guard -# refuses every other one before its request is built. +READ_PREFIXES = ("Get", "List", "Describe", "Head", "Filter") class ReadOnlyError(PermissionError): @@ -56,12 +58,13 @@ def __repr__(self) -> str: class _Guard: """ Refuses every AWS call a tool's client would make that isn't a read, before its request is built, and counts - the ones it lets through. A toolkit hooks it onto its analyzer (attach) or onto a client it made (watch). + the ones it lets through. A toolkit hooks it onto its analyzer (attach) or onto a client it made (watch). Its + hook has one ID, so each client's event emitter keeps it once, however often it's asked to. """ def __init__(self) -> None: self._lock = threading.Lock() - self._hook = f"aws-analyzer-read-only-{id(self)}" # each client's emitter keeps it once, however often asked + self._hook = f"aws-analyzer-read-only-{id(self)}" self.counts: Counter[str] = Counter() self.blocked: list[str] = [] @@ -84,12 +87,12 @@ def _check(self, model: Any = None, **_: Any) -> None: def watch(self, client: Any) -> Any: """ Hooks one boto3 client (once: its event emitter keeps the hook's unique_id, so asking again does nothing) - and returns it. + and returns it. Tools run side by side, so the hook is registered under the lock. """ events = getattr(getattr(client, "meta", None), "events", None) if events is None: return client - with self._lock: # tools run side by side: register each hook once + with self._lock: events.register_first("before-parameter-build.*.*", self._check, unique_id=self._hook) return client @@ -99,6 +102,9 @@ def attach(self, core: Any, factory: str | None = None) -> Any: hooked; the clients it already has are hooked (its main `client` when it's set in __init__, and the ones it keeps in dicts, like `_clients`); and, when it makes clients through a method (`factory`, LambdaAnalyzer's `_service`), every client that returns, which covers clients passed in (clients=) too. Returns `core`. + + It reads the analyzer's attributes as they're stored, never a `client` property, since reading one could + make a client. A tool running beside this one can add a client while they're read; that one was made hooked. """ if factory and not getattr(core, "_agent_guard", None): make = getattr(core, factory) @@ -111,11 +117,11 @@ def made(*args: Any, **kwargs: Any) -> Any: session = getattr(core, "session", None) if session is not None and not isinstance(session, _GuardedSession): core.session = _GuardedSession(session, self) - attributes = getattr(core, "__dict__", {}) # not a `client` property: reading one could make a client + attributes = getattr(core, "__dict__", {}) for value in list(attributes.values()): try: found = list(value.values()) if isinstance(value, dict) else [value] - except RuntimeError: # a tool running beside this one added a client: it was made hooked + except RuntimeError: continue for client in found: if _is_client(client): diff --git a/src/aws_analyzer/agent/harness.py b/src/aws_analyzer/agent/harness.py index a869138..1f98589 100644 --- a/src/aws_analyzer/agent/harness.py +++ b/src/aws_analyzer/agent/harness.py @@ -3,6 +3,9 @@ question streamed through it (stream_turn), counting the model's calls and tokens and stopping when asked or out of time. What changes how the agent works rather than what it can read (a subagent, a middleware, another memory) goes in build_graph(); everything about LangGraph's messages stays in this file. + +_MIN_PYTHON is deepagents' floor; the package's own is 3.10, and everything but the agent itself runs there. +HIDDEN_TOOLS are deepagents' tools the model never sees (see _hide_tools). """ from __future__ import annotations @@ -15,9 +18,10 @@ from .budget import _Run from .llm import answer_cost, explain_model_error -_MIN_PYTHON = (3, 11) # deepagents' floor (the package's own is 3.10: everything but the agent itself runs there) -HIDDEN_TOOLS = frozenset({"task", "execute"}) # deepagents' tools the model never sees: its general-purpose -# subagent, which the agent doesn't need, and shell commands, which its in-memory files can't run anyway +_MIN_PYTHON = (3, 11) +HIDDEN_TOOLS = frozenset({"task", "execute"}) +_OVERTIME_SECONDS = 60 +_AI_MESSAGES = ("AIMessage", "AIMessageChunk") def check_ready() -> None: @@ -51,57 +55,81 @@ def build_graph(model: Any, tools: list[Callable[..., Any]], system_prompt: str, ) -def stream_turn(graph: Any, content: str, *, thread: str, run: _Run, answer: AgentAnswer, +def stream_turn(graph: Any, content: str, *, thread: str, run: _Run, answer: AgentAnswer, # noqa: PLR0913 model_prices: dict[str, tuple[float, float]] | None = None, progress: Callable[..., None] | None = None) -> str: """ Streams one question (`content`) through the graph in the conversation `thread`, and returns the answer's text. Fills `answer` as it goes (model calls, tokens, the model that served it) and keeps `run.cost` current - for the tools' cost budget. Stops when `run.stop` is set, the kernel is interrupted or the time budget is well - past; a model or loop that fails sets answer.error and answer.exception instead of raising. + for the tools' cost budget. Stops when `run.stop` is set, the kernel is interrupted (the notebook's stop button: + it keeps what it found) or the time budget is well past. A model or loop that fails, whatever it raises, sets + answer.error and answer.exception instead of raising. """ - budget = run.budget base = answer.model steps = 0 text = "" message = {"role": "user", "content": content} - config = {"configurable": {"thread_id": thread}, "recursion_limit": budget.steps} + config = {"configurable": {"thread_id": thread}, "recursion_limit": run.budget.steps} try: for update in graph.stream({"messages": [message]}, config=config, stream_mode="updates"): - for node in (update or {}).values(): - for msg in _update_messages(node): - if type(msg).__name__ not in ("AIMessage", "AIMessageChunk"): - continue - answer.model_calls += 1 - usage = getattr(msg, "usage_metadata", None) or {} - answer.input_tokens += int(usage.get("input_tokens") or 0) - answer.output_tokens += int(usage.get("output_tokens") or 0) - served = (getattr(msg, "response_metadata", None) or {}).get("model_name") - if served: # the model behind the deployment: 'prod-gpt-4o (gpt-4o-2024-11-20)' - answer.model = served if base in ("", served) else f"{base} ({served})" - cost = answer_cost(answer.model, answer.input_tokens, answer.output_tokens, model_prices) - with run.lock: - run.cost = cost or 0.0 - if not getattr(msg, "tool_calls", None): - text = _message_text(msg) or text + for msg in _model_messages(update): + _count_usage(msg, answer, base, run, model_prices) + if not getattr(msg, "tool_calls", None): + text = _message_text(msg) or text steps += 1 if progress: progress(steps) - if run.stop.is_set(): - answer.stopped = "stopped" - break - if run.elapsed > budget.seconds + 60: # the tools stopped at the budget; the model kept going - answer.stopped = "time" + reason = _stop_reason(run) + if reason: + answer.stopped = reason break - except KeyboardInterrupt: # the notebook's stop button: keep what it found + except KeyboardInterrupt: answer.stopped = "stopped" - except Exception as exc: # a model or loop error: the answer says what went wrong + except Exception as exc: # noqa: BLE001 answer.stopped = "steps" if type(exc).__name__ == "GraphRecursionError" else "error" answer.error = explain_model_error(exc) answer.exception = exc return text +def _model_messages(update: Any) -> list[Any]: + """ + The model's messages in one LangGraph stream update, from every node in it. + """ + return [msg for node in (update or {}).values() for msg in _update_messages(node) + if type(msg).__name__ in _AI_MESSAGES] + + +def _count_usage(message: Any, answer: AgentAnswer, base: str, run: _Run, + model_prices: dict[str, tuple[float, float]] | None) -> None: + """ + Adds one model message to the answer: a model call, its tokens, and the model that served it, which for a + deployment reads 'prod-gpt-4o (gpt-4o-2024-11-20)'. Keeps run.cost current for the tools' cost budget. + """ + answer.model_calls += 1 + usage = getattr(message, "usage_metadata", None) or {} + answer.input_tokens += int(usage.get("input_tokens") or 0) + answer.output_tokens += int(usage.get("output_tokens") or 0) + served = (getattr(message, "response_metadata", None) or {}).get("model_name") + if served: + answer.model = served if base in ("", served) else f"{base} ({served})" + cost = answer_cost(answer.model, answer.input_tokens, answer.output_tokens, model_prices) + with run.lock: + run.cost = cost or 0.0 + + +def _stop_reason(run: _Run) -> str: + """ + Why the loop stops now: 'stopped' when run.stop is set, 'time' once the time budget is _OVERTIME_SECONDS past + (the tools stopped at the budget, but the model kept going); '' to go on. + """ + if run.stop.is_set(): + return "stopped" + if run.elapsed > run.budget.seconds + _OVERTIME_SECONDS: + return "time" + return "" + + def _update_messages(node: Any) -> list[Any]: """ The messages a LangGraph stream update carries for one node (a list, or an Overwrite wrapping one). diff --git a/src/aws_analyzer/agent/llm.py b/src/aws_analyzer/agent/llm.py index bd11a1c..953659c 100644 --- a/src/aws_analyzer/agent/llm.py +++ b/src/aws_analyzer/agent/llm.py @@ -2,6 +2,9 @@ The model the agent answers with: an Azure OpenAI deployment made from settings or the environment (or any LangChain chat model passed in), its name, what its tokens cost, and what a failed call to it means. Another provider (a model on Amazon Bedrock, OpenAI itself) gets its own maker here, next to azure_chat_model(). + +AZURE_API_VERSION is the Azure OpenAI API version used when neither api_version= nor the environment +(OPENAI_API_VERSION, AZURE_OPENAI_API_VERSION) names one: a GA version that supports tool calling. """ from __future__ import annotations @@ -12,8 +15,9 @@ from .._kit.deps import _require from .._kit.errors import _Hint -AZURE_API_VERSION = "2024-10-21" # the Azure OpenAI API version used when neither api_version= nor the environment -# (OPENAI_API_VERSION, AZURE_OPENAI_API_VERSION) names one: a GA version that supports tool calling +AZURE_API_VERSION = "2024-10-21" +_RETRIES = 3 +_ERROR_CHARACTERS = 300 def azure_settings(deployment: str | None = None, endpoint: str | None = None, @@ -28,8 +32,9 @@ def azure_settings(deployment: str | None = None, endpoint: str | None = None, or AZURE_API_VERSION) -def azure_chat_model(*, endpoint: str | None, deployment: str | None, api_version: str, api_key: str | None = None, - token_provider: Callable[[], str] | None = None, max_tokens: int | None = None) -> Any: +def azure_chat_model(*, endpoint: str | None, deployment: str | None, api_version: str, # noqa: PLR0913 + api_key: str | None = None, token_provider: Callable[[], str] | None = None, + max_tokens: int | None = None) -> Any: """ LangChain's AzureChatOpenAI for a deployment. The key is api_key, else AZURE_OPENAI_API_KEY, or a Microsoft Entra ID token_provider instead. _Hint (a note, not a traceback) says which settings are missing. @@ -47,7 +52,7 @@ def azure_chat_model(*, endpoint: str | None, deployment: str | None, api_versio ) openai = _require("langchain_openai", "The agent's Azure OpenAI model", package="langchain-openai") settings: dict[str, Any] = {"azure_endpoint": endpoint, "azure_deployment": deployment, - "api_version": api_version, "max_retries": 3} + "api_version": api_version, "max_retries": _RETRIES} if max_tokens: settings["max_tokens"] = max_tokens if api_key: @@ -94,8 +99,8 @@ def explain_model_error(exc: BaseException) -> str: """ name = type(exc).__name__ text = " ".join(str(exc).split()) - if len(text) > 300: - text = text[:299] + "…" + if len(text) > _ERROR_CHARACTERS: + text = text[:_ERROR_CHARACTERS - 1] + "…" status = getattr(exc, "status_code", None) if name == "AuthenticationError" or status == 401: why = ("Azure OpenAI refused the key: check AZURE_OPENAI_API_KEY (or api_key=) is a key of this resource, " diff --git a/src/aws_analyzer/agent/prompt.py b/src/aws_analyzer/agent/prompt.py index ae4baed..1c6b099 100644 --- a/src/aws_analyzer/agent/prompt.py +++ b/src/aws_analyzer/agent/prompt.py @@ -19,7 +19,8 @@ def asked_at(now: datetime, zone: tzinfo | str | None = timezone.utc) -> str: """ The line that starts each question sent to the model: when it was asked, in the user's time zone and in UTC, so "last night" and "this morning" mean what the user meant. zone is a tzinfo or anything set_timezone() takes - ('Asia/Kolkata', 'kolkata', '+05:30'); None is this machine's own zone. + ('Asia/Kolkata', 'kolkata', '+05:30'); None is this machine's own zone. The zone is named as the tz database + does ('Asia/Kolkata'), or by its offset ('UTC+05:30'). """ utc = now.astimezone(timezone.utc) tz = _zone_from(zone) if isinstance(zone, str) else zone @@ -27,7 +28,7 @@ def asked_at(now: datetime, zone: tzinfo | str | None = timezone.utc) -> str: if local.utcoffset() == timedelta(0) and _zone_name(local) == "UTC": return (f"[Asked {utc:%A %Y-%m-%d %H:%M} UTC. Times are in UTC (the user can pick their zone with " "set_timezone()): give them in UTC and say so.]") - name = getattr(tz, "key", None) or _zone_name(local) # 'Asia/Kolkata', or 'UTC+05:30' for an offset + name = getattr(tz, "key", None) or _zone_name(local) return (f"[Asked {local:%A %Y-%m-%d %H:%M} in the user's time zone ({name}, UTC{local:%z}), " f"which is {utc:%Y-%m-%d %H:%M} UTC.]") diff --git a/src/aws_analyzer/agent/results.py b/src/aws_analyzer/agent/results.py index e052013..f5ccfef 100644 --- a/src/aws_analyzer/agent/results.py +++ b/src/aws_analyzer/agent/results.py @@ -2,6 +2,10 @@ What a tool hands the model: JSON-ready values, time windows in UTC, text with anything that looks like a secret hidden, and the parts it couldn't read with the permission each one needs. Every toolkit builds its results with these, so the model reads the same shapes whichever service a tool is about. + +_SECRETS is what hide_secrets() takes out of a log line before the model sees it, as (pattern, replacement) pairs: +AWS access key IDs, JWTs, Bearer and Basic credentials, private keys, values after secret-sounding names, and the +password in a user:password@host URL. """ from __future__ import annotations @@ -12,15 +16,15 @@ from .._kit.errors import _why -_SECRETS = ( # (pattern, replacement): what hide_secrets() takes out of a log line before the model sees it +_SECRETS = ( (re.compile(r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b"), "[access key hidden]"), - (re.compile(r"\beyJ[\w-]{8,}\.[\w-]{8,}\.[\w-]{8,}"), "[token hidden]"), # a JWT + (re.compile(r"\beyJ[\w-]{8,}\.[\w-]{8,}\.[\w-]{8,}"), "[token hidden]"), (re.compile(r"(?i)\b(bearer|basic)\s+[\w.~+/=-]{12,}"), r"\1 [hidden]"), (re.compile(r"(?i)(-----BEGIN [A-Z ]*PRIVATE KEY-----).*?(-----END [A-Z ]*PRIVATE KEY-----|$)", re.S), r"\1 [hidden]"), (re.compile(r"(?i)\b([\w-]*(?:password|passwd|secret|token|api[_-]?key|access[_-]?key|private[_-]?key|" r"credential)[\w-]*)(\"?\s*[:=]\s*\"?)([^\s\"',;&}]+)"), r"\1\2[hidden]"), - (re.compile(r"(?i)([a-z][a-z0-9+.-]*://[^\s:/@]+:)([^\s@/]+)(@)"), r"\1[hidden]\3"), # user:password@host + (re.compile(r"(?i)([a-z][a-z0-9+.-]*://[^\s:/@]+:)([^\s@/]+)(@)"), r"\1[hidden]\3"), ) diff --git a/src/aws_analyzer/agent/runtime.py b/src/aws_analyzer/agent/runtime.py index 0a6f0eb..0c289a9 100644 --- a/src/aws_analyzer/agent/runtime.py +++ b/src/aws_analyzer/agent/runtime.py @@ -11,7 +11,7 @@ import time from typing import TYPE_CHECKING, Any, Callable -from botocore.exceptions import BotoCoreError, ClientError +from botocore.exceptions import ClientError from .._kit.errors import _error_code, _why from .._kit.text import _clip @@ -23,7 +23,10 @@ if TYPE_CHECKING: from .toolkits import Toolkit -COVERAGE = ("window", "regions", "region", "cut_short", "covered_from", "scanned") # a result's keys the report shows +COVERAGE = ("window", "regions", "region", "cut_short", "covered_from", "scanned") +_ERROR_CHARACTERS = 200 +_MESSAGE_CHARACTERS = 400 +_AWS_MESSAGE_CHARACTERS = 300 def as_tool(fn: Callable[..., dict[str, Any]], toolkit: Toolkit, @@ -38,12 +41,7 @@ def as_tool(fn: Callable[..., dict[str, Any]], toolkit: Toolkit, @functools.wraps(fn) def tool(*args: Any, **kwargs: Any) -> dict[str, Any]: run = current() - try: - arguments = dict(inspect.signature(fn).bind_partial(*args, **kwargs).arguments) - except TypeError: - arguments = dict(kwargs) - defaults = _defaults(fn) # LangChain passes every default: keep what the model asked for - arguments = {k: v for k, v in arguments.items() if k not in defaults or v != defaults[k]} + arguments = _asked_for(fn, args, kwargs) call = ToolCall(fn.__name__, arguments, first=_first_parameter(fn)) started = time.monotonic() with run.lock: @@ -52,44 +50,80 @@ def tool(*args: Any, **kwargs: Any) -> dict[str, Any]: try: check_budget(run, used) result = fn(*args, **kwargs) - except _Refused as exc: - call.refused, call.summary = exc.reason, _REFUSED.get(exc.reason, "not run") - result = {"refused": exc.reason, "message": exc.message, **exc.extra} - except ClientError as exc: - try: - missing = toolkit.not_found(exc, arguments) - except Exception: # looking for close names failed too: report the error itself - missing = None - if missing is not None: - result = missing - call.summary = str(missing.get("error") or "") - else: - code = _error_code(exc) - permission = toolkit.permission_for(exc) - call.error = _why(code, permission) if permission else code - result = {"error": call.error, - "message": _clip(exc.response.get("Error", {}).get("Message", str(exc)), 300)} - except ReadOnlyError as exc: - call.error = str(exc) - result = {"error": "blocked", "message": str(exc)} - except (BotoCoreError, ValueError, TypeError) as exc: # bad input, a region or time it can't read - call.error = f"{type(exc).__name__}: {_clip(str(exc), 200)}" - result = {"error": type(exc).__name__, "message": _clip(str(exc), 400)} - except Exception as exc: # a tool never brings the agent down; the answer says what failed - call.error = f"{type(exc).__name__}: {_clip(str(exc), 200)}" - result = {"error": type(exc).__name__, "message": _clip(str(exc), 400)} + except Exception as exc: # noqa: BLE001 + result = _failed(exc, call, toolkit, arguments) call.seconds = time.monotonic() - started - unread = result.get("errors") if isinstance(result.get("errors"), dict) else None - if unread and not call.error: - call.error = "; ".join(f"{section}: {why}" for section, why in unread.items()) - call.next = list(result.pop("_next", []) or []) - call.summary = result.pop("_summary", None) or call.summary or (call.error or "") - call.coverage = {k: result[k] for k in COVERAGE if k in result} - return _plain(result) + return _reported(call, result) return tool +def _asked_for(fn: Callable[..., Any], args: tuple, kwargs: dict[str, Any]) -> dict[str, Any]: + """ + The arguments a tool was called with, by name, leaving out the ones equal to their default: LangChain passes + every default, so what's left is what the model asked for. + """ + try: + arguments = dict(inspect.signature(fn).bind_partial(*args, **kwargs).arguments) + except TypeError: + arguments = dict(kwargs) + defaults = _defaults(fn) + return {k: v for k, v in arguments.items() if k not in defaults or v != defaults[k]} + + +def _failed(exc: Exception, call: ToolCall, toolkit: Toolkit, arguments: dict[str, Any]) -> dict[str, Any]: + """ + What a tool that raised hands the model, recorded on its call. A tool never raises into the agent loop: a + refused call says why, an AWS error is the toolkit's to explain (_aws_error), a call the guard blocked says so, + and anything else (bad input, a region or time it can't read, a bug) says what failed. + """ + if isinstance(exc, _Refused): + call.refused, call.summary = exc.reason, _REFUSED.get(exc.reason, "not run") + return {"refused": exc.reason, "message": exc.message, **exc.extra} + if isinstance(exc, ClientError): + return _aws_error(exc, call, toolkit, arguments) + if isinstance(exc, ReadOnlyError): + call.error = str(exc) + return {"error": "blocked", "message": str(exc)} + call.error = f"{type(exc).__name__}: {_clip(str(exc), _ERROR_CHARACTERS)}" + return {"error": type(exc).__name__, "message": _clip(str(exc), _MESSAGE_CHARACTERS)} + + +def _aws_error(exc: ClientError, call: ToolCall, toolkit: Toolkit, arguments: dict[str, Any]) -> dict[str, Any]: + """ + An AWS error as the tool's result: a resource that isn't there as the toolkit's close names, else the error's + code with the permission a refused call needed. When looking for close names fails too, whatever it raises, the + error itself is reported. + """ + try: + missing = toolkit.not_found(exc, arguments) + except Exception: # noqa: BLE001 + missing = None + if missing is not None: + call.summary = str(missing.get("error") or "") + return missing + code = _error_code(exc) + permission = toolkit.permission_for(exc) + call.error = _why(code, permission) if permission else code + message = exc.response.get("Error", {}).get("Message", str(exc)) + return {"error": call.error, "message": _clip(message, _AWS_MESSAGE_CHARACTERS)} + + +def _reported(call: ToolCall, result: dict[str, Any]) -> dict[str, Any]: + """ + Fills in the call from the tool's result, and returns the result as the model reads it: JSON-ready, with the + report's `_summary` and `_next` taken out. Sections the analyzer couldn't read (its `errors`) become the call's + error, and the result's COVERAGE keys (its window, regions, whether it was cut short) are what the report shows. + """ + unread = result.get("errors") if isinstance(result.get("errors"), dict) else None + if unread and not call.error: + call.error = "; ".join(f"{section}: {why}" for section, why in unread.items()) + call.next = list(result.pop("_next", []) or []) + call.summary = result.pop("_summary", None) or call.summary or (call.error or "") + call.coverage = {k: result[k] for k in COVERAGE if k in result} + return _plain(result) + + def _defaults(fn: Callable[..., Any]) -> dict[str, Any]: """ A tool's parameters that have a default, and their defaults. diff --git a/src/aws_analyzer/agent/toolkits/__init__.py b/src/aws_analyzer/agent/toolkits/__init__.py index 352fb26..1318cb9 100644 --- a/src/aws_analyzer/agent/toolkits/__init__.py +++ b/src/aws_analyzer/agent/toolkits/__init__.py @@ -19,8 +19,8 @@ from .lambda_tools import LambdaToolkit from .s3_tools import S3Toolkit -TOOLKITS: dict[str, type[Toolkit]] = {"lambda": LambdaToolkit, "s3": S3Toolkit} # name -> the built-in toolkit -DEFAULT_TOOLKITS = ("lambda", "s3") # what an agent gets unless toolkits= says otherwise +TOOLKITS: dict[str, type[Toolkit]] = {"lambda": LambdaToolkit, "s3": S3Toolkit} +DEFAULT_TOOLKITS = ("lambda", "s3") def make_toolkit(spec: str | Toolkit | type[Toolkit], **settings: Any) -> Toolkit: diff --git a/src/aws_analyzer/agent/toolkits/base.py b/src/aws_analyzer/agent/toolkits/base.py index 2c499e8..d8011be 100644 --- a/src/aws_analyzer/agent/toolkits/base.py +++ b/src/aws_analyzer/agent/toolkits/base.py @@ -6,8 +6,8 @@ each tool's AWS reads and the IAM permissions its operations need, and it defines: make_analyzer(session) the analyzer its tools read through - tools() its tools: plain functions whose names, type hints and docstrings are what the model - reads. Each returns a dict the model can read (results.py's helpers), with + tools() its tools: functions or methods whose names, type hints and docstrings are + what the model reads. Each returns a dict the model can read (results.py's helpers), with coverage keys (window, regions, cut_short, covered_from), and `_summary` (the report's one line) and `_next` ((View call, what it shows) pairs), which the agent takes out instructions(budget) its section of the system prompt: how to use its tools well @@ -40,23 +40,35 @@ class Toolkit: """ The tools the agent has for one AWS service. A subclass sets the class attributes below and defines make_analyzer(), tools() and, when it helps, instructions() and not_found(); see this module's docstring. - `analyzer` is one to read through instead of the one make_analyzer() would make. + `analyzer` is one to read through instead of the one make_analyzer() would make, and `agent` is the + AgentAnalyzer that took the toolkit (bind()). + + name its key in TOOLKITS and AgentAnalyzer(toolkits=[...]): 'lambda' + title how reports name it: 'Lambda' + covers what it lets the agent answer, for the system prompt's first paragraph + privacy what it never reads or sends to the model, for the tools() report's note + reads tool name -> the AWS reads it makes (tools() and README list them) + service the IAM prefix of most of its operations, for permission_for(): 'lambda' + services operation -> its IAM prefix where that's another: {'FilterLogEvents': 'logs'} + permissions operation -> its IAM permission where it isn't :: + {'ListObjectsV2': 's3:ListBucket'} + view the View whose commands its tools suggest next: 'LambdaView' + client_factory the analyzer's method that makes clients, which the guard hooks too """ - name: str = "" # its key in TOOLKITS and AgentAnalyzer(toolkits=[...]): 'lambda' - title: str = "" # how reports name it: 'Lambda' - covers: str = "" # what it lets the agent answer, for the system prompt's first paragraph - privacy: str = "" # what it never reads or sends to the model, for the tools() report's note - reads: dict[str, tuple[str, ...]] = {} # tool name -> the AWS reads it makes (tools() and README list them) - service: str = "" # the IAM prefix of most of its operations, for permission_for(): 'lambda' - services: dict[str, str] = {} # operation -> its IAM prefix where that's another: {'FilterLogEvents': 'logs'} - permissions: dict[str, str] = {} # operation -> its IAM permission where it isn't :: - # {'ListObjectsV2': 's3:ListBucket'} - view: str = "" # the View whose commands its tools suggest next: 'LambdaView' - client_factory: str | None = None # the analyzer's method that makes clients, which the guard hooks too + name: str = "" + title: str = "" + covers: str = "" + privacy: str = "" + reads: dict[str, tuple[str, ...]] = {} + service: str = "" + services: dict[str, str] = {} + permissions: dict[str, str] = {} + view: str = "" + client_factory: str | None = None def __init__(self, analyzer: Any = None): - self.agent: AgentAnalyzer | None = None # set by bind(), when an AgentAnalyzer takes the toolkit + self.agent: AgentAnalyzer | None = None self._analyzer = analyzer def __repr__(self) -> str: @@ -71,7 +83,6 @@ def bind(self, agent: AgentAnalyzer) -> Toolkit: self.agent = agent return self - # ------------------------------------------------------------------ what a toolkit defines def make_analyzer(self, session: Any) -> Any: """ @@ -81,7 +92,7 @@ def make_analyzer(self, session: Any) -> Any: def tools(self) -> list[Callable[..., dict[str, Any]]]: """ - Its tools, as plain functions (see this module's docstring). + Its tools, as functions or methods (see this module's docstring). """ raise NotImplementedError @@ -106,7 +117,6 @@ def request_cost(self, counts: dict[str, int]) -> float: """ return 0.0 - @property def analyzer(self) -> Any: """ @@ -159,7 +169,6 @@ def permission_for(self, exc: ClientError) -> str: service = self.services.get(op, self.service) return f"{service}:{op}" if op and service else "" - # ------------------------------------------------------------------ helpers for its tools def _regions(self, regions: str) -> Any: """ diff --git a/src/aws_analyzer/agent/toolkits/lambda_tools.py b/src/aws_analyzer/agent/toolkits/lambda_tools.py index 01bb24b..d25ed42 100644 --- a/src/aws_analyzer/agent/toolkits/lambda_tools.py +++ b/src/aws_analyzer/agent/toolkits/lambda_tools.py @@ -4,6 +4,11 @@ errors, runs, run times, settings and code (function_errors, function_logs, function_performance, function_details, function_code). The summarize_* functions turn what LambdaAnalyzer returns into the compact dicts the model reads; they're pure, so they run on any Function, ErrorReport, LogPage, Performance, FunctionDetail or CodeMap. + +TOOL_READS is what each tool reads, for tools() and README: what its LambdaAnalyzer method calls. The analyzer's +types are imported for the type hints only, since the analyzer loads when the first tool runs (_lambda_module). +_PERMISSIONS maps a result's error sections to the permission each needs, and _OVERVIEW_PERMISSIONS the sections of +overview()'s 'region:section' errors. """ from __future__ import annotations @@ -23,12 +28,11 @@ from ..results import _iso, _window, _with_permissions, hide_secrets from .base import Toolkit -if TYPE_CHECKING: # for the type hints only: the Lambda analyzer loads when the first tool runs +if TYPE_CHECKING: from ...lambda_functions import ( CodeMap, ErrorReport, Function, FunctionDetail, FunctionMetrics, LambdaAnalyzer, LogPage, Performance, ) -# What each tool reads, for tools() and README. A tool's own reads are what its analyzer method makes. TOOL_READS: dict[str, tuple[str, ...]] = { "find_functions": ("lambda:ListFunctions", "lambda:ListEventSourceMappings", "lambda:GetAccountSettings", "logs:DescribeLogGroups"), @@ -57,9 +61,7 @@ def _lambda_module() -> Any: return lambda_functions -# ============================================================================= -# What the tools return: LambdaAnalyzer's results as compact JSON (pure) -# ============================================================================= +_NEAR_TIMEOUT = 0.9 def _metrics_dict(m: FunctionMetrics | None) -> dict[str, Any] | None: @@ -107,12 +109,11 @@ def summarize_functions(functions: list[Function], regions: list[str], errors: d return out -def summarize_health(functions: list[Function], metrics: dict[str, FunctionMetrics], start: datetime, - end: datetime, regions: list[str], errors: dict[str, str] | None = None, - *, limit: int = 25) -> dict[str, Any]: +def _health_groups(functions: list[Function], + metrics: dict[str, FunctionMetrics]) -> tuple[list[dict], list[dict], list[dict], list[str]]: """ - function_health()'s answer: CloudWatch's counts for every function in a window, grouped into the ones that - failed (most failed calls first), were throttled, ran within 10% of their timeout, and weren't called. + (failed, throttled, near_timeout, not_called): each function with CloudWatch numbers, as a row of them, in + every group it belongs to. Near its timeout is a longest run of _NEAR_TIMEOUT of the timeout or more. """ failed, throttled, near, idle = [], [], [], [] for fn in functions: @@ -126,10 +127,21 @@ def summarize_health(functions: list[Function], metrics: dict[str, FunctionMetri failed.append(row) if m.throttles: throttled.append(row) - if m.duration_max is not None and m.duration_max >= 0.9 * fn.timeout * 1000: + if m.duration_max is not None and m.duration_max >= _NEAR_TIMEOUT * fn.timeout * 1000: near.append(row) if not m.invocations: idle.append(fn.name) + return failed, throttled, near, idle + + +def summarize_health(functions: list[Function], metrics: dict[str, FunctionMetrics], # noqa: PLR0913 + start: datetime, end: datetime, regions: list[str], errors: dict[str, str] | None = None, + *, limit: int = 25) -> dict[str, Any]: + """ + function_health()'s answer: CloudWatch's counts for every function in a window, grouped into the ones that + failed (most failed calls first), were throttled, ran within 10% of their timeout, and weren't called. + """ + failed, throttled, near, idle = _health_groups(functions, metrics) failed.sort(key=lambda r: (-r["errors"], r["name"])) throttled.sort(key=lambda r: (-r["throttles"], r["name"])) near.sort(key=lambda r: -(r["longest_ms"] or 0) / max(r["timeout_ms"], 1)) @@ -295,22 +307,19 @@ def summarize_code(cmap: CodeMap) -> dict[str, Any]: } -_PERMISSIONS = {"logs": "logs:FilterLogEvents", "metrics": "cloudwatch:GetMetricData"} # error section -> permission +_PERMISSIONS = {"logs": "logs:FilterLogEvents", "metrics": "cloudwatch:GetMetricData"} _OVERVIEW_PERMISSIONS = {"list": "lambda:ListFunctions", "triggers": "lambda:ListEventSourceMappings", "account": "lambda:GetAccountSettings", "logs": "logs:DescribeLogGroups", - "metrics": "cloudwatch:GetMetricData"} # overview()'s 'region:section' errors - - - -# ============================================================================= -# LambdaToolkit -# ============================================================================= + "metrics": "cloudwatch:GetMetricData"} class LambdaToolkit(Toolkit): """ The agent's tools for AWS Lambda, reading through LambdaAnalyzer: `analyzer` is one to use (else one is made - on the agent's session), `clients` its clients (for tests) and `prices` its prices. + on the agent's session), `clients` its clients (for tests) and `prices` its prices. LambdaAnalyzer makes a + client per service and region through its _service, which the guard hooks. + + Each tool is a method: its name, type hints and docstring are what the model reads. """ name = "lambda" @@ -324,7 +333,7 @@ class LambdaToolkit(Toolkit): services = {"FilterLogEvents": "logs", "DescribeLogGroups": "logs", "DescribeLogStreams": "logs", "GetMetricData": "cloudwatch", "DescribeRegions": "ec2"} view = "LambdaView" - client_factory = "_service" # LambdaAnalyzer makes a client per service and region through it + client_factory = "_service" def __init__(self, analyzer: LambdaAnalyzer | None = None, *, clients: dict[str, Any] | None = None, prices: dict[str, float] | None = None): @@ -352,162 +361,151 @@ def instructions(self, budget: Budget) -> str: counts (function_health) cover longer windows. If a tool refuses a window, explain why and what the user can do. - Name functions exactly as the tools return them.""" - # ------------------------------------------------------------------ the tools - def tools(self) -> list[Callable[..., dict[str, Any]]]: """ find_functions, function_health, function_errors, function_logs, function_performance, function_details - and function_code: plain functions (their type hints and docstrings tell the model how to call them), each + and function_code: methods whose type hints and docstrings tell the model how to call them, each reading through LambdaAnalyzer and returning a compact, JSON-ready summary with what it covered. """ + return [self.find_functions, self.function_health, self.function_errors, self.function_logs, + self.function_performance, self.function_details, self.function_code] - def find_functions(match: str = "", regions: str = "") -> dict[str, Any]: - """ - Lists the Lambda functions in the account: name, region, runtime (and whether AWS still supports - it), memory, timeout, when each was last changed and its description. match: a name glob like 'etl-*'. - regions: 'us-east-1,eu-west-1' or 'all'; empty for the default regions. - """ - core = self.analyzer - ov = core.overview(match or None, regions=self._regions(regions), metrics=False, details=False) - out = summarize_functions(ov.functions, ov.regions, ov.errors) - out["_summary"] = f"{_plural(len(ov.functions), 'function')} in {', '.join(ov.regions)}" - out["_next"] = [(self._view_call("functions", match or None, regions=regions or None), - "every function with its calls, errors, cost and warnings")] - return out - - def function_health(since: str = "24h", until: str = "", match: str = "", regions: str = "") -> dict[str, Any]: - """ - How every Lambda function did in a time window, from CloudWatch: calls, failed calls, throttles and - the longest run against its timeout, sorted into failed, throttled, near_timeout and not_called. Fast and - cheap: start here for "what failed", "what's slow" or "what's unused". since/until: ISO UTC times - ('2026-10-15T18:00Z') or relative ('24h', '7d'); until defaults to now. match: a name glob like 'etl-*'. - regions: 'us-east-1,eu-west-1' or 'all'; empty for the default regions. - """ - core = self.analyzer - start, end = core._window(since or "24h", until or None) - ov = core.overview(match or None, regions=self._regions(regions), metrics=False, details=False) - found, read, _, errors = core._numbers(ov.functions, since=start, until=end) - self._count_metrics(read, core.prices["metric_request"]) - out = summarize_health(ov.functions, found, start, end, ov.regions, {**ov.errors, **errors}) - failed = out["failed"] - out["_summary"] = (f"{len(failed) + out.get('failed_more', 0)} of {_plural(len(ov.functions), 'function')} " - "had failed calls" + (": " + ", ".join(f"{r['name']} ({r['errors']:,})" for r in failed[:3]) - if failed else "")) - out["_next"] = [(self._next_for(fn, "errors", since=_since(start)), "its errors, grouped by cause") - for fn in ov.functions if fn.name in {r["name"] for r in failed[:2]}] - return out - - def function_errors(function: str, since: str = "24h", until: str = "", region: str = "") -> dict[str, Any]: - """ - One function's errors in a time window, read from its logs and grouped by cause (Timeout, KeyError, - AccessDenied, ...), with CloudWatch's count of failed and throttled calls over the same window. Each - cause has a count, when it first and last happened, an example line and request IDs to look up with - function_logs. since/until: ISO UTC times or relative ('24h'); the window can't be longer than the log - limit. region: empty for the default region. - """ - core = self.analyzer - start, end = self._log_window(function, since, until, region or None) - report = core.errors(function, since=start, until=end, region=region or None) - out = summarize_errors(report) - causes = out["causes"] - kinds = Counter(c["kind"] for c in causes[:3]) - - def label(cause: dict[str, Any]) -> str: # the kind, and what tells it apart when two share a kind - kind = cause["kind"] - rest = re.sub(rf"^\W*{re.escape(kind)}\W*", "", cause["message"] or "").strip() - if kinds[kind] > 1 and rest: - return f"{kind} ({_clip(rest, 40)}) ×{cause['count']}" - return f"{kind} ×{cause['count']}" - - out["_summary"] = (f"{_plural(sum(c['count'] for c in causes), 'error')}, {_plural(len(causes), 'cause')}" - + (": " + ", ".join(label(c) for c in causes[:3]) if causes else "")) - out["_next"] = [(self._next_for(report.function, "errors", since=_since(start)), - "the same errors as a report, with each cause's findings")] - if causes and causes[0]["request_ids"]: - out["_next"].insert(0, (self._next_for(report.function, "logs", request_id=causes[0]["request_ids"][0]), - f"one {causes[0]['kind']} run, start to end")) - return out - - def function_logs(function: str, since: str = "1h", until: str = "", search: str = "", - region: str = "") -> dict[str, Any]: - """ - One function's runs in a time window, newest first, from its logs: each run's status (ok, failed, - timeout, logged an error), run time, memory, whether it was a cold start, and its last lines. search: - text the runs' lines must contain, such as a request ID, an order ID or 'KeyError' (case-sensitive). - since/until: ISO UTC times or relative ('1h'); the window can't be longer than the log limit. region: - empty for the default region. - """ - core = self.analyzer - start, end = self._log_window(function, since, until, region or None) - page = core.log_runs(function, since=start, until=end, search=search or None, region=region or None) - out = summarize_runs(page) - statuses = out["statuses"] - out["_summary"] = (f"{_plural(out['runs_found'], 'run')}" - + (f" ({', '.join(f'{n} {s}' for s, n in statuses.items())})" if statuses else "")) - out["_next"] = [(self._next_for(page.function, "logs", since=_since(start), search=search or None), - "the same lines, as a report")] - return out - - def function_performance(function: str, since: str = "24h", until: str = "", - region: str = "") -> dict[str, Any]: - """ - One function's run times and memory in a time window, from the REPORT line Lambda logs after every - run: median, p95, p99 and longest run against its timeout, timeouts, cold starts, the most memory a run - used and the memory size it needs. since/until: ISO UTC times or relative ('24h'); the window can't be - longer than the log limit. region: empty for the default region. - """ - core = self.analyzer - start, end = self._log_window(function, since, until, region or None) - perf = core.performance(function, since=start, until=end, region=region or None) - out = summarize_performance(perf, core.prices) - p95 = out["p95_ms"] - out["_summary"] = (f"{_plural(out['runs'], 'run')}" - + (f", p95 {p95 / 1000:.1f}s of a {perf.function.timeout}s timeout" if p95 else "")) - out["_next"] = [(self._next_for(perf.function, "performance", since=_since(start)), - "run times, memory and cold starts as a report")] - return out - - def function_details(function: str, region: str = "") -> dict[str, Any]: - """ - Everything about one function: its settings (runtime and its support dates, memory, timeout, - reserved concurrency), what triggers it, its aliases and versions, its last 30 days from CloudWatch, its - log group's retention and size, the estimated monthly cost by part, and every finding about it with the - next step. region: empty for the default region. - """ - core = self.analyzer - detail = core.describe(function, region=region or None, days=30) - out = summarize_detail(detail, core.prices) - total = out["monthly_cost_usd"].get("total") - out["_summary"] = (f"{detail.function.runtime or 'container image'}, {detail.function.memory:,} MB, " - f"{detail.function.timeout}s; {_plural(len(out['findings']), 'finding')}" - + (f"; about {human_money(total)}/month" if total is not None else "")) - out["_next"] = [(self._next_for(detail.function, "function_info"), "the function in plain English")] - return out - - def function_code(function: str, region: str = "") -> dict[str, Any]: - """ - What a Python function's code does, read from its deployment package with Python's ast module - (nothing is run): the AWS calls it makes, the environment variables it reads (names only), the event - fields it uses, the libraries it imports and findings such as HTTP calls without a timeout. Other - runtimes and container images give a note. region: empty for the default region. - """ - core = self.analyzer - cmap = core.code_map(function, region=region or None) - out = summarize_code(cmap) - if "note" in out: - out["_summary"] = _clip(str(out["note"]), 80) - else: - out["_summary"] = (f"{_plural(len(out['aws_calls']), 'AWS call')}" - + (f" ({', '.join(out['aws_services'][:4])})" if out["aws_services"] else "") - + f", {_plural(len(out['env_vars_read']), 'environment variable')}") - if cmap.function is not None: - out["_next"] = [(self._next_for(cmap.function, "how_it_works"), "how the code works, step by step")] - return out - - return [find_functions, function_health, function_errors, function_logs, function_performance, - function_details, function_code] - - # ------------------------------------------------------------------ what the tools share + def find_functions(self, match: str = "", regions: str = "") -> dict[str, Any]: + """ + Lists the Lambda functions in the account: name, region, runtime (and whether AWS still supports + it), memory, timeout, when each was last changed and its description. match: a name glob like 'etl-*'. + regions: 'us-east-1,eu-west-1' or 'all'; empty for the default regions. + """ + core = self.analyzer + ov = core.overview(match or None, regions=self._regions(regions), metrics=False, details=False) + out = summarize_functions(ov.functions, ov.regions, ov.errors) + out["_summary"] = f"{_plural(len(ov.functions), 'function')} in {', '.join(ov.regions)}" + out["_next"] = [(self._view_call("functions", match or None, regions=regions or None), + "every function with its calls, errors, cost and warnings")] + return out + + def function_health(self, since: str = "24h", until: str = "", match: str = "", + regions: str = "") -> dict[str, Any]: + """ + How every Lambda function did in a time window, from CloudWatch: calls, failed calls, throttles and + the longest run against its timeout, sorted into failed, throttled, near_timeout and not_called. Fast and + cheap: start here for "what failed", "what's slow" or "what's unused". since/until: ISO UTC times + ('2026-10-15T18:00Z') or relative ('24h', '7d'); until defaults to now. match: a name glob like 'etl-*'. + regions: 'us-east-1,eu-west-1' or 'all'; empty for the default regions. + """ + core = self.analyzer + start, end = core._window(since or "24h", until or None) + ov = core.overview(match or None, regions=self._regions(regions), metrics=False, details=False) + found, read, _, errors = core._numbers(ov.functions, since=start, until=end) + self._count_metrics(read, core.prices["metric_request"]) + out = summarize_health(ov.functions, found, start, end, ov.regions, {**ov.errors, **errors}) + failed = out["failed"] + out["_summary"] = (f"{len(failed) + out.get('failed_more', 0)} of {_plural(len(ov.functions), 'function')} " + "had failed calls" + (": " + ", ".join(f"{r['name']} ({r['errors']:,})" for r in failed[:3]) + if failed else "")) + out["_next"] = [(self._next_for(fn, "errors", since=_since(start)), "its errors, grouped by cause") + for fn in ov.functions if fn.name in {r["name"] for r in failed[:2]}] + return out + + def function_errors(self, function: str, since: str = "24h", until: str = "", + region: str = "") -> dict[str, Any]: + """ + One function's errors in a time window, read from its logs and grouped by cause (Timeout, KeyError, + AccessDenied, ...), with CloudWatch's count of failed and throttled calls over the same window. Each + cause has a count, when it first and last happened, an example line and request IDs to look up with + function_logs. since/until: ISO UTC times or relative ('24h'); the window can't be longer than the log + limit. region: empty for the default region. + """ + core = self.analyzer + start, end = self._log_window(function, since, until, region or None) + report = core.errors(function, since=start, until=end, region=region or None) + out = summarize_errors(report) + causes = out["causes"] + kinds = Counter(c["kind"] for c in causes[:3]) + out["_summary"] = (f"{_plural(sum(c['count'] for c in causes), 'error')}, {_plural(len(causes), 'cause')}" + + (": " + ", ".join(_cause_label(c, kinds) for c in causes[:3]) if causes else "")) + out["_next"] = [(self._next_for(report.function, "errors", since=_since(start)), + "the same errors as a report, with each cause's findings")] + if causes and causes[0]["request_ids"]: + out["_next"].insert(0, (self._next_for(report.function, "logs", request_id=causes[0]["request_ids"][0]), + f"one {causes[0]['kind']} run, start to end")) + return out + + def function_logs(self, function: str, since: str = "1h", until: str = "", search: str = "", + region: str = "") -> dict[str, Any]: + """ + One function's runs in a time window, newest first, from its logs: each run's status (ok, failed, + timeout, logged an error), run time, memory, whether it was a cold start, and its last lines. search: + text the runs' lines must contain, such as a request ID, an order ID or 'KeyError' (case-sensitive). + since/until: ISO UTC times or relative ('1h'); the window can't be longer than the log limit. region: + empty for the default region. + """ + core = self.analyzer + start, end = self._log_window(function, since, until, region or None) + page = core.log_runs(function, since=start, until=end, search=search or None, region=region or None) + out = summarize_runs(page) + statuses = out["statuses"] + out["_summary"] = (f"{_plural(out['runs_found'], 'run')}" + + (f" ({', '.join(f'{n} {s}' for s, n in statuses.items())})" if statuses else "")) + out["_next"] = [(self._next_for(page.function, "logs", since=_since(start), search=search or None), + "the same lines, as a report")] + return out + + def function_performance(self, function: str, since: str = "24h", until: str = "", + region: str = "") -> dict[str, Any]: + """ + One function's run times and memory in a time window, from the REPORT line Lambda logs after every + run: median, p95, p99 and longest run against its timeout, timeouts, cold starts, the most memory a run + used and the memory size it needs. since/until: ISO UTC times or relative ('24h'); the window can't be + longer than the log limit. region: empty for the default region. + """ + core = self.analyzer + start, end = self._log_window(function, since, until, region or None) + perf = core.performance(function, since=start, until=end, region=region or None) + out = summarize_performance(perf, core.prices) + p95 = out["p95_ms"] + out["_summary"] = (f"{_plural(out['runs'], 'run')}" + + (f", p95 {p95 / 1000:.1f}s of a {perf.function.timeout}s timeout" if p95 else "")) + out["_next"] = [(self._next_for(perf.function, "performance", since=_since(start)), + "run times, memory and cold starts as a report")] + return out + + def function_details(self, function: str, region: str = "") -> dict[str, Any]: + """ + Everything about one function: its settings (runtime and its support dates, memory, timeout, + reserved concurrency), what triggers it, its aliases and versions, its last 30 days from CloudWatch, its + log group's retention and size, the estimated monthly cost by part, and every finding about it with the + next step. region: empty for the default region. + """ + core = self.analyzer + detail = core.describe(function, region=region or None, days=30) + out = summarize_detail(detail, core.prices) + total = out["monthly_cost_usd"].get("total") + out["_summary"] = (f"{detail.function.runtime or 'container image'}, {detail.function.memory:,} MB, " + f"{detail.function.timeout}s; {_plural(len(out['findings']), 'finding')}" + + (f"; about {human_money(total)}/month" if total is not None else "")) + out["_next"] = [(self._next_for(detail.function, "function_info"), "the function in plain English")] + return out + + def function_code(self, function: str, region: str = "") -> dict[str, Any]: + """ + What a Python function's code does, read from its deployment package with Python's ast module + (nothing is run): the AWS calls it makes, the environment variables it reads (names only), the event + fields it uses, the libraries it imports and findings such as HTTP calls without a timeout. Other + runtimes and container images give a note. region: empty for the default region. + """ + core = self.analyzer + cmap = core.code_map(function, region=region or None) + out = summarize_code(cmap) + if "note" in out: + out["_summary"] = _clip(str(out["note"]), 80) + else: + out["_summary"] = (f"{_plural(len(out['aws_calls']), 'AWS call')}" + + (f" ({', '.join(out['aws_services'][:4])})" if out["aws_services"] else "") + + f", {_plural(len(out['env_vars_read']), 'environment variable')}") + if cmap.function is not None: + out["_next"] = [(self._next_for(cmap.function, "how_it_works"), "how the code works, step by step")] + return out def not_found(self, exc: ClientError, arguments: dict[str, Any]) -> dict[str, Any] | None: """ @@ -562,6 +560,18 @@ def _next_for(self, fn: Function, command: str, *args: Any, **kwargs: Any) -> st return self._view_call(command, fn.name, *args, region=fn.region, **kwargs) +def _cause_label(cause: dict[str, Any], kinds: Counter) -> str: + """ + A cause of errors as the report's one line says it: its kind and count ('KeyError ×12'), with what tells it + apart when another of the causes shown has the same kind ('KeyError (order_id) ×12'). + """ + kind = cause["kind"] + rest = re.sub(rf"^\W*{re.escape(kind)}\W*", "", cause["message"] or "").strip() + if kinds[kind] > 1 and rest: + return f"{kind} ({_clip(rest, 40)}) ×{cause['count']}" + return f"{kind} ×{cause['count']}" + + def _since(start: datetime) -> str: """ A window's start as a View's since= takes it. diff --git a/src/aws_analyzer/agent/toolkits/s3_tools.py b/src/aws_analyzer/agent/toolkits/s3_tools.py index 5586006..310e76f 100644 --- a/src/aws_analyzer/agent/toolkits/s3_tools.py +++ b/src/aws_analyzer/agent/toolkits/s3_tools.py @@ -8,6 +8,14 @@ The summarize_* functions, count_arrivals() and trigger_findings() are pure: they run on any BucketConfig, BucketMetrics, Listing, PrefixSummary, list of ObjectInfo or TriggerLink. + +S3_TOOL_READS is what each tool reads (its IAM permissions), for tools() and README; s3_triggers reads Lambda through +the Lambda toolkit. The analyzers' types are imported for the type hints only, since S3's analyzer loads when the +first tool runs (_s3_module). S3_REQUEST_PRICES are USD per 1,000 requests at S3's us-east-1 list prices (2026-10): +LIST requests, and GET, HEAD and the rest; _METRIC_PRICE is USD per 1,000 metrics read with CloudWatch's +GetMetricData. find_buckets(sizes=True) reads CloudWatch for at most _SIZED_BUCKETS buckets, _SIZE_WORKERS at a time, +and s3_triggers looks through at most _SEARCHED_BUCKETS buckets for notifications that invoke a function. +_CONFIG_PERMISSIONS maps a BucketConfig.errors section (and this toolkit's own parts) to the permission it needs. """ from __future__ import annotations @@ -29,13 +37,12 @@ from ..results import _iso, _window, _with_permissions, hide_secrets from .base import Toolkit -if TYPE_CHECKING: # for the type hints only: the S3 analyzer loads when the first tool runs - from ...lambda_functions import Function, FunctionMetrics +if TYPE_CHECKING: + from ...lambda_functions import Function, FunctionMetrics, LambdaAnalyzer from ...s3 import ( BucketConfig, BucketInfo, BucketMetrics, EventNotification, Listing, ObjectInfo, PrefixSummary, S3Analyzer, ) -# What each tool reads (IAM permissions), for tools() and README. s3_triggers reads Lambda through the Lambda toolkit. S3_TOOL_READS: dict[str, tuple[str, ...]] = { "find_buckets": ("s3:ListAllMyBuckets", "s3:GetBucketLocation", "cloudwatch:ListMetrics", "cloudwatch:GetMetricData"), @@ -52,12 +59,14 @@ "s3_triggers": ("s3:GetBucketNotification", "s3:ListBucket", "s3:ListAllMyBuckets", "s3:GetBucketLocation", "lambda:GetFunction", "lambda:GetPolicy", "cloudwatch:GetMetricData"), } -# USD per 1,000 requests (S3's us-east-1 list prices, 2026-10): LIST requests, and GET, HEAD and the rest. S3_REQUEST_PRICES = {"list": 0.005, "get": 0.0004} -_METRIC_PRICE = 0.01 # USD per 1,000 metrics read with CloudWatch GetMetricData -_SIZED_BUCKETS = 200 # find_buckets(sizes=True) reads CloudWatch for at most this many buckets -_SEARCHED_BUCKETS = 50 # s3_triggers looks through at most this many buckets for notifications that call a function -_CONFIG_PERMISSIONS = { # BucketConfig.errors section (and this toolkit's own parts) -> the permission it needs +_REQUESTS_PER_PRICE = 1000 +_METRIC_PRICE = 0.01 +_SIZED_BUCKETS = 200 +_SIZE_WORKERS = 8 +_SEARCHED_BUCKETS = 50 +_MISSED_SHOWN = 3 +_CONFIG_PERMISSIONS = { "versioning": "s3:GetBucketVersioning", "encryption": "s3:GetEncryptionConfiguration", "public_access_block": "s3:GetBucketPublicAccessBlock", "policy": "s3:GetBucketPolicyStatus", "policy_document": "s3:GetBucketPolicy", "ownership": "s3:GetBucketOwnershipControls", @@ -101,11 +110,6 @@ def _between(since: str, until: str, now: datetime | None = None) -> tuple[datet return start, end or now -# ============================================================================= -# What the tools return: S3Analyzer's results as compact JSON (pure) -# ============================================================================= - - def _shown(values: dict[str, Any]) -> dict[str, str]: """ Tags or metadata as the model reads them: each value clipped, and hidden when it looks like a secret, by @@ -160,13 +164,15 @@ def summarize_buckets(buckets: list[BucketInfo], metrics: dict[str, BucketMetric return out -def summarize_bucket(cfg: BucketConfig, metrics: BucketMetrics | None, notifications: list[EventNotification] | None, - *, account_block: dict[str, bool] | None = None, prices: dict[str, float] | None = None, +def summarize_bucket(cfg: BucketConfig, metrics: BucketMetrics | None, # noqa: PLR0913 + notifications: list[EventNotification] | None, *, + account_block: dict[str, bool] | None = None, prices: dict[str, float] | None = None, errors: dict[str, str] | None = None) -> dict[str, Any]: """ bucket_details()'s answer: a bucket's settings in plain words (public access, encryption, versioning, lifecycle rules, replication, logging, tags), its size and estimated monthly storage cost from CloudWatch, the - events it sends where (Lambda functions, queues, topics, EventBridge), and the findings about it. + events it sends where (Lambda functions, queues, topics, EventBridge), and the findings about it. A finding + that a part couldn't be read is left out: that part is in `errors`. """ s3 = _s3_module() errors = {**cfg.errors, **(errors or {})} @@ -201,7 +207,7 @@ def summarize_bucket(cfg: BucketConfig, metrics: BucketMetrics | None, notificat if cfg.policy: found += s3.policy_findings(s3.explain_policy(cfg.policy)) out["findings"] = [{"level": level, "message": text} for level, text in found - if not text.startswith("Couldn't read")] # what couldn't be read is in errors + if not text.startswith("Couldn't read")] if errors: out["errors"] = _with_permissions(errors, _CONFIG_PERMISSIONS) return out @@ -227,7 +233,7 @@ def summarize_listing(listing: Listing, *, folders: int = 100, files: int = 50) return out -def summarize_found(location: str, found: list[ObjectInfo], *, scanned: int, cut_short: bool, +def summarize_found(location: str, found: list[ObjectInfo], *, scanned: int, cut_short: bool, # noqa: PLR0913 start: datetime | None = None, end: datetime | None = None, filters: dict[str, Any] | None = None, limit: int = 30) -> dict[str, Any]: """ @@ -309,40 +315,54 @@ def summarize_file(info: dict[str, Any]) -> dict[str, Any]: return out -# ---- S3 and Lambda: the event notifications that invoke functions, checked from both ends - - @dataclass class Arrivals: """ What a listing of a notification's prefix says about the files written in a window (count_arrivals). + + scanned keys read + matched files written in the window that its prefix and suffix cover + sent of those, the ones uploaded in a way its events fire on: they invoked the function + multipart_missed matched, but uploaded in parts, and it doesn't fire on CompleteMultipartUpload + whole_missed matched, but uploaded whole, and it fires only on multipart uploads + missed_keys a few of the files its events missed + case_missed files in the window whose ending matches only ignoring case + newest the newest files that invoked it (up to 5) + last_match the newest file its filter covers, in the window or before + cut_short the listing stopped at its limit """ - scanned: int = 0 # keys read - matched: int = 0 # files written in the window that its prefix and suffix cover - sent: int = 0 # of those, the ones uploaded in a way its events fire on: they invoked the function - multipart_missed: int = 0 # matched, but uploaded in parts, and it doesn't fire on CompleteMultipartUpload - whole_missed: int = 0 # matched, but uploaded whole, and it fires only on multipart uploads - missed_keys: list[str] = field(default_factory=list) # a few of the files its events missed - case_missed: list[str] = field(default_factory=list) # files in the window whose ending matches only ignoring case - newest: list[ObjectInfo] = field(default_factory=list) # the newest files that invoked it (up to 5) - last_match: ObjectInfo | None = None # the newest file its filter covers, in the window or before - cut_short: bool = False # the listing stopped at its limit + scanned: int = 0 + matched: int = 0 + sent: int = 0 + multipart_missed: int = 0 + whole_missed: int = 0 + missed_keys: list[str] = field(default_factory=list) + case_missed: list[str] = field(default_factory=list) + newest: list[ObjectInfo] = field(default_factory=list) + last_match: ObjectInfo | None = None + cut_short: bool = False @dataclass class TriggerLink: """ One S3 event notification that invokes a Lambda function, checked from both ends (s3_triggers). + + exists the function is there; None when it couldn't be checked + allowed its resource policy lets this bucket invoke it; None when it couldn't be read + arrivals the files written in the window; None when it doesn't fire on new files, or the listing failed + metrics CloudWatch's counts for the function over the window + account the function's account (for the add-permission command) """ bucket: str notification: EventNotification - exists: bool | None = None # the function is there; None when it couldn't be checked - allowed: bool | None = None # its resource policy lets this bucket invoke it; None when it couldn't be read - arrivals: Arrivals | None = None # None when it doesn't fire on new files, or the listing failed - metrics: FunctionMetrics | None = None # CloudWatch's counts for the function over the window - account: str | None = None # the function's account (for the add-permission command) + exists: bool | None = None + allowed: bool | None = None + arrivals: Arrivals | None = None + metrics: FunctionMetrics | None = None + account: str | None = None @property def function(self) -> str: @@ -353,8 +373,8 @@ def region(self) -> str | None: return self.notification.region -def count_arrivals(objects: Iterable[ObjectInfo], notification: EventNotification, start: datetime, end: datetime, - *, limit: int | None = None, show: int = 5) -> Arrivals: +def count_arrivals(objects: Iterable[ObjectInfo], notification: EventNotification, # noqa: PLR0913 + start: datetime, end: datetime, *, limit: int | None = None, show: int = 5) -> Arrivals: """ What a listing of the notification's prefix says about the files written from `start` to `end`: how many its prefix and suffix cover, and how many of those it fired on, by how each was uploaded (an ETag with a part count @@ -374,27 +394,37 @@ def count_arrivals(objects: Iterable[ObjectInfo], notification: EventNotificatio continue written = start <= obj.last_modified < end if obj.key.endswith(suffix): - if out.last_match is None or obj.last_modified > out.last_match.last_modified: - out.last_match = obj - if not written: - continue - out.matched += 1 - if notification.on_created(obj.is_multipart): - out.sent += 1 - sent.append(obj) - continue - if obj.is_multipart: - out.multipart_missed += 1 - else: - out.whole_missed += 1 - if len(out.missed_keys) < 3: - out.missed_keys.append(obj.key) + _count_covered(out, obj, notification, written, sent) elif written and suffix and obj.key.lower().endswith(suffix.lower()): out.case_missed.append(obj.key) out.newest = heapq.nlargest(show, sent, key=lambda o: o.last_modified) return out +def _count_covered(out: Arrivals, obj: ObjectInfo, notification: EventNotification, written: bool, + sent: list[ObjectInfo]) -> None: + """ + Counts one file the notification's prefix and suffix cover: it may be the newest such file, and when it was + written in the window, whether the way it was uploaded fired the notification (it goes in `sent`) or not (a few + of those keys are kept, up to _MISSED_SHOWN). + """ + if out.last_match is None or obj.last_modified > out.last_match.last_modified: + out.last_match = obj + if not written: + return + out.matched += 1 + if notification.on_created(obj.is_multipart): + out.sent += 1 + sent.append(obj) + return + if obj.is_multipart: + out.multipart_missed += 1 + else: + out.whole_missed += 1 + if len(out.missed_keys) < _MISSED_SHOWN: + out.missed_keys.append(obj.key) + + def _where(n: EventNotification) -> str: """ "under 'incoming/' ending in '.csv'", or "anywhere in the bucket". @@ -416,65 +446,20 @@ def trigger_findings(link: TriggerLink) -> list[tuple[str, str]]: """ n, fn, bucket = link.notification, link.function, link.bucket where = _where(n) - found: list[tuple[str, str]] = [] if link.exists is False: - found.append(("warn", f"Bucket {bucket} sends {_events(n)} for files {where} to Lambda function {fn}, which " - f"isn't there in {link.region}: those events invoke nothing. Point the notification at " - "the function that should run, or remove it (aws s3api " - f"get-bucket-notification-configuration --bucket {bucket} shows it).")) - return found + return [("warn", f"Bucket {bucket} sends {_events(n)} for files {where} to Lambda function {fn}, which " + f"isn't there in {link.region}: those events invoke nothing. Point the notification at " + "the function that should run, or remove it (aws s3api " + f"get-bucket-notification-configuration --bucket {bucket} shows it).")] + found: list[tuple[str, str]] = [] if link.allowed is False: - account = f" --source-account {link.account}" if link.account else "" - found.append(("warn", f"{fn}'s resource policy doesn't let bucket {bucket} invoke it: S3 can't deliver the " - f"events for files {where}, so none of them run it. Allow it: aws lambda add-permission " - f"--function-name {fn} --statement-id s3-{bucket} --action lambda:InvokeFunction " - f"--principal s3.amazonaws.com --source-arn arn:aws:s3:::{bucket}{account}" - + (f" --region {link.region}" if link.region else ""))) - a, m = link.arrivals, link.metrics - if a is None and not n.on_created(): + found.append(_not_allowed(link, where)) + if link.arrivals is None and not n.on_created(): found.append(("info", f"Bucket {bucket} invokes {fn} on {_events(n)}, not on new files, so the listing can't " f"show what it sent: CloudWatch's count of {fn}'s runs is the record.")) - if a is not None: - if a.multipart_missed or a.whole_missed: - missed = a.multipart_missed + a.whole_missed - was, they = ("was", "it") if missed == 1 else ("were", "they") - how = ("uploaded in parts (multipart), which sends s3:ObjectCreated:CompleteMultipartUpload" if - a.multipart_missed else "uploaded whole (a PUT), which sends s3:ObjectCreated:Put") - found.append(("warn", f"{_plural(missed, 'file')} {where} written in the window {was} {how}, but the " - f"notification fires only on {_events(n)}: {they} never invoked {fn} (e.g. " - f"{a.missed_keys[0]}). Make it fire on s3:ObjectCreated:* instead.")) - if a.case_missed: - endings = sorted({key[-len(n.suffix):] for key in a.case_missed}) - end, they = ("ends", "it") if len(a.case_missed) == 1 else ("end", "they") - found.append(("warn", f"{_plural(len(a.case_missed), 'file')} under '{n.prefix}' written in the window " - f"{end} in {', '.join(repr(e) for e in endings)}, which the suffix filter " - f"'{n.suffix}' doesn't match (S3 compares case-sensitively): {they} never invoked " - f"{fn} (e.g. {a.case_missed[0]}). Have the uploader write '{n.suffix}', or add a " - "notification for the other ending.")) - if not a.matched and not a.case_missed: - last = (f" The newest file it covers is {a.last_match.key}, written {_iso(a.last_match.last_modified)}." - if a.last_match else " No file there matches its filter at all.") - found.append(("info", f"No file {where} in bucket {bucket} was written in the window" - + (" (in the keys read)" if a.cut_short else "") - + f", so S3 had nothing to send {fn}: if it didn't run, that's why.{last}")) - if m is not None and a.sent: - calls, throttles = int(m.invocations), int(m.throttles) - throttled = f" ({_plural(throttles, 'call')} throttled)" if throttles else "" - if not calls and link.allowed is not False: - found.append(("warn", f"{_plural(a.sent, 'file')} {where} arrived in the window and should have " - f"invoked {fn}, but CloudWatch counted no runs of it{throttled}: S3's events " - f"didn't reach it. function_details('{fn}') shows whether S3 may invoke it " - "and its reserved concurrency (0 stops every run).")) - elif 0 < calls < a.sent: - found.append(("info", f"{_plural(a.sent, 'file')} arrived that should have invoked {fn}, but it ran " - f"{_plural(calls, 'time')} in the window{throttled}: some events may not have " - "reached it yet, since Lambda queues S3's events and retries them for up to 6 " - "hours.")) - if a.cut_short: - found.append(("info", f"The listing of bucket {bucket} {where} stopped at {a.scanned:,} keys (the budget's " - "s3_keys), in key order: the file counts cover those keys only. A notification " - "with a narrower prefix, or AgentView(budget={'s3_keys': " - f"{a.scanned * 5}}}), reads more.")) + if link.arrivals is not None: + found += _arrival_findings(link, where) + m = link.metrics if m is not None and m.errors: found.append(("warn", f"{fn} failed {_plural(int(m.errors), 'time')} in {_plural(int(m.invocations), 'run')} " f"in the window: function_errors('{fn}') shows why. S3 invokes it asynchronously, so a " @@ -483,7 +468,107 @@ def trigger_findings(link: TriggerLink) -> list[tuple[str, str]]: return found -def summarize_triggers(links: list[TriggerLink], start: datetime, end: datetime, *, bucket: str = "", +def _not_allowed(link: TriggerLink, where: str) -> tuple[str, str]: + """ + The finding for a function whose resource policy doesn't let the bucket invoke it, with the add-permission + command that does. + """ + fn, bucket = link.function, link.bucket + account = f" --source-account {link.account}" if link.account else "" + return ("warn", f"{fn}'s resource policy doesn't let bucket {bucket} invoke it: S3 can't deliver the " + f"events for files {where}, so none of them run it. Allow it: aws lambda add-permission " + f"--function-name {fn} --statement-id s3-{bucket} --action lambda:InvokeFunction " + f"--principal s3.amazonaws.com --source-arn arn:aws:s3:::{bucket}{account}" + + (f" --region {link.region}" if link.region else "")) + + +def _arrival_findings(link: TriggerLink, where: str) -> list[tuple[str, str]]: + """ + What the files written in the window say about a link: files its event types or the case of their ending + missed, no file arriving at all, files that arrived without the function running, and a listing cut short. + """ + a = link.arrivals + found: list[tuple[str, str]] = [] + if a.multipart_missed or a.whole_missed: + found.append(_missed_by_upload_kind(link, where)) + if a.case_missed: + found.append(_missed_by_case(link)) + if not a.matched and not a.case_missed: + found.append(_nothing_arrived(link, where)) + if link.metrics is not None and a.sent: + found += _runs_against_arrivals(link, where) + if a.cut_short: + found.append(("info", f"The listing of bucket {link.bucket} {where} stopped at {a.scanned:,} keys (the " + "budget's s3_keys), in key order: the file counts cover those keys only. A notification " + "with a narrower prefix, or AgentView(budget={'s3_keys': " + f"{a.scanned * 5}}}), reads more.")) + return found + + +def _missed_by_upload_kind(link: TriggerLink, where: str) -> tuple[str, str]: + """ + The finding for files the notification's event types missed: uploaded in parts when it fires only on whole + uploads, or the other way round. + """ + a, n = link.arrivals, link.notification + missed = a.multipart_missed + a.whole_missed + was, they = ("was", "it") if missed == 1 else ("were", "they") + how = ("uploaded in parts (multipart), which sends s3:ObjectCreated:CompleteMultipartUpload" if + a.multipart_missed else "uploaded whole (a PUT), which sends s3:ObjectCreated:Put") + return ("warn", f"{_plural(missed, 'file')} {where} written in the window {was} {how}, but the " + f"notification fires only on {_events(n)}: {they} never invoked {link.function} (e.g. " + f"{a.missed_keys[0]}). Make it fire on s3:ObjectCreated:* instead.") + + +def _missed_by_case(link: TriggerLink) -> tuple[str, str]: + """ + The finding for files whose ending matches the suffix filter only ignoring case: S3 compares case-sensitively. + """ + a, n = link.arrivals, link.notification + endings = sorted({key[-len(n.suffix):] for key in a.case_missed}) + end, they = ("ends", "it") if len(a.case_missed) == 1 else ("end", "they") + return ("warn", f"{_plural(len(a.case_missed), 'file')} under '{n.prefix}' written in the window " + f"{end} in {', '.join(repr(e) for e in endings)}, which the suffix filter " + f"'{n.suffix}' doesn't match (S3 compares case-sensitively): {they} never invoked " + f"{link.function} (e.g. {a.case_missed[0]}). Have the uploader write '{n.suffix}', or add a " + "notification for the other ending.") + + +def _nothing_arrived(link: TriggerLink, where: str) -> tuple[str, str]: + """ + The finding for a window in which no file the notification covers was written, with the newest one it covers. + """ + a = link.arrivals + last = (f" The newest file it covers is {a.last_match.key}, written {_iso(a.last_match.last_modified)}." + if a.last_match else " No file there matches its filter at all.") + return ("info", f"No file {where} in bucket {link.bucket} was written in the window" + + (" (in the keys read)" if a.cut_short else "") + + f", so S3 had nothing to send {link.function}: if it didn't run, that's why.{last}") + + +def _runs_against_arrivals(link: TriggerLink, where: str) -> list[tuple[str, str]]: + """ + Files that should have invoked the function against CloudWatch's count of its runs: none at all (S3's events + didn't reach it), or fewer than the files (Lambda queues S3's events and retries them for up to 6 hours). + """ + a, m, fn = link.arrivals, link.metrics, link.function + calls, throttles = int(m.invocations), int(m.throttles) + throttled = f" ({_plural(throttles, 'call')} throttled)" if throttles else "" + if not calls and link.allowed is not False: + return [("warn", f"{_plural(a.sent, 'file')} {where} arrived in the window and should have " + f"invoked {fn}, but CloudWatch counted no runs of it{throttled}: S3's events " + f"didn't reach it. function_details('{fn}') shows whether S3 may invoke it " + "and its reserved concurrency (0 stops every run).")] + if 0 < calls < a.sent: + return [("info", f"{_plural(a.sent, 'file')} arrived that should have invoked {fn}, but it ran " + f"{_plural(calls, 'time')} in the window{throttled}: some events may not have " + "reached it yet, since Lambda queues S3's events and retries them for up to 6 " + "hours.")] + return [] + + +def summarize_triggers(links: list[TriggerLink], start: datetime, end: datetime, *, # noqa: PLR0913 + bucket: str = "", function: str = "", others: list[EventNotification] | None = None, notes: list[tuple[str, str]] | None = None, errors: dict[str, str] | None = None, searched: list[str] | None = None) -> dict[str, Any]: @@ -565,9 +650,113 @@ def _s3_may_invoke(triggers: list[Any], bucket: str) -> bool: return False -# ============================================================================= -# S3Toolkit -# ============================================================================= +@dataclass +class _LinkChecker: + """ + Checks S3 -> Lambda links from both ends for s3_triggers: whether each function is there and lets its bucket + invoke it, the files written in the window, and the functions' CloudWatch numbers. Each function, resource + policy and listing is read once, however many links share it; what couldn't be read goes in `errors`. + + functions a notification's function ARN -> the function, None when it isn't there + unknown function ARNs that couldn't be looked up + """ + + s3_core: S3Analyzer + lam: LambdaAnalyzer + start: datetime + end: datetime + cap: int + errors: dict[str, str] + functions: dict[str, Function | None] = field(default_factory=dict) + unknown: set[str] = field(default_factory=set) + policies: dict[tuple[str, str, str | None], list[Any] | None] = field(default_factory=dict) + listings: dict[tuple[str, str], list[ObjectInfo] | None] = field(default_factory=dict) + + def check(self, bucket: str, n: EventNotification) -> TriggerLink: + arn = n.arn or "" + link = TriggerLink(bucket, n, account=arn.split(":")[4] if arn.count(":") >= 5 else None) + target = self._function(arn, n.name) + link.exists = None if arn in self.unknown else target is not None + if target is not None: + link.allowed = self._allowed(target, arn, bucket) + if link.exists is not False and n.on_created(): + link.arrivals = self._arrivals(bucket, n) + return link + + def add_metrics(self, links: list[TriggerLink]) -> int: + """ + Gives each link its function's CloudWatch numbers over the window. Returns how many metrics were read. + """ + live = list({f.arn: f for f in self.functions.values() if f is not None}.values()) + if not live: + return 0 + try: + numbers, read = self.lam.metrics(live, since=self.start, until=self.end, logs=False) + except (ClientError, BotoCoreError) as exc: + self.errors["metrics"] = _code(exc) + return 0 + for link in links: + target = self.functions.get(link.notification.arn or "") + link.metrics = numbers.get(target.arn) if target is not None else None + return read + + def _function(self, arn: str, name: str) -> Function | None: + """ + The function a notification invokes, or None when it isn't there or couldn't be looked up (`unknown`). + """ + if arn not in self.functions: + try: + self.functions[arn] = self.lam.function(arn) + except (ClientError, BotoCoreError) as exc: + self.functions[arn] = None + if _code(exc) != "ResourceNotFoundException": + self.errors[f"{name}:function"] = _code(exc) + self.unknown.add(arn) + return self.functions[arn] + + def _allowed(self, target: Function, arn: str, bucket: str) -> bool | None: + """ + Whether the function's resource policy (for the version or alias the ARN names) lets the bucket invoke it; + None when the policy couldn't be read. + """ + lf = _lambda_module() + qualifier = lf.parse_function_ref(arn)[1] + key = (target.name, target.region, qualifier) + if key not in self.policies: + try: + self.policies[key] = lf.parse_policy(self.lam.policy(target.name, target.region, qualifier=qualifier)) + except (ClientError, BotoCoreError) as exc: + self.errors[f"{target.name}:policy_of_function"] = _code(exc) + self.policies[key] = None + policy = self.policies[key] + return _s3_may_invoke(policy, bucket) if policy is not None else None + + def _arrivals(self, bucket: str, n: EventNotification) -> Arrivals | None: + """ + What a listing of the notification's prefix says about the files written in the window; None when it + couldn't be listed. It reads up to `cap` keys, and one more to tell when there were more. + """ + s3 = _s3_module() + listed = (bucket, n.prefix) + if listed not in self.listings: + try: + self.listings[listed] = list(self.s3_core.iter_objects(s3.s3_uri(bucket, n.prefix), + limit=self.cap + 1)) + except (ClientError, BotoCoreError) as exc: + self.errors[f"{bucket}:list"] = _code(exc) + self.listings[listed] = None + objects = self.listings[listed] + return count_arrivals(objects, n, self.start, self.end, limit=self.cap) if objects is not None else None + + +def _triggers_summary(links: list[TriggerLink]) -> str: + """ + s3_triggers' one line in the report: its links, and how many have warnings. + """ + warned = [link for link in links if any(level == "warn" for level, _ in trigger_findings(link))] + pairs_text = sorted({f"{link.bucket} → {link.function}" for link in links}) + return (_plural(len(links), "S3 → Lambda link") + (f": {_clip(', '.join(pairs_text), 120)}" if links else "") + + (f"; {_plural(len(warned), 'link')} with warnings" if warned else "")) class S3Toolkit(Toolkit): @@ -641,214 +830,164 @@ def request_cost(self, counts: dict[str, int]) -> float: """ lists = sum(n for op, n in counts.items() if op.startswith("s3:List")) others = sum(n for op, n in counts.items() if op.startswith("s3:") and not op.startswith("s3:List")) - return (lists * S3_REQUEST_PRICES["list"] + others * S3_REQUEST_PRICES["get"]) / 1000 - - # ------------------------------------------------------------------ the tools + return (lists * S3_REQUEST_PRICES["list"] + others * S3_REQUEST_PRICES["get"]) / _REQUESTS_PER_PRICE def tools(self) -> list[Callable[..., dict[str, Any]]]: """ find_buckets, bucket_details, list_folder, find_files, folder_summary and file_info, and s3_triggers when - the agent has the Lambda toolkit too: plain functions (their type hints and docstrings tell the model how to - call them), each reading through S3Analyzer and returning a compact, JSON-ready summary. + the agent has the Lambda toolkit too: methods whose type hints and docstrings tell the model how to call + them, each reading through S3Analyzer and returning a compact, JSON-ready summary. """ - - def find_buckets(match: str = "", sizes: bool = False) -> dict[str, Any]: - """ - Lists the S3 buckets in the account: name, region and when each was made. match: a name glob like - 'acme-*'. sizes=True also reads each bucket's object count and size from CloudWatch's daily storage - metrics and sorts the biggest first (two reads per bucket: slower when there are many). - """ - core = self.analyzer - buckets = [b for b in core.list_buckets() if not match or fnmatch.fnmatchcase(b.name, match)] - metrics: dict[str, BucketMetrics] | None = None - errors: dict[str, str] = {} - if sizes: - metrics, errors = self._sizes(buckets[:_SIZED_BUCKETS]) - out = summarize_buckets(buckets, metrics, errors) - if sizes and len(buckets) > _SIZED_BUCKETS: - out["sizes_note"] += f" Sizes were read for the first {_SIZED_BUCKETS} only: narrow with match=." - out["_summary"] = f"{_plural(len(buckets), 'bucket')}" + ( - f" in {', '.join(out['regions'][:4])}" if out["regions"] else "") - out["_next"] = [(self._view_call("overview", match or None), - "every bucket: size, cost, encryption, public access and warnings")] - return out - - def bucket_details(bucket: str) -> dict[str, Any]: - """ - Everything about one bucket's setup: public access, encryption, versioning, object lock, lifecycle - rules in plain words, replication, access logging, tags, its event notifications (which Lambda - functions, queues and topics it sends which events to, for which keys), its object count, size and - estimated monthly storage cost from CloudWatch, and the findings about it with what to do. - """ - s3 = _s3_module() - core = self.analyzer - name, _ = s3.parse_s3_uri(bucket) - core.bucket_region(name) # a bucket that isn't there fails here, with close names - cfg = core.bucket_config(name) - errors: dict[str, str] = {} - notifications = metrics = account = None - try: - notifications = core.notifications(name) - except (ClientError, BotoCoreError) as exc: - errors["notifications"] = _code(exc) - try: - metrics = core.bucket_metrics(name) - self._count_metrics(_metrics_read(metrics), _METRIC_PRICE) - except (ClientError, BotoCoreError) as exc: - errors["metrics"] = _code(exc) - try: - account = core.account_public_access_block() - except (ClientError, BotoCoreError, ValueError): - account = None # unknown: the findings say account-level settings may still apply - out = summarize_bucket(cfg, metrics, notifications, account_block=account, prices=core.prices, - errors=errors) - size = out.get("size", {}).get("total") - out["_summary"] = ", ".join(filter(None, [ - size and f"{size}", out["public_access"], f"encryption {out['encryption']}", - f"{_plural(len(notifications), 'event notification')}" if notifications is not None else "", - _plural(len(out["findings"]), "finding")])) - out["_next"] = [(self._view_call("bucket_info", name), "its settings, policy and size as a report")] - if cfg.policy: - out["_next"].append((self._view_call("policy", name), "its bucket policy in plain English")) - return out - - def list_folder(location: str) -> dict[str, Any]: - """ - What's directly in a folder (one level, like `aws s3 ls`): its subfolders, and its files newest first - with size, when each was written and storage class. Fast even on huge buckets. location: - 's3://bucket/prefix/' or 'bucket/prefix/'; a bucket name alone lists its top level. - """ - listing = self.analyzer.ls(location, limit=1000) - out = summarize_listing(listing) - out["_summary"] = (f"{_plural(len(listing.folders), 'folder')}, {_plural(len(listing.objects), 'file')}" - + (" (more not listed)" if listing.truncated else "")) - out["_next"] = [(self._view_call("ls", listing.uri), "the same folder as a report")] - return out - - def find_files(location: str, match: str = "", since: str = "", until: str = "", min_size: str = "", - max_size: str = "", limit: int = 30) -> dict[str, Any]: - """ - The files under a location (every level below it) that match a filter: how many, their total size, - when the oldest and newest were written, and the newest `limit` of them. match: a glob on the file name, - or on the whole key when it has a '/' ('*.csv', 'orders_2026-10-1*'). since/until: only files last - written in that window, as ISO UTC times ('2026-10-15T18:00Z') or relative ('24h', '7d'). min_size / - max_size: '10MB', '1GB'. Reads the keys one listing page at a time, up to the budget's limit. - """ - s3 = _s3_module() - core = self.analyzer - start, end = _between(since, until) - keep = s3.make_filter(pattern=match or None, min_size=min_size or None, max_size=max_size or None, - modified_after=start, modified_before=end if until else None) - cap = self.budget.s3_keys - scanned, found, more = 0, [], False - for obj in core.iter_objects(location, limit=cap + 1): # one past the cap says there's more - if scanned >= cap: - more = True - break - scanned += 1 - if keep(obj): - found.append(obj) - bucket, prefix = s3.parse_s3_uri(location) - out = summarize_found(s3.s3_uri(bucket, prefix), found, scanned=scanned, cut_short=more, start=start, - end=end, limit=max(1, min(int(limit or 30), 200)), - filters={"match": match, "min_size": min_size, "max_size": max_size}) - out["_summary"] = (f"{_plural(len(found), 'file')} ({out['total_size']})" - + (f" of {scanned:,} keys" if scanned != len(found) else "") - + (", cut short" if out["cut_short"] else "")) - out["_next"] = [(self._view_call("find", s3.s3_uri(bucket, prefix), pattern=match or None, - min_size=min_size or None, max_size=max_size or None, - modified_after=_iso(start), modified_before=_iso(end) if until else None), - "the same search as a report")] - return out - - def folder_summary(location: str) -> dict[str, Any]: - """ - What a location holds, every level below it: how many files and how much, by file type, storage - class and folder, the oldest and newest file, the largest ones, the estimated monthly storage cost and - findings (many small files, files in Glacier, files untouched for 90+ days...). Reads every key, up to - the budget's limit: use the narrowest location you can. - """ - core = self.analyzer - summary = core.summarize(location, top_n=5, folder_depth=1, limit=self.budget.s3_keys) - out = summarize_prefix(summary, core.prices) - out["_summary"] = (f"{_plural(summary.object_count, 'file')}, {out['total_size']}, about " - f"{human_money(summary.monthly_cost)}/month" + (", cut short" if summary.truncated - else "")) - out["_next"] = [(self._view_call("summary", summary.uri), "the same as a dashboard"), - (self._view_call("tree", summary.uri), "its folders' sizes as a tree")] - return out - - def file_info(location: str) -> dict[str, Any]: - """ - One file's headers, never its content: size, when it was written, content type, storage class (a - file in Glacier must be restored before anything can read it), encryption (a KMS key means readers need - kms:Decrypt on it), version, restore and lock status, its metadata and tags. location: - 's3://bucket/key'. - """ - core = self.analyzer - info = core.head(location) - out = summarize_file(info) - out["_summary"] = f"{out['size']}, {info.get('storage_class', 'STANDARD')}, written " + ( - _iso(info.get("last_modified")) or "?") - out["_next"] = [(self._view_call("head", info["uri"]), "every header as a report"), - (self._view_call("preview", info["uri"]), "what's in it, in your notebook only")] - return out - - tools = [find_buckets, bucket_details, list_folder, find_files, folder_summary, file_info] + tools = [self.find_buckets, self.bucket_details, self.list_folder, self.find_files, self.folder_summary, + self.file_info] if self.peer("lambda") is not None: - tools.append(self._triggers_tool()) + tools.append(self.s3_triggers) return tools - def _triggers_tool(self) -> Callable[..., dict[str, Any]]: - def s3_triggers(bucket: str = "", function: str = "", since: str = "24h", until: str = "", - region: str = "") -> dict[str, Any]: - """ - How S3 and Lambda are wired: the bucket event notifications that invoke Lambda functions. Give a - bucket (the functions its events invoke), a function (the buckets whose events invoke it, from its - resource policy and those buckets' notifications) or both. For each link: the events and the key - prefix and suffix it fires on, whether the function exists and lets that bucket invoke it, how many files - the filter covers were written in the window and how many of them invoked it (and which it missed, by - upload kind or by the case of their ending), and how often the function ran, failed and was throttled in - that window (CloudWatch). Use it for "which Lambda processes files in X", "why didn't Y run after the - upload" or "were last night's files processed". since/until: ISO UTC times or relative ('24h'). region: - the function's region when it isn't the default. - """ - return self._triggers(bucket, function, since, until, region) - - return s3_triggers - - # ------------------------------------------------------------------ what the tools share + def find_buckets(self, match: str = "", sizes: bool = False) -> dict[str, Any]: + """ + Lists the S3 buckets in the account: name, region and when each was made. match: a name glob like + 'acme-*'. sizes=True also reads each bucket's object count and size from CloudWatch's daily storage + metrics and sorts the biggest first (two reads per bucket: slower when there are many). + """ + core = self.analyzer + buckets = [b for b in core.list_buckets() if not match or fnmatch.fnmatchcase(b.name, match)] + metrics: dict[str, BucketMetrics] | None = None + errors: dict[str, str] = {} + if sizes: + metrics, errors = self._sizes(buckets[:_SIZED_BUCKETS]) + out = summarize_buckets(buckets, metrics, errors) + if sizes and len(buckets) > _SIZED_BUCKETS: + out["sizes_note"] += f" Sizes were read for the first {_SIZED_BUCKETS} only: narrow with match=." + out["_summary"] = f"{_plural(len(buckets), 'bucket')}" + ( + f" in {', '.join(out['regions'][:4])}" if out["regions"] else "") + out["_next"] = [(self._view_call("overview", match or None), + "every bucket: size, cost, encryption, public access and warnings")] + return out - def _sizes(self, buckets: list[BucketInfo]) -> tuple[dict[str, BucketMetrics], dict[str, str]]: + def bucket_details(self, bucket: str) -> dict[str, Any]: """ - Each bucket's CloudWatch size and object count, read side by side; errors by 'bucket:metrics'. + Everything about one bucket's setup: public access, encryption, versioning, object lock, lifecycle + rules in plain words, replication, access logging, tags, its event notifications (which Lambda + functions, queues and topics it sends which events to, for which keys), its object count, size and + estimated monthly storage cost from CloudWatch, and the findings about it with what to do. """ + s3 = _s3_module() core = self.analyzer - for region in {b.region for b in buckets if b.region}: - core._cloudwatch_in(region) # boto3 sessions aren't thread-safe: make the clients before the threads + name, _ = s3.parse_s3_uri(bucket) + self._require_bucket(name) + cfg = core.bucket_config(name) + errors: dict[str, str] = {} + notifications = metrics = None + try: + notifications = core.notifications(name) + except (ClientError, BotoCoreError) as exc: + errors["notifications"] = _code(exc) + try: + metrics = core.bucket_metrics(name) + self._count_metrics(_metrics_read(metrics), _METRIC_PRICE) + except (ClientError, BotoCoreError) as exc: + errors["metrics"] = _code(exc) + account = self._account_block() + out = summarize_bucket(cfg, metrics, notifications, account_block=account, prices=core.prices, + errors=errors) + size = out.get("size", {}).get("total") + out["_summary"] = ", ".join(filter(None, [ + size and f"{size}", out["public_access"], f"encryption {out['encryption']}", + f"{_plural(len(notifications), 'event notification')}" if notifications is not None else "", + _plural(len(out["findings"]), "finding")])) + out["_next"] = [(self._view_call("bucket_info", name), "its settings, policy and size as a report")] + if cfg.policy: + out["_next"].append((self._view_call("policy", name), "its bucket policy in plain English")) + return out - def size(bucket: BucketInfo) -> tuple[str, BucketMetrics | None, str | None]: - try: - return bucket.name, core.bucket_metrics(bucket.name), None - except (ClientError, BotoCoreError) as exc: - return bucket.name, None, _code(exc) + def list_folder(self, location: str) -> dict[str, Any]: + """ + What's directly in a folder (one level, like `aws s3 ls`): its subfolders, and its files newest first + with size, when each was written and storage class. Fast even on huge buckets. location: + 's3://bucket/prefix/' or 'bucket/prefix/'; a bucket name alone lists its top level. + """ + listing = self.analyzer.ls(location, limit=1000) + out = summarize_listing(listing) + out["_summary"] = (f"{_plural(len(listing.folders), 'folder')}, {_plural(len(listing.objects), 'file')}" + + (" (more not listed)" if listing.truncated else "")) + out["_next"] = [(self._view_call("ls", listing.uri), "the same folder as a report")] + return out - metrics: dict[str, BucketMetrics] = {} - errors: dict[str, str] = {} - with ThreadPoolExecutor(max_workers=8) as pool: - for name, found, error in pool.map(size, buckets): - if found is not None: - metrics[name] = found - self._count_metrics(_metrics_read(found), _METRIC_PRICE) - if error: - errors[f"{name}:metrics"] = error - return metrics, errors + def find_files(self, location: str, match: str = "", since: str = "", until: str = "", # noqa: PLR0913 + min_size: str = "", max_size: str = "", limit: int = 30) -> dict[str, Any]: + """ + The files under a location (every level below it) that match a filter: how many, their total size, + when the oldest and newest were written, and the newest `limit` of them. match: a glob on the file name, + or on the whole key when it has a '/' ('*.csv', 'orders_2026-10-1*'). since/until: only files last + written in that window, as ISO UTC times ('2026-10-15T18:00Z') or relative ('24h', '7d'). min_size / + max_size: '10MB', '1GB'. Reads the keys one listing page at a time, up to the budget's limit. + """ + s3 = _s3_module() + start, end = _between(since, until) + keep = s3.make_filter(pattern=match or None, min_size=min_size or None, max_size=max_size or None, + modified_after=start, modified_before=end if until else None) + found, scanned, more = self._matching(location, keep) + bucket, prefix = s3.parse_s3_uri(location) + out = summarize_found(s3.s3_uri(bucket, prefix), found, scanned=scanned, cut_short=more, start=start, + end=end, limit=max(1, min(int(limit or 30), 200)), + filters={"match": match, "min_size": min_size, "max_size": max_size}) + out["_summary"] = (f"{_plural(len(found), 'file')} ({out['total_size']})" + + (f" of {scanned:,} keys" if scanned != len(found) else "") + + (", cut short" if out["cut_short"] else "")) + out["_next"] = [(self._view_call("find", s3.s3_uri(bucket, prefix), pattern=match or None, + min_size=min_size or None, max_size=max_size or None, + modified_after=_iso(start), modified_before=_iso(end) if until else None), + "the same search as a report")] + return out + + def folder_summary(self, location: str) -> dict[str, Any]: + """ + What a location holds, every level below it: how many files and how much, by file type, storage + class and folder, the oldest and newest file, the largest ones, the estimated monthly storage cost and + findings (many small files, files in Glacier, files untouched for 90+ days...). Reads every key, up to + the budget's limit: use the narrowest location you can. + """ + core = self.analyzer + summary = core.summarize(location, top_n=5, folder_depth=1, limit=self.budget.s3_keys) + out = summarize_prefix(summary, core.prices) + out["_summary"] = (f"{_plural(summary.object_count, 'file')}, {out['total_size']}, about " + f"{human_money(summary.monthly_cost)}/month" + (", cut short" if summary.truncated + else "")) + out["_next"] = [(self._view_call("summary", summary.uri), "the same as a dashboard"), + (self._view_call("tree", summary.uri), "its folders' sizes as a tree")] + return out + + def file_info(self, location: str) -> dict[str, Any]: + """ + One file's headers, never its content: size, when it was written, content type, storage class (a + file in Glacier must be restored before anything can read it), encryption (a KMS key means readers need + kms:Decrypt on it), version, restore and lock status, its metadata and tags. location: + 's3://bucket/key'. + """ + core = self.analyzer + info = core.head(location) + out = summarize_file(info) + out["_summary"] = f"{out['size']}, {info.get('storage_class', 'STANDARD')}, written " + ( + _iso(info.get("last_modified")) or "?") + out["_next"] = [(self._view_call("head", info["uri"]), "every header as a report"), + (self._view_call("preview", info["uri"]), "what's in it, in your notebook only")] + return out - def _triggers(self, bucket: str, function: str, since: str, until: str, region: str) -> dict[str, Any]: + def s3_triggers(self, bucket: str = "", function: str = "", since: str = "24h", # noqa: PLR0913 + until: str = "", region: str = "") -> dict[str, Any]: """ - s3_triggers: the links for a bucket, a function or both, each checked from both ends (see its docstring). + How S3 and Lambda are wired: the bucket event notifications that invoke Lambda functions. Give a + bucket (the functions its events invoke), a function (the buckets whose events invoke it, from its + resource policy and those buckets' notifications) or both. For each link: the events and the key + prefix and suffix it fires on, whether the function exists and lets that bucket invoke it, how many files + the filter covers were written in the window and how many of them invoked it (and which it missed, by + upload kind or by the case of their ending), and how often the function ran, failed and was throttled in + that window (CloudWatch). Use it for "which Lambda processes files in X", "why didn't Y run after the + upload" or "were last night's files processed". since/until: ISO UTC times or relative ('24h'). region: + the function's region when it isn't the default. """ - s3, lf = _s3_module(), _lambda_module() + s3 = _s3_module() lam_kit = self.peer("lambda") if lam_kit is None: raise ValueError("s3_triggers needs the Lambda toolkit too: AgentView(toolkits=['lambda', 's3'])") @@ -858,90 +997,71 @@ def _triggers(self, bucket: str, function: str, since: str, until: str, region: raise ValueError("s3_triggers needs a bucket, a function, or both") start, end = _between(since or "24h", until) name = s3.parse_s3_uri(bucket)[0] if bucket else "" - fn = lam.function(function, region=region or None) if function else None # missing: its close names + fn = lam.function(function, region=region or None) if function else None errors: dict[str, str] = {} notes: list[tuple[str, str]] = [] others: list[EventNotification] = [] pairs: list[tuple[str, EventNotification]] = [] searched: list[str] = [] if name: - for n in core.notifications(name): # a missing bucket fails here: its close names - if n.target != "Lambda function": - others.append(n) - elif fn is None or _points_at(n, fn): - pairs.append((name, n)) + pairs, others = self._pairs_in_bucket(name, fn) elif fn is not None: - searched, named, notes = self._buckets_for(fn, errors) - for candidate in searched: - try: - pairs += [(candidate, n) for n in core.notifications(candidate) if _points_at(n, fn)] - except (ClientError, BotoCoreError) as exc: - errors[f"{candidate}:notifications"] = _code(exc) - for leftover in sorted(set(named) - {b for b, _ in pairs}): - notes.append(("info", f"{fn.name}'s resource policy lets bucket {leftover} invoke it, but " - f"{leftover} has no event notification that does: a leftover permission, or a " - "notification that was removed.")) - - functions: dict[str, Function | None] = {} # notification ARN -> the function, None when it isn't there - unknown: set[str] = set() # ARNs that couldn't be looked up - policies: dict[tuple[str, str, str | None], list[Any] | None] = {} - listings: dict[tuple[str, str], list[ObjectInfo] | None] = {} - links: list[TriggerLink] = [] - cap = self.budget.s3_keys - for where, n in pairs: - arn = n.arn or "" - link = TriggerLink(where, n, account=arn.split(":")[4] if arn.count(":") >= 5 else None) - if arn not in functions: - try: - functions[arn] = lam.function(arn) - except (ClientError, BotoCoreError) as exc: - functions[arn] = None - if _code(exc) != "ResourceNotFoundException": - errors[f"{n.name}:function"] = _code(exc) - unknown.add(arn) - target = functions[arn] - link.exists = None if arn in unknown else target is not None - if target is not None: - qualifier = lf.parse_function_ref(arn)[1] - key = (target.name, target.region, qualifier) - if key not in policies: - try: - policies[key] = lf.parse_policy(lam.policy(target.name, target.region, qualifier=qualifier)) - except (ClientError, BotoCoreError) as exc: - errors[f"{target.name}:policy_of_function"] = _code(exc) - policies[key] = None - if policies[key] is not None: - link.allowed = _s3_may_invoke(policies[key], where) - if link.exists is not False and n.on_created(): - listed = (where, n.prefix) - if listed not in listings: - try: - listings[listed] = list(core.iter_objects(s3.s3_uri(where, n.prefix), limit=cap + 1)) - except (ClientError, BotoCoreError) as exc: - errors[f"{where}:list"] = _code(exc) - listings[listed] = None - if listings[listed] is not None: - link.arrivals = count_arrivals(listings[listed], n, start, end, limit=cap) - links.append(link) - - live = list({f.arn: f for f in functions.values() if f is not None}.values()) - if live: - try: - numbers, read = lam.metrics(live, since=start, until=end, logs=False) - self._count_metrics(read, lam.prices["metric_request"]) - for link in links: - target = functions.get(link.notification.arn or "") - link.metrics = numbers.get(target.arn) if target is not None else None - except (ClientError, BotoCoreError) as exc: - errors["metrics"] = _code(exc) + pairs, searched, notes = self._pairs_for_function(fn, errors) + + checker = _LinkChecker(core, lam, start, end, self.budget.s3_keys, errors) + links = [checker.check(where, n) for where, n in pairs] + read = checker.add_metrics(links) + if read: + self._count_metrics(read, lam.prices["metric_request"]) out = summarize_triggers(links, start, end, bucket=name, function=fn.name if fn else "", others=others, notes=notes, errors=errors, searched=searched) - warned = [link for link in links if any(level == "warn" for level, _ in trigger_findings(link))] - pairs_text = sorted({f"{link.bucket} → {link.function}" for link in links}) - out["_summary"] = (_plural(len(links), "S3 → Lambda link") + (f": {_clip(', '.join(pairs_text), 120)}" - if links else "") - + (f"; {_plural(len(warned), 'link')} with warnings" if warned else "")) + out["_summary"] = _triggers_summary(links) + out["_next"] = self._triggers_next(links, checker.functions, lam_kit, start) + return out + + def _pairs_in_bucket(self, bucket: str, fn: Function | None) -> tuple[list[tuple[str, EventNotification]], + list[EventNotification]]: + """ + (bucket, notification) for each of a bucket's notifications that invokes a Lambda function (`fn` only, when + it's given), and its notifications to anything else. A bucket that isn't there fails here, and the tool's + answer gives its close names. + """ + pairs, others = [], [] + for n in self.analyzer.notifications(bucket): + if n.target != "Lambda function": + others.append(n) + elif fn is None or _points_at(n, fn): + pairs.append((bucket, n)) + return pairs, others + + def _pairs_for_function(self, fn: Function, errors: dict[str, str]) -> tuple[ + list[tuple[str, EventNotification]], list[str], list[tuple[str, str]]]: + """ + (bucket, notification) for each notification that invokes a function, from the buckets that may (_buckets_for), + the buckets looked through, and notes: a bucket the function's resource policy names that has no + notification for it is a leftover permission, or a notification that was removed. + """ + searched, named, notes = self._buckets_for(fn, errors) + pairs: list[tuple[str, EventNotification]] = [] + for candidate in searched: + try: + pairs += [(candidate, n) for n in self.analyzer.notifications(candidate) if _points_at(n, fn)] + except (ClientError, BotoCoreError) as exc: + errors[f"{candidate}:notifications"] = _code(exc) + for leftover in sorted(set(named) - {b for b, _ in pairs}): + notes.append(("info", f"{fn.name}'s resource policy lets bucket {leftover} invoke it, but " + f"{leftover} has no event notification that does: a leftover permission, or a " + "notification that was removed.")) + return pairs, searched, notes + + def _triggers_next(self, links: list[TriggerLink], functions: dict[str, Function | None], lam_kit: Toolkit, + start: datetime) -> list[tuple[str, str]]: + """ + The View calls worth running after s3_triggers: for each link, why its function's runs failed, the files + written in the window, and the function itself. + """ + s3 = _s3_module() nxt: list[tuple[str, str]] = [] for link in links: target = functions.get(link.notification.arn or "") @@ -954,8 +1074,66 @@ def _triggers(self, bucket: str, function: str, since: str, until: str, region: modified_after=_iso(start)), "the files written in the window")) if target is not None: nxt.append((lam_kit._next_for(target, "function_info"), "the function, its triggers and settings")) - out["_next"] = nxt - return out + return nxt + + def _require_bucket(self, name: str) -> None: + """ + Fails when the bucket isn't there, so the tool's answer is its close names (not_found) rather than a + report of every part that couldn't be read. + """ + self.analyzer.bucket_region(name) + + def _account_block(self) -> dict[str, bool] | None: + """ + The account's public access block, or None when it can't be read: the findings then say account-level + settings may still apply. + """ + try: + return self.analyzer.account_public_access_block() + except (ClientError, BotoCoreError, ValueError): + return None + + def _matching(self, location: str, keep: Callable[[ObjectInfo], bool]) -> tuple[list[ObjectInfo], int, bool]: + """ + (the files under a location that `keep` accepts, how many keys were read, whether there were more): it reads + up to the budget's s3_keys, and one more to tell when there were more. + """ + cap = self.budget.s3_keys + scanned, found, more = 0, [], False + for obj in self.analyzer.iter_objects(location, limit=cap + 1): + if scanned >= cap: + more = True + break + scanned += 1 + if keep(obj): + found.append(obj) + return found, scanned, more + + def _sizes(self, buckets: list[BucketInfo]) -> tuple[dict[str, BucketMetrics], dict[str, str]]: + """ + Each bucket's CloudWatch size and object count, read side by side; errors by 'bucket:metrics'. boto3 + sessions aren't thread-safe, so the CloudWatch clients are made before the threads start. + """ + core = self.analyzer + for region in {b.region for b in buckets if b.region}: + core._cloudwatch_in(region) + + def size(bucket: BucketInfo) -> tuple[str, BucketMetrics | None, str | None]: + try: + return bucket.name, core.bucket_metrics(bucket.name), None + except (ClientError, BotoCoreError) as exc: + return bucket.name, None, _code(exc) + + metrics: dict[str, BucketMetrics] = {} + errors: dict[str, str] = {} + with ThreadPoolExecutor(max_workers=_SIZE_WORKERS) as pool: + for name, found, error in pool.map(size, buckets): + if found is not None: + metrics[name] = found + self._count_metrics(_metrics_read(found), _METRIC_PRICE) + if error: + errors[f"{name}:metrics"] = error + return metrics, errors def _buckets_for(self, fn: Function, errors: dict[str, str]) -> tuple[list[str], list[str], list[tuple[str, str]]]: diff --git a/src/aws_analyzer/agent/view.py b/src/aws_analyzer/agent/view.py index b39692c..883e106 100644 --- a/src/aws_analyzer/agent/view.py +++ b/src/aws_analyzer/agent/view.py @@ -2,6 +2,9 @@ AgentView: the agent's notebook UI. ask() shows the answer first, then what it checked to get it (each tool call with its window and regions, what it couldn't read or was refused, the AWS reads, the tokens and the cost) and the View commands to look closer; transcript() and tools() show the questions so far and what the agent can call. + +Its reports sit in a box of the class _ROOT, which scopes its stylesheet's rules, and each title starts with +_BADGE, so its reports are easy to tell from other analyzers'. """ from __future__ import annotations @@ -20,7 +23,7 @@ from .analyzer import AgentAnalyzer from .answers import AgentAnswer, ToolCall, answer_findings, answer_next -_ROOT = "agt" # the class around each report's HTML: its stylesheet's rules are scoped to it +_ROOT = "agt" _EXTRA_CSS = """\ .agt .ans{font-size:14px;line-height:1.6;margin:8px 0 12px;max-width:900px;overflow-wrap:anywhere} .agt .ans>:first-child{margin-top:0} @@ -43,7 +46,7 @@ .agt .ans tbody tr:last-child td{border-bottom:0} .agt .ans th.r,.agt .ans td.r{text-align:right;font-variant-numeric:tabular-nums}""" _CSS = _base_css(_ROOT, _EXTRA_CSS) -_BADGE = "🤖 Agent" # the chip before each report's title, so reports from different analyzers are easy to tell apart +_BADGE = "🤖 Agent" def _render_html(blocks: list[Any], max_rows: int) -> str: @@ -113,8 +116,8 @@ class AgentView(_BaseView): progress: 'auto' (a progress bar while it works, when tqdm is installed), 'plain' or 'off'. """ - _render_html = staticmethod(_render_html) # its look: the stylesheet, root class and badge - _GROUPS = { # help() lists the commands in these groups, in this order + _render_html = staticmethod(_render_html) + _GROUPS = { "💬 Ask": ("ask", "new_conversation"), "🔎 Look back": ("transcript",), "🧰 What it can do": ("tools",), @@ -143,7 +146,6 @@ def __init__( f"{', '.join(f'{k}=...' for k in settings)})") self.core: AgentAnalyzer = core or AgentAnalyzer(**settings) - # ------------------------------------------------------------------ plumbing def _explain(self, code: str, message: str) -> str: """ @@ -154,7 +156,6 @@ def _explain(self, code: str, message: str) -> str: return f"{text} README lists the read-only IAM permissions each tool needs." return message - # ------------------------------------------------------------------ asking @_friendly_errors def ask(self, question: str, *, new: bool = False) -> None: @@ -235,7 +236,6 @@ def new_conversation(self) -> None: self.core.new_conversation() self._show([_Note("New conversation: the next ask() starts fresh.", "ok")]) - # ------------------------------------------------------------------ looking back @_friendly_errors def transcript(self) -> None: @@ -267,7 +267,6 @@ def transcript(self) -> None: _Note(answer.error or "No answer: it stopped before writing one.", "warn")) self._show(blocks) - # ------------------------------------------------------------------ what it can do @_friendly_errors def tools(self) -> None: diff --git a/tests/test_style.py b/tests/test_style.py index fcd4e7d..a018f1d 100644 --- a/tests/test_style.py +++ b/tests/test_style.py @@ -29,20 +29,6 @@ ".claude/skills/release/release.py": 4, "src/aws_analyzer/__init__.py": 3, "src/aws_analyzer/_bedrock.py": 83, - "src/aws_analyzer/agent/analyzer.py": 10, - "src/aws_analyzer/agent/answers.py": 18, - "src/aws_analyzer/agent/budget.py": 10, - "src/aws_analyzer/agent/guard.py": 7, - "src/aws_analyzer/agent/harness.py": 7, - "src/aws_analyzer/agent/llm.py": 2, - "src/aws_analyzer/agent/prompt.py": 1, - "src/aws_analyzer/agent/results.py": 3, - "src/aws_analyzer/agent/runtime.py": 5, - "src/aws_analyzer/agent/toolkits/__init__.py": 2, - "src/aws_analyzer/agent/toolkits/base.py": 14, - "src/aws_analyzer/agent/toolkits/lambda_tools.py": 10, - "src/aws_analyzer/agent/toolkits/s3_tools.py": 36, - "src/aws_analyzer/agent/view.py": 8, "src/aws_analyzer/bedrock_chat.py": 281, "src/aws_analyzer/bedrock_kb.py": 235, "src/aws_analyzer/dynamodb.py": 82,