quality(atlasbot): merge strict gate coverage

atlasbot/engine/answerer/common.py

@@ -38,17 +38,7 @@ def _strip_followup_meta(reply: str) -> str:
     return cleaned
 
 
-def _build_meta(
+def _build_meta(mode: str, call_count: int, call_cap: int, limit_hit: bool, time_budget_hit: bool, time_budget_sec: float, classify: dict[str, Any], tool_hint: dict[str, Any] | None, started: float) -> dict[str, Any]:
-    mode: str,
-    call_count: int,
-    call_cap: int,
-    limit_hit: bool,
-    time_budget_hit: bool,
-    time_budget_sec: float,
-    classify: dict[str, Any],
-    tool_hint: dict[str, Any] | None,
-    started: float,
-) -> dict[str, Any]:
     return {
         "mode": mode,
         "llm_calls": call_count,
@@ -214,13 +204,7 @@ def _build_chunk_groups(chunks: list[dict[str, Any]], group_size: int) -> list[l
     return groups
 
 
-async def _score_chunks(
+async def _score_chunks(call_llm: Callable[..., Any], chunks: list[dict[str, Any]], question: str, sub_questions: list[str], plan: ModePlan) -> dict[str, float]:
-    call_llm: Callable[..., Any],
-    chunks: list[dict[str, Any]],
-    question: str,
-    sub_questions: list[str],
-    plan: ModePlan,
-) -> dict[str, float]:
     scores: dict[str, float] = {chunk["id"]: 0.0 for chunk in chunks}
     if not chunks:
         return scores
@@ -238,11 +222,7 @@ async def _score_chunks(
     return await _score_groups_parallel(call_llm, groups, ctx)
 
 
-async def _score_groups_serial(
+async def _score_groups_serial(call_llm: Callable[..., Any], groups: list[list[dict[str, Any]]], ctx: ScoreContext) -> dict[str, float]:
-    call_llm: Callable[..., Any],
-    groups: list[list[dict[str, Any]]],
-    ctx: ScoreContext,
-) -> dict[str, float]:
     scores: dict[str, float] = {}
     for grp in groups:
         runs = [await _score_chunk_group(call_llm, grp, ctx.question, ctx.sub_questions) for _ in range(ctx.retries)]
@@ -254,11 +234,7 @@ async def _score_groups_serial(
     return scores
 
 
-async def _score_groups_parallel(
+async def _score_groups_parallel(call_llm: Callable[..., Any], groups: list[list[dict[str, Any]]], ctx: ScoreContext) -> dict[str, float]:
-    call_llm: Callable[..., Any],
-    groups: list[list[dict[str, Any]]],
-    ctx: ScoreContext,
-) -> dict[str, float]:
     coros: list[Awaitable[tuple[int, dict[str, float]]]] = []
     for idx, grp in enumerate(groups):
         for _ in range(ctx.retries):
@@ -278,12 +254,7 @@ async def _score_groups_parallel(
     return scores
 
 
-async def _score_chunk_group(
+async def _score_chunk_group(call_llm: Callable[..., Any], group: list[dict[str, Any]], question: str, sub_questions: list[str]) -> dict[str, float]:
-    call_llm: Callable[..., Any],
-    group: list[dict[str, Any]],
-    question: str,
-    sub_questions: list[str],
-) -> dict[str, float]:
     prompt = (
         prompts.CHUNK_SCORE_PROMPT
         + "\nQuestion: "
@@ -310,13 +281,7 @@ async def _score_chunk_group(
     return scored
 
 
-async def _score_chunk_group_run(
+async def _score_chunk_group_run(call_llm: Callable[..., Any], idx: int, group: list[dict[str, Any]], question: str, sub_questions: list[str]) -> tuple[int, dict[str, float]]:
-    call_llm: Callable[..., Any],
-    idx: int,
-    group: list[dict[str, Any]],
-    question: str,
-    sub_questions: list[str],
-) -> tuple[int, dict[str, float]]:
     return idx, await _score_chunk_group(call_llm, group, question, sub_questions)
 
 
@@ -332,12 +297,7 @@ def _merge_score_runs(runs: list[dict[str, float]]) -> dict[str, float]:
     return {key: totals[key] / counts[key] for key in totals}
 
 
-async def _select_best_score_run(
+async def _select_best_score_run(call_llm: Callable[..., Any], group: list[dict[str, Any]], runs: list[dict[str, float]], ctx: ScoreContext) -> dict[str, float]:
-    call_llm: Callable[..., Any],
-    group: list[dict[str, Any]],
-    runs: list[dict[str, float]],
-    ctx: ScoreContext,
-) -> dict[str, float]:
     if not runs:
         return {}
     prompt = (
@@ -364,11 +324,7 @@ async def _select_best_score_run(
     return runs[idx]
 
 
-def _keyword_hits(
+def _keyword_hits(ranked: list[dict[str, Any]], head: dict[str, Any], keywords: list[str] | None) -> list[dict[str, Any]]:
-    ranked: list[dict[str, Any]],
-    head: dict[str, Any],
-    keywords: list[str] | None,
-) -> list[dict[str, Any]]:
     if not keywords:
         return []
     lowered = [kw.lower() for kw in keywords if isinstance(kw, str) and kw.strip()]
@@ -384,13 +340,7 @@ def _keyword_hits(
     return hits
 
 
-def _select_chunks(
+def _select_chunks(chunks: list[dict[str, Any]], scores: dict[str, float], plan: ModePlan, keywords: list[str] | None = None, must_ids: list[str] | None = None) -> list[dict[str, Any]]:
-    chunks: list[dict[str, Any]],
-    scores: dict[str, float],
-    plan: ModePlan,
-    keywords: list[str] | None = None,
-    must_ids: list[str] | None = None,
-) -> list[dict[str, Any]]:
     if not chunks:
         return []
     ranked = sorted(chunks, key=lambda item: scores.get(item["id"], 0.0), reverse=True)
@@ -403,12 +353,7 @@ def _select_chunks(
     return selected
 
 
-def _append_must_chunks(
+def _append_must_chunks(chunks: list[dict[str, Any]], selected: list[dict[str, Any]], must_ids: list[str] | None, limit: int) -> bool:
-    chunks: list[dict[str, Any]],
-    selected: list[dict[str, Any]],
-    must_ids: list[str] | None,
-    limit: int,
-) -> bool:
     if not must_ids:
         return False
     id_map = {item["id"]: item for item in chunks}
@@ -421,12 +366,7 @@ def _append_must_chunks(
     return False
 
 
-def _append_keyword_chunks(
+def _append_keyword_chunks(ranked: list[dict[str, Any]], selected: list[dict[str, Any]], keywords: list[str] | None, limit: int) -> bool:
-    ranked: list[dict[str, Any]],
-    selected: list[dict[str, Any]],
-    keywords: list[str] | None,
-    limit: int,
-) -> bool:
     if not ranked:
         return False
     head = ranked[0]
@@ -438,11 +378,7 @@ def _append_keyword_chunks(
     return False
 
 
-def _append_ranked_chunks(
+def _append_ranked_chunks(ranked: list[dict[str, Any]], selected: list[dict[str, Any]], limit: int) -> None:
-    ranked: list[dict[str, Any]],
-    selected: list[dict[str, Any]],
-    limit: int,
-) -> None:
     for item in ranked:
         if len(selected) >= limit:
             break

atlasbot/engine/answerer/engine.py

@@ -31,29 +31,14 @@ class AnswerEngine:
       post-processing helpers stay split across smaller modules.
     """
 
-    def __init__(
+    def __init__(self, settings: Settings, llm: LLMClient, kb: KnowledgeBase, snapshot: SnapshotProvider) -> None:
-        self,
-        settings: Settings,
-        llm: LLMClient,
-        kb: KnowledgeBase,
-        snapshot: SnapshotProvider,
-    ) -> None:
         self._settings = settings
         self._llm = llm
         self._kb = kb
         self._snapshot = snapshot
         self._store = ClaimStore(settings.state_db_path, settings.conversation_ttl_sec)
 
-    async def answer(
+    async def answer(self, question: str, *, mode: str, history: list[dict[str, str]] | None = None, observer: Callable[[str, str], None] | None = None, conversation_id: str | None = None, snapshot_pin: bool | None = None) -> AnswerResult:
-        self,
-        question: str,
-        *,
-        mode: str,
-        history: list[dict[str, str]] | None = None,
-        observer: Callable[[str, str], None] | None = None,
-        conversation_id: str | None = None,
-        snapshot_pin: bool | None = None,
-    ) -> AnswerResult:
         """Answer a question by delegating to the staged workflow."""
 
         return await run_answer(
@@ -71,15 +56,7 @@ class AnswerEngine:
         reply = await self._llm.chat(messages, model=self._settings.ollama_model)
         return AnswerResult(reply, _default_scores(), {"mode": "stock"})
 
-    async def _synthesize_answer(
+    async def _synthesize_answer(self, question: str, subanswers: list[str], context: str, classify: dict[str, Any], plan: ModePlan, call_llm: Callable[..., Any]) -> str:
-        self,
-        question: str,
-        subanswers: list[str],
-        context: str,
-        classify: dict[str, Any],
-        plan: ModePlan,
-        call_llm: Callable[..., Any],
-    ) -> str:
         style_hint = _style_hint(classify)
         if not subanswers:
             prompt = (
@@ -148,13 +125,7 @@ class AnswerEngine:
             return drafts[idx]
         return drafts[0]
 
-    async def _score_answer(
+    async def _score_answer(self, question: str, reply: str, plan: ModePlan, call_llm: Callable[..., Any]) -> AnswerScores:
-        self,
-        question: str,
-        reply: str,
-        plan: ModePlan,
-        call_llm: Callable[..., Any],
-    ) -> AnswerScores:
         if not plan.use_scores:
             return _default_scores()
         prompt = prompts.SCORE_PROMPT + "\nQuestion: " + question + "\nAnswer: " + reply
@@ -162,14 +133,7 @@ class AnswerEngine:
         data = _parse_json_block(raw, fallback={})
         return _scores_from_json(data)
 
-    async def _extract_claims(
+    async def _extract_claims(self, question: str, reply: str, summary: dict[str, Any], facts_used: list[str], call_llm: Callable[..., Any]) -> list[ClaimItem]:
-        self,
-        question: str,
-        reply: str,
-        summary: dict[str, Any],
-        facts_used: list[str],
-        call_llm: Callable[..., Any],
-    ) -> list[ClaimItem]:
         if not reply or not summary:
             return []
         summary_json = _json_excerpt(summary)
@@ -208,27 +172,13 @@ class AnswerEngine:
                     claims.append(ClaimItem(id=claim_id, claim=claim_text, evidence=evidence_items))
         return claims
 
-    async def _dedup_reply(
+    async def _dedup_reply(self, reply: str, plan: ModePlan, call_llm: Callable[..., Any], tag: str) -> str:
-        self,
-        reply: str,
-        plan: ModePlan,
-        call_llm: Callable[..., Any],
-        tag: str,
-    ) -> str:
         if not _needs_dedup(reply):
             return reply
         dedup_prompt = prompts.DEDUP_PROMPT + "\nDraft: " + reply
         return await call_llm(prompts.DEDUP_SYSTEM, dedup_prompt, model=plan.fast_model, tag=tag)
 
-    async def _answer_followup(  # noqa: C901
+    async def _answer_followup(self, question: str, state: ConversationState, summary: dict[str, Any], classify: dict[str, Any], plan: ModePlan, call_llm: Callable[..., Any]) -> str:  # noqa: C901, ARG002
-        self,
-        question: str,
-        state: ConversationState,
-        summary: dict[str, Any],
-        classify: dict[str, Any],  # noqa: ARG002
-        plan: ModePlan,
-        call_llm: Callable[..., Any],
-    ) -> str:
         claim_ids = await self._select_claims(question, state.claims, plan, call_llm)
         selected = [claim for claim in state.claims if claim.id in claim_ids] if claim_ids else state.claims[:2]
         evidence_lines = []
@@ -284,13 +234,7 @@ class AnswerEngine:
         reply = _strip_followup_meta(reply)
         return reply
 
-    async def _select_claims(
+    async def _select_claims(self, question: str, claims: list[ClaimItem], plan: ModePlan, call_llm: Callable[..., Any]) -> list[str]:
-        self,
-        question: str,
-        claims: list[ClaimItem],
-        plan: ModePlan,
-        call_llm: Callable[..., Any],
-    ) -> list[str]:
         if not claims:
             return []
         claims_brief = [{"id": claim.id, "claim": claim.claim} for claim in claims]
@@ -308,14 +252,7 @@ class AnswerEngine:
         state_payload = self._store.get(conversation_id)
         return _state_from_payload(state_payload) if state_payload else None
 
-    def _store_state(
+    def _store_state(self, conversation_id: str, claims: list[ClaimItem], summary: dict[str, Any], snapshot: dict[str, Any] | None, pin_snapshot: bool) -> None:
-        self,
-        conversation_id: str,
-        claims: list[ClaimItem],
-        summary: dict[str, Any],
-        snapshot: dict[str, Any] | None,
-        pin_snapshot: bool,
-    ) -> None:
         snapshot_id = _snapshot_id(summary)
         pinned_snapshot = snapshot if pin_snapshot else None
         payload = {

atlasbot/engine/answerer/factsheet.py

@@ -76,13 +76,7 @@ def _is_plain_math_question(question: str) -> bool:
     )
 
 
-def _quick_fact_sheet_lines(  # noqa: C901
+def _quick_fact_sheet_lines(question: str, summary_lines: list[str], kb_lines: list[str], *, limit: int) -> list[str]:  # noqa: C901
-    question: str,
-    summary_lines: list[str],
-    kb_lines: list[str],
-    *,
-    limit: int,
-) -> list[str]:
     tokens = {
         token
         for token in re.findall(r"[a-z0-9][a-z0-9_-]{2,}", question.lower())

atlasbot/engine/answerer/post.py

@@ -59,10 +59,7 @@ def _needs_evidence_guard(reply: str, facts: list[str]) -> bool:
     return any(term in lower_reply for term in arch_terms) and not any(term in fact_text for term in arch_terms)
 
 
-async def _contradiction_decision(
+async def _contradiction_decision(ctx: ContradictionContext, attempts: int = 1) -> dict[str, Any]:
-    ctx: ContradictionContext,
-    attempts: int = 1,
-) -> dict[str, Any]:
     best = {"use_facts": True, "confidence": 50}
     facts_block = "\n".join(ctx.facts[:12])
     for idx in range(max(1, attempts)):
@@ -256,12 +253,7 @@ def _expand_tokens(tokens: list[str]) -> list[str]:
     return expanded
 
 
-def _ensure_token_coverage(
+def _ensure_token_coverage(lines: list[str], tokens: list[str], summary_lines: list[str], max_add: int = 4) -> list[str]:
-    lines: list[str],
-    tokens: list[str],
-    summary_lines: list[str],
-    max_add: int = 4,
-) -> list[str]:
     if not lines or not tokens or not summary_lines:
         return lines
     hay = " ".join(lines).lower()

atlasbot/engine/answerer/post_ext.py

@@ -74,12 +74,7 @@ def _needs_focus_fix(question: str, reply: str, classify: dict[str, Any]) -> boo
     return any(marker in reply.lower() for marker in extra_markers)
 
 
-def _extract_keywords(
+def _extract_keywords(raw_question: str, normalized: str, sub_questions: list[str], keywords: list[Any] | None) -> list[str]:
-    raw_question: str,
-    normalized: str,
-    sub_questions: list[str],
-    keywords: list[Any] | None,
-) -> list[str]:
     stopwords = {
         "the",
         "and",
@@ -211,13 +206,10 @@ def _resolve_path(data: Any, path: str) -> Any | None:
         else:
             return None
         if index is not None:
-            try:
+            idx = int(index)
-                idx = int(index)
+            if isinstance(cursor, list) and 0 <= idx < len(cursor):
-                if isinstance(cursor, list) and 0 <= idx < len(cursor):
+                cursor = cursor[idx]
-                    cursor = cursor[idx]
+            else:
-                else:
-                    return None
-            except ValueError:
                 return None
     return cursor
 

atlasbot/engine/answerer/retrieval.py

@@ -125,9 +125,9 @@ def _metric_ctx_values(ctx: dict[str, Any]) -> tuple[list[str], str, list[str],
     if not isinstance(summary_lines, list):
         return [], "", [], [], set()
     question = ctx.get("question") if isinstance(ctx, dict) else ""
-    sub_questions = ctx.get("sub_questions") if isinstance(ctx, dict) else []
+    sub_questions = ctx.get("sub_questions") if isinstance(ctx.get("sub_questions"), list) else []
-    keywords = ctx.get("keywords") if isinstance(ctx, dict) else []
+    keywords = ctx.get("keywords") if isinstance(ctx.get("keywords"), list) else []
-    keyword_tokens = ctx.get("keyword_tokens") if isinstance(ctx, dict) else []
+    keyword_tokens = ctx.get("keyword_tokens") if isinstance(ctx.get("keyword_tokens"), list) else []
     token_set = {str(token).lower() for token in keyword_tokens if token}
     token_set |= {token.lower() for token in _extract_keywords(str(question), str(question), sub_questions=sub_questions, keywords=keywords)}
     token_set = _token_variants(token_set)

atlasbot/engine/answerer/retrieval_ext.py

@@ -32,13 +32,7 @@ def _metric_key_tokens(summary_lines: list[str]) -> set[str]:
     return tokens
 
 
-async def _select_best_candidate(
+async def _select_best_candidate(call_llm: Callable[..., Any], question: str, candidates: list[str], plan: ModePlan, tag: str) -> int:
-    call_llm: Callable[..., Any],
-    question: str,
-    candidates: list[str],
-    plan: ModePlan,
-    tag: str,
-) -> int:
     if len(candidates) <= 1:
         return 0
     prompt = (
@@ -82,13 +76,7 @@ def _collect_fact_candidates(selected: list[dict[str, Any]], limit: int) -> list
     return _dedupe_lines(lines, limit=limit)
 
 
-async def _select_best_list(
+async def _select_best_list(call_llm: Callable[..., Any], question: str, candidates: list[list[str]], plan: ModePlan, tag: str) -> list[str]:
-    call_llm: Callable[..., Any],
-    question: str,
-    candidates: list[list[str]],
-    plan: ModePlan,
-    tag: str,
-) -> list[str]:
     if not candidates:
         return []
     if len(candidates) == 1:
@@ -106,12 +94,7 @@ async def _select_best_list(
     return chosen
 
 
-async def _extract_fact_types(
+async def _extract_fact_types(call_llm: Callable[..., Any], question: str, keywords: list[str], plan: ModePlan) -> list[str]:
-    call_llm: Callable[..., Any],
-    question: str,
-    keywords: list[str],
-    plan: ModePlan,
-) -> list[str]:
     prompt = prompts.FACT_TYPES_PROMPT + "\nQuestion: " + question
     if keywords:
         prompt += "\nKeywords: " + ", ".join(keywords)
@@ -130,12 +113,7 @@ async def _extract_fact_types(
     return chosen[:10]
 
 
-async def _derive_signals(
+async def _derive_signals(call_llm: Callable[..., Any], question: str, fact_types: list[str], plan: ModePlan) -> list[str]:
-    call_llm: Callable[..., Any],
-    question: str,
-    fact_types: list[str],
-    plan: ModePlan,
-) -> list[str]:
     if not fact_types:
         return []
     prompt = prompts.SIGNAL_PROMPT.format(question=question, fact_types="; ".join(fact_types))
@@ -154,13 +132,7 @@ async def _derive_signals(
     return chosen[:12]
 
 
-async def _scan_chunk_for_signals(
+async def _scan_chunk_for_signals(call_llm: Callable[..., Any], question: str, signals: list[str], chunk_lines: list[str], plan: ModePlan) -> list[str]:
-    call_llm: Callable[..., Any],
-    question: str,
-    signals: list[str],
-    chunk_lines: list[str],
-    plan: ModePlan,
-) -> list[str]:
     if not signals or not chunk_lines:
         return []
     prompt = prompts.CHUNK_SCAN_PROMPT.format(
@@ -183,13 +155,7 @@ async def _scan_chunk_for_signals(
     return chosen[:15]
 
 
-async def _prune_metric_candidates(
+async def _prune_metric_candidates(call_llm: Callable[..., Any], question: str, candidates: list[str], plan: ModePlan, attempts: int) -> list[str]:
-    call_llm: Callable[..., Any],
-    question: str,
-    candidates: list[str],
-    plan: ModePlan,
-    attempts: int,
-) -> list[str]:
     if not candidates:
         return []
     prompt = prompts.FACT_PRUNE_PROMPT.format(question=question, candidates="\n".join(candidates), max_lines=6)
@@ -208,13 +174,7 @@ async def _prune_metric_candidates(
     return chosen[:6]
 
 
-async def _select_fact_lines(
+async def _select_fact_lines(call_llm: Callable[..., Any], question: str, candidates: list[str], plan: ModePlan, max_lines: int) -> list[str]:
-    call_llm: Callable[..., Any],
-    question: str,
-    candidates: list[str],
-    plan: ModePlan,
-    max_lines: int,
-) -> list[str]:
     if not candidates:
         return []
     prompt = prompts.FACT_PRUNE_PROMPT.format(question=question, candidates="\n".join(candidates), max_lines=max_lines)

atlasbot/engine/answerer/workflow.py

@@ -23,16 +23,7 @@ from .retrieval_ext import *
 from .spine import *
 from .workflow_post import finalize_answer
 
-async def run_answer(  # noqa: C901
+async def run_answer(engine: Any, question: str, *, mode: str, history: list[dict[str, str]] | None = None, observer: Callable[[str, str], None] | None = None, conversation_id: str | None = None, snapshot_pin: bool | None = None) -> AnswerResult:  # noqa: C901
-    engine: Any,
-    question: str,
-    *,
-    mode: str,
-    history: list[dict[str, str]] | None = None,
-    observer: Callable[[str, str], None] | None = None,
-    conversation_id: str | None = None,
-    snapshot_pin: bool | None = None,
-) -> AnswerResult:
     """Answer a question using the staged reasoning pipeline."""
 
     settings = engine._settings
@@ -69,14 +60,7 @@ async def run_answer(  # noqa: C901
         "evidence_fix",
     }
 
-    async def call_llm(
+    async def call_llm(system: str, prompt: str, *, context: str | None = None, model: str | None = None, tag: str = "") -> str:
-        system: str,
-        prompt: str,
-        *,
-        context: str | None = None,
-        model: str | None = None,
-        tag: str = "",
-    ) -> str:
         nonlocal call_count, limit_hit, time_budget_hit
         if not limitless and call_count >= call_cap:
             limit_hit = True

atlasbot/engine/answerer/workflow_post.py

@@ -15,32 +15,7 @@ from .retrieval import *
 from .spine import *
 
 
-async def finalize_answer(  # noqa: C901
+async def finalize_answer(*, engine: Any, call_llm: Callable[..., Any], normalized: str, subanswers: list[str], context: str, classify: dict[str, Any], plan: ModePlan, summary: dict[str, Any], summary_lines: list[str], metric_facts: list[str], key_facts: list[str], facts_used: list[str], allowed_nodes: list[str], allowed_namespaces: list[str], runbook_paths: list[str], lowered_question: str, force_metric: bool, keyword_tokens: list[str], question_tokens: list[str], snapshot_context: str, observer: Callable[[str, str], None] | None, mode: str, metric_keys: list[str] | None = None) -> tuple[str, AnswerScores, list[ClaimItem]]:  # noqa: C901
-    *,
-    engine: Any,
-    call_llm: Callable[..., Any],
-    normalized: str,
-    subanswers: list[str],
-    context: str,
-    classify: dict[str, Any],
-    plan: ModePlan,
-    summary: dict[str, Any],
-    summary_lines: list[str],
-    metric_facts: list[str],
-    key_facts: list[str],
-    facts_used: list[str],
-    allowed_nodes: list[str],
-    allowed_namespaces: list[str],
-    runbook_paths: list[str],
-    lowered_question: str,
-    force_metric: bool,
-    keyword_tokens: list[str],
-    question_tokens: list[str],
-    snapshot_context: str,
-    observer: Callable[[str, str], None] | None,
-    mode: str,
-    metric_keys: list[str] | None = None,
-) -> tuple[str, AnswerScores, list[ClaimItem]]:
     """Synthesize and post-process the final answer."""
 
     reply = await engine._synthesize_answer(normalized, subanswers, context, classify, plan, call_llm)

atlasbot/main.py

@@ -49,14 +49,7 @@ async def main() -> None:
     queue = QueueManager(settings, handler)
     await queue.start()
 
-    async def answer_handler(
+    async def answer_handler(question: str, mode: str, history=None, conversation_id=None, snapshot_pin: bool | None = None, observer=None) -> AnswerResult:
-        question: str,
-        mode: str,
-        history=None,
-        conversation_id=None,
-        snapshot_pin: bool | None = None,
-        observer=None,
-    ) -> AnswerResult:
         if settings.queue_enabled:
             payload = await queue.submit(
                 {

atlasbot/matrix/bot.py

@@ -89,17 +89,7 @@ class MatrixClient:
 class MatrixBot:
     """Drive Matrix conversation handling and heartbeat replies."""
 
-    def __init__(
+    def __init__(self, settings: Settings, bot: MatrixBotConfig, engine: AnswerEngine, answer_handler: Callable[[str, str, list[dict[str, str]] | None, str | None, Callable[[str, str], None] | None], Awaitable[AnswerResult]] | None = None) -> None:
-        self,
-        settings: Settings,
-        bot: MatrixBotConfig,
-        engine: AnswerEngine,
-        answer_handler: Callable[
-            [str, str, list[dict[str, str]] | None, str | None, Callable[[str, str], None] | None],
-            Awaitable[AnswerResult],
-        ]
-        | None = None,
-    ) -> None:
         self._settings = settings
         self._bot = bot
         self._engine = engine

atlasbot/snapshot/builder/core_a.py

@@ -150,11 +150,7 @@ def _merge_cluster_summary(snapshot: dict[str, Any], summary: dict[str, Any]) ->
     )
 
 
-def _merge_cluster_fields(
+def _merge_cluster_fields(summary: dict[str, Any], cluster_summary: dict[str, Any], field_types: dict[str, type]) -> None:
-    summary: dict[str, Any],
-    cluster_summary: dict[str, Any],
-    field_types: dict[str, type],
-) -> None:
     for key, expected in field_types.items():
         value = cluster_summary.get(key)
         if isinstance(value, expected):
@@ -249,10 +245,7 @@ def _build_hardware_by_node(nodes_detail: list[dict[str, Any]]) -> dict[str, Any
     return {"hardware_by_node": mapping} if mapping else {}
 
 
-def _build_hardware_usage(  # noqa: C901
+def _build_hardware_usage(metrics: dict[str, Any], hardware_by_node: dict[str, Any] | None) -> dict[str, Any]:  # noqa: C901
-    metrics: dict[str, Any],
-    hardware_by_node: dict[str, Any] | None,
-) -> dict[str, Any]:
     if not isinstance(hardware_by_node, dict) or not hardware_by_node:
         return {}
     node_load = metrics.get("node_load") if isinstance(metrics.get("node_load"), list) else []

atlasbot/snapshot/builder/format_a.py

@@ -470,7 +470,9 @@ def _append_pvc_usage(lines: list[str], summary: dict[str, Any]) -> None:
         return
     parts = []
     for entry in pvc_usage:
-        metric = entry.get("metric") if isinstance(entry, dict) else {}
+        if not isinstance(entry, dict):
+            continue
+        metric = entry.get("metric") if isinstance(entry.get("metric"), dict) else {}
         namespace = metric.get("namespace")
         pvc = metric.get("persistentvolumeclaim")
         value = entry.get("value")
@@ -478,8 +480,6 @@ def _append_pvc_usage(lines: list[str], summary: dict[str, Any]) -> None:
             parts.append(f"{namespace}/{pvc}={_format_float(value)}%")
     if parts:
         lines.append("pvc_usage_top: " + "; ".join(parts))
-
-
 def _append_root_disk_headroom(lines: list[str], summary: dict[str, Any]) -> None:
     headroom = summary.get("root_disk_low_headroom")
     if not isinstance(headroom, list) or not headroom:

atlasbot/snapshot/builder/format_b.py

@@ -6,6 +6,25 @@ from .core_a import _VALUE_PAIR_LEN
 from .format_a import *
 
 
+def _append_namespace_metric_series(
+    lines: list[str],
+    label: str,
+    entries: list[Any],
+    formatter: Any,
+) -> None:
+    parts = []
+    for entry in entries:
+        if not isinstance(entry, dict):
+            continue
+        metric = entry.get("metric") if isinstance(entry.get("metric"), dict) else {}
+        namespace = metric.get("namespace")
+        value = entry.get("value")
+        if namespace:
+            parts.append(f"{namespace}={formatter(value)}")
+    if parts:
+        lines.append(f"{label}: " + "; ".join(parts))
+
+
 def _append_longhorn(lines: list[str], summary: dict[str, Any]) -> None:  # noqa: C901
     longhorn = summary.get("longhorn") if isinstance(summary.get("longhorn"), dict) else {}
     if not longhorn:
@@ -52,78 +71,24 @@ def _append_namespace_usage(lines: list[str], summary: dict[str, Any]) -> None:
     metrics = summary.get("metrics") if isinstance(summary.get("metrics"), dict) else {}
     cpu_top = metrics.get("namespace_cpu_top") if isinstance(metrics.get("namespace_cpu_top"), list) else []
     mem_top = metrics.get("namespace_mem_top") if isinstance(metrics.get("namespace_mem_top"), list) else []
-    if cpu_top:
+    _append_namespace_metric_series(lines, "namespace_cpu_top", cpu_top, _format_float)
-        parts = []
+    _append_namespace_metric_series(lines, "namespace_mem_top", mem_top, _format_bytes)
-        for entry in cpu_top:
-            metric = entry.get("metric") if isinstance(entry, dict) else {}
-            namespace = metric.get("namespace")
-            value = entry.get("value")
-            if namespace:
-                parts.append(f"{namespace}={_format_float(value)}")
-        if parts:
-            lines.append("namespace_cpu_top: " + "; ".join(parts))
-    if mem_top:
-        parts = []
-        for entry in mem_top:
-            metric = entry.get("metric") if isinstance(entry, dict) else {}
-            namespace = metric.get("namespace")
-            value = entry.get("value")
-            if namespace:
-                parts.append(f"{namespace}={_format_bytes(value)}")
-        if parts:
-            lines.append("namespace_mem_top: " + "; ".join(parts))
 
 
 def _append_namespace_requests(lines: list[str], summary: dict[str, Any]) -> None:
     metrics = summary.get("metrics") if isinstance(summary.get("metrics"), dict) else {}
     cpu_req = metrics.get("namespace_cpu_requests_top") if isinstance(metrics.get("namespace_cpu_requests_top"), list) else []
     mem_req = metrics.get("namespace_mem_requests_top") if isinstance(metrics.get("namespace_mem_requests_top"), list) else []
-    if cpu_req:
+    _append_namespace_metric_series(lines, "namespace_cpu_requests_top", cpu_req, _format_float)
-        parts = []
+    _append_namespace_metric_series(lines, "namespace_mem_requests_top", mem_req, _format_bytes)
-        for entry in cpu_req:
-            metric = entry.get("metric") if isinstance(entry, dict) else {}
-            namespace = metric.get("namespace")
-            value = entry.get("value")
-            if namespace:
-                parts.append(f"{namespace}={_format_float(value)}")
-        if parts:
-            lines.append("namespace_cpu_requests_top: " + "; ".join(parts))
-    if mem_req:
-        parts = []
-        for entry in mem_req:
-            metric = entry.get("metric") if isinstance(entry, dict) else {}
-            namespace = metric.get("namespace")
-            value = entry.get("value")
-            if namespace:
-                parts.append(f"{namespace}={_format_bytes(value)}")
-        if parts:
-            lines.append("namespace_mem_requests_top: " + "; ".join(parts))
 
 
 def _append_namespace_io_net(lines: list[str], summary: dict[str, Any]) -> None:
     metrics = summary.get("metrics") if isinstance(summary.get("metrics"), dict) else {}
     net_top = metrics.get("namespace_net_top") if isinstance(metrics.get("namespace_net_top"), list) else []
     io_top = metrics.get("namespace_io_top") if isinstance(metrics.get("namespace_io_top"), list) else []
-    if net_top:
+    _append_namespace_metric_series(lines, "namespace_net_top", net_top, _format_rate_bytes)
-        parts = []
+    _append_namespace_metric_series(lines, "namespace_io_top", io_top, _format_rate_bytes)
-        for entry in net_top:
-            metric = entry.get("metric") if isinstance(entry, dict) else {}
-            namespace = metric.get("namespace")
-            value = entry.get("value")
-            if namespace:
-                parts.append(f"{namespace}={_format_rate_bytes(value)}")
-        if parts:
-            lines.append("namespace_net_top: " + "; ".join(parts))
-    if io_top:
-        parts = []
-        for entry in io_top:
-            metric = entry.get("metric") if isinstance(entry, dict) else {}
-            namespace = metric.get("namespace")
-            value = entry.get("value")
-            if namespace:
-                parts.append(f"{namespace}={_format_rate_bytes(value)}")
-        if parts:
-            lines.append("namespace_io_top: " + "; ".join(parts))
 
 
 def _append_pod_usage(lines: list[str], summary: dict[str, Any]) -> None:  # noqa: C901
@@ -143,7 +108,9 @@ def _append_pod_usage(lines: list[str], summary: dict[str, Any]) -> None:  # noq
     if cpu_top:
         parts = []
         for entry in cpu_top:
-            metric = entry.get("metric") if isinstance(entry, dict) else {}
+            if not isinstance(entry, dict):
+                continue
+            metric = entry.get("metric") if isinstance(entry.get("metric"), dict) else {}
             namespace = metric.get("namespace")
             pod = metric.get("pod")
             value = entry.get("value")
@@ -154,7 +121,9 @@ def _append_pod_usage(lines: list[str], summary: dict[str, Any]) -> None:  # noq
     if cpu_top_node:
         parts = []
         for entry in cpu_top_node:
-            metric = entry.get("metric") if isinstance(entry, dict) else {}
+            if not isinstance(entry, dict):
+                continue
+            metric = entry.get("metric") if isinstance(entry.get("metric"), dict) else {}
             namespace = metric.get("namespace")
             pod = metric.get("pod")
             node = metric.get("node")
@@ -166,7 +135,9 @@ def _append_pod_usage(lines: list[str], summary: dict[str, Any]) -> None:  # noq
     if mem_top:
         parts = []
         for entry in mem_top:
-            metric = entry.get("metric") if isinstance(entry, dict) else {}
+            if not isinstance(entry, dict):
+                continue
+            metric = entry.get("metric") if isinstance(entry.get("metric"), dict) else {}
             namespace = metric.get("namespace")
             pod = metric.get("pod")
             value = entry.get("value")
@@ -177,7 +148,9 @@ def _append_pod_usage(lines: list[str], summary: dict[str, Any]) -> None:  # noq
     if mem_top_node:
         parts = []
         for entry in mem_top_node:
-            metric = entry.get("metric") if isinstance(entry, dict) else {}
+            if not isinstance(entry, dict):
+                continue
+            metric = entry.get("metric") if isinstance(entry.get("metric"), dict) else {}
             namespace = metric.get("namespace")
             pod = metric.get("pod")
             node = metric.get("node")
@@ -230,7 +203,9 @@ def _append_job_failures(lines: list[str], summary: dict[str, Any]) -> None:
         return
     parts = []
     for entry in failures:
-        metric = entry.get("metric") if isinstance(entry, dict) else {}
+        if not isinstance(entry, dict):
+            continue
+        metric = entry.get("metric") if isinstance(entry.get("metric"), dict) else {}
         namespace = metric.get("namespace")
         job_name = metric.get("job_name") or metric.get("job")
         value = entry.get("value")
@@ -323,7 +298,9 @@ def _append_postgres(lines: list[str], summary: dict[str, Any]) -> None:
     if isinstance(by_db, list) and by_db:
         parts = []
         for entry in by_db:
-            metric = entry.get("metric") if isinstance(entry, dict) else {}
+            if not isinstance(entry, dict):
+                continue
+            metric = entry.get("metric") if isinstance(entry.get("metric"), dict) else {}
             value = entry.get("value")
             if isinstance(value, list) and len(value) >= _VALUE_PAIR_LEN:
                 value = value[1]

atlasbot/snapshot/builder/format_c.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from typing import Any
 
+from .core_a import PVC_USAGE_CRITICAL
 from .format_b import *
 def _append_signals(lines: list[str], summary: dict[str, Any]) -> None:
     signals = summary.get("signals") if isinstance(summary.get("signals"), list) else []

scripts/check_coverage.py

@@ -5,35 +5,20 @@ from __future__ import annotations
 
 import argparse
 import json
-from datetime import date
 from pathlib import Path
 
 
-def main() -> int:  # noqa: C901
+def main() -> int:
     """Check each production file against a minimum coverage percentage."""
 
     parser = argparse.ArgumentParser()
     parser.add_argument("coverage_json")
     parser.add_argument("--root", default="atlasbot")
     parser.add_argument("--threshold", type=float, default=95.0)
-    parser.add_argument("--exceptions-file", default="testing/coverage_exceptions.json")
     args = parser.parse_args()
 
     data = json.loads(Path(args.coverage_json).read_text(encoding="utf-8"))
     files = data.get("files") if isinstance(data, dict) else {}
-    exceptions_path = Path(args.exceptions_file)
-    per_file_thresholds: dict[str, float] = {}
-    if exceptions_path.exists():
-        payload = json.loads(exceptions_path.read_text(encoding="utf-8"))
-        expires_on = str(payload.get("expires_on") or "").strip()
-        if expires_on and date.today() > date.fromisoformat(expires_on):
-            print(f"coverage exceptions expired on {expires_on}: {exceptions_path}")
-            return 1
-        overrides = payload.get("per_file_thresholds")
-        if isinstance(overrides, dict):
-            for path, threshold in overrides.items():
-                if isinstance(path, str) and isinstance(threshold, (int, float)):
-                    per_file_thresholds[path] = float(threshold)
     violations: list[tuple[float, str]] = []
     for path, payload in sorted(files.items()):
         if not path.startswith(f"{args.root}/"):
@@ -42,9 +27,8 @@ def main() -> int:  # noqa: C901
         percent = summary.get("percent_covered") if isinstance(summary, dict) else None
         if not isinstance(percent, (int, float)):
             continue
-        threshold = per_file_thresholds.get(path, float(args.threshold))
+        if float(percent) < args.threshold:
-        if float(percent) < threshold:
+            violations.append((float(percent), path))
-            violations.append((float(percent), f"{path} (min {threshold:.2f}%)"))
 
     if violations:
         for percent, path in sorted(violations):
@@ -55,3 +39,4 @@ def main() -> int:  # noqa: C901
 
 if __name__ == "__main__":
     raise SystemExit(main())
+