diff --git a/src/wikibricks/automation.py b/src/wikibricks/automation.py index 4dc1cdf..3b85b54 100644 --- a/src/wikibricks/automation.py +++ b/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", ] diff --git a/src/wikibricks/client.py b/src/wikibricks/client.py index 1c0e3f3..fa5e588 100644 --- a/src/wikibricks/client.py +++ b/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) diff --git a/src/wikibricks/config/__init__.py b/src/wikibricks/config/__init__.py index b9956b4..5aaa109 100644 --- a/src/wikibricks/config/__init__.py +++ b/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 diff --git a/src/wikibricks/curation/application.py b/src/wikibricks/curation/application.py index e91a2ed..ef2f2c4 100644 --- a/src/wikibricks/curation/application.py +++ b/src/wikibricks/curation/application.py @@ -164,40 +164,6 @@ 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"] = ( @@ -205,6 +171,23 @@ def _apply_patch( 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"], diff --git a/src/wikibricks/curation/backlog.py b/src/wikibricks/curation/backlog.py index 7dbd3fb..21bcf6c 100644 --- a/src/wikibricks/curation/backlog.py +++ b/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": ( diff --git a/src/wikibricks/export_karpathy.py b/src/wikibricks/export_karpathy.py index 754b8d5..55e23f9 100644 --- a/src/wikibricks/export_karpathy.py +++ b/src/wikibricks/export_karpathy.py @@ -75,8 +75,6 @@ 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 @@ -84,8 +82,6 @@ def main() -> int: 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() diff --git a/src/wikibricks/import_karpathy.py b/src/wikibricks/import_karpathy.py index 4ebc998..85735ef 100644 --- a/src/wikibricks/import_karpathy.py +++ b/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, diff --git a/src/wikibricks/karpathy_logic.py b/src/wikibricks/karpathy_logic.py index 13fcc03..aeb7be1 100644 --- a/src/wikibricks/karpathy_logic.py +++ b/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: diff --git a/src/wikibricks/migrate_postgres.py b/src/wikibricks/migrate_postgres.py index 2fca002..191a343 100644 --- a/src/wikibricks/migrate_postgres.py +++ b/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)): diff --git a/src/wikibricks/models.py b/src/wikibricks/models.py index c9d38db..d3491dd 100644 --- a/src/wikibricks/models.py +++ b/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) diff --git a/src/wikibricks/omnigent_install.py b/src/wikibricks/omnigent_install.py index d00773f..9151942 100644 --- a/src/wikibricks/omnigent_install.py +++ b/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)), diff --git a/src/wikibricks/remote/lakebase.py b/src/wikibricks/remote/lakebase.py index 952a883..c943096 100644 --- a/src/wikibricks/remote/lakebase.py +++ b/src/wikibricks/remote/lakebase.py @@ -5,7 +5,7 @@ import hashlib import json from collections.abc import Callable -from dataclasses import dataclass +from dataclasses import asdict, dataclass from typing import TYPE_CHECKING, Any from uuid import NAMESPACE_URL, UUID, uuid5 @@ -85,6 +85,10 @@ def _canonical_json(value: Any) -> str: ) +def _row_dict(cursor: Any, row: tuple[Any, ...]) -> dict[str, Any]: + return dict(zip([column[0] for column in cursor.description], row)) + + def _resolve_event( store: PostgresStore | SQLiteStore, row: dict[str, Any], @@ -94,78 +98,52 @@ def _resolve_event( placeholder = "?" if isinstance(store, SQLiteStore) else "%s" with store.connection() as conn: if row["entity_kind"] == "page_version": - payload_row = conn.execute( - "SELECT p.path, v.version, v.title, v.page_type, v.content, v.content_text, " - "v.tags, v.source_ids, v.parent_id, v.chunk_index, v.created_by, v.created_at, " - "v.curation_patch_id " + cursor = conn.execute( + "SELECT p.path AS path, v.version AS version, v.title AS title, " + "v.page_type AS page_type, v.content AS content, " + "v.content_text AS content_text, v.tags AS tags, " + "v.source_ids AS source_ids, v.parent_id AS parent_id, " + "v.chunk_index AS chunk_index, v.created_by AS created_by, " + "v.created_at AS created_at, v.curation_patch_id AS curation_patch_id " "FROM page_versions v JOIN pages p ON p.page_id = v.page_id " f"WHERE v.version_id = {placeholder}", (row["version_id"],), - ).fetchone() + ) + payload_row = cursor.fetchone() if not payload_row: raise RuntimeError(f"missing immutable page version {row['version_id']}") - payload = { - "path": payload_row[0], - "version": payload_row[1], - "title": payload_row[2], - "page_type": payload_row[3], - "content": payload_row[4], - "content_text": payload_row[5], - "tags": payload_row[6], - "source_ids": payload_row[7], - "parent_id": payload_row[8], - "chunk_index": payload_row[9], - "created_by": payload_row[10], - "created_at": payload_row[11], - "curation_patch_id": payload_row[12], - } + payload = _row_dict(cursor, payload_row) elif row["entity_kind"] == "session_event_version": - payload_row = conn.execute( - "SELECT s.harness, s.external_id, s.user_id, s.agent, s.workspace, " - "s.page_path, e.external_id, e.position, v.version, v.kind, v.content, " - "v.metadata, v.source_created_at, v.created_at " + cursor = conn.execute( + "SELECT s.harness AS harness, s.external_id AS session_external_id, " + "s.user_id AS user_id, s.agent AS agent, s.workspace AS workspace, " + "s.page_path AS page_path, e.external_id AS event_external_id, " + "e.position AS position, v.version AS version, v.kind AS kind, " + "v.content AS content, v.metadata AS metadata, " + "v.source_created_at AS source_created_at, v.created_at AS created_at " "FROM session_event_versions v " "JOIN session_events e ON e.event_id = v.event_id " "JOIN sessions s ON s.session_id = e.session_id " f"WHERE v.version_id = {placeholder}", (row["version_id"],), - ).fetchone() + ) + payload_row = cursor.fetchone() if not payload_row: raise RuntimeError(f"missing immutable session event version {row['version_id']}") - payload = { - "harness": payload_row[0], - "session_external_id": payload_row[1], - "user_id": payload_row[2], - "agent": payload_row[3], - "workspace": payload_row[4], - "page_path": payload_row[5], - "event_external_id": payload_row[6], - "position": payload_row[7], - "version": payload_row[8], - "kind": payload_row[9], - "content": payload_row[10], - "metadata": payload_row[11], - "source_created_at": payload_row[12], - "created_at": payload_row[13], - } + payload = _row_dict(cursor, payload_row) elif row["entity_kind"] == "curation_receipt": - payload_row = conn.execute( - "SELECT run_id, patch_id, status, result_version_id, local_content_hash, " - "details, applied_at FROM curation_receipts " + cursor = conn.execute( + "SELECT run_id AS run_id, patch_id AS patch_id, status AS status, " + "result_version_id AS result_version_id, " + "local_content_hash AS local_content_hash, details AS details, " + "applied_at AS applied_at FROM curation_receipts " f"WHERE patch_id = {placeholder}", (row["version_id"],), - ).fetchone() + ) + payload_row = cursor.fetchone() if not payload_row: raise RuntimeError(f"missing immutable curation receipt {row['version_id']}") - payload = { - "run_id": payload_row[0], - "patch_id": payload_row[1], - "status": payload_row[2], - "result_version_id": payload_row[3], - "local_content_hash": payload_row[4], - "details": payload_row[5], - "applied_at": payload_row[6], - } + payload = _row_dict(cursor, payload_row) else: raise ValueError(f"unsupported outbox entity kind: {row['entity_kind']}") json_fields = { @@ -204,19 +182,7 @@ def build_batch( events = tuple( _resolve_event(store, row, replica_id=replica_id) for row in pending ) - digest_input = [ - { - "replica_id": str(event.replica_id), - "sequence": event.sequence, - "event_id": str(event.event_id), - "entity_kind": event.entity_kind, - "entity_id": str(event.entity_id), - "version_id": str(event.version_id), - "payload_hash": event.payload_hash, - "payload": event.payload, - } - for event in events - ] + digest_input = [asdict(event) for event in events] digest = hashlib.sha256(_canonical_json(digest_input).encode("utf-8")).hexdigest() batch_id = uuid5( NAMESPACE_URL, diff --git a/src/wikibricks/session_ingest.py b/src/wikibricks/session_ingest.py index 0f645c0..b61061d 100644 --- a/src/wikibricks/session_ingest.py +++ b/src/wikibricks/session_ingest.py @@ -46,3 +46,13 @@ def session_tags(record: SessionRecord) -> list[str]: tags.append(f"agent:{record.agent}") tags.append(f"user:{record.user_id}") return tags + + +def session_title(record: SessionRecord) -> str: + configured = record.metadata.get("title") + if configured: + return str(configured)[:120] + for event in record.events: + if event.kind == "user" and event.content.strip(): + return event.content.strip().splitlines()[0][:120] + return f"Session {record.external_id[:8]}" diff --git a/src/wikibricks/storage/search.py b/src/wikibricks/storage/search.py index 6689aa5..7ff35e4 100644 --- a/src/wikibricks/storage/search.py +++ b/src/wikibricks/storage/search.py @@ -9,9 +9,8 @@ from wikibricks.storage.store import PostgresStore -def _snippet(text: str, query: str, *, length: int = 300) -> str: +def _snippet(text: str, tokens: list[str], *, length: int = 300) -> str: normalized = " ".join(str(text or "").split()) - tokens = re.findall(r"[\w-]+", query, flags=re.UNICODE) if not tokens: return normalized[:length] expression = "|".join(re.escape(token) for token in tokens) @@ -144,7 +143,10 @@ def query(self, query: str, *, num_results: int = 5) -> list[dict[str, Any]]: "path": row[1], "title": row[2], "page_type": row[3], - "snippet": _snippet(row[4], query), + "snippet": _snippet( + row[4], + re.findall(r"[\w-]+", query, flags=re.UNICODE), + ), "tags": list(row[5] or []), "version": row[6], "score": float(row[7]), diff --git a/src/wikibricks/storage/sessions.py b/src/wikibricks/storage/sessions.py index 3ef6470..12ce385 100644 --- a/src/wikibricks/storage/sessions.py +++ b/src/wikibricks/storage/sessions.py @@ -13,6 +13,7 @@ session_content_hash, session_identity, session_page_path, + session_title, ) from wikibricks.storage.content import canonical_hash, insert_search_chunks @@ -24,16 +25,6 @@ class SessionRepository: def __init__(self, store: PostgresStore) -> None: self.store = store - @staticmethod - def title(record: SessionRecord) -> str: - configured = record.metadata.get("title") - if configured: - return str(configured)[:120] - for event in record.events: - if event.kind == "user" and event.content.strip(): - return event.content.strip().splitlines()[0][:120] - return f"Session {record.external_id[:8]}" - @staticmethod def event_hash(event: SessionEvent) -> str: return canonical_hash(event.to_dict()) @@ -64,7 +55,7 @@ def ingest(self, record: SessionRecord) -> IngestResult: record.started_at, record.updated_at, session_page_path(record), - self.title(record), + session_title(record), Jsonb(record.metadata), record_hash, stable_id, @@ -87,7 +78,7 @@ def ingest(self, record: SessionRecord) -> IngestResult: record.started_at, record.updated_at, session_page_path(record), - self.title(record), + session_title(record), Jsonb(record.metadata), record_hash, ), diff --git a/src/wikibricks/storage/sqlite_store.py b/src/wikibricks/storage/sqlite_store.py index ba84b3b..d86ad56 100644 --- a/src/wikibricks/storage/sqlite_store.py +++ b/src/wikibricks/storage/sqlite_store.py @@ -18,13 +18,14 @@ session_content_hash, session_identity, session_page_path, + session_title, ) from wikibricks.storage.content import ( canonical_hash, iter_search_chunks, page_content_hash, ) -from wikibricks.storage.search import select_hits +from wikibricks.storage.search import _snippet, select_hits DEFAULT_DATABASE_PATH = Path.home() / ".wikibricks" / "wikibricks.db" _UNSET = object() @@ -38,18 +39,6 @@ def _json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) -def _snippet(text: str, tokens: list[str], *, length: int = 300) -> str: - normalized = " ".join(text.split()) - expression = "|".join(re.escape(token) for token in tokens) - match = re.search(expression, normalized, flags=re.IGNORECASE | re.UNICODE) - if match is None: - return normalized[:length] - padding = max(0, (length - len(match.group(0))) // 2) - start = max(0, match.start() - padding) - end = min(len(normalized), start + length) - return normalized[start:end] - - def _insert_chunks( conn: sqlite3.Connection, *, @@ -163,81 +152,27 @@ def write_page( content_text = " ".join( str(content.get(key) or "") for key in ("summary", "body") ).strip() - page_hash = page_content_hash( - title=title, - page_type=page_type, - content=content, - content_text=content_text, - tags=tags, - source_ids=source_ids, - parent_id=parent_id, - chunk_index=chunk_index, - ) with self.connection(write=True) as conn: existing = conn.execute( - "SELECT p.page_id, p.status, v.version, v.content_hash " - "FROM pages p LEFT JOIN page_versions v " - "ON v.version_id = p.current_version_id WHERE p.path = ?", - (path,), + "SELECT status FROM pages WHERE path = ?", (path,) ).fetchone() if existing and existing["status"] != "active": raise ValueError(f"wiki page is superseded: {path}") - if existing and existing["content_hash"] == page_hash: - return f"Wiki page unchanged: {path}" - timestamp = _now() - if existing: - page_id = existing["page_id"] - version = int(existing["version"]) + 1 - else: - page_id = str(uuid4()) - version = 1 - conn.execute( - "INSERT INTO pages(page_id, path, created_at, updated_at) " - "VALUES (?, ?, ?, ?)", - (page_id, path, timestamp, timestamp), - ) - version_id = str(uuid4()) - conn.execute( - "INSERT INTO page_versions(" - "version_id, page_id, version, title, page_type, content, " - "content_text, tags, source_ids, parent_id, chunk_index, created_by, " - "content_hash, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", - ( - version_id, - page_id, - version, - title, - page_type, - _json(content), - content_text, - _json(tags or []), - _json(source_ids) if source_ids is not None else None, - parent_id, - chunk_index, - created_by, - page_hash, - timestamp, - ), - ) - conn.execute( - "UPDATE pages SET current_version_id = ?, updated_at = ? WHERE page_id = ?", - (version_id, timestamp, page_id), - ) - _insert_chunks( + message, _version_id = self.write_page_in_connection( conn, - table="page_search_chunks", - fts_table="page_search_fts", - version_id=version_id, - text=content_text, - ) - conn.execute( - "INSERT INTO sync_outbox(" - "event_id, entity_kind, entity_id, version_id, payload_hash, created_at" - ") VALUES (?, 'page_version', ?, ?, ?, ?)", - (str(uuid4()), page_id, version_id, page_hash, timestamp), + path, + title, + content, + page_type=page_type, + created_by=created_by, + tags=tags or [], + source_ids=source_ids, + parent_id=parent_id, + chunk_index=chunk_index, + content_text=content_text, ) self._failpoint("before_commit") - return f"Wrote wiki page: {path}" + return message def write_page_in_connection( self, @@ -253,7 +188,7 @@ def write_page_in_connection( parent_id: str | None, chunk_index: int | None, content_text: str, - curation_patch_id: str, + curation_patch_id: str | None = None, expected_base_content_hash: str | None | object = _UNSET, ) -> tuple[str, str]: """Write one curated page version inside an existing transaction.""" @@ -508,13 +443,11 @@ def assign_outbox_batch(self, sequences: list[int], batch_id: UUID) -> None: def acknowledge_outbox_batch(self, batch_id: UUID) -> int: with self.connection(write=True) as conn: - before = conn.total_changes - conn.execute( + return conn.execute( "UPDATE sync_outbox SET acknowledged_at = ? " "WHERE batch_id = ? AND acknowledged_at IS NULL", (_now(), str(batch_id)), - ) - return conn.total_changes - before + ).rowcount def get_sync_cursor(self, target: str) -> dict[str, Any]: with self.connection() as conn: @@ -727,16 +660,6 @@ def repair_search_indexes(self) -> tuple[int, int]: ) return len(page_rows), len(session_rows) - @staticmethod - def _session_title(record: SessionRecord) -> str: - configured = record.metadata.get("title") - if configured: - return str(configured)[:120] - for event in record.events: - if event.kind == "user" and event.content.strip(): - return event.content.strip().splitlines()[0][:120] - return f"Session {record.external_id[:8]}" - def ingest_session(self, record: SessionRecord) -> IngestResult: stable_id = str(session_identity(record)) record_hash = session_content_hash(record) @@ -763,7 +686,7 @@ def ingest_session(self, record: SessionRecord) -> IngestResult: record.started_at, record.updated_at, session_page_path(record), - self._session_title(record), + session_title(record), _json(record.metadata), record_hash, timestamp, @@ -786,7 +709,7 @@ def ingest_session(self, record: SessionRecord) -> IngestResult: record.started_at, record.updated_at, session_page_path(record), - self._session_title(record), + session_title(record), _json(record.metadata), record_hash, timestamp, diff --git a/src/wikibricks_curator/curator.py b/src/wikibricks_curator/curator.py index 9514c29..089e8fe 100644 --- a/src/wikibricks_curator/curator.py +++ b/src/wikibricks_curator/curator.py @@ -2,7 +2,6 @@ from __future__ import annotations -import hashlib from dataclasses import replace from datetime import datetime, timedelta, timezone from importlib.resources import files @@ -13,11 +12,11 @@ from wikibricks.curation import ( apply_run, build_manifest, - canonical_json, get_or_create_replica_id, store_manifest, ) from wikibricks.curation.backlog import load_curation_backlog +from wikibricks.curation.protocol import content_hash from wikibricks.storage.sqlite_store import SQLiteStore from wikibricks_curator.evidence import build_request from wikibricks_curator.router import route_sessions @@ -171,9 +170,7 @@ def run_curator( results.append(result) continue result["events"] = len(built["request"]["evidence"]) - digest = hashlib.sha256( - canonical_json(built["request"]).encode("utf-8") - ).hexdigest() + digest = content_hash(built["request"]) run_id = uuid5(replica_id, f"wikibricks:local-curator:{project}:{digest}") # Model output varies between calls; one retry absorbs an invalid reply. for attempt in (1, 2): diff --git a/src/wikibricks_curator/evidence.py b/src/wikibricks_curator/evidence.py index e0635eb..1f6e8c4 100644 --- a/src/wikibricks_curator/evidence.py +++ b/src/wikibricks_curator/evidence.py @@ -169,9 +169,8 @@ def build_request( ) -> dict[str, Any] | None: """Build one bounded curator request for a backlog item.""" project = item["project"] - requested_paths = list(dict.fromkeys(item["pages"])) living_page = item["living_page"] - paths = set(requested_paths) | {living_page} + paths = set(item["pages"]) | {living_page} pages = _active_pages( conn, paths=paths, diff --git a/src/wikibricks_curator/gateway.py b/src/wikibricks_curator/gateway.py index d4a2d73..272d00d 100644 --- a/src/wikibricks_curator/gateway.py +++ b/src/wikibricks_curator/gateway.py @@ -38,8 +38,7 @@ def resolve_token(profile: str | None) -> str: def _json_content(response: Any) -> dict[str, Any]: - content = response.read() - text = content.decode("utf-8") if isinstance(content, bytes) else content + text = response.read().decode("utf-8") try: payload = json.loads(text) except json.JSONDecodeError as error: diff --git a/src/wikibricks_curator/router.py b/src/wikibricks_curator/router.py index c5b013e..5e48bed 100644 --- a/src/wikibricks_curator/router.py +++ b/src/wikibricks_curator/router.py @@ -73,7 +73,7 @@ def _candidates(conn: Any, generic: set[str]) -> list[dict[str, str]]: def _new_path(path: str, *, candidates: set[str], generic: set[str]) -> bool: - if not isinstance(path, str) or path in candidates: + if path in candidates: return False last_segment = path.rsplit("/", 1)[-1] return _NEW_TOPIC.fullmatch(path) is not None and _normalize(last_segment) not in generic | {"home"} @@ -117,8 +117,7 @@ def route_sessions( prompt = _ROUTER_PROMPT.read_text(encoding="utf-8") schema = _schema() empty = {session["session_id"] for session in sessions if not session["text"]} - model_ids = [session_id for session_id in ids if session_id not in empty] - model_id_set = set(model_ids) + model_id_set = {session_id for session_id in ids if session_id not in empty} request = { "sessions": [session for session in sessions if session["session_id"] in model_id_set], "candidates": candidates, @@ -127,7 +126,7 @@ def route_sessions( error: str | None = None for _attempt in (1, 2): try: - raw = chat(prompt, request, schema) if model_ids else {"routes": []} + raw = chat(prompt, request, schema) if model_id_set else {"routes": []} error = None break except Exception as exc: @@ -165,7 +164,7 @@ def route_sessions( for session_id in empty: accepted[session_id] = None if invalid_routes: - invalid += len(model_ids) + invalid += len(model_id_set) new_topics = sorted( {path for path in accepted.values() if path is not None and path not in candidate_paths} ) diff --git a/src/wikibricks_remote/maintenance.py b/src/wikibricks_remote/maintenance.py index 3b4b278..431bcee 100644 --- a/src/wikibricks_remote/maintenance.py +++ b/src/wikibricks_remote/maintenance.py @@ -12,7 +12,7 @@ from wikibricks.curation import build_manifest, publish_manifest from wikibricks.postgres_store import PostgresStore -from wikibricks_remote.proposals import _canonical_json, build_patches +from wikibricks_remote.proposals import _canonical_json, archive_page, build_patches from wikibricks_remote.resources import ( RemotePolicy, load_prompt, @@ -119,19 +119,8 @@ def _current_pages( ).fetchall() pages = [] for event_id, version_id, content_hash, raw_payload in rows: - payload = dict(raw_payload) pages.append( - { - "evidence_id": f"archive-event:{event_id}", - "path": payload["path"], - "title": payload["title"], - "page_type": payload["page_type"], - "content": payload["content"], - "tags": list(payload.get("tags") or []), - "source_ids": list(payload.get("source_ids") or []), - "base_version_id": str(version_id), - "base_content_hash": content_hash, - } + archive_page(event_id, version_id, content_hash, raw_payload) ) return pages diff --git a/src/wikibricks_remote/proposals.py b/src/wikibricks_remote/proposals.py index f8c0c18..a9a4642 100644 --- a/src/wikibricks_remote/proposals.py +++ b/src/wikibricks_remote/proposals.py @@ -27,6 +27,26 @@ _PROPOSAL_FIELDS = _PROPOSAL_REQUIRED_FIELDS | {"link_type"} +def archive_page( + event_id: UUID, + version_id: UUID, + content_hash: str, + raw_payload: dict[str, Any], +) -> dict[str, Any]: + payload = dict(raw_payload) + return { + "evidence_id": f"archive-event:{event_id}", + "path": payload["path"], + "title": payload["title"], + "page_type": payload["page_type"], + "content": payload["content"], + "tags": list(payload.get("tags") or []), + "source_ids": list(payload.get("source_ids") or []), + "base_version_id": str(version_id), + "base_content_hash": content_hash, + } + + def _canonical_json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) diff --git a/src/wikibricks_remote/search/lakebase.py b/src/wikibricks_remote/search/lakebase.py index abdd1f8..b8e2bf4 100644 --- a/src/wikibricks_remote/search/lakebase.py +++ b/src/wikibricks_remote/search/lakebase.py @@ -8,6 +8,7 @@ from uuid import UUID from wikibricks.postgres_store import PostgresStore +from wikibricks_remote.proposals import archive_page from wikibricks_remote.search.documents import SearchDocument, project_event from wikibricks_remote.search.embeddings import ( Embedder, @@ -436,18 +437,8 @@ def _current_pages( ).fetchall() pages = {} for event_id, version_id, content_hash, raw_payload in rows: - payload = dict(raw_payload) - pages[payload["path"]] = { - "evidence_id": f"archive-event:{event_id}", - "path": payload["path"], - "title": payload["title"], - "page_type": payload["page_type"], - "content": payload["content"], - "tags": list(payload.get("tags") or []), - "source_ids": list(payload.get("source_ids") or []), - "base_version_id": str(version_id), - "base_content_hash": content_hash, - } + page = archive_page(event_id, version_id, content_hash, raw_payload) + pages[page["path"]] = page return pages def candidates( diff --git a/src/wikibricks_remote/search/ranking.py b/src/wikibricks_remote/search/ranking.py index ae0d3d2..abfc620 100644 --- a/src/wikibricks_remote/search/ranking.py +++ b/src/wikibricks_remote/search/ranking.py @@ -32,9 +32,8 @@ def reciprocal_rank_fusion( keyword_paths: Sequence[str], *, maximum: int, - constant: int = 60, ) -> tuple[CandidateRank, ...]: - if maximum < 1 or constant < 1: + if maximum < 1: raise ValueError("RRF bounds must be positive") vector_ranks: dict[str, int] = {} keyword_ranks: dict[str, int] = {} @@ -47,7 +46,7 @@ def reciprocal_rank_fusion( vector_rank = vector_ranks.get(path) keyword_rank = keyword_ranks.get(path) score = sum( - 1.0 / (constant + rank) + 1.0 / (60 + rank) for rank in (vector_rank, keyword_rank) if rank is not None )