fix(ci): clean main pipeline — ruff, tiktoken-skip, deterministic timeout

Three concerns resolved in one pass:

1. ruff — 46 lint violations (31 auto-fixable F541/F401/UP035, 15 manual:
   E402 import-reorder in scaffold/_core.py + server.py, F841 unused vars,
   E702 semicolon-statement splits in tests).

2. tiktoken — tests/scaffold/test_real_tokenizer.py imported tiktoken at
   module top, hard-failing CI collection. tiktoken is not a declared dev
   dep (it is a tokenizer-validation utility). Switched to
   pytest.importorskip so the file runs locally where tiktoken is
   installed and skips cleanly in CI.

3. timeout race — test_decompose_timeout_returns_degraded_flag and
   test_decompose_timeout_vector_only_results_returned forced the
   fallback by setting FIDELIS_DECOMPOSE_TIMEOUT_SECS=0.001, hoping
   do_recall would take longer than 1ms. Fast CI runners win that race;
   the fallback never executes; degraded-flag missing; assert fails.
   Fix: patch fidelis.server.do_recall with a stub that sleeps 1.0s,
   raise the timeout to 0.1s. 10x margin is race-immune.

Verified locally: 481 passed, 1 xpassed, 0 failures (excluding Ollama
integration suites which skip cleanly in CI).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: roli-lpci

src/fidelis/cli.py

@@ -38,7 +38,7 @@ def _server_error(exc: BaseException) -> None:
     print(f"Error: fidelis-server unreachable or unhealthy at {_base_url()}", file=sys.stderr)
     print(f"  reason: {msg}", file=sys.stderr)
     print("  • If you haven't installed the service: `fidelis init`", file=sys.stderr)
-    print(f"  • If the service is installed: `tail ~/.fidelis/server.log`", file=sys.stderr)
+    print("  • If the service is installed: `tail ~/.fidelis/server.log`", file=sys.stderr)
     sys.exit(1)
 
 

src/fidelis/init_cmd.py

@@ -342,11 +342,11 @@ def cmd_init(args) -> int:
         print(f"  log: {Path.home() / '.fidelis' / 'server.log'}")
         print()
         print("next steps:")
-        print(f"  fidelis health                  # confirm")
-        print(f"  fidelis watch ~/notes           # auto-ingest a directory")
-        print(f"  fidelis mcp install             # wire up Claude Code")
+        print("  fidelis health                  # confirm")
+        print("  fidelis watch ~/notes           # auto-ingest a directory")
+        print("  fidelis mcp install             # wire up Claude Code")
         return 0
     else:
-        print(f"✗ service installed but /health did not respond within 10s")
+        print("✗ service installed but /health did not respond within 10s")
         print(f"  check log: {Path.home() / '.fidelis' / 'server.log'}")
         return 2

src/fidelis/scaffold/_core.py

@@ -17,6 +17,8 @@
 
 from __future__ import annotations
 
+import math as _math
+
 # Version banner — used as scaffold-presence marker for drift measurement.
 SCAFFOLD_VERSION = "v0.1.0"
 SCAFFOLD_OPEN = f"[FIDELIS-SCAFFOLD-{SCAFFOLD_VERSION}]"
@@ -34,8 +36,6 @@
 # Confidence signal — inline meta-information about retrieval quality.
 # Helps the LLM calibrate its confidence to actual retrieval quality.
 
-import math as _math
-
 
 def _sanitize_top_score(top_score: float | None) -> float | None:
     """Sanitize a raw retrieval score to [0, 1] or None.

src/fidelis/server.py

@@ -47,8 +47,6 @@
 import sys
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
-logger = logging.getLogger("cogito.server")
-
 from fidelis import __version__
 from fidelis.config import load, mem0_config
 from fidelis.degrade import queued_count, replay_queue, safe_add
@@ -57,6 +55,8 @@
 from fidelis.recall_hybrid import recall_hybrid as do_recall_hybrid
 from fidelis.snapshot import _read_snapshot, _snapshot_path
 
+logger = logging.getLogger("cogito.server")
+
 
 def _boot(cfg: dict) -> object:
     """Import mem0 from wherever it's installed and return a Memory instance."""
@@ -73,7 +73,6 @@ def _boot(cfg: dict) -> object:
 
 def make_handler(memory: object, cfg: dict) -> type:
     user_id: str = cfg["user_id"]
-    query_threshold: float = cfg.get("query_threshold", 250.0)
     # FIDELIS_DECOMPOSE_TIMEOUT_SECS: max seconds for /recall sub-query pipeline.
     # Default 8s preserves existing behavior in normal cases; kicks in only on slow-call edges.
     _decompose_timeout: float = float(os.environ.get("FIDELIS_DECOMPOSE_TIMEOUT_SECS", 8))

tests/scaffold/test_backend_portability.py

@@ -23,7 +23,6 @@
 
 import json
 import subprocess
-import time
 import urllib.error
 import urllib.request
 from typing import NamedTuple
@@ -397,7 +396,7 @@ def call(sys_prompt, user_msg):
     hedge_rate  = results["hedge_compliance_rate"]
     answer_rate = results["answer_compliance_rate"]
 
-    print(f"\n=== claude-cli ===")
+    print("\n=== claude-cli ===")
     print(f"Hedge compliance : {hedge_rate:.0%} ({sum(r['compliant'] for r in results['hedge_results'])}/{len(results['hedge_results'])})")
     print(f"Answer compliance: {answer_rate:.0%} ({sum(r['compliant'] for r in results['answer_results'])}/{len(results['answer_results'])})")
     print("\nHedge questions:")

tests/scaffold/test_consumer_surface.py

@@ -8,7 +8,6 @@
 from __future__ import annotations
 
 import json
-from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -149,10 +148,13 @@ def test_mcp_uninstall_removes_entry(tmp_path):
     settings_path = tmp_path / "settings.local.json"
     settings_path.write_text(json.dumps({"mcpServers": {}}))
 
-    args_install = MagicMock(); args_install.settings = str(settings_path); args_install.force = False
+    args_install = MagicMock()
+    args_install.settings = str(settings_path)
+    args_install.force = False
     cmd_mcp_install(args_install)
 
-    args_uninstall = MagicMock(); args_uninstall.settings = str(settings_path)
+    args_uninstall = MagicMock()
+    args_uninstall.settings = str(settings_path)
     rc = cmd_mcp_uninstall(args_uninstall)
     assert rc == 0
 
@@ -163,7 +165,8 @@ def test_mcp_uninstall_removes_entry(tmp_path):
 def test_mcp_uninstall_handles_missing_settings(tmp_path):
     from fidelis.mcp_cmd import cmd_mcp_uninstall
     settings_path = tmp_path / "no-such-file.json"
-    args = MagicMock(); args.settings = str(settings_path)
+    args = MagicMock()
+    args.settings = str(settings_path)
     rc = cmd_mcp_uninstall(args)
     assert rc == 0  # graceful no-op
 

tests/scaffold/test_e2e_store_query.py

@@ -15,7 +15,6 @@
 import os
 import subprocess
 import sys
-import tempfile
 import time
 import urllib.error
 import urllib.request

tests/scaffold/test_openai_format_compatibility.py

@@ -140,8 +140,8 @@ def test_scaffold_markers_appear_verbatim(self):
         serialized = json.dumps(payload)
         # JSON-encode the markers to see what they look like inside the string
         # The markers contain only ASCII, so they should appear literally
-        assert SCAFFOLD_OPEN in serialized, f"SCAFFOLD_OPEN not found in serialized payload"
-        assert SCAFFOLD_CLOSE in serialized, f"SCAFFOLD_CLOSE not found in serialized payload"
+        assert SCAFFOLD_OPEN in serialized, "SCAFFOLD_OPEN not found in serialized payload"
+        assert SCAFFOLD_CLOSE in serialized, "SCAFFOLD_CLOSE not found in serialized payload"
 
     def test_scaffold_content_roundtrips_through_json(self):
         """Scaffold content must survive a JSON round-trip without corruption."""
@@ -322,15 +322,14 @@ def test_answer_compliance(self):
     def test_full_classification_summary(self, capsys):
         """Run all 10 questions, print per-question classification for the report."""
         print(f"\n{'='*60}")
-        print(f"gpt-oss:20b — OpenAI wire format smoke test")
+        print("gpt-oss:20b — OpenAI wire format smoke test")
         print(f"{'='*60}")
 
         all_results = []
         for q in QUESTIONS:
             messages = self._build_messages(q)
             response = _call_openai_chat(messages)
             classification = _classify_response(response)
-            expected = "HEDGED" if q.should_hedge else "ANSWERED/OTHER"
             correct = (
                 (q.should_hedge and classification == "HEDGED")
                 or (not q.should_hedge and classification != "HEDGED")

tests/scaffold/test_real_tokenizer.py

@@ -14,9 +14,10 @@
 from __future__ import annotations
 
 import pytest
-import tiktoken
 
-from fidelis.scaffold._core import wrap_system_prompt, _QTYPE_PROC
+tiktoken = pytest.importorskip("tiktoken")
+
+from fidelis.scaffold._core import wrap_system_prompt, _QTYPE_PROC  # noqa: E402
 
 # --------------------------------------------------------------------------- #
 # Fixtures / helpers

tests/scaffold/test_streaming_marker_integrity.py

@@ -19,10 +19,8 @@
 from __future__ import annotations
 
 import json
-from typing import Iterator
 
 import httpx
-import pytest
 
 import anthropic
 
@@ -236,7 +234,7 @@ def test_scaffold_open_marker_present_and_intact_in_wire_body(self):
             f"Got: {wire_system[:120]!r}"
         )
         assert SCAFFOLD_CLOSE in wire_system, (
-            f"SCAFFOLD_CLOSE marker missing from wire body."
+            "SCAFFOLD_CLOSE marker missing from wire body."
         )
 
     def test_scaffold_open_marker_at_wire_body_start(self):