refactor: remove unnecessary complexity (ponytail review)

src/wikibricks/automation.py

@@ -5,7 +5,6 @@
 import asyncio
 import logging
 import os
-import threading
 import time
 from collections.abc import Callable
 from contextlib import contextmanager
@@ -209,23 +208,8 @@ async def run_background_loop() -> None:
         await asyncio.sleep(delay)
 
 
-def run_background_worker(stop: threading.Event) -> None:
-    """Run maintenance for a native host until its lifecycle ends."""
-    while not stop.is_set():
-        delay = 300
-        try:
-            config = load_config()
-            delay = config.automation_poll_seconds
-            if config.automation_enabled:
-                run_background_cycle(config)
-        except Exception as exc:
-            _LOGGER.warning("WikiBricks background cycle failed: %s", exc)
-        stop.wait(delay)
-
-
 __all__ = [
     "run_background_cycle",
     "run_background_loop",
-    "run_background_worker",
     "run_remote_cycle",
 ]

src/wikibricks/client.py

@@ -6,7 +6,6 @@
 import json
 import os
 import re
-import warnings
 from pathlib import Path
 from typing import Any
 
@@ -40,7 +39,6 @@ def write_page(self, *args, **kwargs) -> str:
         return self.store.write_page(*args, **kwargs)
 
     def write_pages(self, pages: list[dict[str, Any]]) -> int:
-        written = 0
         for page in pages:
             content = page.get("content", page.get("content_json"))
             self.write_page(
@@ -55,8 +53,7 @@ def write_pages(self, pages: list[dict[str, Any]]) -> int:
                 chunk_index=page.get("chunk_index"),
                 content_text_override=page.get("content_text_override"),
             )
-            written += 1
-        return written
+        return len(pages)
 
     def read_page(self, path: str) -> dict[str, Any] | None:
         page = self.store.read_page(path)
@@ -81,13 +78,8 @@ def history(self, path: str) -> list[dict[str, Any]]:
     def search(
         self,
         query: str,
-        mode: str = "HYBRID",
         num_results: int = 5,
-        rerank_by_citations: bool | None = None,
-        rerank_with_pagerank: bool | None = None,
-        include_ephemeral: bool = False,
     ) -> list[dict[str, Any]]:
-        del mode, rerank_by_citations, rerank_with_pagerank, include_ephemeral
         hits = self.store.search(query, num_results=num_results)
         self._log("search", query=query)
         return hits
@@ -231,40 +223,6 @@ def materialize_index(self) -> str:
         )
         return f"Materialized index with {len(pages)} pages"
 
-    def sync_index(self) -> None:
-        warnings.warn(
-            "sync_index is unnecessary for local search",
-            DeprecationWarning,
-            stacklevel=2,
-        )
-
-    def index_row_count(self) -> int:
-        return len(self.list_pages(include_ephemeral=True))
-
-    def reconcile_vs_source(self) -> int:
-        return 0
-
-    def list_recent_by_cwd_tag(
-        self, cwd_basename: str, limit: int = 3
-    ) -> list[dict[str, Any]]:
-        if not cwd_basename:
-            return []
-        with self.store.connection() as conn:
-            rows = conn.execute(
-                "SELECT page_path, title, updated_at FROM sessions "
-                "WHERE workspace LIKE ? ORDER BY updated_at DESC LIMIT ?",
-                (f"%/{cwd_basename}", limit),
-            ).fetchall()
-        return [
-            {
-                "path": row[0],
-                "title": row[1],
-                "summary": row[1],
-                "updated_at": row[2],
-            }
-            for row in rows
-        ]
-
     def _log(self, op_type, path=None, query=None, details=None) -> None:
         try:
             self.store.log(op_type, path=path, query=query, details=details)

src/wikibricks/config/__init__.py

@@ -168,12 +168,7 @@ def _environment_overlay(environ: Mapping[str, str]) -> dict[str, Any]:
             value = convert(raw)
         except ValueError as exc:
             raise ValueError(f"{name} has an invalid value") from exc
-        target = result.setdefault(section, {})
-        if "." in key:
-            parent, child = key.split(".", 1)
-            target.setdefault(parent, {})[child] = value
-        else:
-            target[key] = value
+        result.setdefault(section, {})[key] = value
     return result
 
 

src/wikibricks/curation/application.py

@@ -164,47 +164,30 @@ def _apply_patch(
     proposal = patch["proposal"]
     patch_id = UUID(patch["patch_id"])
     if operation in {"create_page", "update_page"}:
-        if isinstance(store, SQLiteStore):
-            expected = None if operation == "create_page" else patch["base_content_hash"]
-            if force:
-                _message, version_id = store.write_page_in_connection(
-                    conn,
-                    patch["path"],
-                    proposal["title"],
-                    proposal["content"],
-                    page_type=proposal["page_type"],
-                    created_by=created_by,
-                    tags=proposal["tags"],
-                    source_ids=proposal["source_ids"],
-                    parent_id=proposal["parent_id"],
-                    chunk_index=proposal["chunk_index"],
-                    content_text=proposal["content_text"],
-                    curation_patch_id=str(patch_id),
-                )
-            else:
-                _message, version_id = store.write_page_in_connection(
-                    conn,
-                    patch["path"],
-                    proposal["title"],
-                    proposal["content"],
-                    page_type=proposal["page_type"],
-                    created_by=created_by,
-                    tags=proposal["tags"],
-                    source_ids=proposal["source_ids"],
-                    parent_id=proposal["parent_id"],
-                    chunk_index=proposal["chunk_index"],
-                    content_text=proposal["content_text"],
-                    curation_patch_id=str(patch_id),
-                    expected_base_content_hash=expected,
-                )
-            return UUID(version_id)
         kwargs: dict[str, Any] = {}
         if not force:
             kwargs["expected_base_content_hash"] = (
                 None
                 if operation == "create_page"
                 else patch["base_content_hash"]
             )
+        if isinstance(store, SQLiteStore):
+            _message, version_id = store.write_page_in_connection(
+                conn,
+                patch["path"],
+                proposal["title"],
+                proposal["content"],
+                page_type=proposal["page_type"],
+                created_by=created_by,
+                tags=proposal["tags"],
+                source_ids=proposal["source_ids"],
+                parent_id=proposal["parent_id"],
+                chunk_index=proposal["chunk_index"],
+                content_text=proposal["content_text"],
+                curation_patch_id=str(patch_id),
+                **kwargs,
+            )
+            return UUID(version_id)
         _message, version_id = store.pages.write_in_connection(
             conn,
             patch["path"],

src/wikibricks/curation/backlog.py

@@ -108,7 +108,6 @@ def curation_backlog(
             {
                 **item,
                 "workspaces": sorted(item["workspaces"]),
-                "workspace": item["workspace"],
                 "new_sessions": new_sessions,
                 "pages": project_pages[:3],
                 "last_page_update": (

src/wikibricks/export_karpathy.py

@@ -75,17 +75,13 @@ def main() -> int:
     p = argparse.ArgumentParser(description=__doc__)
     p.add_argument("target_dir", help="where to write the markdown tree")
     p.add_argument("--database-url", help="Optional PostgreSQL connection URL")
-    p.add_argument("--limit", type=int, default=None,
-                   help="cap number of pages exported (testing)")
     args = p.parse_args()
 
     from wikibricks import WikiClient
 
     wiki = WikiClient(args.database_url)
 
     pages, edges = fetch_pages_and_edges(wiki)
-    if args.limit is not None:
-        pages = pages[: args.limit]
     print(f"exporting {len(pages)} pages, {len(edges)} edges to {args.target_dir}")
 
     target = Path(args.target_dir).resolve()

src/wikibricks/import_karpathy.py

@@ -39,27 +39,24 @@ def _read_page_record(file: Path, base_dir: Path) -> dict:
     meta, body = parse_frontmatter(text)
 
     path = wiki_path_for(str(file), base_dir=str(base_dir), frontmatter=meta)
-    title = meta.get("title")
-    if not title:
-        # First H1 in body, or filename slug as fallback.
-        for line in body.splitlines():
-            if line.startswith("# "):
-                title = line[2:].strip()
-                break
-        if not title:
-            title = file.stem.replace("-", " ").title()
+    title = (
+        meta.get("title")
+        or next(
+            (line[2:].strip() for line in body.splitlines() if line.startswith("# ")),
+            "",
+        )
+        or file.stem.replace("-", " ").title()
+    )
 
     tags = meta.get("tags", [])
     if isinstance(tags, str):
         tags = [t.strip() for t in tags.split(",") if t.strip()]
     tags = [str(t) for t in tags]
 
-    summary = ""
-    for line in body.splitlines():
-        s = line.strip()
-        if s and not s.startswith("#"):
-            summary = s[:200]
-            break
+    summary = next(
+        (s[:200] for s in (line.strip() for line in body.splitlines()) if s and not s.startswith("#")),
+        "",
+    )
 
     return {
         "path": path,

src/wikibricks/karpathy_logic.py

@@ -119,9 +119,7 @@ def map_wiki_path_to_file(wiki_path: str) -> str:
 
     Subfolders are preserved. Existing `.md` suffix is not duplicated.
     """
-    if wiki_path.endswith(".md"):
-        return wiki_path
-    return f"{wiki_path}.md"
+    return wiki_path.removesuffix(".md") + ".md"
 
 
 def _render_yaml_value(v) -> str:

src/wikibricks/migrate_postgres.py

@@ -41,8 +41,6 @@ class MigrationReport:
 def _sqlite_value(value: Any) -> Any:
     if value is None or isinstance(value, (str, int, float, bytes)):
         return value
-    if isinstance(value, bool):
-        return int(value)
     if isinstance(value, UUID):
         return str(value)
     if isinstance(value, (date, datetime)):

src/wikibricks/models.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import re
-from dataclasses import dataclass, field
+from dataclasses import asdict, dataclass, field
 from typing import Any
 
 SESSION_EVENT_KINDS = frozenset(
@@ -92,13 +92,7 @@ def from_dict(cls, value: dict[str, Any]) -> SessionEvent:
         )
 
     def to_dict(self) -> dict[str, Any]:
-        return {
-            "external_id": self.external_id,
-            "kind": self.kind,
-            "content": self.content,
-            "created_at": self.created_at,
-            "metadata": self.metadata,
-        }
+        return asdict(self)
 
 
 @dataclass(frozen=True, slots=True)
@@ -146,14 +140,4 @@ def from_dict(cls, value: dict[str, Any]) -> SessionRecord:
         )
 
     def to_dict(self) -> dict[str, Any]:
-        return {
-            "harness": self.harness,
-            "external_id": self.external_id,
-            "user_id": self.user_id,
-            "agent": self.agent,
-            "workspace": self.workspace,
-            "started_at": self.started_at,
-            "updated_at": self.updated_at,
-            "events": [event.to_dict() for event in self.events],
-            "metadata": self.metadata,
-        }
+        return asdict(self)

src/wikibricks/omnigent_install.py

@@ -30,6 +30,18 @@
 _MINIMUM_VERSION = (0, 11, 0)
 _VERSION_PATTERN = re.compile(r"\b(\d+)\.(\d+)\.(\d+)(?:[^\s]*)?")
 _MCP_NAME = "wikibricks"
+_OWNED_SETTINGS = {
+    "debby": "Claude mcpServers.wikibricks",
+    "polly": "Claude mcpServers.wikibricks",
+    "goose": "extensions.wikibricks",
+    "hermes": "mcp_servers.wikibricks",
+    "opencode": "mcp.wikibricks",
+    "kimi": "mcpServers.wikibricks",
+    "kiro": "mcpServers.wikibricks",
+    "qwen": "mcpServers.wikibricks",
+    "claude": "mcpServers.wikibricks",
+    "codex": "mcp_servers.wikibricks",
+}
 _FILE_HARNESSES = {
     "goose": "goose",
     "hermes": "hermes",
@@ -431,20 +443,9 @@ def install_integrations(
         owned_files.extend((str(paths["pi_config"]), str(paths["pi_extension"])))
     owned_files.extend(str(resolved_home / ".wikibricks" / "bin" / name) for name in launchers)
     owned_files.append(str(paths["manifest"]))
-    owned_settings: dict[str, list[str]] = {}
-    for name in statuses:
-        if name in {"debby", "polly"}:
-            owned_settings[name] = ["Claude mcpServers.wikibricks"]
-        elif name == "goose":
-            owned_settings[name] = ["extensions.wikibricks"]
-        elif name == "hermes":
-            owned_settings[name] = ["mcp_servers.wikibricks"]
-        elif name == "opencode":
-            owned_settings[name] = ["mcp.wikibricks"]
-        elif name in {"kimi", "kiro", "qwen", "claude"}:
-            owned_settings[name] = ["mcpServers.wikibricks"]
-        elif name == "codex":
-            owned_settings[name] = ["mcp_servers.wikibricks"]
+    owned_settings = {
+        name: [_OWNED_SETTINGS[name]] for name in statuses if name in _OWNED_SETTINGS
+    }
     if wrapper_paths:
         owned_settings["omnigent"] = [
             *(f"harness.{name}.command" for name in sorted(wrapper_paths)),