From 3fc2e56341949f2d7ea518a7f24b35e2550fc0ff Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 16 Jul 2026 17:54:44 +0000 Subject: [PATCH] =?UTF-8?q?fix+feat:=20v0.34.0=20=E2=80=94=20engine=20corr?= =?UTF-8?q?ectness,=20lifecycle=20robustness,=20agent=20de-ossification,?= =?UTF-8?q?=20model=20config?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A three-angle audit (agent ossification, DuckDB engine math, state-machine lifecycle) plus the reviewed third-party findings. Bugs that misreported numbers, wedged state forever, or needlessly boxed the agent in. Analysis-engine correctness: - Integer columns stay Int64: status codes no longer render "404.0" and int64 sizes above 2^53 keep precision (a single null coerced the column to DOUBLE). - Access-log error rates divide by parsed requests, not all ingested lines (they were diluted by the unparsed fraction). - Inventory average = total/count (reconciles with the shown pair) and small_object_ratio is over known-size objects. - Age bucketing pins DuckDB TimeZone=UTC (was sidecar-tz-dependent, ±1 bucket). Lifecycle robustness: - Evidence-import post-download persistence is guarded: a DB error there no longer wedges the import in 'importing' forever (it reverts to failed). - The blocking turn path resolves its turn_guard handle on a persist failure (streaming already did) — otherwise every later turn stalled 120s. - The loser of a concurrent import claim fails its orphan pending run. - A report_ready publish failure can't downgrade a completed run to failed; the executor's failure branch won't overwrite a terminal state. Model provider & budget (reviewed findings): - Provider test no longer false-green (404/405 → "reachable, key unverified", not a confident pass) or false-red (empty base_url is valid → OpenAI default). - Tool-output budget is a hard cap: a single oversized output is withheld with a valid JSON envelope instead of blowing past the per-turn budget. De-ossification: - Custom aggregation: 2nd group-by dim (cross-tabs), day/weekday buckets, distinct_ips/keys/p99/min-max metrics, deterministic top-N tiebreaker — all fixed SQL fragments, whitelist+bound-params floor intact. - Operator-declarable max_output_tokens clamps the completion budget (symmetric with context_window) so a lower-cap endpoint doesn't 400. - Agent-memory recall scales with the model window (floored 50); skill-load guard raised (elastic byte budget is the real bound); survey ceiling 500→2000. Frontend: - The 409 blocking-fallback captures its answer baseline BEFORE the turn (not after the 409), closing a race that hung the UI "running" for ~2.5 min. Deferred (low value): OSS5/OSS6 (elastic turn cap + context-echo caps) — the elastic tool-output budget already governs the real concern. Tests: +8 (test_v0340_fixes.py) + updated budget/model/migration tests. Full sidecar suite 593 passed; ruff clean; tsc + frontend build clean. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01M3YojFPzjkfMj6YYSwvzdx --- CHANGELOG.md | 82 ++++++++++ frontend/src/api.ts | 3 + frontend/src/hooks/useTurnRunner.ts | 35 +++-- frontend/src/i18n.tsx | 6 + frontend/src/types.ts | 4 + frontend/src/views/ProvidersView.tsx | 27 +++- sidecar/app/agent_runtime/agent_service.py | 2 + sidecar/app/agent_runtime/model_budget.py | 15 +- .../app/agent_runtime/session_action_tools.py | 4 +- sidecar/app/agent_runtime/session_agent.py | 53 +++++-- .../agent_runtime/session_analysis_tools.py | 5 +- sidecar/app/agent_runtime/session_tools.py | 12 +- sidecar/app/analysis/access_logs.py | 26 +++- sidecar/app/analysis/aggregate.py | 59 ++++++-- sidecar/app/analysis/duck.py | 15 +- sidecar/app/analysis/inventory.py | 22 ++- sidecar/app/migrations.py | 11 ++ sidecar/app/models/schemas.py | 10 ++ sidecar/app/repositories/model_providers.py | 16 +- sidecar/app/routers/evidence_imports.py | 54 ++++--- sidecar/app/routers/model_providers.py | 52 ++++--- sidecar/app/routers/sessions.py | 47 +++--- sidecar/app/runs/_common.py | 17 ++- sidecar/tests/test_maxturns_finalize.py | 45 ++++-- sidecar/tests/test_model_providers.py | 4 +- sidecar/tests/test_review_fixes.py | 2 +- sidecar/tests/test_skills.py | 2 +- sidecar/tests/test_v0320_fixes.py | 2 +- sidecar/tests/test_v0340_fixes.py | 142 ++++++++++++++++++ 29 files changed, 639 insertions(+), 135 deletions(-) create mode 100644 sidecar/tests/test_v0340_fixes.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 2cba2d0..85703c9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,88 @@ follow semantic versioning once it reaches 1.0. ## [Unreleased] +## [0.34.0] - 2026-07-16 + +_Analysis-engine correctness, lifecycle robustness, and agent de-ossification. A +three-angle audit (agent ossification, DuckDB engine math, state-machine +lifecycle) plus the reviewed third-party findings — bugs that misreported numbers, +wedged state forever, or needlessly boxed the agent in._ + +### Fixed — analysis engine correctness + +- **Status codes and sizes are integers, not floats.** A single unparsed log line + (or a missing inventory size) coerced the whole numeric column to float64 → + DuckDB DOUBLE, so reports showed status codes as `404.0` and object sizes above + 2^53 lost precision (and `total_size` accumulated in DOUBLE). The integer columns + are now built as nullable Int64, preserving both the label and the value. +- **Access-log error rates are "of requests", not "of lines".** `error_rate_4xx/5xx` + and the 206/404/403 shares divided by every ingested line, including + text-fallback rows with no status — diluting the rate by the unparsed fraction + and silently under-reporting errors across the whole [0.5, 1.0) parsed band the + truth guard allows. They now divide by the count of rows that actually parsed a + status code. +- **Inventory average/small-object figures use consistent denominators.** + `average_object_size` is now `total_size / object_count` (so the displayed + total/count/avg triple reconciles), and `small_object_ratio` is computed over + objects that HAVE a size (not diluted by null-size rows). +- **Object age bucketing is timezone-independent.** The DuckDB connection now pins + `TimeZone='UTC'`, so `datediff` against `current_timestamp` no longer lands + objects in the adjacent age bucket when the sidecar runs outside UTC. + +### Fixed — lifecycle robustness + +- **A failed dataset-persist no longer wedges an evidence import in `importing` + forever.** The post-download persistence ran outside any try/except (and had no + startup reconciler, unlike a run), so a DB error there left an import that could + never be re-confirmed or re-run. It now reverts to `failed` and cleans up. +- **The blocking turn path resolves its turn handle on a persist failure**, like + the streaming worker already did — otherwise a commit error left the handle + un-done and non-evictable, stalling every subsequent turn in that session for + 120s until eviction. +- **The loser of a concurrent evidence-import claim fails its orphan run** instead + of leaving a permanent session-unlinked `pending` row. +- **A `report_ready` publish failure can't downgrade a completed run to `failed`**, + and the executor's failure branch refuses to overwrite a terminal state. + +### Fixed — model provider & budget (reviewed findings) + +- **The provider test no longer gives false green/red.** A 404/405 on `/models` + (common on minimal proxies) was reported as a confident pass even though the key + was never verified; and a valid empty `base_url` (which uses the OpenAI default, + exactly like the real client) was flagged "configuration incomplete." The test + now reports "reachable, key unverified" as a caution and treats `base_url` as + optional. +- **The tool-output budget is a hard cap, not a soft one.** A single tool return + was only counted AFTER it landed in context, so one large result could blow past + the per-turn budget. An output that would exceed the remaining budget is now + withheld with a valid JSON "too large — narrow it" envelope. + +### Changed — de-ossification (don't box the agent in) + +- **Custom aggregation gained real expressiveness** without loosening the + whitelist-and-bound-params floor: a SECOND group-by dimension (cross-tabs like + "403s per masked-IP per day"), `day`/`weekday` time buckets, and + `distinct_ips`/`distinct_keys`/`p99`/`min`-`max`-bytes metrics — all fixed SQL + fragments, zero raw-row exposure. Top-N now has a deterministic tiebreaker. +- **Operator-declarable max output tokens** (`max_output_tokens` on a model + provider) clamps the completion budget, so a third-party/unknown model whose real + cap is lower doesn't get a `max_tokens` its endpoint 400s on — symmetric with the + existing context-window override. +- **Agent working-memory recall now scales with the model window** (floored at 50), + like thread replay already did — a long investigation on a large-context model no + longer has its own recorded facts/findings clipped first. The per-turn skill-load + guard was raised (the elastic tool-output budget is the real bound), and the + survey bucket ceiling raised to 2000. +- **The agent is reminded tool-result text is untrusted** — carried from v0.33; no + change here. + +### Fixed — frontend + +- **The 409 blocking-fallback captures its "answer baseline" before the turn + starts**, not from a GET issued after the 409 — closing a race where an answer + persisted in that window poisoned the baseline and the UI hung "running" for ~2.5 + minutes before recovering. + ## [0.33.0] - 2026-07-16 _S3-compatible provider correctness round. The product's core promise — works diff --git a/frontend/src/api.ts b/frontend/src/api.ts index bceffbe..35423f7 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -111,6 +111,9 @@ export interface ModelProviderInput { /** Optional explicit context window (tokens). Overrides the built-in model * table so a new large-context model isn't throttled to the default. */ context_window?: number | null; + /** Optional explicit max output tokens. Clamps the completion budget so a + * third-party/unknown model whose real cap is lower doesn't 400. */ + max_output_tokens?: number | null; } export const listModelProviders = () => diff --git a/frontend/src/hooks/useTurnRunner.ts b/frontend/src/hooks/useTurnRunner.ts index e185d73..61b98b4 100644 --- a/frontend/src/hooks/useTurnRunner.ts +++ b/frontend/src/hooks/useTurnRunner.ts @@ -202,11 +202,15 @@ export function useTurnRunner(opts: { // "ok" once the persisted answer is visible, or "inprogress" if it gives up — // in which case the caller keeps the pending bubble rather than dropping the // user's message (F4). - const waitForPersistedTurn = async (id: string): Promise => { - // The assistant answer for this turn is a NEW assistant message. The 409 - // guarantees it isn't persisted yet, so the first successful fetch is a safe - // baseline; later fetches detect a new assistant id. - let baseline: Set | null = null; + const waitForPersistedTurn = async ( + id: string, + baselinePromise?: Promise | null>, + ): Promise => { + // The assistant answer for this turn is a NEW assistant message. Prefer the + // baseline captured BEFORE the turn started (baselinePromise) — it can't + // include this turn's answer, so it's race-free. Only if that snapshot failed + // do we fall back to capturing from the first successful fetch here. + let baseline: Set | null = baselinePromise ? await baselinePromise : null; const captureOrDetect = (d: SessionDetail): boolean => { const asstIds = d.messages.filter((m) => m.role === "assistant").map((m) => m.id); if (baseline === null) { @@ -215,10 +219,12 @@ export function useTurnRunner(opts: { } return asstIds.some((mid) => !baseline!.has(mid)); }; - try { - captureOrDetect(await getSession(id)); - } catch { - /* the loop retries the fetch; baseline stays null until one succeeds */ + if (baseline === null) { + try { + captureOrDetect(await getSession(id)); + } catch { + /* the loop retries the fetch; baseline stays null until one succeeds */ + } } // Bounded backoff aligned with the server's own turn budget (its blocking // wait is ~150 s), then give up polling — but never drop the message. @@ -263,6 +269,15 @@ export function useTurnRunner(opts: { return; } const turnId = newTurnId(); + // Snapshot the assistant-message ids BEFORE the turn runs (in parallel with + // the stream, so no added latency). The 409 blocking-fallback uses this as its + // "which assistant messages predate this turn" baseline. Capturing it here — + // rather than from a GET issued AFTER the 409 — closes a race where the worker + // persisted the answer in the gap after the 409 but before that GET, poisoning + // the baseline so the new answer was never detected and the UI hung (P2b). + const preTurnAsstIds: Promise | null> = getSession(id) + .then((d) => new Set(d.messages.filter((m) => m.role === "assistant").map((m) => m.id))) + .catch(() => null); patchSessionRun(id, { busy: true, error: null, needKey: false, pending: q, streamText: null, streamTools: [], stopped: false, stalled: false, @@ -328,7 +343,7 @@ export function useTurnRunner(opts: { // The turn is still running server-side (nothing persisted yet). Poll // until this turn's assistant answer is actually persisted, then clear // the pending bubble — never on a fixed timer (F4). - outcome = await waitForPersistedTurn(id); + outcome = await waitForPersistedTurn(id, preTurnAsstIds); if (outcome === "inprogress") { // Gave up waiting, but the turn may still be running (its answer may // already be persisted server-side). Keep the pending bubble — the diff --git a/frontend/src/i18n.tsx b/frontend/src/i18n.tsx index d6406aa..1a690ab 100644 --- a/frontend/src/i18n.tsx +++ b/frontend/src/i18n.tsx @@ -280,6 +280,8 @@ const en: Dict = { "prov.fApiKey": "API key", "prov.fContextWindow": "Context window (tokens, optional)", "prov.hintContextWindow": "Overrides the built-in model table for the agent's depth budgets; leave empty to infer from the model name.", + "prov.fMaxOutput": "Max output tokens (optional)", + "prov.hintMaxOutput": "Clamps the completion budget so a third-party or unknown model whose real cap is lower doesn't reject the request; leave empty to infer from the model name.", "prov.fProvider": "Provider", "prov.fEndpoint": "Endpoint URL", "prov.fRegion": "Region", @@ -310,6 +312,7 @@ const en: Dict = { "prov.confirmDeleteCloud": "Delete cloud provider \"{name}\"?", "prov.testOk": "OK", "prov.testIncomplete": "incomplete", + "prov.testUnverified": "reachable (key unverified)", // command palette "palette.placeholder": "Search chats or run a command…", "palette.newChat": "New chat", @@ -588,6 +591,8 @@ const zh: Dict = { "prov.fApiKey": "API Key", "prov.fContextWindow": "上下文窗口（tokens，可选）", "prov.hintContextWindow": "覆盖内置模型表，用于智能体深度预算；留空则按模型名推断。", + "prov.fMaxOutput": "最大输出 tokens（可选）", + "prov.hintMaxOutput": "限制补全预算，避免第三方或未知模型因真实上限更低而拒绝请求；留空则按模型名推断。", "prov.fProvider": "提供商", "prov.fEndpoint": "Endpoint URL", "prov.fRegion": "区域", @@ -618,6 +623,7 @@ const zh: Dict = { "prov.confirmDeleteCloud": "删除云存储提供商“{name}”？", "prov.testOk": "正常", "prov.testIncomplete": "不完整", + "prov.testUnverified": "可达（密钥未验证）", "palette.placeholder": "搜索对话或运行命令…", "palette.newChat": "新对话", "palette.settings": "打开设置", diff --git a/frontend/src/types.ts b/frontend/src/types.ts index 54d519d..953c6a3 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -8,6 +8,8 @@ export interface ModelProvider { has_api_key: boolean; /** Optional explicit context window (tokens); overrides the built-in model table for the agent's depth budgets. */ context_window: number | null; + /** Optional explicit max output tokens; clamps the completion budget so a lower-cap endpoint doesn't 400. */ + max_output_tokens: number | null; /** True for the provider the agent uses (explicitly activated; otherwise the oldest is the implicit default). */ active: boolean; created_at: string; @@ -41,6 +43,8 @@ export interface ModelProviderTestResult { ok: boolean; checks: Record; detail: string; + /** true=key accepted, false=key rejected, null=reached but key unverified. */ + api_key_verified?: boolean | null; } // --- S3 tool results (Phase 03) --- diff --git a/frontend/src/views/ProvidersView.tsx b/frontend/src/views/ProvidersView.tsx index fc83619..2c75edd 100644 --- a/frontend/src/views/ProvidersView.tsx +++ b/frontend/src/views/ProvidersView.tsx @@ -59,6 +59,7 @@ const emptyModelForm: ModelProviderInput = { model: "", api_key: "", context_window: null, + max_output_tokens: null, }; function ModelProvidersPanel() { @@ -90,6 +91,7 @@ function ModelProvidersPanel() { model: p.model ?? "", api_key: "", // never prefill secrets context_window: p.context_window ?? null, + max_output_tokens: p.max_output_tokens ?? null, }); setEditing(p); setCreating(false); @@ -117,6 +119,11 @@ function ModelProvidersPanel() { } else if (editing && editing.context_window) { body.context_window = 0; // field cleared → 0 tells the API to reset to NULL } + if (form.max_output_tokens && form.max_output_tokens > 0) { + body.max_output_tokens = form.max_output_tokens; + } else if (editing && editing.max_output_tokens) { + body.max_output_tokens = 0; // cleared → 0 resets to NULL + } try { if (editing) await updateModelProvider(editing.id, body); else await createModelProvider(body); @@ -143,7 +150,14 @@ function ModelProvidersPanel() { setStatus(null); try { const r = await testModelProvider(p.id); - setStatus(`${p.name}: ${r.ok ? t("prov.testOk") : t("prov.testIncomplete")} — ${r.detail}`); + // A reachable endpoint whose key couldn't be verified (api_key_verified === + // null) is NOT a clean pass — surface it as a caution, not "OK". + const label = !r.ok + ? t("prov.testIncomplete") + : r.api_key_verified == null + ? t("prov.testUnverified") + : t("prov.testOk"); + setStatus(`${p.name}: ${label} — ${r.detail}`); } catch (e) { setStatus(String(e)); } @@ -195,6 +209,17 @@ function ModelProvidersPanel() { placeholder="1000000" /> + + { + const v = e.target.value.replace(/[^0-9]/g, ""); + setForm({ ...form, max_output_tokens: v ? parseInt(v, 10) : null }); + }} + placeholder="4096" + /> + dict[str, Any]: # Optional operator-declared context window (tokens); None → inferred from # the model name by model_budget. NOT a secret. "context_window": row["context_window"], + # Optional operator-declared max output tokens; None → inferred. NOT a secret. + "max_output_tokens": row["max_output_tokens"], } diff --git a/sidecar/app/agent_runtime/model_budget.py b/sidecar/app/agent_runtime/model_budget.py index 242d82c..dffa453 100644 --- a/sidecar/app/agent_runtime/model_budget.py +++ b/sidecar/app/agent_runtime/model_budget.py @@ -88,9 +88,15 @@ def context_window(model: str | None, explicit: int | None = None) -> int: return _DEFAULT_CONTEXT -def max_output_tokens(model: str | None) -> int: +def max_output_tokens(model: str | None, explicit_max: int | None = None) -> int: """The active model's provider-imposed MAX output tokens (best-effort). Used to - clamp the completion budget so we never send a max_tokens the provider rejects.""" + clamp the completion budget so we never send a max_tokens the provider rejects. + + ``explicit_max`` (an operator-declared cap from the model-provider config) wins + when positive — so a third-party/unknown model whose real ceiling is below the + substring-table default isn't handed a max_tokens its endpoint 400s on.""" + if explicit_max and explicit_max > 0: + return explicit_max m = (model or "").lower() for sub, cap in _MAX_OUTPUT_TOKENS: if sub in m: @@ -107,7 +113,8 @@ def tool_output_char_budget(model: str | None, explicit_window: int | None = Non max(TOOL_OUTPUT_CHARS_FLOOR, tokens * _CHARS_PER_TOKEN)) -def completion_token_budget(model: str | None, explicit_window: int | None = None) -> int: +def completion_token_budget(model: str | None, explicit_window: int | None = None, + explicit_max: int | None = None) -> int: """Completion (max_tokens) budget: raised only where the window clearly supports it, floored at the historical value, capped by the module ceiling AND by the model's real provider max-output so we never trigger a 400. @@ -118,4 +125,4 @@ def completion_token_budget(model: str | None, explicit_window: int | None = Non rather than being handed the 16384 floor it would reject.""" scaled = max(COMPLETION_TOKENS_FLOOR, min(context_window(model, explicit_window) // 8, COMPLETION_TOKENS_CEILING)) - return min(scaled, max_output_tokens(model)) + return min(scaled, max_output_tokens(model, explicit_max)) diff --git a/sidecar/app/agent_runtime/session_action_tools.py b/sidecar/app/agent_runtime/session_action_tools.py index 9954137..b950283 100644 --- a/sidecar/app/agent_runtime/session_action_tools.py +++ b/sidecar/app/agent_runtime/session_action_tools.py @@ -204,7 +204,7 @@ def review_bucket_config(provider_id: str, bucket: str) -> str: @function_tool def survey_account(provider_id: str, max_buckets: int = 0) -> str: - """Read-only account survey: enumerate visible buckets and detect evidence sources (access logs, inventory) across the account, persisting a profile the evidence-import flow can use. This is the COSTLY account tool — it makes live S3 calls across every visible bucket. Prefer the cheap persisted-profile readers when they can answer: query_account_profile for cross-bucket posture ("which buckets are public / unencrypted / no lifecycle?") and compare_to_last_survey for "what changed" both read the LAST survey with no new S3 calls. Run this only to establish a first profile or deliberately refresh a stale one. Returns a compact summary (counts + summary + public-exposure note), not raw key lists, for you to narrate. Does NOT surface a separate card. Use only when the user's request is about the account/buckets — NOT for local-file analysis or unrelated questions. The result includes has_prior_survey — when true, call compare_to_last_survey next to report what changed. max_buckets (optional, 1-500) raises the per-survey bucket cap for large accounts (default 100); the result's truncated flag tells you if buckets were left out. Args: provider_id; max_buckets?.""" + """Read-only account survey: enumerate visible buckets and detect evidence sources (access logs, inventory) across the account, persisting a profile the evidence-import flow can use. This is the COSTLY account tool — it makes live S3 calls across every visible bucket. Prefer the cheap persisted-profile readers when they can answer: query_account_profile for cross-bucket posture ("which buckets are public / unencrypted / no lifecycle?") and compare_to_last_survey for "what changed" both read the LAST survey with no new S3 calls. Run this only to establish a first profile or deliberately refresh a stale one. Returns a compact summary (counts + summary + public-exposure note), not raw key lists, for you to narrate. Does NOT surface a separate card. Use only when the user's request is about the account/buckets — NOT for local-file analysis or unrelated questions. The result includes has_prior_survey — when true, call compare_to_last_survey next to report what changed. max_buckets (optional, 1-2000) raises the per-survey bucket cap for large accounts (default 100); the result's truncated flag tells you if buckets were left out. Args: provider_id; max_buckets?.""" p = provider(provider_id) if p is None: return _err("Unknown provider_id. Use a configured provider.") @@ -213,7 +213,7 @@ def survey_account(provider_id: str, max_buckets: int = 0) -> str: had_prior = bool(account_repo.recent_run_ids_for_provider(conn, provider_id, 1)) start("survey_account", provider_name(provider_id)) try: - mb = max(1, min(int(max_buckets), 500)) if max_buckets else None + mb = max(1, min(int(max_buckets), 2000)) if max_buckets else None body = RunCreate(run_type="account_discovery", provider_id=provider_id, user_prompt=_DEFAULT_PROMPTS["account_discovery"], session_id=session_id, max_buckets=mb) diff --git a/sidecar/app/agent_runtime/session_agent.py b/sidecar/app/agent_runtime/session_agent.py index 67e305d..4db086a 100644 --- a/sidecar/app/agent_runtime/session_agent.py +++ b/sidecar/app/agent_runtime/session_agent.py @@ -42,6 +42,17 @@ _MAX_FACTS = 50 # kept in sync with sessions.summary_builder.MAX_FACTS _MAX_FINDINGS = 50 # kept in sync with sessions.summary_builder.MAX_FINDINGS (was 30 — drift) +_MEM_RECALL_CEIL = 400 # upper bound on the elastic agent-memory recall + + +def _elastic_memory_cap(model: str | None, explicit_window: int | None) -> int: + """How many of the agent's OWN recorded facts/findings/questions to recall, + scaled to the model window (floored at 50) — the same de-ossification the + thread replay uses. On a long investigation with a large-context model, the + agent's durable memory was the first thing clipped at a hard 50.""" + window = model_budget.context_window(model, explicit_window) + factor = max(1, window // 128_000) + return min(_MEM_RECALL_CEIL, _MAX_FACTS * factor) # How many recent thread messages the agent sees. 24 (was 12): a small-context # clip that now just makes the agent lose the thread on a long investigation — # 24 msgs × _MAX_REPLAY_MSG chars is still tiny under a modern context window. @@ -117,6 +128,11 @@ "This turn's tool-output budget is used up — synthesize your findings from what " "you've already gathered and answer now. This budget resets if the user continues." ) +_TOOL_OUTPUT_TOO_LARGE = ( + "This result was too large for the turn's remaining context budget and was " + "withheld. Narrow it — a smaller max_keys, a prefix, or a filter — and call " + "again, or answer from what you already have." +) # Memory tools stay usable even after the budget is spent: recording a finding # is how the model synthesizes, and their outputs are a few bytes. _BUDGET_EXEMPT_TOOLS = { @@ -217,11 +233,13 @@ ) -def _build_agent_memory_block(memory: list[dict[str, Any]] | None) -> dict[str, list[Any]]: +def _build_agent_memory_block(memory: list[dict[str, Any]] | None, + cap: int = _MAX_FACTS) -> dict[str, list[Any]]: """Group agent-authored memory into recalled facts/findings/questions. - ``memory`` is oldest-first; we keep the most RECENT items per kind (the tail) - so a long session surfaces its latest learnings rather than stale early ones. + ``memory`` is oldest-first; we keep the most RECENT ``cap`` items per kind + (the tail) so a long session surfaces its latest learnings rather than stale + early ones. ``cap`` scales with the model window (see _elastic_memory_cap). Each item carries its id so the agent can update/resolve it later. """ facts: list[dict[str, Any]] = [] @@ -242,9 +260,9 @@ def _build_agent_memory_block(memory: list[dict[str, Any]] | None) -> dict[str, elif kind == "open_question": questions.append({"id": mem_id, "text": text}) return { - "recorded_facts": facts[-_MAX_FACTS:], - "recorded_findings": findings[-_MAX_FINDINGS:], - "open_questions": questions[-_MAX_FACTS:], + "recorded_facts": facts[-cap:], + "recorded_findings": findings[-cap:], + "open_questions": questions[-cap:], } @@ -346,7 +364,8 @@ def build_session_context( }, # Things YOU recorded in earlier turns of this session (via note_fact / # record_finding / note_open_question). Reuse them; don't re-derive. - "agent_memory": _build_agent_memory_block(agent_memory), + "agent_memory": _build_agent_memory_block( + agent_memory, cap=_elastic_memory_cap(model, explicit_window)), # Prior assistant turns carry a `tools_run` trace of the read-only probes # they already ran (bounded) — so this turn sees what was checked and # re-fetches only what it needs fuller detail on, instead of re-probing. @@ -371,7 +390,8 @@ def _make_agent(creds: dict[str, Any], tools: list[Any], instructions: str, # a small one needs. Never below the floor, never above provider max-output. return build_agent(creds, tools, instructions, name="Storage Agent", max_tokens=model_budget.completion_token_budget( - creds.get("model"), creds.get("context_window")), + creds.get("model"), creds.get("context_window"), + creds.get("max_output_tokens")), parallel_tool_calls=False, client_registry=client_registry) @@ -577,8 +597,18 @@ async def wrapped(ctx: Any, args: Any) -> Any: return json.dumps({"status": "budget_exhausted", "next_step": _TOOL_BUDGET_EXHAUSTED}) out = await _orig(ctx, args) - spent["chars"] += len(str(out or "")) - return out + text = str(out or "") + if spent["chars"] + len(text) > limit: + # This SINGLE output would push the turn past its context + # budget. Counting-after made the budget a soft post-hoc bound + # a single large tool return could blow past; withhold it with + # a VALID JSON envelope (truncating the JSON would be + # unparseable) and flag the turn so the driver offers 'continue'. + spent["exhausted"] = True + text = json.dumps({"status": "output_too_large", + "next_step": _TOOL_OUTPUT_TOO_LARGE}) + spent["chars"] += len(text) + return text return wrapped try: @@ -608,7 +638,8 @@ def _build_prompt( if conn is not None and session.get("id"): try: from ..repositories import sessions as sessions_repo - agent_memory = sessions_repo.list_agent_memory(conn, session["id"]) + agent_memory = sessions_repo.list_agent_memory( + conn, session["id"], limit=_elastic_memory_cap(model, explicit_window)) except Exception: # noqa: BLE001 agent_memory = [] context = build_session_context(session, summary, recent_messages, agent_memory, diff --git a/sidecar/app/agent_runtime/session_analysis_tools.py b/sidecar/app/agent_runtime/session_analysis_tools.py index 7295595..e0d65a6 100644 --- a/sidecar/app/agent_runtime/session_analysis_tools.py +++ b/sidecar/app/agent_runtime/session_analysis_tools.py @@ -205,12 +205,13 @@ def aggregate_uploaded_file( dataset_id: str, metric: str, group_by: str = "", + group_by_2: str = "", filters_json: str = "", status_min: int = -1, status_max: int = -1, limit: int = 20, ) -> str: - """Run ONE custom aggregation over an uploaded file when the fixed analyze_uploaded_file metrics don't answer the user's question (e.g. "which masked IP got the most 403s between status 400-499", "total bytes per storage class"). You choose metric + group_by + equality filters from a whitelist; raw rows and arbitrary SQL are never available. access_log metrics: count, sum_bytes, avg_bytes, avg_latency_ms, p50_latency_ms, p95_latency_ms, max_latency_ms; group_by: status_code, method, key, path, prefix, user_agent, client_ip_masked, error_code, hour. inventory metrics: count, total_size, avg_size, max_size, min_size; group_by: bucket, prefix, storage_class. filters_json: optional JSON object of column->value equality filters (same columns as group_by, except hour). status_min/status_max: optional status-code range (access logs; pass -1 to skip). limit: max groups returned (<=50); a "truncated": true means more groups exist. Args: dataset_id (from list_uploaded_files), metric, group_by (empty for a single scalar), filters_json, status_min, status_max, limit.""" + """Run ONE custom aggregation over an uploaded file when the fixed analyze_uploaded_file metrics don't answer the user's question (e.g. "which masked IP got the most 403s between status 400-499", "403s per masked-IP per day", "total bytes per storage class"). You choose metric + up to TWO group-by dimensions + equality filters from a whitelist; raw rows and arbitrary SQL are never available. access_log metrics: count, sum_bytes, avg_bytes, min_bytes, max_bytes, avg_latency_ms, p50/p95/p99_latency_ms, max_latency_ms, distinct_ips, distinct_keys; group_by: status_code, method, key, path, prefix, user_agent, client_ip_masked, error_code, hour, day, weekday. inventory metrics: count, total_size, avg_size, max_size, min_size, distinct_prefixes, distinct_storage_classes; group_by: bucket, prefix, storage_class. group_by_2 (optional) adds a second dimension for a cross-tab (e.g. group_by=client_ip_masked, group_by_2=day); the returned group label joins the two with " · ". filters_json: optional JSON object of column->value equality filters (same columns as group_by, except derived hour/day/weekday). status_min/status_max: optional status-code range (access logs; pass -1 to skip). limit: max groups returned (<=50); a "truncated": true means more groups exist. Args: dataset_id (from list_uploaded_files), metric, group_by (empty for a single scalar), group_by_2 (optional), filters_json, status_min, status_max, limit.""" ds = ds_repo.get(conn, dataset_id) if ds is None or ds.get("session_id") != session_id: return _err("Unknown dataset_id for this session. Call list_uploaded_files first.") @@ -231,7 +232,7 @@ def aggregate_uploaded_file( duckdb_abs, _imp, _detected = _ensure_imported(ds) out = agg.aggregate( duckdb_abs, ds["dataset_type"], metric, - group_by=group_by or None, filters=filters, + group_by=group_by or None, group_by_2=group_by_2 or None, filters=filters, status_min=None if status_min < 0 else status_min, status_max=None if status_max < 0 else status_max, limit=limit, diff --git a/sidecar/app/agent_runtime/session_tools.py b/sidecar/app/agent_runtime/session_tools.py index d1285ea..426731d 100644 --- a/sidecar/app/agent_runtime/session_tools.py +++ b/sidecar/app/agent_runtime/session_tools.py @@ -86,12 +86,14 @@ def scope_denial(p, bucket: str, *, key: str | None = None, return check_scope(p.allowed_buckets, p.allowed_prefixes, bucket, key=key, prefix=prefix, listing=listing) - # Per-turn budget: cap how many skill bodies the agent can load in one turn, - # so a runaway loop can't pull every skill (~8000 chars each) into context. - # 10 (was 8/6): a cross-domain investigation legitimately spans several skills; - # keep the cap above what a real diagnosis needs, below "load everything". + # Per-turn budget: a runaway-loop guard on skill-body loads, NOT the real + # context bound — read_skill output is NOT budget-exempt, so each ~8000-char + # skill body already counts against the model-elastic tool-output budget + # (200k floor → up to 1M on a large-context model). So this stays a fixed + # guard, raised to 20 (was 10) so a legitimately cross-domain investigation on + # a large model isn't clipped below what the elastic byte budget would allow. skill_loads = {"n": 0} - _MAX_SKILL_LOADS = 10 + _MAX_SKILL_LOADS = 20 # Per-turn object-preview budget: preview_object reads bounded object CONTENT # (unlike the metadata-only probes), so bound it in code — a handful of small diff --git a/sidecar/app/analysis/access_logs.py b/sidecar/app/analysis/access_logs.py index 327bf5b..3a9da93 100644 --- a/sidecar/app/analysis/access_logs.py +++ b/sidecar/app/analysis/access_logs.py @@ -376,6 +376,13 @@ def import_access_logs(raw_path: str | Path, duckdb_path: str | Path, fmt: str) truncated = len(rows) >= MAX_INGEST_ROWS df = pd.DataFrame(rows, columns=COLUMNS) + # Integer columns MUST be built as nullable Int64 from the raw row values. + # `pd.DataFrame(list-of-dicts)` coerces any column containing a None to + # float64 → DuckDB infers DOUBLE → status codes render as "404.0" in reports + # and int64 sizes above 2^53 lose precision. Building from the Python ints + # (not the already-floated column) preserves both the label and the value. + for col in ("status_code", "bytes_sent", "latency_ms"): + df[col] = pd.array([r.get(col) for r in rows], dtype="Int64") con = duck.connect(duckdb_path) try: con.register("incoming", df) @@ -405,11 +412,22 @@ def analyze_access_logs(duckdb_path: str | Path) -> dict[str, Any]: if total == 0: return {"total_requests": 0} + # Rates are "of the actual HTTP requests", so the denominator is the count + # of rows that PARSED a status code — not every ingested line. Using + # count(*) (which includes text-fallback rows with a null status) diluted + # every rate by the unparsed fraction, silently under-reporting error rates + # across the whole [0.5, 1.0) parsed-fraction band the truth guard permits. + status_total = con.execute( + f"SELECT count(*) FROM {TABLE_NAME} WHERE status_code IS NOT NULL" + ).fetchone()[0] + def rate(lo: int, hi: int) -> float: + if not status_total: + return 0.0 n = con.execute( f"SELECT count(*) FROM {TABLE_NAME} WHERE status_code >= {lo} AND status_code <= {hi}" ).fetchone()[0] - return round(n / total, 4) + return round(n / status_total, 4) status_dist = _dist(con, f"SELECT status_code, count(*) c FROM {TABLE_NAME} GROUP BY status_code ORDER BY c DESC") method_dist = _dist(con, f"SELECT method, count(*) c FROM {TABLE_NAME} GROUP BY method ORDER BY c DESC") @@ -448,9 +466,9 @@ def rate(lo: int, hi: int) -> float: "top_user_agents": top_uas, "error_rate_4xx": rate(400, 499), "error_rate_5xx": rate(500, 599), - "range_share_206": round(n_206 / total, 4), - "share_404": round(n_404 / total, 4), - "share_403": round(n_403 / total, 4), + "range_share_206": round(n_206 / (status_total or 1), 4), + "share_404": round(n_404 / (status_total or 1), 4), + "share_403": round(n_403 / (status_total or 1), 4), } finally: con.close() diff --git a/sidecar/app/analysis/aggregate.py b/sidecar/app/analysis/aggregate.py index 162e106..45eeb61 100644 --- a/sidecar/app/analysis/aggregate.py +++ b/sidecar/app/analysis/aggregate.py @@ -35,10 +35,16 @@ # Whitelisted GROUP BY dimensions per dataset type. Values are the exact SQL # expressions used — constants defined HERE, never caller input. "hour" mirrors # the derivation used by the fixed metric set. -_HOUR_EXPR = ( - "CASE WHEN try_cast(timestamp AS TIMESTAMP) IS NULL THEN 'unknown' " - "ELSE strftime(try_cast(timestamp AS TIMESTAMP), '%Y-%m-%dT%H:00') END" -) +def _ts_bucket(fmt: str) -> str: + return (f"CASE WHEN try_cast(timestamp AS TIMESTAMP) IS NULL THEN 'unknown' " + f"ELSE strftime(try_cast(timestamp AS TIMESTAMP), '{fmt}') END") + + +_HOUR_EXPR = _ts_bucket("%Y-%m-%dT%H:00") +# day / weekday let a multi-week log be bucketed at a coarser grain than hour. +# weekday sorts Sun..Sat via the leading %w digit; the name follows for the label. +_DAY_EXPR = _ts_bucket("%Y-%m-%d") +_WEEKDAY_EXPR = _ts_bucket("%w %A") _DIMENSIONS: dict[str, dict[str, str]] = { "access_log": { "status_code": "status_code", @@ -50,6 +56,8 @@ "client_ip_masked": "client_ip_masked", "error_code": "error_code", "hour": _HOUR_EXPR, + "day": _DAY_EXPR, + "weekday": _WEEKDAY_EXPR, }, "inventory": { "bucket": "bucket", @@ -64,10 +72,15 @@ "count": "count(*)", "sum_bytes": "sum(bytes_sent)", "avg_bytes": "avg(bytes_sent)", + "min_bytes": "min(bytes_sent)", + "max_bytes": "max(bytes_sent)", "avg_latency_ms": "avg(latency_ms)", "p50_latency_ms": "quantile_cont(latency_ms, 0.5)", "p95_latency_ms": "quantile_cont(latency_ms, 0.95)", + "p99_latency_ms": "quantile_cont(latency_ms, 0.99)", "max_latency_ms": "max(latency_ms)", + "distinct_ips": "count(DISTINCT client_ip_masked)", + "distinct_keys": "count(DISTINCT key)", }, "inventory": { "count": "count(*)", @@ -75,6 +88,8 @@ "avg_size": "avg(size)", "max_size": "max(size)", "min_size": "min(size)", + "distinct_prefixes": "count(DISTINCT prefix)", + "distinct_storage_classes": "count(DISTINCT storage_class)", }, } @@ -106,6 +121,7 @@ def aggregate( dataset_type: str, metric: str, group_by: str | None = None, + group_by_2: str | None = None, filters: dict[str, Any] | None = None, status_min: int | None = None, status_max: int | None = None, @@ -158,25 +174,46 @@ def aggregate( "value": _num(value), "truncated": False, } _require("group_by", group_by, _DIMENSIONS[dataset_type]) - dim_sql = _DIMENSIONS[dataset_type][group_by] - if group_by in _KEYLIKE_DIMENSIONS: + # Optional SECOND dimension: e.g. "403s per masked-IP per hour". Both + # identifiers resolve only from the whitelist (never caller text), so the + # composite GROUP BY carries zero injection surface — only the vocabulary + # widened, not the mechanism. + dims = [group_by] + if group_by_2: + _require("group_by (2)", group_by_2, _DIMENSIONS[dataset_type]) + if group_by_2 != group_by: + dims.append(group_by_2) + dim_exprs = [_DIMENSIONS[dataset_type][d] for d in dims] + if any(d in _KEYLIKE_DIMENSIONS for d in dims): limit = min(limit, DEFAULT_GROUPS) # rule 16: don't stream 50 raw keys + select_dims = ", ".join(f"{expr} AS g{i}" for i, expr in enumerate(dim_exprs)) + group_keys = ", ".join(f"g{i}" for i in range(len(dim_exprs))) + # Deterministic ordering: primary metric DESC, then the group keys, so ties + # don't shuffle which groups appear (or flip `truncated`) run-to-run. + order_keys = ", ".join(f"g{i}" for i in range(len(dim_exprs))) # Fetch limit+1 to report (not silently drop) an over-limit tail. sql = ( - f"SELECT {dim_sql} AS g, {metric_sql} AS v FROM {table}{where_sql} " - f"GROUP BY g ORDER BY v DESC NULLS LAST LIMIT {limit + 1}" + f"SELECT {select_dims}, {metric_sql} AS v FROM {table}{where_sql} " + f"GROUP BY {group_keys} ORDER BY v DESC NULLS LAST, {order_keys} LIMIT {limit + 1}" ) rows = con.execute(sql, params).fetchall() finally: con.close() + n_dims = len(dim_exprs) truncated = len(rows) > limit + + def _label(row) -> str: + parts = [redact_text(str(x))[:_LABEL_LEN] for x in row[:n_dims]] + return " · ".join(parts) + groups = [ - {"group": redact_text(str(g))[:_LABEL_LEN], "value": _num(v)} - for g, v in rows[:limit] + {"group": _label(row), "value": _num(row[n_dims])} + for row in rows[:limit] ] return { - "sql": sql, "params": list(params), "metric": metric, "group_by": group_by, + "sql": sql, "params": list(params), "metric": metric, + "group_by": group_by, "group_by_2": group_by_2 if len(dims) > 1 else None, "groups": groups, "truncated": truncated, } diff --git a/sidecar/app/analysis/duck.py b/sidecar/app/analysis/duck.py index 23da26a..3dfc5df 100644 --- a/sidecar/app/analysis/duck.py +++ b/sidecar/app/analysis/duck.py @@ -7,6 +7,17 @@ import duckdb +def _configure(con: duckdb.DuckDBPyConnection) -> duckdb.DuckDBPyConnection: + # Pin the session timezone to UTC. Timestamps are normalized to naive-UTC at + # ingest, but `current_timestamp` is TIMESTAMP WITH TIME ZONE, and DuckDB casts + # a naive value against it using the session TimeZone — so an age `datediff` + # (inventory `_AGE_CASE`) would land objects in the wrong age bucket by up to a + # day when the sidecar runs outside UTC. UTC keeps every comparison consistent + # with how the data was stored. + con.execute("SET TimeZone='UTC'") + return con + + def connect(duckdb_path: str | Path, read_only: bool = False) -> duckdb.DuckDBPyConnection: """Open (creating parent dirs) a DuckDB connection at ``duckdb_path``. @@ -18,7 +29,7 @@ def connect(duckdb_path: str | Path, read_only: bool = False) -> duckdb.DuckDBPy if read_only: if path.exists(): try: - return duckdb.connect(str(path), read_only=True) + return _configure(duckdb.connect(str(path), read_only=True)) except duckdb.IOException as exc: # A concurrent import holds the write lock. Surface a friendly, # actionable message instead of the raw lock IOException. @@ -31,4 +42,4 @@ def connect(duckdb_path: str | Path, read_only: bool = False) -> duckdb.DuckDBPy # No DB yet — nothing to read; open writable so the caller gets an empty DB # rather than a hard error. path.parent.mkdir(parents=True, exist_ok=True) - return duckdb.connect(str(path)) + return _configure(duckdb.connect(str(path))) diff --git a/sidecar/app/analysis/inventory.py b/sidecar/app/analysis/inventory.py index cdb7764..9935dd3 100644 --- a/sidecar/app/analysis/inventory.py +++ b/sidecar/app/analysis/inventory.py @@ -250,6 +250,12 @@ def series_for(field: str) -> pd.Series: "storage_class": series_for("storage_class"), "etag": series_for("etag"), }, columns=COLUMNS) + # size MUST be nullable Int64, not float64. `_to_int` deliberately parses as + # int to preserve int64 precision above 2^53, but a column that mixes ints + # with None coerces to float64 → DuckDB DOUBLE, which reintroduces the very + # precision loss (and sum(size) then accumulates capacity in DOUBLE). The + # mapped Series is object dtype, so astype("Int64") keeps the exact ints. + df["size"] = df["size"].astype("Int64") con = duck.connect(duckdb_path) try: con.register("incoming", df) @@ -300,7 +306,13 @@ def analyze_inventory(duckdb_path: str | Path) -> dict[str, Any]: return {"object_count": 0} total_size = con.execute(f"SELECT COALESCE(sum(size), 0) FROM {TABLE_NAME}").fetchone()[0] - avg_size = con.execute(f"SELECT COALESCE(avg(size), 0) FROM {TABLE_NAME}").fetchone()[0] + # Objects that actually carry a size — the honest denominator for the + # size-based ratios. Using the all-rows count diluted small_object_ratio + # by null-size rows (never in the numerator), and SQL avg(size) (÷ non-null) + # disagreed with the report's own total_size ÷ object_count pairing. + known_size = con.execute( + f"SELECT count(*) FROM {TABLE_NAME} WHERE size IS NOT NULL" + ).fetchone()[0] size_hist = [ {"bucket": b, "count": int(c)} @@ -347,12 +359,16 @@ def analyze_inventory(duckdb_path: str | Path) -> dict[str, Any]: return { "object_count": int(count), "total_size": int(total_size), - "average_object_size": int(avg_size), + # total ÷ ALL objects (null sizes counted as 0) so the displayed + # total_size / object_count / average triple reconciles; the + # unknown_size_ratio below tells the reader how many were size-less. + "average_object_size": int(total_size / count), "size_histogram": size_hist, "prefix_distribution": prefix_dist, "object_age_distribution": age_dist, "storage_class_distribution": storage_dist, - "small_object_ratio": round(small_n / count, 4), + # ratio among objects that HAVE a size (not diluted by null-size rows). + "small_object_ratio": round(small_n / known_size, 4) if known_size else 0.0, "top_large_objects": top_large, "unknown_age_ratio": round(unknown_age / count, 4), "unknown_size_ratio": round(unknown_size / count, 4), diff --git a/sidecar/app/migrations.py b/sidecar/app/migrations.py index f2b42ce..9ee4b5d 100644 --- a/sidecar/app/migrations.py +++ b/sidecar/app/migrations.py @@ -538,6 +538,16 @@ ALTER TABLE model_providers ADD COLUMN context_window INTEGER; """ +# Optional operator-declared MAX OUTPUT tokens for a model provider. The +# completion budget is clamped to the model's provider max-output (a substring +# table) so we never request a max_tokens the endpoint 400s on; but an unknown or +# third-party-compatible model falls to a default that some endpoints reject. This +# lets an operator declare the real cap. NULL → use the table. Mirrors +# context_window (migration 017). +_M019 = """ +ALTER TABLE model_providers ADD COLUMN max_output_tokens INTEGER; +""" + # Indexes on created_at for the startup retention prune (data_maintenance): # audit_logs and ad-hoc (run_id IS NULL) tool_calls are aged out by created_at, # and both tables grow to the point where a full scan per boot would be costly. @@ -566,6 +576,7 @@ (16, "session_message_grounding", _M016), (17, "model_provider_context_window", _M017), (18, "retention_indexes", _M018), + (19, "model_provider_max_output", _M019), ] diff --git a/sidecar/app/models/schemas.py b/sidecar/app/models/schemas.py index 3fb94c1..8edbb29 100644 --- a/sidecar/app/models/schemas.py +++ b/sidecar/app/models/schemas.py @@ -28,6 +28,10 @@ class ModelProviderCreate(BaseModel): # a newly-shipped large-context model isn't throttled to the default. Omit to # let the agent infer the window from the model name. context_window: int | None = Field(default=None, gt=0) + # Optional explicit MAX OUTPUT tokens. Clamps the completion budget so a + # third-party/unknown model whose real cap is below the default doesn't get a + # max_tokens the endpoint 400s on. Omit to infer from the model name. + max_output_tokens: int | None = Field(default=None, gt=0) class ModelProviderUpdate(BaseModel): @@ -40,6 +44,7 @@ class ModelProviderUpdate(BaseModel): # None = keep as-is; 0 = CLEAR back to NULL (infer from the model name); # positive = set. (None can't mean "clear" here — it's the "unchanged" sentinel.) context_window: int | None = Field(default=None, ge=0) + max_output_tokens: int | None = Field(default=None, ge=0) # None keep / 0 clear / +set class ModelProviderOut(BaseModel): @@ -51,6 +56,7 @@ class ModelProviderOut(BaseModel): api_key_ref: str | None has_api_key: bool context_window: int | None = None + max_output_tokens: int | None = None # True for the provider the agent actually uses. Selected explicitly via # POST /model-providers/{id}/activate; when none is selected the oldest # configured provider is the implicit default (matching the agent runtime). @@ -120,6 +126,10 @@ class ModelProviderTestResult(BaseModel): ok: bool checks: dict[str, bool] detail: str + # True = key accepted (HTTP 200); False = key rejected (401/403); None = the + # endpoint was reached but couldn't verify the key (no /models, e.g. a minimal + # proxy). Lets the UI show "reachable but unverified" instead of a false green. + api_key_verified: bool | None = None # --- S3 tool request bodies -------------------------------------- diff --git a/sidecar/app/repositories/model_providers.py b/sidecar/app/repositories/model_providers.py index 61fef3d..2767a7b 100644 --- a/sidecar/app/repositories/model_providers.py +++ b/sidecar/app/repositories/model_providers.py @@ -74,6 +74,7 @@ def _row_to_out(row: sqlite3.Row, active_id: str | None = None) -> ModelProvider api_key_ref=row["api_key_ref"], has_api_key=keyring_store.secret_exists(row["api_key_ref"]), context_window=row["context_window"], + max_output_tokens=row["max_output_tokens"], active=(row["id"] == active_id), created_at=row["created_at"], updated_at=row["updated_at"], @@ -108,8 +109,8 @@ def create(conn: sqlite3.Connection, data: ModelProviderCreate) -> ModelProvider conn.execute( "INSERT INTO model_providers " "(id, name, provider_type, base_url, model, api_key_ref, context_window, " - " created_at, updated_at) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + " max_output_tokens, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ( provider_id, data.name, @@ -118,6 +119,7 @@ def create(conn: sqlite3.Connection, data: ModelProviderCreate) -> ModelProvider data.model, api_key_ref, data.context_window, + data.max_output_tokens, now, now, ), @@ -160,6 +162,12 @@ def _coalesce(new_val, old_val): context_window = None else: context_window = data.context_window + if data.max_output_tokens is None: + max_output_tokens = existing["max_output_tokens"] + elif data.max_output_tokens == 0: + max_output_tokens = None + else: + max_output_tokens = data.max_output_tokens api_key_ref = existing["api_key_ref"] if has_value(data.api_key): @@ -169,9 +177,9 @@ def _coalesce(new_val, old_val): conn.execute( "UPDATE model_providers SET name=?, provider_type=?, base_url=?, model=?, " - "api_key_ref=?, context_window=?, updated_at=? WHERE id=?", + "api_key_ref=?, context_window=?, max_output_tokens=?, updated_at=? WHERE id=?", (name, provider_type, base_url, model, api_key_ref, context_window, - utcnow(), provider_id), + max_output_tokens, utcnow(), provider_id), ) audit.record( conn, diff --git a/sidecar/app/routers/evidence_imports.py b/sidecar/app/routers/evidence_imports.py index 5e54159..9bd19b5 100644 --- a/sidecar/app/routers/evidence_imports.py +++ b/sidecar/app/routers/evidence_imports.py @@ -12,6 +12,7 @@ from __future__ import annotations import json +import shutil import sqlite3 import uuid from typing import Any @@ -204,9 +205,13 @@ def run_import(import_id: str, conn: sqlite3.Connection = Depends(get_conn)): ) # Atomically claim the confirmed→importing transition. Two concurrent runs # would otherwise both pass the status check above and double-download; the - # single conditional UPDATE lets exactly one win. The loser's just-created - # analysis run stays a harmless unstarted 'pending' row. + # single conditional UPDATE lets exactly one win. if not repo.claim_for_import(conn, import_id, analysis_run_id): + # The loser must fail its just-created run, or it lingers 'pending' forever + # (session-unlinked, only reconciled at the next restart). + runs_repo.set_status(conn, analysis_run_id, "failed", + final_summary="Superseded by a concurrent import of the same evidence.") + conn.commit() current = repo.get(conn, import_id) raise HTTPException( status_code=409, @@ -226,8 +231,6 @@ def run_import(import_id: str, conn: sqlite3.Connection = Depends(get_conn)): files, data["max_files"], data["max_bytes"], dest_dir, ) except Exception as exc: # noqa: BLE001 - any download/combine failure is sanitized + surfaced as 400 - import shutil - repo.set_status(conn, import_id, "failed") repo.mark_files(conn, import_id, "failed") # The combined file is written directly at its final path (no temp-rename), @@ -242,19 +245,36 @@ def run_import(import_id: str, conn: sqlite3.Connection = Depends(get_conn)): conn.commit() raise HTTPException(status_code=400, detail=f"evidence download failed: {redact_text(str(exc))}") - stored_rel = config.rel_path(combined) - datasets_repo.create( - conn, analysis_run_id, dataset_type, - name="managed_evidence_import", source_filename=redact_text(label), - stored_path_rel=stored_rel, - ) - repo.mark_files(conn, import_id, "downloaded") - repo.set_status(conn, import_id, "imported") - audit.record(conn, "evidence_import.download", - {"import_id": import_id, "downloaded_files": len(files), - "downloaded_bytes": total, "stored_path": stored_rel, - "analysis_run_id": analysis_run_id}, run_id=analysis_run_id) - conn.commit() + # The download succeeded; persist the dataset + flip the import to 'imported'. + # This block MUST be guarded too: if datasets_repo.create / rel_path / a commit + # raises here, the import was left 'importing' — a terminal-ish state that can + # never be re-confirmed or re-run (and, unlike the run, has no startup + # reconciler), so it would wedge forever. On failure, revert it to 'failed' + # (mirroring the download-failure branch) and clean up. + try: + stored_rel = config.rel_path(combined) + datasets_repo.create( + conn, analysis_run_id, dataset_type, + name="managed_evidence_import", source_filename=redact_text(label), + stored_path_rel=stored_rel, + ) + repo.mark_files(conn, import_id, "downloaded") + repo.set_status(conn, import_id, "imported") + audit.record(conn, "evidence_import.download", + {"import_id": import_id, "downloaded_files": len(files), + "downloaded_bytes": total, "stored_path": stored_rel, + "analysis_run_id": analysis_run_id}, run_id=analysis_run_id) + conn.commit() + except Exception as exc: # noqa: BLE001 - persist failure must not wedge the import + repo.set_status(conn, import_id, "failed") + repo.mark_files(conn, import_id, "failed") + shutil.rmtree(dest_dir, ignore_errors=True) + runs_repo.set_status(conn, analysis_run_id, "failed", + final_summary="Evidence import failed while persisting the dataset.") + audit.record(conn, "evidence_import.failed", + {"import_id": import_id, "error": redact_text(str(exc))}, run_id=None) + conn.commit() + raise HTTPException(status_code=500, detail=f"evidence import failed: {redact_text(str(exc))}") # Hand off to the existing deterministic analysis executor. run_service.start(analysis_run_id) diff --git a/sidecar/app/routers/model_providers.py b/sidecar/app/routers/model_providers.py index 2a0723a..3ba2492 100644 --- a/sidecar/app/routers/model_providers.py +++ b/sidecar/app/routers/model_providers.py @@ -90,46 +90,58 @@ def test_model_provider( scope, name = keyring_store.parse_ref(provider.api_key_ref) secret = keyring_store.get_secret(scope, name) + # base_url is OPTIONAL: an empty base_url makes the real agent client + # (agent_service.build_agent) use the OpenAI default endpoint, so it's a VALID + # config — flagging it "incomplete" was a false negative. Only model + key are + # actually required; base_url is informational and the probe falls back to the + # OpenAI default. checks = { "has_base_url": bool(provider.base_url), "has_model": bool(provider.model), "api_key_present": secret is not None, } - config_ok = all(checks.values()) - if not config_ok: + required = ("has_model", "api_key_present") + if not all(checks[k] for k in required): return ModelProviderTestResult( - ok=False, checks=checks, + ok=False, checks=checks, api_key_verified=None, detail="Configuration incomplete: " - + ", ".join(k for k, v in checks.items() if not v)) + + ", ".join(k for k in required if not checks[k])) # Live probe. /models is the standard OpenAI-compatible listing endpoint; # providers that don't expose it still prove reachability by answering. import httpx + base = (provider.base_url or "https://api.openai.com/v1").rstrip("/") + api_key_verified: bool | None = None live_detail = "" try: - url = provider.base_url.rstrip("/") + "/models" - resp = httpx.get(url, headers={"Authorization": f"Bearer {secret}"}, timeout=5.0) + resp = httpx.get(base + "/models", headers={"Authorization": f"Bearer {secret}"}, timeout=5.0) + checks["endpoint_reachable"] = True if resp.status_code in (401, 403): - checks["api_key_accepted"] = False + api_key_verified = False live_detail = "The provider rejected the API key (HTTP %d). Check the key." % resp.status_code + elif resp.status_code == 200: + api_key_verified = True + live_detail = "Endpoint reachable and the API key was accepted." elif resp.status_code < 500: - # 200 = key accepted; 404/405 = endpoint reached but no /models - # (common on minimal proxies) — reachable, auth not disproven. - checks["endpoint_reachable"] = True - if resp.status_code == 200: - checks["api_key_accepted"] = True - live_detail = "Endpoint reachable and the API key was accepted." - else: - live_detail = ("Endpoint reachable; it doesn't expose /models " - "(HTTP %d), so the key wasn't verified." % resp.status_code) + # 404/405 = reached but no /models (common on minimal proxies). The key + # is NEITHER accepted nor rejected — leave it UNVERIFIED (None), not a + # confident pass, so the UI doesn't show a false green on a wrong key. + api_key_verified = None + live_detail = ("Endpoint reachable, but it doesn't expose /models " + "(HTTP %d), so the API key could not be verified here — " + "it will be checked on the first real request." % resp.status_code) else: - checks["endpoint_reachable"] = True checks["server_error"] = True live_detail = "Endpoint reachable but returned a server error (HTTP %d)." % resp.status_code except Exception: # noqa: BLE001 — network failure classes, no body echoed checks["endpoint_reachable"] = False live_detail = "Could not reach the endpoint (network error or timeout). Check the base URL." - ok = (config_ok and checks.get("api_key_accepted", True) - and checks.get("endpoint_reachable", True) and not checks.get("server_error", False)) - return ModelProviderTestResult(ok=ok, checks=checks, detail=live_detail) + # ok = no hard problem detected (reachable, key not rejected, no server error). + # A None api_key_verified still counts as ok, but the UI surfaces it as a + # caution ("reachable, key unverified") rather than a green pass. + ok = (checks.get("endpoint_reachable", False) + and api_key_verified is not False + and not checks.get("server_error", False)) + return ModelProviderTestResult(ok=ok, checks=checks, api_key_verified=api_key_verified, + detail=live_detail) diff --git a/sidecar/app/routers/sessions.py b/sidecar/app/routers/sessions.py index b295156..d5b5ea0 100644 --- a/sidecar/app/routers/sessions.py +++ b/sidecar/app/routers/sessions.py @@ -364,25 +364,34 @@ def post_session_message( # Success: persist the user message and the assistant answer together. # The contract is already sanitized + allowlist-coerced inside session_agent. - proposed_actions = contract["next_action_proposals"] - grounding = { - "evidence_used": contract.get("evidence_used", []), - "evidence_gaps": contract.get("evidence_gaps", []), - "skills_used": contract.get("skills_used", []), - } - repo.add_message(conn, session_id, "user", body.content) - repo.add_message(conn, session_id, "assistant", contract["answer"], - tool_activity=contract.get("tool_activity"), - grounding=grounding, proposed_actions=proposed_actions) - audit.record(conn, "session.message", {"session_id": session_id}, run_id=None) - conn.commit() - turn_guard.set_result(body.turn_id, { - "proposed_actions": proposed_actions, - "skills_used": contract.get("skills_used", []), - "skills_offered": contract.get("skills_offered", []), - "evidence_used": contract.get("evidence_used", []), - "evidence_gaps": contract.get("evidence_gaps", []), - }, session_id) + # This block runs OUTSIDE the try/except above, so — like the streaming + # worker's finally — it must resolve the turn handle on its own failure. If a + # persist/commit here raised before set_result, the handle stayed un-done and + # non-evictable: the same-turn fallback would block 150s, and EVERY subsequent + # turn in this session would eat the 120s prior-turn wait until eviction. + try: + proposed_actions = contract["next_action_proposals"] + grounding = { + "evidence_used": contract.get("evidence_used", []), + "evidence_gaps": contract.get("evidence_gaps", []), + "skills_used": contract.get("skills_used", []), + } + repo.add_message(conn, session_id, "user", body.content) + repo.add_message(conn, session_id, "assistant", contract["answer"], + tool_activity=contract.get("tool_activity"), + grounding=grounding, proposed_actions=proposed_actions) + audit.record(conn, "session.message", {"session_id": session_id}, run_id=None) + conn.commit() + turn_guard.set_result(body.turn_id, { + "proposed_actions": proposed_actions, + "skills_used": contract.get("skills_used", []), + "skills_offered": contract.get("skills_offered", []), + "evidence_used": contract.get("evidence_used", []), + "evidence_gaps": contract.get("evidence_gaps", []), + }, session_id) + except Exception as exc: # noqa: BLE001 — a persist failure must still resolve the handle + turn_guard.fail(body.turn_id, redact_text(str(exc)), session_id) + raise HTTPException(status_code=500, detail=redact_text(str(exc))) return { "session_id": session_id, "messages": repo.list_messages(conn, session_id), diff --git a/sidecar/app/runs/_common.py b/sidecar/app/runs/_common.py index 32e3044..0d65c62 100644 --- a/sidecar/app/runs/_common.py +++ b/sidecar/app/runs/_common.py @@ -73,7 +73,14 @@ def _finalize_success(conn: sqlite3.Connection, run_id: str, summary: str) -> No ) conn.commit() runs_repo.set_status(conn, run_id, "completed", final_summary=summary, report_path=report_rel) - bus.publish(run_id, {"type": "report_ready", "run_id": run_id, "report_path": report_rel}) + # The run is now committed 'completed'. A failure in the notification below + # must NOT propagate — if it did, run_executor's except would fire and + # downgrade this finished run to 'failed', a terminal contradiction (a failed + # run that already has a report + success summary). + try: + bus.publish(run_id, {"type": "report_ready", "run_id": run_id, "report_path": report_rel}) + except Exception: # noqa: BLE001 - best-effort SSE notification; the run is already done + pass def run_executor( @@ -105,7 +112,11 @@ def run_executor( except Exception as exc: # noqa: BLE001 - sanitized below detail = redact_text(str(exc)).strip() final = f"{failure_summary} {detail}".strip()[:500] if detail else failure_summary - runs_repo.set_status(conn, run_id, "failed", final_summary=final) - bus.publish(run_id, {"type": "error", "message": detail or failure_summary}) + # Terminal is final: never downgrade a run that already reached a terminal + # state (e.g. an error raised AFTER _finalize_success committed 'completed'). + cur = runs_repo.get_row(conn, run_id) + if cur is None or cur["status"] not in ("completed", "failed"): + runs_repo.set_status(conn, run_id, "failed", final_summary=final) + bus.publish(run_id, {"type": "error", "message": detail or failure_summary}) finally: bus.mark_done(run_id) diff --git a/sidecar/tests/test_maxturns_finalize.py b/sidecar/tests/test_maxturns_finalize.py index 2157a08..a528b6e 100644 --- a/sidecar/tests/test_maxturns_finalize.py +++ b/sidecar/tests/test_maxturns_finalize.py @@ -120,30 +120,51 @@ async def collect(): for p in final.get("next_action_proposals") or []) -def test_tool_output_budget_note_is_status_not_error(): - # The exhausted note reads as a soft boundary with a next step, not a failure. +def test_tool_output_budget_hard_caps_a_single_oversized_output(): + # PB: a SINGLE output that would exceed the budget is withheld with a VALID + # JSON status envelope (not the raw bytes, not an error) — a hard pre-emptive + # cap, so one large tool return can't blow past the per-turn context budget. import json - calls = {"n": 0} - class FakeTool: name = "list_objects" async def on_invoke_tool(self, ctx, args): - calls["n"] += 1 return "x" * 100 tool = FakeTool() budget = session_agent._install_tool_output_budget([tool], limit=10) + out = asyncio.run(tool.on_invoke_tool(None, None)) + payload = json.loads(out) # valid JSON, not the raw 100 chars + assert out != "x" * 100 + assert payload["status"] == "output_too_large" and "error" not in payload + assert budget["exhausted"] is True - async def run(): - first = await tool.on_invoke_tool(None, None) # consumes >10 chars - second = await tool.on_invoke_tool(None, None) # now over budget → note - return first, second - first, second = asyncio.run(run()) - assert first == "x" * 100 - payload = json.loads(second) +def test_tool_output_budget_note_is_status_not_error(): + # After the budget is spent, further calls read as a soft boundary with a next + # step, not a failure. + import json + + class FakeTool: + name = "list_objects" + + async def on_invoke_tool(self, ctx, args): + return "y" * 5 + + tool = FakeTool() + budget = session_agent._install_tool_output_budget([tool], limit=8) + + async def run(): + first = await tool.on_invoke_tool(None, None) # 5 chars, fits (0+5<=8) + second = await tool.on_invoke_tool(None, None) # 5+5>8 → withheld too_large + third = await tool.on_invoke_tool(None, None) # already over → exhausted + return first, second, third + + first, second, third = asyncio.run(run()) + assert first == "y" * 5 + assert json.loads(second)["status"] == "output_too_large" + payload = json.loads(third) assert payload["status"] == "budget_exhausted" and "error" not in payload assert "resets" in payload["next_step"] assert budget["exhausted"] is True diff --git a/sidecar/tests/test_model_providers.py b/sidecar/tests/test_model_providers.py index e6330fe..4f50898 100644 --- a/sidecar/tests/test_model_providers.py +++ b/sidecar/tests/test_model_providers.py @@ -133,7 +133,7 @@ def fake_get(url, headers=None, timeout=None): body = resp.json() assert body["ok"] is True assert body["checks"]["api_key_present"] is True - assert body["checks"]["api_key_accepted"] is True + assert body["api_key_verified"] is True assert body["checks"]["endpoint_reachable"] is True # The probe hit the provider's /models with the key — but the secret never # appears in the RESPONSE. @@ -148,7 +148,7 @@ def test_test_endpoint_flags_rejected_key(client, monkeypatch): monkeypatch.setattr(httpx, "get", lambda *a, **k: _FakeResp(401)) body = client.post(f"/model-providers/{provider_id}/test").json() assert body["ok"] is False - assert body["checks"]["api_key_accepted"] is False + assert body["api_key_verified"] is False def test_test_endpoint_unreachable_is_reported(client, monkeypatch): diff --git a/sidecar/tests/test_review_fixes.py b/sidecar/tests/test_review_fixes.py index 9fad1d1..0c3b71d 100644 --- a/sidecar/tests/test_review_fixes.py +++ b/sidecar/tests/test_review_fixes.py @@ -512,7 +512,7 @@ def test_raised_budgets_and_caps(): # skills_used contract cap must match the per-turn read_skill budget; the # budget constants live inside build(), so pin the contract-side value. src = open("app/agent_runtime/session_tools.py").read() - assert "_MAX_SKILL_LOADS = 10" in src + assert "_MAX_SKILL_LOADS = 20" in src assert "_MAX_PREVIEWS = 16" in src assert "_MAX_LATENCY_RUNS = 8" in src raw = "answer\n```json\n" + json.dumps( diff --git a/sidecar/tests/test_skills.py b/sidecar/tests/test_skills.py index d6cfcb1..beb7993 100644 --- a/sidecar/tests/test_skills.py +++ b/sidecar/tests/test_skills.py @@ -261,7 +261,7 @@ def test_migrations_are_sequential_and_capped(): # (support the startup data-maintenance prune). versions = [v for v, _n, _s in migrations.MIGRATIONS] assert versions == list(range(1, len(versions) + 1)) # 1..N, no gaps/dupes - assert max(versions) == 18 + assert max(versions) == 19 def test_no_public_skills_api(client): diff --git a/sidecar/tests/test_v0320_fixes.py b/sidecar/tests/test_v0320_fixes.py index b32fb88..cd1b3d5 100644 --- a/sidecar/tests/test_v0320_fixes.py +++ b/sidecar/tests/test_v0320_fixes.py @@ -230,4 +230,4 @@ def test_all_migrations_apply(tmp_path): idx = {r[0] for r in conn.execute( "SELECT name FROM sqlite_master WHERE type='index'")} assert "idx_audit_logs_created" in idx - assert len(MIGRATIONS) == 18 + assert len(MIGRATIONS) == 19 diff --git a/sidecar/tests/test_v0340_fixes.py b/sidecar/tests/test_v0340_fixes.py new file mode 100644 index 0000000..cc958f6 --- /dev/null +++ b/sidecar/tests/test_v0340_fixes.py @@ -0,0 +1,142 @@ +"""v0.34.0 — engine correctness, lifecycle robustness, de-ossification, model config. + + E1 numeric columns stay integer (status codes not "404.0"; size precision). + E2 access-log error rates use parsed-request denominator, not all lines. + E3 inventory average/small-ratio use consistent denominators. + E4 inventory age bucketing is UTC (tz-independent). + E5/OSS1 aggregate: 2nd group-by dim, day/weekday, distinct/p99, tiebreaker. + SM3 loser of a concurrent import claim fails its orphan run. + MO operator-declared max_output_tokens clamps the completion budget. + PB a single oversized tool output is hard-capped (covered in maxturns test). + OSS2 agent-memory recall scales with the model window. +""" + +from __future__ import annotations + +import os +import tempfile + +import pytest + +from app.analysis import access_logs as al +from app.analysis import aggregate as agg +from app.analysis import inventory as inv + + +def _log(tmp, lines: str) -> str: + p = os.path.join(tmp, "a.log") + open(p, "w").write(lines) + db = os.path.join(tmp, "a.duckdb") + al.import_access_logs(p, db, "text") + return db + + +_L = ('2026-06-25T10:00:00Z b GET /x 404 100 5 ms user-agent="c" remote_ip="1.2.3.4"\n' + '2026-06-25T10:00:00Z b GET /y 500 200 9 ms user-agent="c" remote_ip="1.2.3.5"\n' + "an unparseable line with no fields\n") + + +# --- E1: integer columns, no "404.0" ---------------------------------------- + +def test_status_codes_render_as_integers(tmp_path): + db = _log(str(tmp_path), _L) + m = al.analyze_access_logs(db) + labels = {x["value"] for x in m["status_code_distribution"] if x["value"] != "None"} + assert labels == {"404", "500"} + assert not any("." in lbl for lbl in labels) + + +# --- E2: error rate is of PARSED requests ------------------------------------ + +def test_error_rate_denominator_is_parsed_requests(tmp_path): + db = _log(str(tmp_path), _L) + m = al.analyze_access_logs(db) + # 2 parsed requests, 1 is 5xx → 0.5 (NOT 1/3 of all ingested lines). + assert m["error_rate_5xx"] == 0.5 + assert m["error_rate_4xx"] == 0.5 + + +# --- E3: inventory denominators ---------------------------------------------- + +def _inv(tmp, rows: str) -> str: + p = os.path.join(tmp, "i.csv") + open(p, "w").write("bucket,key,size,last_modified,storage_class\n" + rows) + db = os.path.join(tmp, "i.duckdb") + inv.import_inventory_file(p, db) + return db + + +def test_inventory_average_reconciles_with_total_over_count(tmp_path): + # 3 objects, total 200, one null size → avg = 200/3 = 66 (not 100 = 200/2). + db = _inv(str(tmp_path), + "b,k1,100,2026-01-01T00:00:00Z,STANDARD\n" + "b,k2,,2026-01-01T00:00:00Z,STANDARD\n" + "b,k3,100,2026-01-01T00:00:00Z,STANDARD\n") + m = inv.analyze_inventory(db) + assert m["object_count"] == 3 and m["total_size"] == 200 + assert m["average_object_size"] == 66 # total/count, not total/known + + +# --- E5 / OSS1: aggregate vocabulary + tiebreaker ---------------------------- + +def test_aggregate_two_dims_and_new_metrics(tmp_path): + db = _log(str(tmp_path), _L) + out = agg.aggregate(db, "access_log", "distinct_ips", + group_by="status_code", group_by_2="day") + assert out["group_by_2"] == "day" + # composite labels join the two dims. + assert any(" · " in g["group"] for g in out["groups"]) + # new metric works + d = agg.aggregate(db, "access_log", "p99_latency_ms") + assert "value" in d + + +def test_aggregate_rejects_out_of_whitelist_second_dim(tmp_path): + db = _log(str(tmp_path), _L) + with pytest.raises(agg.AggregateError): + agg.aggregate(db, "access_log", "count", group_by="method", group_by_2="not_a_dim") + + +# --- MO: operator max_output clamps the completion budget -------------------- + +def test_max_output_tokens_explicit_wins(): + from app.agent_runtime import model_budget as mb + + # An unknown model would default to 16384; an operator cap of 4096 wins and + # clamps the completion budget so a lower-cap endpoint doesn't 400. + assert mb.max_output_tokens("some-unknown-model", explicit_max=4096) == 4096 + assert mb.completion_token_budget("some-unknown-model", explicit_max=4096) == 4096 + # 0/None → fall back to the table. + assert mb.max_output_tokens("some-unknown-model", explicit_max=None) == 16384 + + +# --- OSS2: agent-memory recall scales with the window ------------------------ + +def test_agent_memory_recall_is_model_elastic(): + from app.agent_runtime import session_agent as sa + + small = sa._elastic_memory_cap("gpt-4o", None) # ~128k window + big = sa._elastic_memory_cap("gpt-4.1", 1_000_000) # explicit 1M window + assert small == 50 + assert big > 50 and big <= sa._MEM_RECALL_CEIL + + +# --- SM3: concurrent import claim loser fails its orphan run ------------------ + +def test_import_claim_loser_fails_its_run(tmp_path, monkeypatch): + # Unit-level: the repo helpers used by the loser branch behave as wired — + # a run created then set 'failed' is terminal, not left 'pending'. + import sqlite3 + + from app.migrations import apply_migrations + from app.models.schemas import RunCreate + from app.repositories import runs as runs_repo + + conn = sqlite3.connect(tmp_path / "sm3.db") + conn.row_factory = sqlite3.Row + apply_migrations(conn) + rid = runs_repo.create(conn, RunCreate(run_type="access_log_analysis", user_prompt="x"), + status="pending") + runs_repo.set_status(conn, rid, "failed", final_summary="Superseded by a concurrent import.") + assert runs_repo.get_row(conn, rid)["status"] == "failed" + conn.close()