feat: add scoped diagnosis discovery and evidence core

agent/lambda/aws_finops_mcp.py

@@ -53,7 +53,10 @@ def _recommendation_evidence(record, option, resource_type):
     names = {metric["name"] for metric in metrics}
     coverage = {
         "cpu": bool(names & {"Cpu", "CPU"}), "memory": "Memory" in names,
-        "io": any(name.startswith(("EBS_", "DISK_", "NETWORK_", "EBSVolume", "Network", "StorageNetwork"))
+        # I/O pressure only: storage *capacity* (EBSVolumeStorageSpaceUtilization, VolumeBytesUsed)
+        # is not I/O evidence and must not satisfy the downsize gate.
+        "io": any((name.startswith(("EBS_", "DISK_", "NETWORK_", "EBSVolume", "Network", "StorageNetwork"))
+                   and name != "EBSVolumeStorageSpaceUtilization")
                   or name in {"VolumeReadIOPs", "VolumeWriteIOPs", "ReadIOPSEphemeralStorage", "WriteIOPSEphemeralStorage"}
                   for name in names),
     }

agent/lambda/test_aws_finops_mcp.py

@@ -225,6 +225,16 @@ def test_preserves_provider_freshness_lookback_and_missing_utilization(stubber,
     assert recommendation["missingUtilizationMetrics"] == ["memory", "io"]
 
 
+def test_storage_capacity_metric_is_not_io_evidence(stubber):
+    response = response_for("rds")
+    record = response[CASES["rds"]["collection"]][0]
+    record["utilizationMetrics"] = [{"name": "CPU", "statistic": "MAXIMUM", "value": 17.0},
+                                    {"name": "EBSVolumeStorageSpaceUtilization", "statistic": "MAXIMUM", "value": 40.0}]
+    enqueue(stubber, "rds", response)
+    recommendation = invoke("rds")["results"]["rds"]["recommendations"][0]
+    assert recommendation["missingUtilizationMetrics"] == ["memory", "io"]
+
+
 @pytest.mark.parametrize("resource_type", ["ec2", "rds"])
 def test_selects_the_lowest_provider_rank_independently_of_response_order(stubber, resource_type):
     response = response_for(resource_type)

docs/api-reference.md

@@ -538,6 +538,25 @@ runtime payloads for compatibility with older or malformed responses.
 | `readStatus`, `readReason`, `readTruncated` | API read availability/coverage, independent of collector status: `ok`, `partial` (`row_limit`), or `unavailable` (`busy`/`timeout`/`query_failed`). |
 | `sources[].scope/capturedAtMs/lastSuccessAtMs`, `publishedSources[]` | Current/saved source status, scope and reasons, plus optional capture/sweep clocks and saved provenance. Missing status is explicitly unknown; absent clocks are not fabricated. |
 
+### Trace edge instance observations
+
+Trace `queries` edges (service → `db`) may carry three additive, optional `meta` fields built by
+`web/lib/trace-graph.ts`. Older rows and responses without them remain compatible; absence means
+no bound observation was recorded, not that no instance queried the database.
+
+| Field | Meaning |
+|---|---|
+| `instance_observations[]` | Distinct (`instance_ref`, `database_ref`) pairs from outbound `CLIENT` spans only. Each entry carries `instance_id`, `instance_ref`, `database_ref`, `account`, `region`, `db_host`, `observed_at_ms` (latest span start), `first_observed_at_ms`, `observation_count`, an opaque `correlation_id` naming the latest span, optional `source_id`, and `provenance: "trace_span"`. At most 200 pairs per edge. |
+| `instance_observations_omitted` | Count of further distinct pairs dropped by the 200-pair cap. |
+| `instance_observations_unbound` | Count of spans whose EC2 claim could not be bound: missing or invalid account/region/clock, or no unique inventory EC2 instance and RDS instance endpoint in the same account and region. |
+
+These are sampled telemetry claims bound to inventory, not authority: a pair shows that the
+sampled spans named that instance and endpoint within the trace window. It does not prove
+current connectivity, credentials or complete coverage of instances querying the database.
+Binding requires an exact RDS instance endpoint; Aurora cluster/reader and RDS Proxy endpoints
+stay unbound. The agent SQL view `sql_reader.topology_edges` does not project these fields; see
+[agent SQL reader](runbooks/agent-sql-reader.md#trace-queue-projection--트레이스-큐-투영).
+
 The UI supports the existing trace envelope and optional inventory/saved-source
 fields emitted by the bounded publication implementation in `web/lib/graph-store.ts`.
 Source integration does not establish successful producer rollout or migration. Source details are collapsed and height-bounded; their count

docs/runbooks/agent-sql-reader.md

@@ -402,6 +402,12 @@ copies stay absent for trace queues. SELECT is granted only on the view, never t
 `scripts/v2/workers/test_graph_collection.py` exercises this contract on disposable PostgreSQL,
 including grant restoration, reapplication, denied base-table reads and denied view writes.
 
+Trace edge instance observations (`instance_observations[]`, `instance_observations_omitted`,
+`instance_observations_unbound`; see [API reference](../api-reference.md#trace-edge-instance-observations))
+are not part of the agent evidence contract: `sql_reader.topology_edges` projects only numeric
+`spanCount` and `metricCount` from edge `meta`, so `execute_sql` cannot read them. Exposing them
+requires a new reviewed view migration; never grant the base table.
+
 그래프 근거 마이그레이션 이후 `01M27B...`가 노드 투영을 갱신한다. 적용 후 보존된 행을 포함해
 큐 claim이 destination ARN의 한정자와만 일치하는지 확인한다. 비-ARN·잘못된 ARN은 null이고
 호출자 폴백은 없어야 한다. 출처는 항상 `telemetry_claim`이며 trace 큐의 `accountId`·`region`·

scripts/v2/steampipe/sync_lambda.py

@@ -89,7 +89,8 @@ def _failure_label_is_throttling(label):
         "vpc_id, db_subnet_group_name, availability_zone, endpoint_address, endpoint_port, "
         "backup_retention_period, preferred_backup_window, latest_restorable_time, vpc_security_groups, "
         "auto_minor_version_upgrade, copy_tags_to_snapshot, deletion_protection, "
-        "iam_database_authentication_enabled, performance_insights_enabled, create_time, tags "
+        "iam_database_authentication_enabled, performance_insights_enabled, create_time, tags, "
+        "read_replica_source_db_instance_identifier "
         "FROM aws_rds_db_instance ORDER BY db_instance_identifier",
         "db_instance_identifier",
         "region",

scripts/v2/workers/diagnosis/budget.py

@@ -0,0 +1,130 @@
+"""One thread-safe allocation ledger from handler entry through artifact export."""
+from copy import deepcopy
+import math
+import os
+import threading
+import time
+
+
+# The single parse of the task ledger; report.py imports these so the render
+# deadline and this analysis ledger cannot desynchronize.
+TASK_SECONDS = int(os.environ.get("DIAGNOSIS_TASK_BUDGET_S", "3300"))
+EXPORT_RESERVE_SECONDS = int(os.environ.get("DIAGNOSIS_POST_RENDER_RESERVE_S", "600"))
+MAX_INPUT_BYTES = 96 * 1024
+MAX_OUTPUT_TOKENS = 8192
+# Opus 5.x cannot disable thinking, which spends from max_tokens; report.py shares this parse.
+OPUS_THINKING_HEADROOM = max(0, int(os.environ.get("DIAGNOSIS_OPUS_THINKING_HEADROOM", "8000")))
+
+
+def opus_headroom(model_id):
+    """Thinking headroom for Opus 5.x only; an Opus 4.x rollback override takes thinking off."""
+    return OPUS_THINKING_HEADROOM if "claude-opus-5" in model_id else 0
+
+
+def attempt_tokens(model_id):
+    return MAX_OUTPUT_TOKENS + opus_headroom(model_id)
+
+
+TIER_LIMITS = {
+    "light": {"attempts": 20, "input_bytes": 2 * 1024 * 1024, "output_tokens": 100000},
+    "mid": {"attempts": 24, "input_bytes": 3 * 1024 * 1024, "output_tokens": 120000},
+    "deep": {"attempts": 32, "input_bytes": 4 * 1024 * 1024, "output_tokens": 200000},
+}
+
+
+class BudgetExceeded(RuntimeError):
+    """A visible allocation/deadline failure; callers return partial coverage."""
+
+
+class ReportBudget:
+    def __init__(self, tier, started_at=None):
+        if tier not in TIER_LIMITS:
+            raise ValueError("unknown diagnosis tier")
+        self.tier = tier
+        self.started_at = time.monotonic() if started_at is None else started_at
+        if not isinstance(self.started_at, (int, float)) or not math.isfinite(self.started_at):
+            raise ValueError("started_at must be a monotonic timestamp")
+        self.limits = dict(TIER_LIMITS[tier])
+        self._lock = threading.Lock()
+        self._attempts = {}
+        self._input_bytes = 0
+        self._output_tokens = 0
+
+    def remaining(self, analysis=True):
+        reserve = EXPORT_RESERVE_SECONDS if analysis else 0
+        return max(0.0, self.started_at + TASK_SECONDS - reserve - time.monotonic())
+
+    def reserve(self, stage, model_id, input_bytes, max_tokens):
+        if not isinstance(stage, str) or not stage or not isinstance(model_id, str) or not model_id:
+            raise ValueError("stage and model_id must be named")
+        if type(input_bytes) is not int or type(max_tokens) is not int or input_bytes < 0 or max_tokens < 1:
+            raise ValueError("attempt allocations must be nonnegative integer bytes and positive tokens")
+        if input_bytes > MAX_INPUT_BYTES or max_tokens > attempt_tokens(model_id):
+            raise BudgetExceeded("per-attempt allocation exceeded")
+        with self._lock:
+            if self.remaining() <= 0:
+                raise BudgetExceeded("analysis deadline exhausted")
+            # Every specialist/retry leaves capacity for a bounded final review.
+            # Review attempts can consume this reserve. Nothing refunds a failed attempt.
+            final = stage in {"review", "final_review"}
+            held_calls = 0 if final else 1
+            held_bytes = 0 if final else MAX_INPUT_BYTES
+            held_tokens = 0 if final else attempt_tokens(model_id)
+            if len(self._attempts) + 1 + held_calls > self.limits["attempts"]:
+                raise BudgetExceeded("model attempt ceiling exhausted")
+            if self._input_bytes + input_bytes + held_bytes > self.limits["input_bytes"]:
+                raise BudgetExceeded("submitted input ceiling exhausted")
+            if self._output_tokens + max_tokens + held_tokens > self.limits["output_tokens"]:
+                raise BudgetExceeded("reserved output ceiling exhausted")
+            attempt_id = f"attempt-{len(self._attempts) + 1}"
+            self._attempts[attempt_id] = {
+                "id": attempt_id, "stage": stage, "model_id": model_id,
+                "input_bytes": input_bytes, "max_tokens": max_tokens,
+                "outcome": "pending", "usage": None, "elapsed_seconds": None,
+                "_started_at": time.monotonic(),
+            }
+            self._input_bytes += input_bytes
+            self._output_tokens += max_tokens
+            return attempt_id
+
+    def record(self, attempt_id, outcome, usage=None):
+        if not isinstance(outcome, str) or not outcome:
+            raise ValueError("outcome must be named")
+        with self._lock:
+            attempt = self._attempts.get(attempt_id)
+            if attempt is None:
+                raise ValueError("unknown attempt")
+            if attempt["outcome"] != "pending":
+                raise ValueError("attempt already recorded")
+            normalized = None
+            if isinstance(usage, dict):
+                counts = {key: usage.get(key) for key in ("input_tokens", "output_tokens")}
+                if all(type(value) is int and value >= 0 for value in counts.values()):
+                    normalized = counts
+            attempt["outcome"] = outcome
+            attempt["usage"] = normalized
+            attempt["elapsed_seconds"] = max(0.0, time.monotonic() - attempt["_started_at"])
+
+    def snapshot(self):
+        with self._lock:
+            attempts = [{key: deepcopy(value) for key, value in row.items() if not key.startswith("_")}
+                        for row in self._attempts.values()]
+            known = {"input_tokens": 0, "output_tokens": 0}
+            unknown = 0
+            for row in attempts:
+                if row["usage"] is None:
+                    unknown += 1
+                else:
+                    for key in known:
+                        known[key] += row["usage"][key]
+            return {
+                "tier": self.tier, "limits": dict(self.limits),
+                "attempt_count": len(attempts), "attempts": attempts,
+                "submitted_input_bytes": self._input_bytes,
+                "reserved_output_tokens": self._output_tokens,
+                "actual_usage": None if unknown else dict(known),
+                "known_usage": known, "unknown_usage_attempts": unknown,
+                "analysis_remaining_seconds": self.remaining(),
+                "task_remaining_seconds": self.remaining(analysis=False),
+                "export_reserve_seconds": EXPORT_RESERVE_SECONDS,
+            }