diff --git a/CLAUDE.md b/CLAUDE.md index 60373cb2aa..053265b09a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -101,8 +101,15 @@ permanently break composition. `session_links` is also the topology-edge table (the docs' older `topology_edges` name): it persists every parent reference a parser asserts, even when the parent isn't ingested yet, keyed `(src_session_id, dst_origin, dst_native_id, link_type)`, resolved on each save -by `resolve_session_links_for_session`. `TopologyEdgeStatus` = -unresolved/resolved/repaired/**quarantined** (cycle-break). +by `_resolve_session_graph`/`_resolve_outbound_session_links` +(`storage/sqlite/archive_tiers/write.py`) — the sole production +implementation, invoked unconditionally from `write_parsed_session_to_archive` +(the single choke point both live incremental ingest and full raw +replay/reindex go through). `storage/sqlite/queries/session_links.py`'s +similarly-named async `resolve_session_links_for_session` has no production +caller; it exists only as test infrastructure (`polylogue-enium`). +`TopologyEdgeStatus` = unresolved/resolved/repaired/**quarantined** +(cycle-break). ### The five tiers (durability is the axis) diff --git a/polylogue/archive/provider/semantics.py b/polylogue/archive/provider/semantics.py index d7517993d5..e2249be62e 100644 --- a/polylogue/archive/provider/semantics.py +++ b/polylogue/archive/provider/semantics.py @@ -157,24 +157,6 @@ def extract_content_blocks(content: Sequence[object] | None) -> list[ContentBloc return blocks -def extract_display_text_from_content_blocks(content: Sequence[object] | None) -> str: - """Rebuild human-readable text from stored structured content blocks.""" - if not content: - return "" - - parts: list[str] = [] - for raw_block in content: - block = _content_block_record(raw_block) - if block is None: - continue - if block.get("type") not in {"text", "code", "tool_result", "thinking"}: - continue - text = _string_field(block, "text") - if text: - parts.append(text) - return "\n".join(parts) - - def extract_claude_code_text(content: Sequence[object] | None) -> str: """Extract text from Claude Code content blocks, excluding non-text blocks.""" if not content: diff --git a/polylogue/archive/semantic/pricing.py b/polylogue/archive/semantic/pricing.py index 31da7e072d..04163b6d27 100644 --- a/polylogue/archive/semantic/pricing.py +++ b/polylogue/archive/semantic/pricing.py @@ -530,42 +530,6 @@ def resolve_model_identity( ) -def _usage_payload(value: object) -> CostUsagePayload: - usage = _record(value) - input_tokens = _coerce_int( - usage.get("input_tokens") - or usage.get("prompt_tokens") - or usage.get("inputTokenCount") - or usage.get("promptTokenCount") - ) - output_tokens = _coerce_int( - usage.get("output_tokens") - or usage.get("completion_tokens") - or usage.get("outputTokenCount") - or usage.get("candidatesTokenCount") - ) - cache_read_tokens = _coerce_int( - usage.get("cache_read_tokens") - or usage.get("cache_read_input_tokens") - or usage.get("cached_tokens") - or usage.get("cachedContentTokenCount") - ) - cache_write_tokens = _coerce_int( - usage.get("cache_write_tokens") - or usage.get("cache_creation_input_tokens") - or usage.get("cache_creation_tokens") - ) - explicit_total = _coerce_int(usage.get("total_tokens") or usage.get("totalTokenCount")) - total = explicit_total or input_tokens + output_tokens + cache_read_tokens + cache_write_tokens - return CostUsagePayload( - input_tokens=input_tokens, - output_tokens=output_tokens, - cache_read_tokens=cache_read_tokens, - cache_write_tokens=cache_write_tokens, - total_tokens=total, - ) - - def _token_usage_payload(tokens: TokenUsage | None) -> CostUsagePayload: if tokens is None: return CostUsagePayload() diff --git a/polylogue/cli/archive_query.py b/polylogue/cli/archive_query.py index d9da464a17..e0b8ac5d6d 100644 --- a/polylogue/cli/archive_query.py +++ b/polylogue/cli/archive_query.py @@ -35,8 +35,6 @@ from polylogue.archive.query.spec import ( QuerySpecError, SessionQuerySpec, - normalize_action_sequence, - normalize_action_terms, parse_query_date, resolve_default_root_filter, session_count_unit_label, @@ -1832,10 +1830,6 @@ def _optional_str(value: object) -> str | None: return text or None -def _tags(value: object) -> tuple[str, ...]: - return _csv_tokens(value) - - def _csv_tokens(value: object) -> tuple[str, ...]: if value is None: return () @@ -1876,20 +1870,6 @@ def _tool_tokens(value: object) -> tuple[str, ...]: return tuple(token.lower() for token in _csv_tokens(value)) -def _action_tokens(field: str, value: object) -> tuple[str, ...]: - try: - return normalize_action_terms(field, value) - except QuerySpecError as exc: - raise click.UsageError(f"invalid {exc.field}: {exc.value}") from exc - - -def _action_sequence_tokens(value: object) -> tuple[str, ...]: - try: - return normalize_action_sequence("action_sequence", value) - except QuerySpecError as exc: - raise click.UsageError(f"invalid {exc.field}: {exc.value}") from exc - - def _message_type(value: object) -> str | None: if not value: return None diff --git a/polylogue/cli/click_option_groups.py b/polylogue/cli/click_option_groups.py index 095300d7cc..da004d6bcb 100644 --- a/polylogue/cli/click_option_groups.py +++ b/polylogue/cli/click_option_groups.py @@ -68,18 +68,6 @@ def get_metavar(self, param: click.Parameter, ctx: click.Context | None = None) return "TEXT" -def _load_message_types() -> list[str]: - from polylogue.archive.message.types import MessageType - - return [m.value for m in MessageType] - - -def _load_material_origins() -> list[str]: - from polylogue.core.enums import MaterialOrigin - - return [m.value for m in MaterialOrigin] - - def _load_retrieval_lanes() -> list[str]: from polylogue.archive.query.spec import QUERY_RETRIEVAL_LANES diff --git a/polylogue/cli/commands/check.py b/polylogue/cli/commands/check.py index 57b6c36415..3728164c83 100644 --- a/polylogue/cli/commands/check.py +++ b/polylogue/cli/commands/check.py @@ -4,28 +4,11 @@ import click -from polylogue.cli.shared.check_models import VacuumResult from polylogue.cli.shared.check_options import apply_check_command_options from polylogue.cli.shared.check_rendering_json import emit_json_output from polylogue.cli.shared.check_rendering_plain import render_plain_output -from polylogue.cli.shared.check_support import ( - format_count_mapping as _format_count_mapping_impl, -) -from polylogue.cli.shared.check_support import ( - make_schema_progress_callback as _make_schema_progress_callback_impl, -) -from polylogue.cli.shared.check_support import ( - parse_schema_samples as _parse_schema_samples_impl, -) -from polylogue.cli.shared.check_support import run_vacuum as _run_vacuum_impl -from polylogue.cli.shared.check_support import vacuum_database as _vacuum_database_impl from polylogue.cli.shared.check_workflow import CheckCommandOptions, run_check_workflow, validate_check_options from polylogue.cli.shared.types import AppEnv -from polylogue.core.protocols import ProgressCallback - - -def _format_count_mapping(counts: dict[str, int]) -> str: - return _format_count_mapping_impl(counts) @click.command("doctor") @@ -97,20 +80,4 @@ def check_command( render_plain_output(env, result, options) -def _make_schema_progress_callback() -> ProgressCallback: - return _make_schema_progress_callback_impl() - - -def _run_vacuum(env: AppEnv) -> None: - _run_vacuum_impl(env) - - -def _vacuum_database(env: AppEnv) -> VacuumResult: - return _vacuum_database_impl(env) - - -def _parse_schema_samples(raw: str) -> int | None: - return _parse_schema_samples_impl(raw) - - __all__ = ["check_command"] diff --git a/polylogue/cli/shared/check_support.py b/polylogue/cli/shared/check_support.py index 0650a29105..e8b3d27d0c 100644 --- a/polylogue/cli/shared/check_support.py +++ b/polylogue/cli/shared/check_support.py @@ -15,17 +15,6 @@ def format_count_mapping(counts: dict[str, int]) -> str: return ", ".join(f"{key}={value:,}" for key, value in sorted(counts.items())) -def format_semantic_metric_summary(metric_summary: dict[str, dict[str, int]]) -> str: - return ", ".join( - ( - f"{metric}(preserved={counts.get('preserved', 0):,}, " - f"declared_loss={counts.get('declared_loss', 0):,}, " - f"critical_loss={counts.get('critical_loss', 0):,})" - ) - for metric, counts in sorted(metric_summary.items()) - ) - - def parse_schema_samples(raw: str) -> int | None: value = raw.strip().lower() if value == "all": diff --git a/polylogue/cli/shared/types.py b/polylogue/cli/shared/types.py index 9ae1c3918b..23c26476bb 100644 --- a/polylogue/cli/shared/types.py +++ b/polylogue/cli/shared/types.py @@ -20,12 +20,6 @@ from polylogue.ui import UI -def _lazy_ui() -> UI: - from polylogue.ui import UI as _UI - - return _UI(plain=True) - - def _lazy_services(runtime: ResolvedRuntimeConfig | None) -> RuntimeServices: from polylogue.services import build_runtime_services diff --git a/polylogue/context/selection.py b/polylogue/context/selection.py index c006e56145..75e25cb770 100644 --- a/polylogue/context/selection.py +++ b/polylogue/context/selection.py @@ -10,16 +10,9 @@ from collections.abc import Awaitable, Callable, Sequence from dataclasses import dataclass from pathlib import Path -from types import SimpleNamespace -from typing import TYPE_CHECKING, Any, TypedDict +from typing import TYPE_CHECKING, Any -from polylogue.core.timestamps import parse_archive_datetime -from polylogue.mcp.archive_support import archive_index_active_paths, archive_query_filters -from polylogue.storage.sqlite.archive_tiers.archive import ( - ArchiveSessionSearchHit, - ArchiveSessionSummary, - ArchiveStore, -) +from polylogue.mcp.archive_support import archive_index_active_paths if TYPE_CHECKING: from polylogue.archive.query.spec import SessionQuerySpec @@ -43,17 +36,6 @@ class ContextImageSelection: query_total: int = 0 -class ArchiveContextImageFilters(TypedDict): - origins: tuple[str, ...] - excluded_origins: tuple[str, ...] - tags: tuple[str, ...] - excluded_tags: tuple[str, ...] - repo_names: tuple[str, ...] - cwd_prefix: str | None - since_ms: int | None - until_ms: int | None - - @dataclass(frozen=True, slots=True) class _ContextImageQueryAttempt: query: str | None @@ -195,88 +177,3 @@ def archive_context_image_active( archive_root=archive_root, db_anchor_path=db_anchor_path, ) - - -def query_archive_context_image( - archive: ArchiveStore, - spec: SessionQuerySpec, - *, - default_limit: int, -) -> list[SimpleNamespace]: - """Project archive sessions into the context-image summary surface.""" - query = " ".join(spec.query_terms).strip() - kwargs = archive_context_image_filters(spec) - if query: - rows: list[ArchiveSessionSummary | ArchiveSessionSearchHit] = list( - archive.search_summaries( - query, - limit=spec.limit or default_limit, - offset=spec.offset, - sort="date", - reverse=spec.reverse, - **kwargs, - ) - ) - else: - rows = list( - archive.list_summaries( - limit=spec.limit or default_limit, - offset=spec.offset, - sort="date", - reverse=spec.reverse, - **kwargs, - ) - ) - - summaries: list[ArchiveSessionSummary] = [] - for row in dedupe_archive_context_image_rows(rows): - if isinstance(row, ArchiveSessionSearchHit): - try: - summaries.append(archive.read_summary(row.session_id)) - except KeyError: - continue - else: - summaries.append(row) - return [archive_context_image_summary(row) for row in summaries] - - -def archive_context_image_filters(spec: SessionQuerySpec) -> ArchiveContextImageFilters: - filters = archive_query_filters(spec) - return { - "origins": filters["origins"], - "excluded_origins": filters["excluded_origins"], - "tags": filters["tags"], - "excluded_tags": filters["excluded_tags"], - "repo_names": filters["repo_names"], - "cwd_prefix": filters["cwd_prefix"], - "since_ms": filters["since_ms"], - "until_ms": filters["until_ms"], - } - - -def archive_context_image_summary(row: ArchiveSessionSummary) -> SimpleNamespace: - return SimpleNamespace( - id=row.session_id, - origin=row.origin, - title=row.title, - display_title=row.title, - created_at=parse_archive_datetime(row.created_at), - updated_at=parse_archive_datetime(row.updated_at), - message_count=row.message_count, - messages=(), - tool_use_count=0, - ) - - -def dedupe_archive_context_image_rows( - rows: list[ArchiveSessionSummary | ArchiveSessionSearchHit], -) -> list[ArchiveSessionSummary | ArchiveSessionSearchHit]: - deduped: list[ArchiveSessionSummary | ArchiveSessionSearchHit] = [] - seen: set[str] = set() - for row in rows: - session_id = row.session_id - if session_id in seen: - continue - seen.add(session_id) - deduped.append(row) - return deduped diff --git a/polylogue/core/dates.py b/polylogue/core/dates.py index 4954fcfce8..ed441ba4f7 100644 --- a/polylogue/core/dates.py +++ b/polylogue/core/dates.py @@ -47,15 +47,3 @@ def parse_date(date_str: str) -> datetime | None: # Ensure result is always UTC-aware result = result.replace(tzinfo=timezone.utc) return result - - -def format_date_iso(dt: datetime) -> str: - """Format datetime as ISO string compatible with storage layer. - - Args: - dt: datetime to format - - Returns: - ISO 8601 formatted string (YYYY-MM-DD HH:MM:SS) - """ - return dt.strftime("%Y-%m-%d %H:%M:%S") diff --git a/polylogue/daemon/backup.py b/polylogue/daemon/backup.py index 74493d9332..a6d8ce1aeb 100644 --- a/polylogue/daemon/backup.py +++ b/polylogue/daemon/backup.py @@ -193,21 +193,6 @@ def _json_str_list(value: object) -> list[str]: return [str(item) for item in value] if isinstance(value, list) else [] -def _precious_archive_tiers(root: Path) -> dict[str, Path]: - return { - "source": root / "source.db", - "user": root / "user.db", - "embeddings": root / "embeddings.db", - } - - -def _omitted_archive_tiers(root: Path) -> dict[str, Path]: - return { - "index": root / "index.db", - "ops": root / "ops.db", - } - - def _all_archive_tiers(root: Path) -> dict[str, Path]: return { "source": root / "source.db", @@ -241,11 +226,6 @@ def _optional_profile_tiers(profile: BackupProfile) -> set[str]: return set() -def _profile_omitted_tiers(root: Path, profile: BackupProfile) -> dict[str, Path]: - included = set(_profile_archive_tiers(root, profile)) - return {tier: path for tier, path in _all_archive_tiers(root).items() if tier not in included} - - def _archive_layout_present(root: Path) -> bool: return any(path.exists() for path in _all_archive_tiers(root).values()) diff --git a/polylogue/daemon/status.py b/polylogue/daemon/status.py index dbe76f3eec..83b1095108 100644 --- a/polylogue/daemon/status.py +++ b/polylogue/daemon/status.py @@ -1083,11 +1083,6 @@ def _fmt_bytes(value: int) -> str: return "0 KB" -def _failing_files_info() -> list[str]: - """Return live-source files currently marked failed or excluded.""" - return [item.source_path for item in _live_cursor_summary_info().failing_files] - - def _live_cursor_summary_info() -> LiveCursorSummary: """Return live cursor backlog/failure state without source-tree scans.""" dbf = _active_status_db_path() diff --git a/polylogue/insights/transforms.py b/polylogue/insights/transforms.py index 0211af69f6..9207ad6b3e 100644 --- a/polylogue/insights/transforms.py +++ b/polylogue/insights/transforms.py @@ -2127,15 +2127,6 @@ def _normal_read_text(messages: Sequence[Message]) -> str: return "\n\n".join(f"{_role_value(message)}: {message.text or ''}" for message in messages) -def _message_text_fragments(message: Message) -> Iterable[str]: - if message.text: - yield message.text - for block in message.blocks: - text = _block_text(block) - if text: - yield text - - def _block_ref(session: Session, message: Message, block_index: int, block: Mapping[str, object]) -> TransformRawRef: return TransformRawRef( session_id=str(session.id), diff --git a/polylogue/mcp/server_prompts.py b/polylogue/mcp/server_prompts.py index 362b78fd11..fb7c0a672c 100644 --- a/polylogue/mcp/server_prompts.py +++ b/polylogue/mcp/server_prompts.py @@ -29,7 +29,6 @@ ArchiveSessionSummary, ArchiveStore, ) - from polylogue.storage.sqlite.archive_tiers.write import ArchiveSessionEnvelope class ErrorContextPayload(TypedDict): @@ -143,22 +142,6 @@ def _code_snippet_payload(block: MCPFencedCodeBlock, session_id: str) -> Extract } -def _archive_prompt_session(session: ArchiveSessionEnvelope) -> PromptSession: - return PromptSession( - id=session.session_id, - origin=session.origin, - display_title=session.title or "(untitled)", - messages=tuple( - PromptMessage( - role=message.role, - text="\n\n".join(block.text for block in message.blocks if block.text), - timestamp=message.occurred_at, - ) - for message in session.messages - ), - ) - - def _archive_prompt_session_page(archive: ArchiveStore, session_id: str, *, limit: int = 20) -> PromptSession: """Build prompt context from a bounded message projection.""" summary = archive.read_summary(session_id) diff --git a/polylogue/pipeline/ids.py b/polylogue/pipeline/ids.py index 22b0a96f73..b1b121efa9 100644 --- a/polylogue/pipeline/ids.py +++ b/polylogue/pipeline/ids.py @@ -11,7 +11,7 @@ from polylogue.core.hashing import hash_bytes, hash_payload from polylogue.core.json import JSONValue from polylogue.core.sources import origin_from_provider -from polylogue.core.types import ContentHash, MessageId, SessionEventId, SessionId +from polylogue.core.types import ContentHash, MessageId, SessionId # ParsedMessage/ParsedSession/ParsedAttachment/ParsedContentBlock are used only # as parameter/return type annotations below (never constructed or @@ -175,10 +175,6 @@ def message_id(session_id: SessionId, provider_message_id: str) -> MessageId: return MessageId(f"{session_id}:{provider_message_id}") -def session_event_id(session_id: SessionId, event_index: int) -> SessionEventId: - return SessionEventId(f"{session_id}:session-event:{event_index:06d}") - - def _content_block_payload(block: ParsedContentBlock) -> dict[str, JSONValue]: """Build a hash-stable payload for a single content block.""" payload: dict[str, JSONValue] = { diff --git a/polylogue/rendering/core_markdown.py b/polylogue/rendering/core_markdown.py index 0af942c69c..c6cac1782e 100644 --- a/polylogue/rendering/core_markdown.py +++ b/polylogue/rendering/core_markdown.py @@ -15,7 +15,6 @@ if TYPE_CHECKING: from polylogue.archive.models import Session - from polylogue.storage.archive_views import SessionRenderProjection @dataclass(frozen=True, slots=True) @@ -157,21 +156,6 @@ def _normalize_markdown_message( ) -def _group_projection_attachments( - projection: SessionRenderProjection, -) -> dict[str | None, list[MarkdownAttachment]]: - attachments_by_message: dict[str | None, list[MarkdownAttachment]] = {} - for attachment in projection.attachments: - attachments_by_message.setdefault(attachment.message_id, []).append( - _normalize_markdown_attachment( - attachment_id=attachment.attachment_id, - path=attachment.path, - display_name=attachment.display_name, - ) - ) - return attachments_by_message - - def format_session_markdown(conv: Session) -> str: """Format a loaded Session domain object to markdown.""" attachments_by_message: dict[str | None, list[MarkdownAttachment]] = {} diff --git a/polylogue/schemas/generation/archive_workload_profile.py b/polylogue/schemas/generation/archive_workload_profile.py index 0ce6242fde..03e4ceeb7a 100644 --- a/polylogue/schemas/generation/archive_workload_profile.py +++ b/polylogue/schemas/generation/archive_workload_profile.py @@ -59,19 +59,6 @@ def _mix(conn: sqlite3.Connection, table: str, column: str) -> JSONDocument: return {"" if row[0] is None else str(row[0]): int(row[1]) for row in rows} -def _mixes(conn: sqlite3.Connection, table: str, columns: Sequence[str]) -> JSONDocument: - available = _columns(conn, table) - selected = [column for column in columns if column in available] - if not selected: - return {} - counters = {column: Counter[str]() for column in selected} - query = "SELECT " + ", ".join(f'"{column}"' for column in selected) + f' FROM "{table}"' - for row in conn.execute(query): - for index, column in enumerate(selected): - counters[column]["" if row[index] is None else str(row[index])] += 1 - return {column: dict(sorted(counts.items(), key=lambda item: item[0])) for column, counts in counters.items()} - - def _sketch_rows(rows: Iterable[Sequence[object]], index: int = 0) -> JSONDocument: sketch = DistributionSketch() null_count = 0 @@ -113,33 +100,6 @@ def _column_distributions( return payload -def _length_distributions( - conn: sqlite3.Connection, - table: str, - columns: Sequence[str], -) -> JSONDocument: - available = _columns(conn, table) - selected = [column for column in columns if column in available] - if not selected: - return {} - sketches = {column: DistributionSketch() for column in selected} - null_counts = dict.fromkeys(selected, 0) - expressions = ", ".join(f'length(CAST("{column}" AS BLOB))' for column in selected) - for row in conn.execute(f'SELECT {expressions} FROM "{table}"'): - for index, column in enumerate(selected): - value = row[index] - if value is None: - null_counts[column] += 1 - elif isinstance(value, int | float) and not isinstance(value, bool): - sketches[column].observe(value) - payload: JSONDocument = {} - for column in selected: - distribution = sketches[column].to_payload() - distribution["null_count"] = null_counts[column] - payload[f"{column}_bytes"] = distribution - return payload - - def _scan_table_profile( conn: sqlite3.Connection, table: str, diff --git a/polylogue/schemas/generation/packages.py b/polylogue/schemas/generation/packages.py index ec0d827f57..49015ce9f6 100644 --- a/polylogue/schemas/generation/packages.py +++ b/polylogue/schemas/generation/packages.py @@ -112,16 +112,6 @@ def _observe_package_membership(package: _PackageAccumulator, membership: _UnitM _update_observed_window(package, membership.unit.observed_at) -def _observe_journal_package_membership( - package: _PackageAccumulator, - membership: _UnitMembership, -) -> None: - """Update bounded package metadata; exact distinct values remain in SQLite.""" - if membership.unit.source_path: - _merge_representative_paths(package.representative_paths, [membership.unit.source_path]) - _update_observed_window(package, membership.unit.observed_at) - - def _package_bundle_scope_count(package: _PackageAccumulator) -> int: if package.journal_bundle_scope_count is not None: return package.journal_bundle_scope_count diff --git a/polylogue/schemas/operator/workflow.py b/polylogue/schemas/operator/workflow.py index 62d44c40e0..0f591c5af6 100644 --- a/polylogue/schemas/operator/workflow.py +++ b/polylogue/schemas/operator/workflow.py @@ -2,7 +2,6 @@ from __future__ import annotations -from polylogue.schemas.operator.annotations import collect_annotation_summary from polylogue.schemas.operator.commit import ( commit_provider_schema as _commit_provider_schema, ) @@ -21,7 +20,6 @@ from polylogue.schemas.operator.inference import ( promote_schema_cluster as _promote_schema_cluster, ) -from polylogue.schemas.operator.models import SchemaAnnotationSummary from polylogue.schemas.operator.resolution import ( explain_schema as _explain_schema, ) @@ -53,7 +51,3 @@ run_artifact_coverage = _run_artifact_coverage list_artifact_observations = _list_artifact_observations list_artifact_cohorts = _list_artifact_cohorts - - -def _collect_annotation_summary(schema: dict[str, object]) -> SchemaAnnotationSummary: - return collect_annotation_summary(schema) diff --git a/polylogue/sinex/material_adapter.py b/polylogue/sinex/material_adapter.py index ed7f4876be..ef8595ac16 100644 --- a/polylogue/sinex/material_adapter.py +++ b/polylogue/sinex/material_adapter.py @@ -240,29 +240,6 @@ def _dropped_block_gap( ) -def _block_fidelity_gap( - session_id: str, - message_index: int, - block_position: int, - block: object, -) -> FidelityGapInput | None: - unsupported: list[str] = [] - metadata = _attr(block, "metadata", default={}) - if isinstance(metadata, Mapping): - represented_metadata = {"language", "semantic_type"} - unsupported.extend(f"metadata.{key}" for key in sorted(set(map(str, metadata)) - represented_metadata)) - if _items(_attr(block, "web_constructs", default=())): - unsupported.append("web_constructs") - if not unsupported: - return None - return FidelityGapInput( - scope="block", - record_id=f"{session_id}:message[{message_index}]:block[{block_position}]", - gap_kind="unsupported_normalized_fields", - detail="material-protocol v1 has no field for: " + ", ".join(unsupported), - ) - - def _parsed_block_input(position: int, block: ParsedContentBlock) -> BlockInput: metadata = block.metadata or {} semantic_type = metadata.get("semantic_type") diff --git a/polylogue/storage/artifacts/inspection.py b/polylogue/storage/artifacts/inspection.py index bee68827f8..3797110bad 100644 --- a/polylogue/storage/artifacts/inspection.py +++ b/polylogue/storage/artifacts/inspection.py @@ -208,19 +208,6 @@ def _support_status( _FULL_JSON_INSPECTION_MAX_BYTES = 8 * 1024 * 1024 # 8 MB — bounded fallback for large JSON documents -def _inspection_prefix_from_bytes(raw_content: bytes, source_path: str | None) -> bytes: - """Extract a small prefix of in-memory raw content for artifact classification.""" - if len(raw_content) <= _INSPECTION_PREFIX_BYTES: - return raw_content - normalized = (source_path or "").lower() - is_jsonl = normalized.endswith((".jsonl", ".jsonl.txt", ".ndjson")) - if is_jsonl: - newline_pos = raw_content.find(b"\n") - if newline_pos > 0: - return raw_content[: newline_pos + 1] - return raw_content[:_INSPECTION_PREFIX_BYTES] - - def _inspection_prefix(record: RawSessionRecord) -> bytes: """Extract a small prefix of raw content sufficient for classification. diff --git a/polylogue/storage/blob_integrity.py b/polylogue/storage/blob_integrity.py index b05499876c..dda030a3f6 100644 --- a/polylogue/storage/blob_integrity.py +++ b/polylogue/storage/blob_integrity.py @@ -1092,18 +1092,6 @@ def _split_container_source_path(source_path: str) -> tuple[Path, str] | None: return Path(outer), member -def _json_payload_at_index(raw_bytes: bytes, source_index: int) -> object: - loaded = json_loads(raw_bytes) - if isinstance(loaded, list): - try: - return loaded[source_index] - except IndexError as exc: - raise IndexError(f"source_index {source_index} outside JSON array of {len(loaded)} payloads") from exc - if source_index == 0: - return loaded - raise IndexError("non-array JSON payload only supports source_index 0") - - def _jsonl_payload_at_index(raw_bytes: bytes, source_index: int) -> object: for idx, line in enumerate(raw_bytes.splitlines()): if idx == source_index: diff --git a/polylogue/storage/blob_repair.py b/polylogue/storage/blob_repair.py index 8acf50ccb5..d1eb9a78ee 100644 --- a/polylogue/storage/blob_repair.py +++ b/polylogue/storage/blob_repair.py @@ -81,12 +81,6 @@ def _referenced_blob_hashes( return hashes, surfaces -def _surface_detail(surfaces: list[str]) -> str: - if not surfaces: - return "references: none" - return "references: " + ", ".join(sorted(dict.fromkeys(surfaces))) - - def count_orphaned_blobs_sync( conn: sqlite3.Connection, *, db_path: Path | str | None = None, configured_root: Path | None = None ) -> int: diff --git a/polylogue/storage/derived/insights.py b/polylogue/storage/derived/insights.py index 920e7504d7..8c9d0735c8 100644 --- a/polylogue/storage/derived/insights.py +++ b/polylogue/storage/derived/insights.py @@ -112,24 +112,6 @@ def build_action_statuses(metrics: Metrics) -> dict[str, DerivedModelStatus]: # --------------------------------------------------------------------------- -def build_profile_fts_status( - metrics: Metrics, - *, - key_prefix: str, - name: str, - label: str, -) -> DerivedModelStatus: - return _fts_status( - metrics, - name=name, - label=label, - ready_key=f"{key_prefix}_ready", - source_rows_key="profile_rows", - materialized_rows_key=f"{key_prefix}_rows", - duplicate_key=f"{key_prefix}_duplicates", - ) - - def _profile_rows_status(metrics: Metrics) -> DerivedModelStatus: ready = _metric_bool(metrics, "profile_rows_ready") profile_rows = _metric_int(metrics, "profile_rows") diff --git a/polylogue/storage/fts/fts_lifecycle.py b/polylogue/storage/fts/fts_lifecycle.py index 72cd1fb4ca..9db5f72350 100644 --- a/polylogue/storage/fts/fts_lifecycle.py +++ b/polylogue/storage/fts/fts_lifecycle.py @@ -196,29 +196,6 @@ async def _triggers_present_async(conn: aiosqlite.Connection, names: tuple[str, return row is not None and row[0] == len(names) -async def suspend_fts_triggers_async(conn: aiosqlite.Connection, *, mark_stale: bool = True) -> None: - """Drop FTS triggers to avoid per-row overhead during bulk inserts. - - Call rebuild_fts_index_async() after to repopulate the FTS index. - """ - if mark_stale: - from polylogue.storage.fts.freshness import mark_all_fts_stale_async - - await mark_all_fts_stale_async(conn, detail="FTS triggers suspended for bulk write") - for name in _FTS_TRIGGER_NAMES: - await conn.execute(f"DROP TRIGGER IF EXISTS {name}") - - -async def restore_fts_triggers_async(conn: aiosqlite.Connection) -> None: - """Re-create FTS triggers after bulk insert.""" - await suspend_fts_triggers_async(conn) - for ddl in await _fts_trigger_ddl_for_existing_surfaces_async(conn): - if ";" in ddl: - await conn.executescript(ddl) - else: - await conn.execute(ddl) - - # polylogue-a7xr.5: FTS trigger DDL is now sourced from storage/fts/sql.py as the single # source of truth. Aliases below preserve backward compatibility with code that # references the private _*_TRIGGER_DDL names. diff --git a/polylogue/storage/insights/feedback/__init__.py b/polylogue/storage/insights/feedback/__init__.py index 5fa0007740..8981f03364 100644 --- a/polylogue/storage/insights/feedback/__init__.py +++ b/polylogue/storage/insights/feedback/__init__.py @@ -19,7 +19,6 @@ import json import uuid -from collections.abc import Sequence from datetime import UTC, datetime from pathlib import Path from typing import TYPE_CHECKING @@ -412,11 +411,3 @@ def supports_kind(kind: CorrectionKind | str) -> bool: "supports_kind", "upsert_correction", ] - - -def _list_corrections_sequence_typed( - items: Sequence[LearningCorrection], -) -> list[LearningCorrection]: - """Type-only helper used in tests to surface ``Sequence`` semantics.""" - - return list(items) diff --git a/polylogue/storage/insights/session/storage.py b/polylogue/storage/insights/session/storage.py index d58d544eac..d30ae0c8a3 100644 --- a/polylogue/storage/insights/session/storage.py +++ b/polylogue/storage/insights/session/storage.py @@ -6,8 +6,6 @@ from collections.abc import Mapping, Sequence from typing import TypeVar -from pydantic import BaseModel - from polylogue.core.timestamps import parse_timestamp from polylogue.storage.runtime import ( SessionLatencyProfileRecord, @@ -398,18 +396,6 @@ def session_latency_profile_insert_values(record: SessionLatencyProfileRecord) - ) -def _fallback_timeline_payload_json( - evidence_payload: BaseModel, - inference_payload: BaseModel, -) -> str | None: - return _json_or_none( - { - **evidence_payload.model_dump(mode="json"), - **inference_payload.model_dump(mode="json"), - } - ) - - def _epoch_ms_or_none(value: str | None) -> int | None: if not value: return None diff --git a/polylogue/storage/repair.py b/polylogue/storage/repair.py index 855f3d5c5b..7a2ea5ac38 100644 --- a/polylogue/storage/repair.py +++ b/polylogue/storage/repair.py @@ -5191,20 +5191,6 @@ def count_orphaned_messages_sync(conn: sqlite3.Connection) -> int: ) -def has_orphaned_messages_sync(conn: sqlite3.Connection) -> bool: - return bool( - conn.execute( - """ - SELECT 1 - FROM messages m - LEFT JOIN sessions s ON s.session_id = m.session_id - WHERE s.session_id IS NULL - LIMIT 1 - """ - ).fetchone() - ) - - def count_empty_sessions_sync(conn: sqlite3.Connection) -> int: """Count session rows that are debris: no *content* (zero messages, or every message carries zero words -- see ``_empty_session_candidate_ids``) @@ -5895,93 +5881,6 @@ def preview_superseded_raw_snapshots(*, count: int) -> RepairResult: ) -def count_stale_supersession_receipts_sync(source_conn: sqlite3.Connection, *, index_db_path: Path) -> int: - """Count superseded raws whose receipt is stale but reissuable against the - current head (polylogue-ktwa). Read-only; never writes a receipt.""" - from polylogue.storage.raw_retention import plan_stale_supersession_reissue - - plan = plan_stale_supersession_reissue(source_conn, index_db_path=index_db_path) - return len(plan.eligible) - - -def repair_stale_supersession_receipts(config: Config, dry_run: bool = False) -> RepairResult: - """Reissue fresh supersession receipts against the current head (polylogue-ktwa). - - Writes ONLY new ``raw_revision_applications`` rows; never deletes a blob - and never mutates ``raw_revision_heads`` or an existing receipt. Actual - blob release remains solely the job of ``repair_superseded_raw_snapshots`` - under its own ``active_raw_retention_authority`` gate. - """ - from polylogue.storage.raw_retention import ( - RawRetentionSafetyError, - plan_stale_supersession_reissue, - reissue_stale_supersession_receipts, - ) - from polylogue.storage.sqlite.connection_profile import open_connection, open_readonly_connection - - archive_root = _raw_materialization_archive_root(config) - source_db_path = archive_root / "source.db" - index_db_path = archive_root / "index.db" - if not source_db_path.is_file() or not index_db_path.is_file(): - return _repair_result( - "stale_supersession_receipts", - repaired_count=0, - success=False, - detail=f"Skipped: archive tier file(s) not found: {source_db_path}, {index_db_path}", - ) - with closing(open_readonly_connection(source_db_path)) as source_conn: - source_conn.row_factory = sqlite3.Row - if dry_run: - try: - plan = plan_stale_supersession_reissue(source_conn, index_db_path=index_db_path) - except RawRetentionSafetyError as exc: - return _repair_result( - "stale_supersession_receipts", repaired_count=0, success=False, detail=f"Skipped: {exc}" - ) - ineligible_total = sum(plan.ineligible_reason_counts.values()) - return _repair_result( - "stale_supersession_receipts", - repaired_count=len(plan.eligible), - success=True, - detail=( - f"Would: reissue {len(plan.eligible):,} stale supersession receipt(s) " - f"({plan.stale_count:,} stale, {plan.already_current_count:,} already current, " - f"{ineligible_total:,} ineligible)" - ), - ) - with closing(open_connection(index_db_path)) as index_conn: - try: - result = reissue_stale_supersession_receipts( - source_conn, index_conn, index_db_path=index_db_path, dry_run=False - ) - except RawRetentionSafetyError as exc: - return _repair_result( - "stale_supersession_receipts", repaired_count=0, success=False, detail=f"Skipped: {exc}" - ) - return _repair_result( - "stale_supersession_receipts", - repaired_count=result.reissued_count, - success=not result.errors, - detail=( - f"Reissued {result.reissued_count:,} of {result.eligible_count:,} eligible stale " - "supersession receipt(s)" + (f"; errors: {'; '.join(result.errors[:3])}" if result.errors else "") - ), - ) - - -def preview_stale_supersession_receipts(*, count: int) -> RepairResult: - return _repair_result( - "stale_supersession_receipts", - repaired_count=count, - success=True, - detail=( - f"Would: reissue {count} stale supersession receipts" - if count - else "Would: No stale supersession receipts found" - ), - ) - - def _relink_orphaned_attachments_best_effort(config: Config, index_conn: sqlite3.Connection) -> int: """polylogue-w06b: try to recover ref-less ``attachments`` rows by re-parsing ``source.db`` raw sessions before the destructive cleanup diff --git a/polylogue/storage/sqlite/archive_tiers/archive.py b/polylogue/storage/sqlite/archive_tiers/archive.py index 7d617ed327..12bb991f21 100644 --- a/polylogue/storage/sqlite/archive_tiers/archive.py +++ b/polylogue/storage/sqlite/archive_tiers/archive.py @@ -9363,14 +9363,6 @@ def _session_cost_insight_from_archive_row(conn: sqlite3.Connection, row: sqlite ) -def _json_object_from_text(value: object) -> dict[str, object]: - try: - decoded = json.loads(str(value or "{}")) - except json.JSONDecodeError: - return {} - return decoded if isinstance(decoded, dict) else {} - - def _json_value(value: object, *, default: JSONValue) -> JSONValue: try: decoded = json.loads(str(value or json.dumps(default))) @@ -10885,10 +10877,6 @@ def _no_action_between_predicate(previous: int, current: int, action_relation: s ) -def _count_rows(conn: sqlite3.Connection, table: str) -> int: - return _count_scalar(conn, f"SELECT COUNT(*) FROM {table}") - - def _count_scalar(conn: sqlite3.Connection, sql: str, params: tuple[object, ...] = ()) -> int: row = conn.execute(sql, params).fetchone() return int(row[0] or 0) if row is not None else 0 diff --git a/polylogue/storage/sqlite/archive_tiers/user_write.py b/polylogue/storage/sqlite/archive_tiers/user_write.py index c0e35591e7..67d75f78a9 100644 --- a/polylogue/storage/sqlite/archive_tiers/user_write.py +++ b/polylogue/storage/sqlite/archive_tiers/user_write.py @@ -126,10 +126,6 @@ def _immediate_user_write_transaction(conn: sqlite3.Connection) -> Iterator[None conn.execute(f"RELEASE {savepoint}") -def _default_context_policy() -> dict[str, JSONValue]: - return dict(ASSERTION_DEFAULT_CONTEXT_POLICY) - - def _normalize_assertion_kind(kind: str | AssertionKind) -> AssertionKind: return AssertionKind.from_string(kind) @@ -179,18 +175,6 @@ def _now_ms() -> int: return int(datetime.now(UTC).timestamp() * 1000) -def _read_payload_text(value: str | None) -> dict[str, object]: - if not value: - return {} - try: - parsed = json.loads(value) - except json.JSONDecodeError: - return {} - if isinstance(parsed, dict): - return dict(parsed) - return {} - - def _json_dumps(payload: dict[str, object] | None) -> str: if payload is None: return "{}" @@ -1032,12 +1016,6 @@ def read_archive_workspace_envelope(conn: sqlite3.Connection, name: str) -> Arch ) -def _read_mirrored_assertion(conn: sqlite3.Connection, assertion_id: str) -> ArchiveAssertionEnvelope | None: - if not _table_exists(conn, "assertions"): - return None - return read_assertion_envelope(conn, assertion_id) - - def _split_target_ref(target_ref: str) -> tuple[str, str]: target_type, sep, target_id = target_ref.partition(":") if not sep: diff --git a/polylogue/ui/theme.py b/polylogue/ui/theme.py index 97ee82a12b..d4ca5f5d5c 100644 --- a/polylogue/ui/theme.py +++ b/polylogue/ui/theme.py @@ -313,12 +313,6 @@ def role_color(role: str | Role) -> RoleColor: } -def webui_theme_tokens(mode: ThemeMode) -> dict[str, str]: - """Return one complete WebUI theme token mapping for generation.""" - - return {**WEBUI_SHARED_TOKENS, **WEBUI_THEME_TOKENS[mode]} - - # ============================================================================= # Thinking / reasoning block styling # =============================================================================