"""Command-palette projections over admitted workbench search and navigation.
The application search service owns the safe snapshot, matching, ranking, and
stable result identity. The navigation catalogue owns destination and action
admission. This module only adapts those already-authoritative answers to
Textual command-palette hits; selecting a hit asks the injected root host to
navigate and cannot invoke a business action.
"""
from __future__ import annotations
from collections.abc import Callable, Mapping
from typing import Final, Protocol, override, runtime_checkable
from textual.command import DiscoveryHit, Hit, Hits, Provider
from ...application.search.workbench import (
WorkbenchSearchLabelKey,
WorkbenchSearchRequest,
WorkbenchSearchResponse,
WorkbenchSearchResult,
WorkbenchSearchSource,
WorkbenchSearchStatus,
)
from ...core.errors.hierarchy import CadrumoError
from ...core.i18n.render import tr
from .navigation import (
DestinationUnavailableError,
NavigationContractError,
TuiDestinationCatalogueV1,
TuiFocusIdentityV1,
TuiNavigationTargetV1,
)
[docs]
@runtime_checkable
class WorkbenchSearchDoorV1(Protocol):
"""The application-owned search door consumed by the palette."""
[docs]
def search(self, request: WorkbenchSearchRequest, /) -> WorkbenchSearchResponse:
"""Return the authoritative bounded result page for a transient query."""
...
[docs]
@runtime_checkable
class TuiSearchHostV1(Protocol):
"""The root seam through which palette hits request navigation."""
@property
def workbench_search_service(self) -> WorkbenchSearchDoorV1:
"""Return the already-composed application search service."""
...
@property
def workbench_search_refusal_code(self) -> str | None:
"""Return why search is unavailable, or ``None`` when it is composed."""
...
@property
def destination_catalogue(self) -> TuiDestinationCatalogueV1:
"""Return the already-admitted destination catalogue."""
...
[docs]
def navigate_to(self, target: TuiNavigationTargetV1, /) -> None:
"""Navigate to an already-admitted semantic target."""
...
[docs]
class TuiSearchHostError(CadrumoError):
"""Raised when a palette provider is mounted outside the workbench root."""
def _require_host(app: object) -> TuiSearchHostV1:
"""Return the root host or fail before reading search state."""
if not isinstance(app, TuiSearchHostV1):
raise TuiSearchHostError("the workbench palette requires a TuiSearchHostV1 root")
return app
def _host_or_none(app: object) -> TuiSearchHostV1 | None:
"""Return a usable palette host, withholding entries outside the workbench."""
try:
return _require_host(app)
except TuiSearchHostError:
return None
_RESULT_LABEL_LOCALE_KEYS: Final[Mapping[WorkbenchSearchLabelKey, str]] = {
WorkbenchSearchLabelKey.LEDGER_ENTRY: "tui.search.result.label.ledger_entry",
WorkbenchSearchLabelKey.LEDGER_EVIDENCE: "tui.search.result.label.ledger_evidence",
WorkbenchSearchLabelKey.DECLARATION: "tui.search.result.label.declaration",
WorkbenchSearchLabelKey.MODELO: "tui.search.result.label.modelo",
WorkbenchSearchLabelKey.REVISION: "tui.search.result.label.revision",
WorkbenchSearchLabelKey.FILING: "tui.search.result.label.filing",
WorkbenchSearchLabelKey.HISTORY: "tui.search.result.label.history",
WorkbenchSearchLabelKey.RECONCILIATION: "tui.search.result.label.reconciliation",
WorkbenchSearchLabelKey.NOTIFICATION: "tui.search.result.label.notification",
}
_RESULT_SOURCE_LOCALE_KEYS: Final[Mapping[WorkbenchSearchSource, str]] = {
WorkbenchSearchSource.LEDGER_ENTRY: "tui.search.result.source.ledger_entry",
WorkbenchSearchSource.LEDGER_EVIDENCE: "tui.search.result.source.ledger_evidence",
WorkbenchSearchSource.DECLARATION: "tui.search.result.source.declaration",
WorkbenchSearchSource.MODELO: "tui.search.result.source.modelo",
WorkbenchSearchSource.REVISION: "tui.search.result.source.revision",
WorkbenchSearchSource.FILING: "tui.search.result.source.filing",
WorkbenchSearchSource.HISTORY: "tui.search.result.source.history",
WorkbenchSearchSource.RECONCILIATION: "tui.search.result.source.reconciliation",
WorkbenchSearchSource.NOTIFICATION: "tui.search.result.source.notification",
}
_RESULT_STATUS_LOCALE_KEYS: Final[Mapping[WorkbenchSearchStatus, str]] = {
WorkbenchSearchStatus.LEDGER_ENTRY_READY: "tui.search.result.status.ledger_entry_ready",
WorkbenchSearchStatus.LEDGER_ENTRY_NEEDS_REVIEW: "tui.search.result.status.ledger_entry_needs_review",
WorkbenchSearchStatus.LEDGER_ENTRY_CLASSIFIED: "tui.search.result.status.ledger_entry_classified",
WorkbenchSearchStatus.LEDGER_EVIDENCE_CAPTURED: "tui.search.result.status.ledger_evidence_captured",
WorkbenchSearchStatus.LEDGER_EVIDENCE_MISSING: "tui.search.result.status.ledger_evidence_missing",
WorkbenchSearchStatus.LEDGER_EVIDENCE_STALE: "tui.search.result.status.ledger_evidence_stale",
WorkbenchSearchStatus.DECLARATION_DRAFT: "tui.search.result.status.declaration_draft",
WorkbenchSearchStatus.DECLARATION_IN_PROGRESS: "tui.search.result.status.declaration_in_progress",
WorkbenchSearchStatus.DECLARATION_NEEDS_ATTENTION: "tui.search.result.status.declaration_needs_attention",
WorkbenchSearchStatus.DECLARATION_READY: "tui.search.result.status.declaration_ready",
WorkbenchSearchStatus.DECLARATION_FILED: "tui.search.result.status.declaration_filed",
WorkbenchSearchStatus.MODELO_AVAILABLE: "tui.search.result.status.modelo_available",
WorkbenchSearchStatus.MODELO_UNAVAILABLE: "tui.search.result.status.modelo_unavailable",
WorkbenchSearchStatus.REVISION_CURRENT: "tui.search.result.status.revision_current",
WorkbenchSearchStatus.REVISION_SUPERSEDED: "tui.search.result.status.revision_superseded",
WorkbenchSearchStatus.FILING_SUBMITTED: "tui.search.result.status.filing_submitted",
WorkbenchSearchStatus.FILING_ACCEPTED: "tui.search.result.status.filing_accepted",
WorkbenchSearchStatus.FILING_REJECTED: "tui.search.result.status.filing_rejected",
WorkbenchSearchStatus.HISTORY_OBSERVED: "tui.search.result.status.history_observed",
WorkbenchSearchStatus.HISTORY_NOT_OBSERVED: "tui.search.result.status.history_not_observed",
WorkbenchSearchStatus.RECONCILIATION_OPEN: "tui.search.result.status.reconciliation_open",
WorkbenchSearchStatus.RECONCILIATION_RESOLVED: "tui.search.result.status.reconciliation_resolved",
WorkbenchSearchStatus.NOTIFICATION_UNREAD: "tui.search.result.status.notification_unread",
WorkbenchSearchStatus.NOTIFICATION_READ: "tui.search.result.status.notification_read",
}
_DESTINATION_LOCALE_KEYS: Final[Mapping[str, str]] = {
"workbench.home": "tui.search.destination.home",
"workbench.ledger": "tui.search.destination.ledger",
"workbench.declarations": "tui.search.destination.declarations",
"workbench.aeat_sync": "tui.search.destination.aeat_sync",
"workbench.profile": "tui.search.destination.profile",
}
_DESTINATION_HELP_LOCALE_KEYS: Final[Mapping[str, str]] = {
"workbench.home": "tui.search.destination_help.home",
"workbench.ledger": "tui.search.destination_help.ledger",
"workbench.declarations": "tui.search.destination_help.declarations",
"workbench.aeat_sync": "tui.search.destination_help.aeat_sync",
"workbench.profile": "tui.search.destination_help.profile",
}
_ACTION_LOCALE_KEYS: Final[Mapping[str, str]] = {
"operator.profile.edit": "tui.search.action.edit_profile",
"operator.modelo.work.revisions": "tui.search.action.list_work_revisions",
"operator.ledger.review": "tui.search.action.review_ledger",
"operator.ledger.classify": "tui.search.action.classify_ledger",
"operator.ledger.evidence.review.list": "tui.search.action.review_ledger_evidence",
"operator.ledger.link": "tui.search.action.link_ledger",
"operator.ledger.preflight": "tui.search.action.validate_ledger",
"operator.live.filed.pull": "tui.aeat_sync.action.pull_filed",
"operator.live.filed.pull_all": "tui.aeat_sync.action.pull_filed_all",
"operator.live.notifications.list": "tui.search.action.list_notifications",
"operator.overview.explain": "tui.search.action.explain_overview",
"operator.modelo.filing_record.list": "tui.search.action.list_filing_records",
}
def _render_locale(key: str, locale: str | None, **values: object) -> str:
"""Render one palette label, optionally under a test-selected locale."""
if locale is None:
return tr(key, **values)
return tr(key, locale=locale, **values)
def _result_text(result: WorkbenchSearchResult, *, locale: str | None = None) -> str:
"""Render safe search metadata through localized human-facing labels."""
label = _render_locale(_RESULT_LABEL_LOCALE_KEYS[result.label_key], locale)
source = _render_locale(_RESULT_SOURCE_LOCALE_KEYS[result.source], locale)
status = _render_locale(_RESULT_STATUS_LOCALE_KEYS[result.status], locale)
parts = [label, source, status]
if result.address is not None:
parts.append(
_render_locale(
"tui.search.result.address",
locale,
modelo=result.address.modelo,
filing_year=result.address.filing_year,
period=result.address.period.registry_token,
)
)
return " · ".join(parts)
def _destination_text(destination: str, *, locale: str | None = None) -> str:
"""Produce localized wording for one admitted destination."""
key = _DESTINATION_LOCALE_KEYS.get(destination, "tui.search.destination.unknown")
return _render_locale(key, locale)
def _command_help(target: TuiNavigationTargetV1, *, locale: str | None = None) -> str:
"""Say what a palette entry leads to, without repeating its own name.
An action's help names the area it runs in. A destination's name already
is its area, so its help describes what the area holds instead; using the
area name for both printed every destination twice.
"""
if target.action_candidate_id is not None:
return _destination_text(target.destination, locale=locale)
key = _DESTINATION_HELP_LOCALE_KEYS.get(target.destination)
return "" if key is None else _render_locale(key, locale)
_ACTION_UNKNOWN_LOCALE_KEY: Final[str] = "tui.search.action.unknown"
"""How an action this build cannot name is worded.
An unnamed identifier used to render as generic "available action" copy, which
made an unknown action indistinguishable from a correctly-named, genuinely
offerable one -- a drifted identifier could then sit in a surface for days
looking like working copy. The unknown state is worded as unknown instead, the
way an unrecognised destination and an unrecognised refusal code already are.
"""
[docs]
def workbench_action_label(action_id: str, *, locale: str | None = None) -> str:
"""Render one catalogue action without exposing its internal identifier.
This is the single naming authority for an operator action across the
workbench: the palette and Home resolve the same identifier to the same
words, so a suggested task and the command that performs it cannot be
described differently.
An identifier this build has no wording for renders as explicitly unknown
rather than as a plausible label: it still renders, because a table row is
not the place to raise, but it never claims to be a named action.
"""
key = _ACTION_LOCALE_KEYS.get(action_id, _ACTION_UNKNOWN_LOCALE_KEY)
return _render_locale(key, locale)
_REFUSAL_LOCALE_KEY: Final[str] = "tui.search.refusal.unavailable"
_REFUSAL_UNKNOWN_LOCALE_KEY: Final[str] = "tui.search.refusal.unknown"
_REFUSAL_REASON_LOCALE_KEYS: Final[Mapping[str, str]] = {
"workbench.search.unavailable": "tui.search.refusal.not_composed",
"workbench.search.refresh_unavailable": "tui.search.refusal.refresh_failed",
}
"""Sanitized refusal codes the root publishes, and how each is worded.
An unrecognised code degrades to the generic wording rather than being shown
raw: a reason code is an internal token, not operator copy.
"""
def _no_navigation() -> None:
"""A refusal hit navigates nowhere; selecting it is a no-op."""
[docs]
class WorkbenchSearchProviderV1(Provider):
"""Expose application-ranked workbench results in the command palette."""
[docs]
@override
async def search(self, query: str) -> Hits:
"""Yield only results that still resolve through the current catalogue."""
try:
request = WorkbenchSearchRequest(query=query)
except ValueError:
return
host = _host_or_none(self.app)
if host is None:
return
if host.workbench_search_refusal_code is not None:
# A refused search is a state the root already computed and named.
# Dereferencing the service here would raise on the FIRST-RUN case
# -- a profile whose sources are not all readable yet -- so the
# palette says why instead, and offers nothing it cannot resolve.
yield Hit(
score=0.0,
match_display=_render_locale(_REFUSAL_LOCALE_KEY, None),
command=_no_navigation,
text=_render_locale(_REFUSAL_LOCALE_KEY, None),
help=_render_locale(
_REFUSAL_REASON_LOCALE_KEYS.get(host.workbench_search_refusal_code, _REFUSAL_UNKNOWN_LOCALE_KEY),
None,
),
)
return
response = host.workbench_search_service.search(request)
for result in response.results:
try:
target = host.destination_catalogue.target_for_search_result(result)
except (DestinationUnavailableError, NavigationContractError):
continue
text = _result_text(result)
yield Hit(
score=1.0 / (result.rank + 1),
match_display=text,
command=_navigation_command(host.navigate_to, target),
text=text,
help=_destination_text(result.admission.destination),
)
[docs]
class WorkbenchCommandProviderV1(Provider):
"""Expose admitted destinations and registered action identities as commands."""
[docs]
@override
async def search(self, query: str) -> Hits:
"""Fuzzy-match the current catalogue's admitted routes and actions."""
matcher = self.matcher(query)
host = _host_or_none(self.app)
if host is None:
return
for text, target in _command_entries(host.destination_catalogue):
if (score := matcher.match(text)) > 0:
yield Hit(
score=score,
match_display=matcher.highlight(text),
command=_navigation_command(host.navigate_to, target),
text=text,
help=_command_help(target),
)
[docs]
@override
async def discover(self) -> Hits:
"""List the current admitted destinations and actions before typing."""
host = _host_or_none(self.app)
if host is None:
return
for text, target in _command_entries(host.destination_catalogue):
yield DiscoveryHit(
display=text,
command=_navigation_command(host.navigate_to, target),
text=text,
help=_command_help(target),
)
def _command_entries(
catalogue: TuiDestinationCatalogueV1,
) -> tuple[tuple[str, TuiNavigationTargetV1], ...]:
"""Project every currently admitted destination and registered action once."""
entries: list[tuple[str, TuiNavigationTargetV1]] = []
for route in catalogue.routes:
if route.admission.state.value != "available":
continue
destination = route.descriptor.destination
destination_target = TuiNavigationTargetV1(
destination=destination,
focus=TuiFocusIdentityV1(
destination=destination,
semantic_key=f"navigation.{destination.removeprefix('workbench.')}",
),
)
entries.append((_destination_text(destination), destination_target))
for candidate in route.action_candidates:
action_target = TuiNavigationTargetV1(
destination=destination,
focus=TuiFocusIdentityV1(
destination=destination,
semantic_key=f"action.{candidate.action_candidate_id}",
),
action_candidate_id=candidate.action_candidate_id,
)
entries.append((workbench_action_label(candidate.action_candidate_id), action_target))
return tuple(entries)
def _navigation_command(
navigate_to: Callable[[TuiNavigationTargetV1], None], target: TuiNavigationTargetV1
) -> Callable[[], None]:
"""Bind a validated target without turning a palette choice into an action."""
def navigate() -> None:
"""Ask the root to mount the admitted destination."""
navigate_to(target)
return navigate
__all__ = [
"TuiSearchHostError",
"TuiSearchHostV1",
"WorkbenchCommandProviderV1",
"WorkbenchSearchDoorV1",
"WorkbenchSearchProviderV1",
"workbench_action_label",
]