"""Where the parts are assembled. One place that knows how the platform is wired, so nothing else has to. Every module below this is written against interfaces and takes its collaborators as arguments; this is the only file that names concrete implementations together. Construction is lazy, one property at a time, so a component that is expensive or absent is only built when something asks for it. That matters for OCR in particular: loading models takes seconds and hundreds of megabytes, and a container that did it eagerly would make `--check` slow and `migrate` impossible on a machine without them. """ from __future__ import annotations import logging import os from functools import cached_property from app.api.client import CmsClient from app.config.settings import Settings from app.memory.repository import InMemoryRepository from app.ocr.config import OcrConfig from app.ocr.engine import build_engine from app.tools.client_tools import ( FindClientByIdentifierTool, GetClientTool, SearchClientTool, ) from app.tools.document_tools import ( ExtractDocumentFieldsTool, ClassifyDocumentTool, ExtractFieldsTool, OcrTool, PrepareImageTool, SubmitFilingProposalTool, ) from app.tools.memory_tools import OpenGapsTool, RecallTool, RememberTool from app.tools.registry import ToolRegistry from app.workflow import catalogue from app.workflow.decisions import DecisionPoller from app.workflow.engine import InMemoryRunStore, WorkflowEngine logger = logging.getLogger(__name__) class Container: """Everything this deployment runs, built once.""" def __init__(self, env: dict[str, str] | None = None) -> None: self._env = env if env is not None else dict(os.environ) # ── Configuration ───────────────────────────────────────────────────── @cached_property def settings(self) -> Settings: return Settings.from_env(self._env) @cached_property def ocr_config(self) -> OcrConfig: return OcrConfig.from_env(self._env) @cached_property def database_url(self) -> str | None: return self._env.get("TAXPILOT_DATABASE_URL") or None @cached_property def inbox(self): """Which WhatsApp conversations this deployment may touch. One: the connected account's own self-chat. There is no setting here — the owner is discovered from the linked session and is itself the policy, so nothing has to be typed in and nothing can be typed wrong. Fails closed. This attaches to a real person's account, carrying their family, their friends and every other client they have, so until the connected number is known nothing is processed at all. """ from app.whatsapp.inbox import InboxFilter, SelfChatPolicy self._refuse_retired_allow_list() def owner() -> str | None: """The connected number, over the provider's authenticated API. Deliberately not the number the CMS last reported, and not anything out of a webhook: both are claims from elsewhere about whose account this is, and that is the one fact the policy cannot afford to take on trust. """ from app.whatsapp.session import status_of provider = self.whatsapp return status_of(provider).number if provider is not None else None return InboxFilter(SelfChatPolicy(), owner_resolver=owner) def _refuse_retired_allow_list(self) -> None: """Complain loudly about settings that no longer do anything. These used to widen what the AI would read. They are gone, and silence would be the dangerous outcome: an operator who set them believes a conversation is being watched, and would go looking at the provider rather than here when nothing arrived from it. """ retired = [ name for name in ( "TAXPILOT_WHATSAPP_ALLOWED_NUMBERS", "TAXPILOT_WHATSAPP_ALLOWED_CHATS", "TAXPILOT_WHATSAPP_ALLOWED_SENDERS", ) if self._env.get(name, "").strip() ] if retired: logger.warning( "Ignoring %s: the inbox is the connected account's own self-chat and " "cannot be widened by configuration. Remove these to avoid confusion.", ", ".join(retired), ) @cached_property def meta_config(self): """Meta credentials, or None when this deployment runs on Evolution.""" from app.whatsapp.meta import MetaConfig return MetaConfig.from_env(self._env) @cached_property def whatsapp(self): """The provider this deployment talks through (ADR-0006). Meta when credentials are present, Evolution otherwise. Evolution drives WhatsApp through an unofficial path and risks the number being banned — so a deployment falls back to it only by having given no Meta credentials, and is told loudly. """ if self.meta_config is not None: from app.whatsapp.meta import MetaBusinessProvider return MetaBusinessProvider(self.meta_config) from app.whatsapp.evolution import EvolutionProvider base_url = self._env.get("TAXPILOT_EVOLUTION_URL", "") if base_url: logger.warning( "Using Evolution, which drives WhatsApp unofficially and risks the " "number being banned. It must never point at a firm's live number." ) return EvolutionProvider( base_url=base_url, instance=self._env.get("TAXPILOT_EVOLUTION_INSTANCE", ""), api_key=self._env.get("TAXPILOT_EVOLUTION_KEY", ""), ) @cached_property def inbound_queue(self): from app.whatsapp.webhook import InboundQueue return InboundQueue() @cached_property def seen_messages(self): """Shared between the receiver and the sender. Deliberately one instance: every message the sender transmits is recorded here, so the provider's echo of our own reply is recognised as already handled. That is what makes the note-to-self inbox possible — `fromMe` cannot distinguish our replies from the owner's own documents. """ from app.whatsapp.receiver import SeenMessages return SeenMessages() @cached_property def message_sender(self): from app.whatsapp.provider import MessageSender return MessageSender(self.whatsapp, sent=self.seen_messages) @cached_property def receiver(self): from app.whatsapp.receiver import WebhookReceiver return WebhookReceiver( self.whatsapp, self.message_sender, # The webhook only queues; the daemon does the work, so the handler # here is deliberately empty. OCR inside a webhook would exceed the # provider's patience and earn a retry of the document already # being processed. lambda _message: None, self.seen_messages, self.inbox, ) @cached_property def http_host(self) -> str: """Where the health server binds. Loopback by default. Inside a container that is enough for Docker's own health check, and it means an accidental port publish does not put an infrastructure status page on the internet. A deployment fronting the webhook with a reverse proxy sets 0.0.0.0 deliberately. """ return self._env.get("TAXPILOT_HTTP_HOST", "127.0.0.1") @cached_property def http_port(self) -> int: return int(self._env.get("TAXPILOT_HTTP_PORT", "8080")) @cached_property def poll_seconds(self) -> float: """How often to ask the CMS what reviewers have decided. Thirty seconds by default. A reviewer approving a document does not expect the AI to notice instantly, and polling harder buys nothing but load on an installation that is also serving people. """ return float(self._env.get("TAXPILOT_POLL_SECONDS", "30")) @cached_property def archive_after_days(self) -> int: return int(self._env.get("TAXPILOT_ARCHIVE_AFTER_DAYS", "90")) @cached_property def evolution_webhook_secret(self) -> str: """The token Evolution must present on every delivery. Evolution signs nothing — unlike Meta, which HMACs the body — so this is the whole of the authentication, and the inbound route refuses to register without it. Empty is not "open": it is "no route". """ return self._env.get("TAXPILOT_EVOLUTION_WEBHOOK_SECRET", "").strip() @cached_property def incoming_dir(self): """Where a document lands between the provider and the reader. These are Level 3 files — a client's CNIC, their bank statement — so they stay on the installation's own disk and never go anywhere else (ADR-0002). Created on demand rather than at boot, because a deployment that never receives a document should not leave a directory behind explaining what it might have held. """ from pathlib import Path directory = Path( self._env.get("TAXPILOT_INCOMING_DIR", "") or Path(self._env.get("TAXPILOT_DATA_DIR", ".")) / "incoming" ) directory.mkdir(parents=True, exist_ok=True) return directory # ── Infrastructure ──────────────────────────────────────────────────── @cached_property def connect(self): """A connection factory, or None when no database is configured. A callable rather than a connection: this is built at start-up, possibly before PostgreSQL is accepting connections, and a container that opened a socket eagerly would fail to construct for a reason nothing could report. """ if not self.database_url: return None import psycopg url = self.database_url def factory(): return psycopg.connect(url) return factory @cached_property def cms(self) -> CmsClient: return CmsClient(self.settings) @cached_property def ocr(self): return build_engine(self.ocr_config) @cached_property def memory(self): # Postgres-backed memory arrives with pgvector; in-process until then, # and honest about it — recall works, it simply does not survive a # restart. return InMemoryRepository() @cached_property def store(self): """Durable when a database is configured, in-process otherwise. The fallback is deliberate and logged loudly. A deployment that has not been given a database still runs — but it loses every in-flight workflow on restart, which for a proposal already sitting in somebody's queue means the approval arrives for a run that no longer exists. """ if self.connect is None: logger.warning( "No TAXPILOT_DATABASE_URL: workflows are held in memory and will " "not survive a restart." ) return InMemoryRunStore() from app.workflow.postgres_store import PostgresRunStore return PostgresRunStore(self.connect) @cached_property def pending_pdfs(self): """PDFs downloaded and waiting for the text that names their client. Durable for the same reason runs are, and more urgently: a queued PDF has already been taken out of WhatsApp's hands, so losing it on restart loses the document itself rather than merely the progress on it. """ from app.whatsapp.pending import InMemoryPendingPdfs, PostgresPendingPdfs if self.connect is None: logger.warning( "No TAXPILOT_DATABASE_URL: PDFs awaiting their identifier are held " "in memory and will not survive a restart." ) return InMemoryPendingPdfs() return PostgresPendingPdfs(self.connect) @cached_property def intake_outbox(self): """Registrations the CMS has not acknowledged yet (ADR-0010). Durable for a reason the others share and this one sharpens: a document held here has arrived and has *not* been recorded anywhere the firm can see. Losing this table on restart loses the only evidence that somebody sent something — which is precisely the failure the Intake Center exists to make impossible. """ from app.intake import InMemoryOutbox, PostgresOutbox if self.connect is None: logger.warning( "No TAXPILOT_DATABASE_URL: documents waiting to be registered with " "the CMS are held in memory and will not survive a restart." ) return InMemoryOutbox() return PostgresOutbox(self.connect) # ── The platform ────────────────────────────────────────────────────── @cached_property def tools(self) -> ToolRegistry: """The catalogue this deployment offers. Registered explicitly rather than discovered: a Tool that becomes available merely by existing on disk is a Tool nobody decided to ship. """ registry = ToolRegistry() registry.register(PrepareImageTool()) registry.register(OcrTool(self.ocr)) registry.register(ClassifyDocumentTool()) registry.register(ExtractFieldsTool()) registry.register(ExtractDocumentFieldsTool()) registry.register(SearchClientTool(self.cms)) registry.register(FindClientByIdentifierTool(self.cms)) registry.register(GetClientTool(self.cms)) registry.register(SubmitFilingProposalTool(self.cms)) registry.register(RememberTool(self.memory)) registry.register(RecallTool(self.memory)) registry.register(OpenGapsTool(self.memory)) # Deliberately absent: UploadDocumentTool, which files without a human. # It exists for ADR-0004's promoted path and is registered only by a # deployment that has been granted `documents.write` on purpose. return registry @cached_property def model(self): """A model that may classify documents, or None. None unless `TAXPILOT_MODEL_URL` and `TAXPILOT_MODEL_NAME` are both set — the stage is off by default — and None again unless that URL is on this host. A document's text is Level 3 and does not leave the installation (ADR-0002, ADR-0009); `model.build` refuses a public endpoint and says so at boot. """ from app.documents import model return model.build( self._env.get("TAXPILOT_MODEL_URL"), self._env.get("TAXPILOT_MODEL_NAME"), float(self._env.get("TAXPILOT_MODEL_TIMEOUT", "20")), ) @cached_property def engine(self) -> WorkflowEngine: return WorkflowEngine(self.tools, self.store) @cached_property def poller(self) -> DecisionPoller: """Every workflow that can pause for a reviewer. All of them, from the catalogue. A decision comes back naming a proposal and nothing else, so which definition to resume against is a property of the paused run — and a poller holding a subset would step a run through another workflow's sequence, or silently ignore its approval. Taken from the catalogue rather than listed here, because a list maintained beside the one the router reads is a list that drifts. """ return DecisionPoller(self.cms, self.engine, catalogue.WORKFLOWS)