Establish enforced logging port (#286)

docs/developers-guide.md

@@ -780,6 +780,11 @@ the importer, imported module, and dependency direction.
 
 The enforced groups are:
 
+- `logging_port`: the public `episodic.logging` seam used by every package.
+- `logging_backend`: the `femtologging` implementation, which only
+  `logging_port` may import.
+- `external_libraries`: approved third-party libraries, tracked as import
+  edges alongside the internal groups.
 - `domain_ports`: canonical domain types, canonical ports, ingestion ports,
   canonical constraint names, and LLM ports.
 - `application`: canonical services, profile/template workflows,
@@ -791,10 +796,13 @@ The enforced groups are:
 - `composition_root`: modules that wire concrete adapters, currently
   `episodic.api.runtime` and `episodic.worker.runtime`.
 
-When adding a new port or adapter, update `[tool.hecate]` in `pyproject.toml`
-in the same change as the package. Keep composition-root prefixes before
-broader adapter prefixes because Hecate uses first-match group ordering. Add or
-adjust fixture coverage in `tests/fixtures/architecture/` and run:
+`include_external_packages = true` makes Hecate classify third-party import
+edges. The groups are ordered by first matching prefix: keep `logging_port`
+and `logging_backend` before the broad `external_libraries` group so the latter
+cannot silently permit direct femtologging imports. When adding a new port or
+adapter, update `[tool.hecate]` in `pyproject.toml` in the same change as the
+package. Add or adjust fixture coverage in `tests/fixtures/architecture/` and
+run:
 
 ```shell
 uv run pytest -q tests/test_architecture_enforcement.py \
@@ -1926,17 +1934,27 @@ normalizer.
 
 ## Logging
 
-Structured logging uses femtologging v0.1.0-style logger methods. Import
-`get_logger` (or `getLogger` when matching stdlib naming) from
-`episodic.logging`, then emit via `logger.info(...)`, `logger.warning(...)`,
-`logger.error(...)`, or `logger.exception(...)`.
-
-Keep `episodic.logging.configure_logging(...)` as the local configuration seam.
-The legacy `log_info`, `log_warning`, and `log_error` helpers remain available
-for compatibility, but new code should prefer calling the logger methods
-directly. Femtologging still expects pre-formatted messages rather than stdlib
-`logger.info("%s", value)` lazy formatting, so build the final string before
-calling the method.
+`episodic.logging` is the project's logging port. Call sites obtain a
+`LoggerHandle` through `get_logger` (or the stdlib-compatible `getLogger` alias)
+and emit through `log_debug`, `log_info`, `log_warning`, `log_error`, or
+`log_exception`. Pass percent-style templates and arguments to the port; it
+eagerly formats the final message for femtologging.
+
+The handle deliberately exposes no level methods. The raw femtologging logger
+is an implementation detail, not a public surface: logging behaviour that
+crosses package boundaries must pass through the port. This gives each call
+site a stable emission contract while allowing the port to attach the request
+correlation identifier and future common fields consistently.
+
+Keep `configure_logging(...)` as the local configuration seam. `log_exception`
+records exception information by default; the other wrappers accept optional
+`exc_info` explicitly when a caller needs it.
+
+The import boundary is enforced by Hecate's `logging_port` and
+`logging_backend` groups. The direct-call boundary has two independent gates:
+the opaque `LoggerHandle` is rejected by `ty` when code accesses `.info` or
+another level method, and `tests/test_logging_port_enforcement.py` scans the
+production tree so a regression reports every offending file.
 
 ### LogLevel
 

episodic/api/authorization.py

@@ -7,7 +7,7 @@
 
 import falcon
 
-from episodic.logging import LogLevel, get_logger, log_warning
+from episodic.logging import get_logger, log_debug, log_warning
 
 logger = get_logger(__name__)
 
@@ -173,7 +173,10 @@ def _log_authorization_denial(
     context: AuthorizationContext,
 ) -> None:
     """Log non-permit decisions without recording credential material."""
-    logger.log(
-        LogLevel.DEBUG,
-        (f"Authorization denied with {decision} for {context.method} {context.path}."),
+    log_debug(
+        logger,
+        "Authorization denied with %s for %s %s.",
+        decision,
+        context.method,
+        context.path,
     )

episodic/canonical/ingestion_service.py

@@ -21,7 +21,7 @@
 import typing as typ
 
 from episodic.asyncio_tasks import TaskMetadata, create_task
-from episodic.logging import get_logger
+from episodic.logging import get_logger, log_info
 
 from .domain import IngestionRequest
 from .services import ingest_sources
@@ -185,10 +185,14 @@ def _log_multi_source_outcome(
     episode: CanonicalEpisode,
 ) -> None:
     """Log the completed multi-source ingestion outcome."""
-    logger.info(
-        f"Multi-source ingestion complete: {len(source_inputs)} sources, "
-        f"{len(outcome.preferred_sources)} preferred, "
-        f"{len(outcome.rejected_sources)} rejected. Episode {episode.id}."
+    log_info(
+        logger,
+        "Multi-source ingestion complete: %s sources, %s preferred, %s rejected. "
+        "Episode %s.",
+        len(source_inputs),
+        len(outcome.preferred_sources),
+        len(outcome.rejected_sources),
+        episode.id,
     )
 
 

episodic/canonical/services.py

@@ -20,7 +20,7 @@
 import typing as typ
 import uuid
 
-from episodic.logging import get_logger
+from episodic.logging import get_logger, log_info
 
 from . import reference_documents
 from .domain import (
@@ -268,8 +268,11 @@ async def ingest_sources(
     await uow.approval_events.add(event)
 
     await uow.commit()
-    logger.info(
-        f"Ingested {len(request.sources)} sources into canonical episode {episode_id}."
+    log_info(
+        logger,
+        "Ingested %s sources into canonical episode %s.",
+        len(request.sources),
+        episode_id,
     )
 
     return episode

episodic/canonical/storage/migration_check.py

@@ -24,7 +24,7 @@
 
 from episodic.canonical.storage.alembic_helpers import apply_migrations
 from episodic.canonical.storage.models import Base
-from episodic.logging import get_logger
+from episodic.logging import get_logger, log_error, log_exception, log_info
 
 if typ.TYPE_CHECKING:
     import sqlalchemy as sa
@@ -75,7 +75,9 @@ async def check_migrations_cli() -> int:
         detected, 2 on infrastructure errors.
     """
     if importlib.util.find_spec("py_pglite") is None:
-        _logger.error("py-pglite is not installed; cannot run migration drift check.")
+        log_error(
+            _logger, "py-pglite is not installed; cannot run migration drift check."
+        )
         return 2
 
     import tempfile
@@ -95,24 +97,24 @@ async def check_migrations_cli() -> int:
                 dsn = config.get_connection_string()
                 engine = create_async_engine(dsn, pool_pre_ping=True)
                 try:
-                    _logger.info("Applying migrations to ephemeral database.")
+                    log_info(_logger, "Applying migrations to ephemeral database.")
                     await apply_migrations(engine)
 
-                    _logger.info("Checking for schema drift.")
+                    log_info(_logger, "Checking for schema drift.")
                     diffs = await detect_schema_drift(engine)
                 finally:
                     await engine.dispose()
     except _INFRASTRUCTURE_ERRORS:
-        _logger.exception("Infrastructure error.")
+        log_exception(_logger, "Infrastructure error.")
         return 2
 
     if diffs:
-        _logger.error(f"Schema drift detected ({len(diffs)} difference(s)):")
+        log_error(_logger, "Schema drift detected (%s difference(s)):", len(diffs))
         for diff in diffs:
-            _logger.error(f"  {diff}")
+            log_error(_logger, "  %s", diff)
         return 1
 
-    _logger.info("No schema drift detected.")
+    log_info(_logger, "No schema drift detected.")
     return 0
 
 

episodic/canonical/storage/uow.py

@@ -16,7 +16,7 @@
 
 from episodic.canonical.unit_of_work_protocols import CanonicalUnitOfWork
 from episodic.cost.storage import SqlAlchemyCostLedgerStore
-from episodic.logging import get_logger
+from episodic.logging import get_logger, log_info
 
 from .episode_repository import SqlAlchemyEpisodeRepository
 from .generation_runs import SqlAlchemyGenerationRunStore
@@ -211,7 +211,7 @@ async def commit(self) -> None:
             If no unit-of-work session is active.
         """  # noqa: DOC502  # Documents an exception propagated by the helper.
         await self._apply_session_action("commit")
-        logger.info("Committed canonical unit of work.")
+        log_info(logger, "Committed canonical unit of work.")
 
     async def flush(self) -> None:
         """Flush pending unit-of-work changes."""