fix(0.5.9): bounded log queries, tool-loop budgets, remove over-promises

Closes the four second-review blockers per owner dispositions.

F-4 (implemented): adds airlock/log_query.py as the single bounded reader.
slow/analyzer._load_logs read every record from every daily file, and
advisor/tools.get_recent_errors passed limit=1000000 -- a limit in name only
that materialized every row before filtering. Both are reachable from the
CLI and TUI, so enough accumulated history would exhaust memory and OOM-kill
the proxy under the unit's MemoryMax.

The reader filters while scanning (a rare match stays cheap), walks days
newest-first (a truncated result keeps the most recent records), and reports
truncation rather than returning partial data silently. That last property
matters most: an analysis that scanned half the window and presents itself
as complete yields confident wrong conclusions about traffic it never saw.
Consumers now surface LogPage.truncated.

F-1 Part A (implemented): _run_tool_loop returns a ToolLoopOutcome carrying
an explicit stop_reason, under a ToolLoopBudget bounding rounds, wall-clock,
total tool calls, and per-result bytes. Previously the only bound was a
round count -- three rounds against a slow model is unbounded in the
dimension that hurts -- and every failure path returned a bare None, making
"aborted on a disallowed tool" indistinguishable from "nothing to say".
Part B (parameterized tool arguments) is deferred to 0.5.10: it should be
served by the F-4 reader, and adding arguments first would let a model
request an arbitrarily large slice.

F-2 (re-scoped): behavior unchanged; the claim is corrected. Airlock does
not manage, verify, or retrieve results from a sandbox -- it declares the
provider's code_execution tool on a Messages API call. The opt-in controls
whether derived aggregates leave the machine and is not a security boundary.
Documented in docs/guide/guardrails.md.

F-3 (out of scope): code inspection stays observational. The hardcoded
"enforcement_weight": 0.0 is removed with owner approval -- a field of that
name in persisted evidence advertised wiring that does not exist, the same
over-promise as F-2. Wiring it to enforcement needs its own observe window:
resource_access matches any open( or requests. in a code block.

Suite: 2986 passed.

Author: coreyt

airlock/advisor/tools.py

@@ -20,12 +20,15 @@
 import json
 import logging
 from collections import Counter
-from datetime import datetime, timedelta
 from pathlib import Path
 from typing import Any, Callable
 
-from airlock.fast.state import StateStore
 from airlock.api.queries import get_request_logs
+from airlock.fast.state import StateStore
+from airlock.log_query import LogPage, LogQuery, query_logs
+
+#: Upper bound on rows pulled from the datastore in one advisor query.
+DATASTORE_QUERY_LIMIT = 50_000
 
 logger = logging.getLogger("airlock.advisor.tools")
 
@@ -35,28 +38,23 @@
 # ---------------------------------------------------------------------------
 
 
-def _load_logs(log_dir: str, days: int = 7) -> list[dict[str, Any]]:
-    """Load JSONL records from the last *days* days."""
-    records: list[dict[str, Any]] = []
-    today = datetime.utcnow().date()
-    log_path = Path(log_dir)
-
-    for i in range(days):
-        day = today - timedelta(days=i)
-        file_path = log_path / f"airlock-{day.isoformat()}.jsonl"
-        if not file_path.exists():
-            continue
-        with open(file_path, encoding="utf-8") as f:
-            for line in f:
-                line = line.strip()
-                if not line:
-                    continue
-                try:
-                    records.append(json.loads(line))
-                except json.JSONDecodeError:
-                    continue
+def _load_page(
+    log_dir: str,
+    days: int = 7,
+    predicate: Callable[[dict[str, Any]], bool] | None = None,
+) -> LogPage:
+    """Bounded log scan. Callers should surface ``page.truncated``.
+
+    Previously an unbounded read of every record in the window. Advisor tools
+    are reachable from the CLI and TUI, so an operator with enough accumulated
+    history could exhaust memory simply by asking a question.
+    """
+    return query_logs(LogQuery(days=days, predicate=predicate, directory=Path(log_dir)))
 
-    return records
+
+def _load_logs(log_dir: str, days: int = 7) -> list[dict[str, Any]]:
+    """Backwards-compatible record list. Prefer :func:`_load_page`."""
+    return _load_page(log_dir, days=days).records
 
 
 # ---------------------------------------------------------------------------
@@ -145,17 +143,31 @@ def get_recent_errors(log_dir: str, days: int = 2) -> dict:
     except Exception:
         engine = None
 
+    truncated: dict[str, Any] = {"truncated": False, "limit_hit": None}
     if engine is not None:
-        nodes = get_request_logs(engine, limit=1000000)
+        # Was limit=1000000 — a limit in name only, and every row was
+        # materialized before filtering.
+        nodes = get_request_logs(engine, limit=DATASTORE_QUERY_LIMIT)
         records = [n.properties if hasattr(n, "properties") else n for n in nodes]
         failures = [
             r
             for r in records
             if (r.get("success") is False) or bool(r.get("error_flag"))
         ]
+        if len(records) >= DATASTORE_QUERY_LIMIT:
+            truncated = {"truncated": True, "limit_hit": "datastore_limit"}
     else:
-        records = _load_logs(log_dir, days=days)
-        failures = [r for r in records if not r.get("success")]
+        # Filter during the scan so a quiet window never materializes the
+        # successful traffic it is dominated by.
+        page = _load_page(
+            log_dir,
+            days=days,
+            predicate=lambda r: (
+                (r.get("success") is False) or bool(r.get("error_flag"))
+            ),
+        )
+        failures = page.records
+        truncated = {"truncated": page.truncated, "limit_hit": page.limit_hit}
 
     by_model: Counter = Counter()
     by_error_type: Counter = Counter()
@@ -182,6 +194,9 @@ def get_recent_errors(log_dir: str, days: int = 2) -> dict:
         "by_error_type": dict(by_error_type),
         "by_client": dict(by_client),
         "recent_samples": recent_samples,
+        # The advisor must be able to say "based on a partial window" rather
+        # than presenting a truncated scan as the whole picture.
+        "window": truncated,
     }
 
 

airlock/guardrails/code_inspection.py

@@ -2,6 +2,17 @@
 
 The result intentionally contains categories and counts, never matched source
 text.  This makes it safe to persist alongside the canonical request record.
+
+**This is observation only — it is not wired to enforcement.** The result
+previously carried an ``enforcement_weight`` of ``0.0`` that nothing read; a
+field by that name in persisted evidence implied a wiring that did not exist,
+so it was removed (0.5.9, owner decision).
+
+Connecting inspection to post-response enforcement would need its own observe
+window first: ``resource_access`` matches any ``open(`` or ``requests.`` in a
+code block, which is entirely ordinary in code-assistance traffic. Enabling it
+as a blocking signal without evidence would generate false positives on normal
+work.
 """
 
 from __future__ import annotations
@@ -63,5 +74,4 @@ def inspect_code(text: str) -> dict[str, Any]:
         "code_blocks": len(blocks),
         "findings": findings,
         "score": min(1.0, sum(findings.values()) / 5.0),
-        "enforcement_weight": 0.0,
     }

airlock/log_query.py

@@ -0,0 +1,200 @@
+"""Bounded reader for Airlock's JSONL request logs.
+
+Every consumer that reads request logs — the slow analyzer, the advisor tools,
+and the TUI log screen — goes through here, so the bounds and the truncation
+semantics live in one place.
+
+Why this exists
+---------------
+The TUI screen was given a record cap in 0.5.9; the other readers were not.
+``slow/analyzer._load_logs`` read every record from every daily file into a
+list, and ``advisor/tools.get_recent_errors`` passed ``limit=1000000`` — a limit
+in name only. Both are reachable from the CLI and the TUI, so a deployment with
+enough accumulated history would try to hold its whole log corpus in memory,
+with no degradation path: it either fits or the process dies. Under the systemd
+unit's ``MemoryMax``, that is an OOM kill of the proxy, not just of the
+analysis.
+
+Three properties matter, and the second matters most:
+
+**Filter while scanning.** The predicate runs per line, so a narrow query never
+materializes the whole corpus. This is what makes a generous record ceiling
+generous rather than restrictive.
+
+**Truncation is reported, never silent.** :attr:`LogPage.truncated` and
+:attr:`LogPage.limit_hit` are meant to reach the caller's output. An analysis
+that scanned half the window and presents itself as complete is worse than one
+that refuses — it yields confident, wrong conclusions about traffic it never
+saw.
+
+**Newest-first.** Days are walked backwards from today, so a truncated result
+retains the most recent records, which is what every consumer actually wants.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any, Callable, Iterator
+
+#: Hard ceiling on retained records. Generous because the predicate filters
+#: during the scan; a narrow query will not approach it.
+DEFAULT_MAX_RECORDS = 50_000
+
+#: Hard ceiling on bytes read from disk in one query.
+DEFAULT_MAX_BYTES = 256 * 1024 * 1024
+
+LIMIT_RECORDS = "max_records"
+LIMIT_BYTES = "max_bytes"
+
+
+def _env_int(name: str, default: int) -> int:
+    raw = (os.getenv(name) or "").strip()
+    if not raw:
+        return default
+    try:
+        value = int(raw)
+    except ValueError:
+        return default
+    return value if value > 0 else default
+
+
+def log_dir() -> Path:
+    return Path(os.getenv("AIRLOCK_LOG_DIR", "./logs"))
+
+
+@dataclass(frozen=True)
+class LogQuery:
+    """A bounded request for log records.
+
+    ``predicate`` is applied per record during the scan. Returning False costs
+    only the parse, so callers should filter here rather than afterwards.
+    """
+
+    days: int = 7
+    max_records: int = 0  # 0 → resolve from environment/default
+    max_bytes: int = 0  # 0 → resolve from environment/default
+    predicate: Callable[[dict[str, Any]], bool] | None = None
+    directory: Path | None = None
+
+    def resolved_max_records(self) -> int:
+        return self.max_records or _env_int(
+            "AIRLOCK_LOG_QUERY_MAX_RECORDS", DEFAULT_MAX_RECORDS
+        )
+
+    def resolved_max_bytes(self) -> int:
+        return self.max_bytes or _env_int(
+            "AIRLOCK_LOG_QUERY_MAX_BYTES", DEFAULT_MAX_BYTES
+        )
+
+
+@dataclass
+class LogPage:
+    """Result of a bounded scan.
+
+    ``truncated`` means a limit stopped the scan before the requested window was
+    exhausted — the caller is holding a partial view and must say so.
+    """
+
+    records: list[dict[str, Any]] = field(default_factory=list)
+    scanned: int = 0
+    bytes_read: int = 0
+    truncated: bool = False
+    limit_hit: str | None = None
+    files_read: list[str] = field(default_factory=list)
+    oldest_day: str | None = None
+
+    def note(self) -> str | None:
+        """One-line, human-facing description of truncation, or None."""
+        if not self.truncated:
+            return None
+        if self.limit_hit == LIMIT_RECORDS:
+            return (
+                f"Results truncated at {len(self.records):,} records; "
+                "older records in the requested window were not read."
+            )
+        return (
+            f"Results truncated after reading {self.bytes_read / 1024 / 1024:.0f} MB; "
+            "older records in the requested window were not read."
+        )
+
+    def as_metadata(self) -> dict[str, Any]:
+        """Truncation state for embedding in reports and tool results."""
+        return {
+            "records": len(self.records),
+            "scanned": self.scanned,
+            "truncated": self.truncated,
+            "limit_hit": self.limit_hit,
+            "oldest_day": self.oldest_day,
+        }
+
+
+def _day_files(query: LogQuery) -> Iterator[tuple[str, Path]]:
+    """Yield ``(iso_day, path)`` newest first for the requested window."""
+    directory = query.directory or log_dir()
+    today = datetime.now(timezone.utc).date()
+    for offset in range(max(1, query.days)):
+        day = today - timedelta(days=offset)
+        path = directory / f"airlock-{day.isoformat()}.jsonl"
+        if path.exists():
+            yield day.isoformat(), path
+
+
+def query_logs(query: LogQuery | None = None) -> LogPage:
+    """Scan request logs newest-first under explicit bounds."""
+    query = query or LogQuery()
+    max_records = query.resolved_max_records()
+    max_bytes = query.resolved_max_bytes()
+    page = LogPage()
+
+    for day, path in _day_files(query):
+        page.files_read.append(path.name)
+        page.oldest_day = day
+        try:
+            handle = path.open("r", encoding="utf-8")
+        except OSError:
+            continue
+        with handle:
+            for line in handle:
+                page.bytes_read += len(line)
+                if page.bytes_read > max_bytes:
+                    page.truncated = True
+                    page.limit_hit = LIMIT_BYTES
+                    return page
+                line = line.strip()
+                if not line:
+                    continue
+                page.scanned += 1
+                try:
+                    record = json.loads(line)
+                except json.JSONDecodeError:
+                    continue
+                if query.predicate is not None and not query.predicate(record):
+                    continue
+                page.records.append(record)
+                if len(page.records) >= max_records:
+                    page.truncated = True
+                    page.limit_hit = LIMIT_RECORDS
+                    return page
+    return page
+
+
+def load_records(
+    days: int = 7,
+    *,
+    directory: Path | str | None = None,
+    predicate: Callable[[dict[str, Any]], bool] | None = None,
+    max_records: int = 0,
+) -> LogPage:
+    """Convenience wrapper returning a :class:`LogPage`."""
+    return query_logs(
+        LogQuery(
+            days=days,
+            predicate=predicate,
+            max_records=max_records,
+            directory=Path(directory) if directory is not None else None,
+        )
+    )