"""Pure controller and common interaction shell for Ledger destinations."""
from __future__ import annotations
import asyncio
from collections.abc import Callable
from dataclasses import replace
from typing import ClassVar, Final, cast
from textual.app import App
from textual.binding import Binding
from textual.message import Message
from textual.widgets import DataTable, Static
from ....application.ledger.models import (
ManualLedgerTransactionPatch,
ManualLedgerTransactionResult,
)
from ....application.ledger.workspace import (
LEDGER_WORKSPACE_CONTRACT_VERSION,
LedgerWorkspaceArea,
LedgerWorkspaceAreaStateV1,
LedgerWorkspaceAvailability,
LedgerWorkspaceProjectionV1,
LedgerWorkspaceStatus,
)
from ....application.operator_actions.models import ActionReference
from ....application.review.filter import LedgerReviewStatus
from ....core.errors.hierarchy import CadrumoError, InternalInvariantError
from ....core.i18n.render import tr
from ....core.identity.hex_ids import InvoiceId
from ....core.identity.transaction_ids import TransactionId
from ..components.account_chrome import AccountChromeScreen
from ..components.theme import BASE_CSS, tokenised
from ..components.workspace_host import replace_workspace_body
from ..navigation import TuiFocusIdentityV1, TuiScreenContextV1
from .models import (
LEDGER_DESTINATION_BY_AREA,
LedgerClassificationSubmissionV1,
LedgerEntryRowV1,
LedgerEvidenceConfirmationV1,
LedgerEvidenceConfirmedV1,
LedgerEvidenceDraftV1,
LedgerEvidenceRecordRowV1,
LedgerEvidenceRowV1,
LedgerExclusionSubmissionV1,
LedgerImportOutcomeV1,
LedgerImportRequestV1,
LedgerInvoiceAddResultV1,
LedgerInvoiceEntryV1,
LedgerLinkResultV1,
LedgerLinkSubmissionV1,
LedgerReaderReadinessV1,
LedgerReviewRowV1,
LedgerRouteRefusalV1,
LedgerRouteTargetV1,
)
from .workspace_injection import LedgerWorkspaceInjection
_AREA_LOCALE_KEYS: Final = {
LedgerWorkspaceArea.OVERVIEW: "tui.ledger.area.overview",
LedgerWorkspaceArea.ENTRIES: "tui.ledger.area.entries",
LedgerWorkspaceArea.REVIEW: "tui.ledger.area.review",
LedgerWorkspaceArea.IMPORT: "tui.ledger.area.import",
LedgerWorkspaceArea.CLASSIFICATION: "tui.ledger.area.classification",
LedgerWorkspaceArea.EVIDENCE: "tui.ledger.area.evidence",
LedgerWorkspaceArea.RECONCILIATION: "tui.ledger.area.reconciliation",
}
_AVAILABILITY_LOCALE_KEYS: Final = {
LedgerWorkspaceAvailability.AVAILABLE: "tui.ledger.availability.available",
LedgerWorkspaceAvailability.LOCKED: "tui.ledger.availability.locked",
LedgerWorkspaceAvailability.STALE: "tui.ledger.availability.stale",
LedgerWorkspaceAvailability.NEVER_CAPTURED: "tui.ledger.availability.never_captured",
LedgerWorkspaceAvailability.UNAVAILABLE: "tui.ledger.availability.unavailable",
}
#: Derived from the enum rather than listed, so a status cannot be produced
#: without a name.
#:
#: The hand-written version held three of the four members. Nothing pointed at
#: the enum, so ``EXCLUDED`` -- which ``ledger_transaction_review_status``
#: returns for a REVIEWED_EXCLUDED row -- had no entry, and
#: :func:`review_status_label` raises on a miss. Both the entries table and the
#: review table label every row, so classifying one row as excluded and opening
#: either body took the workspace down.
_REVIEW_STATUS_LOCALE_KEYS: Final = {
status.value: f"tui.ledger.review_status.{status.value}" for status in LedgerReviewStatus
}
_STATUS_LOCALE_KEYS: Final = {
LedgerWorkspaceStatus.READY: "tui.ledger.status.ready",
LedgerWorkspaceStatus.EMPTY: "tui.ledger.status.empty",
LedgerWorkspaceStatus.NEEDS_ATTENTION: "tui.ledger.status.needs_attention",
LedgerWorkspaceStatus.UNMEASURED: "tui.ledger.status.unmeasured",
}
[docs]
def ledger_copy(key: str, **values: object) -> str:
"""Resolve all operator copy through the canonical catalogue boundary."""
return tr(key, **values)
[docs]
def area_label(area: LedgerWorkspaceArea) -> str:
"""Return an operator label without displaying an internal enum token."""
return ledger_copy(_AREA_LOCALE_KEYS[area])
[docs]
def availability_label(availability: LedgerWorkspaceAvailability) -> str:
"""Render availability with a textual cue independent of colour."""
return ledger_copy(_AVAILABILITY_LOCALE_KEYS[availability])
[docs]
def review_status_label(status: str) -> str:
"""Translate a source status without leaking its transport spelling."""
key = _REVIEW_STATUS_LOCALE_KEYS.get(status)
if key is None:
raise ValueError("unsupported Ledger review status")
return ledger_copy(key)
[docs]
def status_label(status: LedgerWorkspaceStatus) -> str:
"""Render source status through its authored catalogue key."""
return ledger_copy(_STATUS_LOCALE_KEYS[status])
[docs]
def item_count_label(state: LedgerWorkspaceAreaStateV1) -> str:
"""Keep an unmeasured denominator distinct from a measured numeric zero."""
return status_label(state.status) if state.status is LedgerWorkspaceStatus.UNMEASURED else str(state.item_count)
_APPLICATION_STATE_REASON_KEY = "tui.ledger.refusal.application_state"
"""The one line a closed Ledger area renders: the application names no Ledger reason codes."""
[docs]
class LedgerWorkspaceController:
"""Read-only custody of one injected application projection and shell context."""
def __init__(
self,
context: TuiScreenContextV1,
projection: LedgerWorkspaceProjectionV1,
injection: LedgerWorkspaceInjection,
) -> None:
"""Admit an outer Ledger context and retain its immutable snapshot."""
if context.destination != "workbench.ledger":
raise ValueError("Ledger workspace requires the workbench.ledger screen context")
if projection.contract_version != LEDGER_WORKSPACE_CONTRACT_VERSION:
raise ValueError("unsupported Ledger workspace projection contract")
self.context = context
self.projection = projection
self.injection = injection
self.review_action = injection.review_action
self.classify_action = injection.classify_action
self.classification_submitter = injection.classification_submitter
self.import_door = injection.import_door
self.invoice_add_door = injection.invoice_add_door
self.exclusion_submitter = injection.exclusion_submitter
self.evidence_door = injection.evidence_door
self.evidence_action = injection.evidence_action
self.evidence_items = injection.evidence_items
self.link_action = injection.link_action
self.link_submitter = injection.link_submitter
self._states = {row.area: row for row in projection.areas}
[docs]
def classification_target_coordinate(self) -> tuple[int, int, str]:
"""Return a safe position and redacted identifier from the visible projection."""
target = self.classification_target
if target is None:
raise InternalInvariantError("classification target is unavailable")
position = next(
index for index, row in enumerate(self.projection.entries, start=1) if row.transaction_id == target
)
return position, len(self.projection.entries), str(target)[:12]
[docs]
def entry_label(self, transaction_id: TransactionId) -> str:
"""Name one entry the way the operator recorded it: its date and description.
An entry the projection does not carry falls back to its short
identifier rather than to a blank cell.
"""
entry = next((row for row in self.projection.entries if row.transaction_id == transaction_id), None)
if entry is None:
return str(transaction_id)[:12]
return ledger_copy("tui.ledger.entry_label", date=entry.date, description=entry.description)
[docs]
def state_for(self, area: LedgerWorkspaceArea) -> LedgerWorkspaceAreaStateV1:
"""Return the application-owned area state."""
return self._states[area]
[docs]
def route_target(self, area: LedgerWorkspaceArea) -> LedgerRouteTargetV1:
"""Build an internal semantic target without invoking it.
The destination comes from the canonical pairing rather than a copy
held here, so this cannot name a screen the route catalogue does not
have. The lookup needs no ``cast``: the pairing is typed as the
destination alias, so a misspelling is a type error at the
declaration instead of an assertion the checker believes.
"""
return LedgerRouteTargetV1(destination=LEDGER_DESTINATION_BY_AREA[area], area=area)
def _selection_refusal(
self,
area: LedgerWorkspaceArea,
target: LedgerRouteTargetV1,
) -> LedgerRouteRefusalV1 | None:
"""Refuse an area that needs a chosen entry when none has been chosen.
Kept out of ``refusal_for``'s ``missing_door`` expression on purpose,
and out of any boolean joining it to the area comparison. A missing
door is an absent INJECTED dependency — something the launcher failed
to wire — and the wholly-wired-doors gate derives what the launcher
owes by reading that expression. A selection is the operator's, made at
runtime, so joining the two demanded an operator value from the
launcher; that unsatisfiable demand is why the classification door was
removed rather than wired.
It also lets the operator hear what is actually true — choose an entry
— instead of being told submission is unavailable.
"""
if area is not LedgerWorkspaceArea.CLASSIFICATION:
return None
if self.classification_target is not None:
return None
# With no entries there is nothing to choose, and asking for a choice
# sends the operator looking for rows that do not exist.
reason_key = (
"tui.ledger.refusal.selection_required" if self.projection.entries else "tui.ledger.refusal.no_entries"
)
return LedgerRouteRefusalV1(
target=target,
availability=LedgerWorkspaceAvailability.UNAVAILABLE,
reason_key=reason_key,
)
def _submission_door_is_missing(self, area: LedgerWorkspaceArea) -> bool:
"""Return whether an area lacks the injected command it needs.
Each clause joins one area to exactly the fields it needs, which is the
shape the wholly-wired-doors gate reads the launcher's obligations from.
"""
return (
(
area is LedgerWorkspaceArea.CLASSIFICATION
and (self.classify_action is None or self.classification_submitter is None)
)
or (area is LedgerWorkspaceArea.IMPORT and self.import_door is None)
or (area is LedgerWorkspaceArea.EVIDENCE and (self.evidence_action is None or self.evidence_items is None))
)
[docs]
def refusal_for(self, area: LedgerWorkspaceArea) -> LedgerRouteRefusalV1 | None:
"""Preserve application refusal separately from deferred screen availability."""
target = self.route_target(area)
state = self.state_for(area)
if state.availability is not LedgerWorkspaceAvailability.AVAILABLE:
return LedgerRouteRefusalV1(
target=target,
availability=state.availability,
reason_key=_APPLICATION_STATE_REASON_KEY,
)
selection_refusal = self._selection_refusal(area, target)
if selection_refusal is not None:
return selection_refusal
if self._submission_door_is_missing(area):
return LedgerRouteRefusalV1(
target=target,
availability=LedgerWorkspaceAvailability.UNAVAILABLE,
reason_key="tui.ledger.refusal.submission_unavailable",
)
return None
[docs]
def entry_rows(self) -> tuple[LedgerEntryRowV1, ...]:
"""Narrow safe application entry references without recovering payloads."""
return tuple(
LedgerEntryRowV1(
transaction_id=row.transaction_id,
review_status=row.review_status,
source=row,
)
for row in self.projection.entries
)
[docs]
def with_transaction_focus(self, transaction_id: TransactionId) -> LedgerWorkspaceController:
"""Re-address this workspace at one visible entry, carrying nothing else.
Selection travels in ``context.focus`` — the channel
:meth:`restored_transaction_id` already reads for cursor restore — so
there is one answer to "which entry is the operator on". A separate
mutable attribute beside it would let the restored cursor and the
classification target disagree about the same question.
Returns a NEW controller rather than mutating, because the context is
frozen. An internal area move resolves its next screen synchronously
against whatever controller the outgoing screen holds, so rebinding
there is what carries the selection across a body swap — with no extra
screen push, and therefore no result callback left registered against a
pump that is about to stop.
The context is constructed explicitly rather than by ``model_copy``:
``model_copy`` skips validation, and the focus/destination agreement
check is the thing stopping a focus from naming another workspace.
Raises:
ValueError: When the entry is not in the visible projection. An
entry the operator cannot see is one they cannot have chosen.
"""
if all(row.transaction_id != transaction_id for row in self.projection.entries):
raise ValueError("classification target is absent from the visible Ledger projection")
return LedgerWorkspaceController(
TuiScreenContextV1(
destination=self.context.destination,
focus=TuiFocusIdentityV1(
destination=self.context.destination,
semantic_key="ledger.transaction",
restore_token=transaction_id,
),
action_candidate_id=self.context.action_candidate_id,
),
self.projection,
self.injection,
)
@property
def classification_target(self) -> TransactionId | None:
"""The entry the operator selected, read from the one focus channel."""
return self.restored_transaction_id()
[docs]
def restored_transaction_id(self) -> TransactionId | None:
"""Resolve a transaction focus by semantic identity, never by row position."""
focus = self.context.focus
if focus is None or focus.semantic_key != "ledger.transaction" or focus.restore_token is None:
return None
candidate = focus.restore_token
return candidate if any(row.transaction_id == candidate for row in self.projection.entries) else None
[docs]
def review_rows(self) -> tuple[LedgerReviewRowV1, ...]:
"""Join review identities to safe entry references, refusing contradictions."""
by_id = {row.transaction_id: row for row in self.projection.entries}
rows: list[LedgerReviewRowV1] = []
for transaction_id in self.projection.review_transaction_ids:
source = by_id.get(transaction_id)
if source is None:
raise ValueError("Ledger review identity is absent from the projected entry catalogue")
rows.append(
LedgerReviewRowV1(
transaction_id=transaction_id,
review_status=source.review_status,
action=self.review_action,
source=source,
)
)
return tuple(rows)
[docs]
async def submit_classification(self, patch: ManualLedgerTransactionPatch) -> ManualLedgerTransactionResult:
"""Submit an explicit patch through the injected authorized door."""
if self.classify_action is None or self.classification_target is None or self.classification_submitter is None:
raise InternalInvariantError("classification submission is unavailable")
submission = LedgerClassificationSubmissionV1(
action=self.classify_action,
transaction_id=self.classification_target,
patch=patch,
)
result = await self.classification_submitter(submission)
if result.ref.transaction_id != self.classification_target:
raise ValueError("classification result transaction identity disagrees")
return result
[docs]
async def preview_import(self, request: LedgerImportRequestV1) -> LedgerImportOutcomeV1:
"""Ask the injected import door what applying ``request`` would do."""
if self.import_door is None:
raise InternalInvariantError("import submission is unavailable")
return await self.import_door.preview(request)
[docs]
async def apply_import(self, request: LedgerImportRequestV1) -> LedgerImportOutcomeV1:
"""Write a previewed import through the injected door."""
if self.import_door is None:
raise InternalInvariantError("import submission is unavailable")
return await self.import_door.apply(request)
[docs]
def can_exclude(self) -> bool:
"""Report whether excluding an entry is admitted: the classify authority and its writer."""
return self.classify_action is not None and self.exclusion_submitter is not None
[docs]
async def submit_exclusion(self, transaction_id: TransactionId) -> ManualLedgerTransactionResult:
"""Exclude one entry the operator can see in the review list."""
if self.classify_action is None or self.exclusion_submitter is None:
raise InternalInvariantError("exclusion is unavailable")
if transaction_id not in self.projection.review_transaction_ids:
raise ValueError("exclusion target is absent from the visible review list")
result = await self.exclusion_submitter(
LedgerExclusionSubmissionV1(action=self.classify_action, transaction_id=transaction_id)
)
if result.ref.transaction_id != transaction_id:
raise ValueError("exclusion result transaction identity disagrees")
return result
[docs]
def can_add_invoices(self) -> bool:
"""Report whether the invoice writer door is admitted in this session."""
return self.invoice_add_door is not None
[docs]
async def add_invoice(self, entry: LedgerInvoiceEntryV1) -> LedgerInvoiceAddResultV1:
"""Record one typed invoice through the injected catalogue writer."""
if self.invoice_add_door is None:
raise InternalInvariantError("invoice entry is unavailable")
return await self.invoice_add_door(entry)
[docs]
def evidence_records(self) -> tuple[LedgerEvidenceRecordRowV1, ...] | None:
"""Read the local evidence records, or ``None`` when no door reads them."""
return None if self.evidence_door is None else self.evidence_door.list_records()
[docs]
async def add_evidence(self, source_path: str) -> LedgerEvidenceRecordRowV1:
"""Register one document, as the operator typed its path, through the injected door."""
if self.evidence_door is None:
raise InternalInvariantError("evidence registration is unavailable")
return await self.evidence_door.add(source_path)
[docs]
async def confirm_evidence(self, confirmation: LedgerEvidenceConfirmationV1) -> LedgerEvidenceConfirmedV1:
"""Record one read document as an invoice through the injected evidence door."""
if self.evidence_door is None:
raise InternalInvariantError("evidence confirmation is unavailable")
return await self.evidence_door.confirm(confirmation)
[docs]
def reader_readiness(self) -> LedgerReaderReadinessV1 | None:
"""Measure the local reader, or ``None`` when no door can measure it."""
return None if self.evidence_door is None else self.evidence_door.reader_readiness()
[docs]
def can_refresh(self) -> bool:
"""Report whether this workspace can re-read its state after a write."""
return self.injection.refresh is not None
[docs]
def refreshed(self) -> LedgerWorkspaceController:
"""Return this workspace re-read from the current profile state.
The focus is kept only while the entry it names is still visible; a
focus on a row the write removed would otherwise address nothing.
"""
refresh = self.injection.refresh
if refresh is None:
return self
snapshot = refresh()
injection = replace(self.injection, evidence_items=snapshot.evidence_items)
focus = self.context.focus
if (
focus is not None
and focus.semantic_key == "ledger.transaction"
and all(row.transaction_id != focus.restore_token for row in snapshot.projection.entries)
):
focus = None
context = TuiScreenContextV1(
destination=self.context.destination,
focus=focus,
action_candidate_id=self.context.action_candidate_id,
)
return LedgerWorkspaceController(context, snapshot.projection, injection)
[docs]
def evidence_rows(self) -> tuple[LedgerEvidenceRowV1, ...]:
"""Project only canonical review-safe metadata from the injected result."""
if self.evidence_action is None or self.evidence_items is None:
raise InternalInvariantError("evidence review is unavailable")
return tuple(
LedgerEvidenceRowV1(
attachment_id=item.attachment_id,
mime_type=item.mime_type,
bytes_size=item.bytes_size,
captured_at=item.captured_at,
pending_review=item.pending_review,
action=self.evidence_action,
source=item,
)
for item in self.evidence_items
)
[docs]
def restored_evidence_id(self) -> str | None:
"""Restore evidence focus by semantic application identity."""
focus = self.context.focus
if focus is None or focus.semantic_key != "ledger.evidence" or focus.restore_token is None:
return None
return (
focus.restore_token
if any(row.attachment_id == focus.restore_token for row in self.evidence_rows())
else None
)
[docs]
def can_submit_links(self) -> bool:
"""Report whether the optional local link-mutation door is admitted."""
return self.link_action is not None and self.link_submitter is not None
[docs]
async def submit_link(self, transaction_id: TransactionId, invoice_id: InvoiceId) -> LedgerLinkResultV1:
"""Admit a visible suggestion and submit it through the authorized door."""
if self.link_action is None or self.link_submitter is None:
raise InternalInvariantError("Ledger reconciliation is unavailable")
if not any(
row.transaction_id == transaction_id and row.invoice_id == invoice_id
for row in self.projection.invoice_reconciliations
) or not any(row.transaction_id == transaction_id for row in self.projection.entries):
raise ValueError("link target is absent from the visible reconciliation projection")
submission = LedgerLinkSubmissionV1(
action=self.link_action,
transaction_id=transaction_id,
invoice_id=invoice_id,
)
result = await self.link_submitter(submission)
if result.transaction_id != transaction_id or result.invoice_id != invoice_id:
raise ValueError("link result identities disagree with the admitted suggestion")
return result
[docs]
class LedgerRouteRequested(Message):
"""Host-facing request to replace the active Ledger destination body."""
def __init__(self, target: LedgerRouteTargetV1) -> None:
"""Store the internal semantic target for the owning host."""
super().__init__()
self.target = target
[docs]
class LedgerReviewRequested(Message):
"""Host-facing request naming a real application query and transaction focus."""
def __init__(self, *, transaction_id: TransactionId, action: ActionReference) -> None:
"""Store the safe transaction identity and canonical query action."""
super().__init__()
self.transaction_id = transaction_id
self.action = action
[docs]
class LedgerEvidenceReviewRequested(Message):
"""Host-facing request carrying the canonical evidence-review query."""
def __init__(self, *, attachment_id: str, action: ActionReference) -> None:
"""Store one visible attachment identity and its admitted action."""
super().__init__()
self.attachment_id = attachment_id
self.action = action
[docs]
class LedgerEntrySelected(Message):
"""Host-facing semantic selection of one safe Ledger entry reference."""
def __init__(self, transaction_id: TransactionId) -> None:
"""Store only the application projection's safe transaction identity."""
super().__init__()
self.transaction_id = transaction_id
[docs]
class LedgerBackRequested(Message):
"""Request that the owning host return to the parent destination."""
[docs]
class LedgerInvoiceEntryRequested(Message):
"""Request the invoice entry form as the workspace body."""
[docs]
class LedgerWorkspaceScreen(AccountChromeScreen):
"""Shared one-scroll shell and semantic navigation behavior."""
BINDINGS: ClassVar = [Binding("escape", "back", "", show=False)]
IS_WORKSPACE_OVERVIEW: ClassVar[bool] = False
"""Whether Back from this body leaves the workspace rather than returning to its overview."""
CSS = BASE_CSS + tokenised(
"""
.ledger-page { width: 100%; height: 1fr; }
.ledger-section { width: 100%; height: auto; margin-bottom: $cadrumo-stack; }
.ledger-refusal { color: $warning; text-style: bold; height: auto; }
.ledger-empty { color: $text-muted; height: auto; }
"""
)
def __init__(self, controller: LedgerWorkspaceController, *, id: str | None = None) -> None:
"""Retain the read-only controller and interaction observations."""
super().__init__(id=id)
self.controller = controller
self.requested_target: LedgerRouteTargetV1 | None = None
self.refusal: LedgerRouteRefusalV1 | None = None
self.back_requested = False
self.refreshing = False
"""Whether a re-read of the workspace is running off the event loop."""
self._after_refresh: Callable[[], object] | None = None
self.refresh_seconds: list[float] = []
"""How long each completed re-read took, newest last."""
[docs]
def refresh_then(self, then: Callable[[], object]) -> None:
"""Re-read the workspace off the event loop, then continue with ``then``.
A capture reads every catalogue the workspace projects, which takes
seconds on a real ledger; on the loop that froze every key. A request
made while one is already running does not start a second read: it
replaces what happens when the running one lands, so the operator's
last intent wins and the store is read once.
"""
if not self.controller.can_refresh():
then()
return
self._after_refresh = then
if self.refreshing:
return
self.refreshing = True
self._status_line().update(ledger_copy("tui.ledger.flow.refreshing"))
self.run_worker(self._refresh(), group="ledger-refresh")
async def _refresh(self) -> None:
started = asyncio.get_running_loop().time()
try:
self.controller = await asyncio.to_thread(self.controller.refreshed)
except CadrumoError:
self._status_line().update(ledger_copy("tui.ledger.flow.refresh_failed"))
else:
self._status_line().update("")
finally:
self.refreshing = False
self.refresh_seconds.append(asyncio.get_running_loop().time() - started)
then, self._after_refresh = self._after_refresh, None
if then is not None:
then()
def _status_line(self) -> Static:
"""The body's flow status line, or its notice line when it has none."""
found = self.query("#ledger-flow-status")
return cast(Static, found.first()) if found else self.query_one("#ledger-refusal", Static)
[docs]
def populate_navigation(self) -> None:
"""Populate the complete seven-area catalogue in canonical order."""
table = cast("DataTable[str]", self.query_one("#ledger-navigation", DataTable))
table.add_column(ledger_copy("tui.ledger.column.destination"), key="destination")
table.add_column(ledger_copy("tui.ledger.column.availability"), key="availability")
table.add_column(ledger_copy("tui.ledger.column.items"), key="items")
for area in LedgerWorkspaceArea:
state = self.controller.state_for(area)
refusal = self.controller.refusal_for(area)
availability = state.availability if refusal is None else refusal.availability
# An area waiting only for the operator to pick an entry is not
# unavailable; saying so would send them away from the fix.
availability_cell = (
ledger_copy("tui.ledger.availability.choose_entry")
if refusal is not None and refusal.reason_key == "tui.ledger.refusal.selection_required"
else availability_label(availability)
)
table.add_row(area_label(area), availability_cell, item_count_label(state), key=area.value)
[docs]
def handle_navigation_selection(self, event: DataTable.RowSelected) -> bool:
"""Handle the common navigation table and expose refusals as visible copy."""
event_table = cast("DataTable[str]", event.data_table)
if event_table.id != "ledger-navigation":
return False
area = LedgerWorkspaceArea(str(event.row_key.value))
refusal = self.controller.refusal_for(area)
notice = self.query_one("#ledger-refusal", Static)
if refusal is not None:
self.refusal = refusal
notice.update(ledger_copy(refusal.reason_key))
return True
target = self.controller.route_target(area)
self.requested_target = target
notice.update("")
self.post_message(LedgerRouteRequested(target))
return True
[docs]
def action_back(self) -> None:
"""Ask the host to return; never terminate the application."""
self.back_requested = True
self.post_message(LedgerBackRequested())
[docs]
def on_ledger_entry_selected(self, event: LedgerEntrySelected) -> None:
"""Re-address the workspace at the chosen entry and repaint navigation.
Lives on the shared base rather than on the entries screen because the
controller is shared: review and reconciliation restore their cursor
from the same focus, so a selection made anywhere is the selection
everywhere.
The controller is REBOUND, not mutated — ``with_transaction_focus``
returns a new one — so the next area move, which resolves its screen
against ``self.controller``, carries the selection across the body swap
without any extra push. The navigation table is rebuilt because
classification's reachability has just changed, and a stale row would
keep offering the old refusal.
"""
self.controller = self.controller.with_transaction_focus(event.transaction_id)
table = cast("DataTable[str]", self.query_one("#ledger-navigation", DataTable))
table.clear(columns=True)
self.populate_navigation()
[docs]
def on_ledger_route_requested(self, event: LedgerRouteRequested) -> None:
"""Resolve the requested area here and hand the finished body to the host.
Textual delivers a screen's own posted message to it before bubbling,
so the workspace answers its own navigation: it owns the controller and
the route catalogue, and the host owns only the screen stack.
"""
# Imported at call time because the route catalogue imports this module
# for the shared shell; at module scope the two would form a cycle.
from .routes import resolve_ledger_screen
replace_workspace_body(cast(App[object], self.app), resolve_ledger_screen(self.controller, event.target))
[docs]
def on_ledger_invoice_entry_requested(self, _: LedgerInvoiceEntryRequested) -> None:
"""Open the invoice entry form, or say why it cannot open in this session."""
from .invoice_entry import LedgerInvoiceEntryScreen
if not self.controller.can_add_invoices():
self.query_one("#ledger-refusal", Static).update(ledger_copy("tui.ledger.refusal.submission_unavailable"))
return
replace_workspace_body(cast(App[object], self.app), LedgerInvoiceEntryScreen(self.controller))
[docs]
def on_ledger_back_requested(self, _: LedgerBackRequested) -> None:
"""Return an area to the Ledger overview; leave the workspace only from the overview."""
# An overview that cannot open would bounce Back straight back here.
if self.IS_WORKSPACE_OVERVIEW or self.controller.refusal_for(LedgerWorkspaceArea.OVERVIEW) is not None:
self.dismiss(None)
return
self.post_message(LedgerRouteRequested(self.controller.route_target(LedgerWorkspaceArea.OVERVIEW)))
[docs]
def on_ledger_review_requested(self, _: LedgerReviewRequested) -> None:
"""Say that the selection led nowhere, rather than appearing to work.
Selecting a review row names a real query action, but no production
consumer executes one yet. Answering with the pending-area copy is the
honest state: an unhandled message is indistinguishable from a dead
keypress, and that silence is precisely how the workspace navigation
defect stayed invisible through the whole surface's construction.
"""
self._refuse_pending_destination()
[docs]
def on_ledger_evidence_review_requested(self, _: LedgerEvidenceReviewRequested) -> None:
"""Report the same pending state for an evidence-row selection."""
self._refuse_pending_destination()
def _refuse_pending_destination(self) -> None:
"""Surface the canonical pending-area copy on this screen's notice."""
self.query_one("#ledger-refusal", Static).update(ledger_copy("tui.ledger.refusal.destination_pending"))
__all__ = [
"LedgerBackRequested",
"LedgerEntrySelected",
"LedgerEvidenceReviewRequested",
"LedgerInvoiceEntryRequested",
"LedgerReviewRequested",
"LedgerRouteRequested",
"LedgerWorkspaceController",
"LedgerWorkspaceScreen",
"area_label",
"availability_label",
"item_count_label",
"ledger_copy",
"review_status_label",
"status_label",
]