refactor: invert server dependencies onto interfaces (DIP)

src/komet_node/__main__.py

@@ -1,11 +1,15 @@
 from __future__ import annotations
 
 import argparse
+import tempfile
 from pathlib import Path
 
 from stellar_sdk import Network
 
+from komet_node.interpreter import NodeInterpreter
 from komet_node.server import StellarRpcServer
+from komet_node.store import ChainStore
+from komet_node.transaction import TransactionEncoder
 
 _DESCRIPTION = 'Komet Node — a local Stellar testnet backed by the K semantics of Soroban.'
 
@@ -35,14 +39,31 @@ def main() -> None:
     )
     args = parser.parse_args()
 
-    server = StellarRpcServer(
-        host=args.host,
-        port=args.port,
-        io_dir=args.io_dir,
-        network_passphrase=Network.TESTNET_NETWORK_PASSPHRASE,
-    )
+    server = build_server(io_dir=args.io_dir, host=args.host, port=args.port)
     server.serve()
 
 
+def build_server(
+    *,
+    io_dir: Path | None = None,
+    network_passphrase: str = Network.TESTNET_NETWORK_PASSPHRASE,
+    host: str = 'localhost',
+    port: int = 8000,
+) -> StellarRpcServer:
+    """Composition root: build the concrete collaborators and wire up the server.
+
+    With no ``io_dir`` the chain runs against a fresh temporary directory — a throwaway chain
+    that starts empty on every launch and leaves the working directory untouched.
+    """
+    resolved_io_dir = Path(tempfile.mkdtemp(prefix='komet-node-')) if io_dir is None else io_dir
+    return StellarRpcServer(
+        interpreter=NodeInterpreter(),
+        encoder=TransactionEncoder(network_passphrase),
+        store=ChainStore(resolved_io_dir),
+        host=host,
+        port=port,
+    )
+
+
 if __name__ == '__main__':
     main()

src/komet_node/interfaces.py

@@ -0,0 +1,146 @@
+"""The abstractions the server depends on: the shared data shapes and the collaborator protocols.
+
+:class:`~komet_node.server.StellarRpcServer` is written against these, not against the
+concrete `NodeInterpreter` / `TransactionEncoder` / `ChainStore`, so the concretes can be
+injected (and substituted in tests) — the dependency-inversion boundary. The concrete classes
+declare that they implement the protocols, and a composition root wires them together (see
+``build_server`` in ``__main__.py``).
+
+Two kinds of thing live here so both the protocols and the concretes can share them without
+either importing the other:
+
+- the request/response and disk data shapes (``TypedDict``);
+- the ``Interpreter`` / ``Encoder`` / ``Store`` protocols.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Protocol, TypedDict
+
+if TYPE_CHECKING:
+    from collections.abc import Callable, Mapping
+    from pathlib import Path
+
+    from pyk.kast.inner import KInner
+
+
+# ----------------------------------------------------------------------
+# Data shapes (functional TypedDict form: the keys are JSON wire / disk
+# names, some camelCase, so they cannot be class attributes).
+# ----------------------------------------------------------------------
+
+# The request envelopes consumed by node.md. `steps` is the JSON-encoded operation list; for a
+# wasm upload it is empty and the steps ride in the <program> cell instead. Key *order* also
+# matters to K's JSON matcher — which a TypedDict cannot express, only the fields and types.
+TxRequest = TypedDict(
+    'TxRequest',
+    {'method': str, 'id': Any, 'now': str, 'txHash': str, 'envelopeXdr': str, 'steps': list},
+)
+SimulateRequest = TypedDict(
+    'SimulateRequest',
+    {'method': str, 'id': Any, 'now': str, 'steps': list},
+)
+
+# The internal simulateTransaction result K emits: always ``latestLedger``, then either
+# ``error`` (a failed simulation) or ``returnValue`` (a JSON SCVal). total=False — mutually
+# exclusive outcomes.
+SimulateResult = TypedDict(
+    'SimulateResult',
+    {'latestLedger': int, 'error': str, 'returnValue': Any},
+    total=False,
+)
+
+# The ``ledgers/ledger_<seq>.json`` record for one closed ledger; carries the header XDR
+# artifacts (hash/headerXdr/metadataXdr) that only Python can build.
+LedgerRecord = TypedDict(
+    'LedgerRecord',
+    {
+        'sequence': int,
+        'txHash': str,
+        'closedAt': int,
+        'hash': str,
+        'headerXdr': str,
+        'metadataXdr': str,
+    },
+)
+
+# One entry of an ``events/events_<ledger>.json`` array, in the getEvents Event shape.
+EventRecord = TypedDict(
+    'EventRecord',
+    {
+        'type': str,
+        'ledger': int,
+        'ledgerClosedAt': str,
+        'contractId': str,
+        'id': str,
+        'inSuccessfulContractCall': bool,
+        'txHash': str,
+        'topic': list[str],
+        'value': str,
+    },
+)
+
+
+# ----------------------------------------------------------------------
+# Collaborator protocols
+# ----------------------------------------------------------------------
+
+
+class Interpreter(Protocol):
+    """Runs RPC request envelopes through the K node semantics (see ``NodeInterpreter``)."""
+
+    def empty_config(self) -> str: ...
+
+    def run(
+        self,
+        state_file: Path,
+        io_dir: Path,
+        request: Mapping[str, Any],
+        program_steps: list[KInner] | None = ...,
+        *,
+        commit: bool = ...,
+    ) -> str | None: ...
+
+
+class Encoder(Protocol):
+    """Decodes Stellar XDR transactions into node request envelopes (see ``TransactionEncoder``)."""
+
+    network_passphrase: str
+
+    def build_tx_request(
+        self, method: str, rpc_id: Any, transaction_xdr: str, now: str
+    ) -> tuple[TxRequest, list[KInner] | None, dict[str, bytes]]: ...
+
+    def build_simulate_request(self, rpc_id: Any, transaction_xdr: str, now: str) -> SimulateRequest: ...
+
+
+class Store(Protocol):
+    """Reads and writes the files of a komet-node io-dir (see ``ChainStore``)."""
+
+    root: Path
+    state_file: Path
+    wasms_dir: Path
+
+    def initialize(self, empty_config: Callable[[], str]) -> bool: ...
+
+    def latest_ledger(self) -> int: ...
+
+    def has_receipt(self, tx_hash: str) -> bool: ...
+
+    def read_receipt(self, tx_hash: str) -> dict[str, Any]: ...
+
+    def write_receipt(self, tx_hash: str, receipt: dict[str, Any]) -> None: ...
+
+    def read_ledger(self, sequence: int) -> LedgerRecord | None: ...
+
+    def write_ledger(self, record: LedgerRecord) -> None: ...
+
+    def write_wasm(self, wasm_hash: str, wasm: bytes) -> None: ...
+
+    def read_staged_event_lines(self) -> list[str]: ...
+
+    def clear_staged_events(self) -> None: ...
+
+    def write_events(self, ledger: int, events: list[EventRecord]) -> None: ...
+
+    def archive_request(self, method: str | None, params: dict[str, Any], request_id: Any) -> None: ...

src/komet_node/interpreter.py

@@ -14,6 +14,7 @@
 from pyk.utils import check_file_path, run_process_2
 
 from .errors import NodeInterpreterError
+from .interfaces import Interpreter
 from .utils import simbolik_definition
 
 if TYPE_CHECKING:
@@ -79,7 +80,7 @@ def _set_cell(pattern: Pattern, cell_symbol: str, value: Pattern) -> Pattern:
     return pattern
 
 
-class NodeInterpreter:
+class NodeInterpreter(Interpreter):
     """
     Runs the K node semantics against a saved KORE world-state configuration.
 

src/komet_node/server.py

@@ -5,28 +5,27 @@
 import logging
 import re
 import sys
-import tempfile
 import time
 import traceback
 from datetime import datetime, timezone
 from http.server import BaseHTTPRequestHandler, HTTPServer
-from pathlib import Path
-from typing import TYPE_CHECKING, Any, Final, TypedDict
+from typing import TYPE_CHECKING, Any, Final
 
-from stellar_sdk import Network, StrKey, TransactionEnvelope, xdr
+from stellar_sdk import StrKey, TransactionEnvelope, xdr
 
 from komet_node.errors import RpcError
-from komet_node.interpreter import NodeInterpreter
 from komet_node.ledger import build_ledger_artifacts
 from komet_node.ledger_entries import InvalidParamsError, format_ledger_entries_response, ledger_key_descriptors
 from komet_node.result_xdr import transaction_meta_xdr, transaction_result_xdr
 from komet_node.scval import scval_from_json
-from komet_node.store import ChainStore, EventRecord, LedgerRecord
-from komet_node.transaction import SimulationRejected, TransactionEncoder, malformed_tx_result_xdr
+from komet_node.transaction import SimulationRejected, malformed_tx_result_xdr
 
 if TYPE_CHECKING:
     from collections.abc import Mapping
     from http.server import HTTPServer as HTTPServerType
+    from pathlib import Path
+
+    from komet_node.interfaces import Encoder, EventRecord, Interpreter, LedgerRecord, SimulateResult, Store
 
 _PROTOCOL_VERSION: Final = 22
 
@@ -103,53 +102,50 @@ def _empty_transaction_data() -> str:
 
 _log = logging.getLogger('komet_node')
 
-# The internal simulateTransaction result K emits (node.md): always ``latestLedger``, then
-# either ``error`` (a failed simulation) or ``returnValue`` (a JSON SCVal for a successful
-# one). ``total=False`` — the two outcomes are mutually exclusive.
-SimulateResult = TypedDict(
-    'SimulateResult',
-    {'latestLedger': int, 'error': str, 'returnValue': Any},
-    total=False,
-)
-
 
 class StellarRpcServer:
     """
     Long-running HTTP/JSON-RPC server that wraps the one-shot K node semantics.
 
     The compiled semantics run one request per process invocation and hold no state
     between runs, so this server supplies what they lack: it keeps the HTTP socket open,
-    persists state to disk, and decodes the Stellar XDR envelope (:class:`TransactionEncoder`)
-    that K cannot parse. It then runs the request envelope through the semantics
-    (:class:`NodeInterpreter`). All RPC dispatch, receipt bookkeeping, ledger accounting,
-    and response formatting are performed in K (``node.md``). The on-disk artifacts — input
-    and output — belong to the :class:`~komet_node.store.ChainStore`, which owns the io-dir
-    layout; this server holds one and asks it for receipts, ledgers, events, and the ledger
-    counter rather than touching paths itself.
+    persists state to disk, and decodes the Stellar XDR envelope (an injected
+    :class:`~komet_node.interfaces.Encoder`) that K cannot parse. It then runs the request
+    envelope through the semantics (an :class:`~komet_node.interfaces.Interpreter`). All RPC
+    dispatch, receipt bookkeeping,
+    ledger accounting, and response formatting are performed in K (``node.md``). The on-disk
+    artifacts — input and output — belong to the injected :class:`~komet_node.interfaces.Store`,
+    which owns the io-dir layout; this server holds one and asks it for receipts, ledgers,
+    events, and the ledger counter rather than touching paths itself.
+
+    The three collaborators (interpreter, encoder, store) are injected rather than constructed
+    here, so the server depends only on the protocols in :mod:`komet_node.interfaces` and a test
+    can drive it with fakes. A composition root wires the concretes — see ``build_server`` in
+    ``__main__.py``.
     """
 
-    interpreter: NodeInterpreter
-    encoder: TransactionEncoder
-    store: ChainStore
+    interpreter: Interpreter
+    encoder: Encoder
+    store: Store
     io_dir: Path
     state_file: Path
 
     def __init__(
         self,
+        *,
+        interpreter: Interpreter,
+        encoder: Encoder,
+        store: Store,
         host: str = 'localhost',
         port: int = 8000,
-        io_dir: Path | None = None,
-        network_passphrase: str = Network.TESTNET_NETWORK_PASSPHRASE,
     ) -> None:
+        self.interpreter = interpreter
+        self.encoder = encoder
+        self.store = store
         self.host = host
         self._port = port
-        self.interpreter = NodeInterpreter()
-        self.encoder = TransactionEncoder(network_passphrase)
         self._httpd: HTTPServerType | None = None
 
-        # With no io-dir given, run against a fresh temporary directory: a throwaway chain
-        # that starts empty on every launch and leaves the working directory untouched.
-        self.store = ChainStore(Path(tempfile.mkdtemp(prefix='komet-node-')) if io_dir is None else io_dir)
         self._fresh = self.store.initialize(self.interpreter.empty_config)
         # state_file and io_dir are the paths the interpreter subprocess resolves against; kept
         # as attributes so callers (and tests) can reach the io-dir root directly.

src/komet_node/store.py

@@ -23,53 +23,24 @@
 from __future__ import annotations
 
 import json
-from typing import TYPE_CHECKING, Any, TypedDict
+from typing import TYPE_CHECKING, Any
+
+from komet_node.interfaces import Store
 
 if TYPE_CHECKING:
     from collections.abc import Callable
     from pathlib import Path
 
-
-# The ``ledgers/ledger_<seq>.json`` record for one closed ledger. Carries the ledger-header
-# XDR artifacts (hash/headerXdr/metadataXdr) that only Python can build; the semantics read
-# them back to serve getTransactions/getLedgers. The functional TypedDict form is used because
-# the keys are the spec's camelCase JSON wire names, not Python identifiers.
-LedgerRecord = TypedDict(
-    'LedgerRecord',
-    {
-        'sequence': int,
-        'txHash': str,
-        'closedAt': int,
-        'hash': str,
-        'headerXdr': str,
-        'metadataXdr': str,
-    },
-)
-
-# One entry of an ``events/events_<ledger>.json`` array, in the getEvents Event shape.
-EventRecord = TypedDict(
-    'EventRecord',
-    {
-        'type': str,
-        'ledger': int,
-        'ledgerClosedAt': str,
-        'contractId': str,
-        'id': str,
-        'inSuccessfulContractCall': bool,
-        'txHash': str,
-        'topic': list[str],
-        'value': str,
-    },
-)
+    from komet_node.interfaces import EventRecord, LedgerRecord
 
 
 # Where the K semantics stage the contract events of the currently executing transaction
 # (one JSON record per line, appended by the `contract_event` interception in node.md).
 _EVENTS_STAGING = 'events_staged.jsonl'
 
 
-class ChainStore:
-    """Reads and writes the files of a single komet-node io-dir."""
+class ChainStore(Store):
+    """Reads and writes the files of a single komet-node io-dir (the :class:`Store` concretion)."""
 
     def __init__(self, io_dir: Path) -> None:
         self.root = io_dir.resolve()