refactor(protocol): enforce abstract contracts and static checks

src/graphon/dsl/entities.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import abc
 from collections.abc import Mapping
 from enum import StrEnum, auto
 from typing import Any, Protocol
@@ -125,4 +126,5 @@ def loadable(self) -> bool:
 
 
 class TypedNodeFactory(Protocol):
+    @abc.abstractmethod
     def create_node(self, node_config: NodeConfigDict) -> Any: ...

src/graphon/dsl/slim/llm.py

@@ -3,7 +3,7 @@
 import json
 from collections.abc import Generator, Iterable, Mapping, Sequence
 from dataclasses import dataclass, field
-from typing import Any, Literal, overload, override
+from typing import TYPE_CHECKING, Any, Literal, overload, override
 
 from pydantic import StrictStr, TypeAdapter, ValidationError
 
@@ -682,3 +682,11 @@ def _parse_optional_llm_usage(payload: object) -> LLMUsage | None:
             raise TypeError(msg)
         normalized_payload[key] = value
     return LLMUsage.from_metadata(normalized_payload)
+
+
+if TYPE_CHECKING:
+    # static assertion to ensure SlimLLM implements LLMProtocol.
+    def _assert_slim_llm_protocol(
+        runtime: SlimLLM,
+    ) -> LLMProtocol:  # pyright: ignore[reportUnusedFunction]
+        return runtime

src/graphon/dsl/tool_runtime.py

@@ -4,7 +4,7 @@
 from collections.abc import Callable, Generator, Iterable, Mapping, Sequence
 from dataclasses import dataclass, field
 from enum import StrEnum
-from typing import Any, cast
+from typing import TYPE_CHECKING, Any, cast
 
 from graphon.file.models import File
 from graphon.model_runtime.entities.llm_entities import LLMUsage
@@ -896,3 +896,12 @@ def _decode_blob(value: object) -> bytes:
             raise ToolNodeError(msg) from error
     msg = "Slim blob payload must be bytes or base64 text."
     raise ToolNodeError(msg)
+
+
+if TYPE_CHECKING:
+    # static assertion to ensure SlimToolNodeRuntime implements
+    # ToolNodeRuntimeProtocol.
+    def _assert_slim_tool_node_runtime_protocol(
+        runtime: SlimToolNodeRuntime,
+    ) -> ToolNodeRuntimeProtocol:  # pyright: ignore[reportUnusedFunction]
+        return runtime

src/graphon/file/protocols.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import abc
 from collections.abc import Generator
 from typing import TYPE_CHECKING, Literal, Protocol
 
@@ -18,26 +19,32 @@ class WorkflowFileRuntimeProtocol(Protocol):
     """
 
     @property
+    @abc.abstractmethod
     def multimodal_send_format(self) -> str: ...
 
+    @abc.abstractmethod
     def http_get(
         self,
         url: str,
         *,
         follow_redirects: bool = True,
     ) -> HttpResponseProtocol: ...
 
+    @abc.abstractmethod
     def storage_load(self, path: str, *, stream: bool = False) -> bytes | Generator: ...
 
+    @abc.abstractmethod
     def load_file_bytes(self, *, file: File) -> bytes: ...
 
+    @abc.abstractmethod
     def resolve_file_url(
         self,
         *,
         file: File,
         for_external: bool = True,
     ) -> str | None: ...
 
+    @abc.abstractmethod
     def resolve_upload_file_url(
         self,
         *,
@@ -46,6 +53,7 @@ def resolve_upload_file_url(
         for_external: bool = True,
     ) -> str: ...
 
+    @abc.abstractmethod
     def resolve_tool_file_url(
         self,
         *,
@@ -54,6 +62,7 @@ def resolve_tool_file_url(
         for_external: bool = True,
     ) -> str: ...
 
+    @abc.abstractmethod
     def verify_preview_signature(
         self,
         *,

src/graphon/graph/graph.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import abc
 import logging
 from collections import defaultdict
 from collections.abc import Mapping, Sequence
@@ -27,6 +28,7 @@ class NodeFactory(Protocol):
     allowing for different node creation strategies while maintaining type safety.
     """
 
+    @abc.abstractmethod
     def create_node(self, node_config: NodeConfigDict) -> Node:
         """Create a Node instance from node configuration data.
 

src/graphon/graph/validation.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import abc
 from collections.abc import Sequence
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Protocol
@@ -34,6 +35,7 @@ def __init__(self, issues: Sequence[GraphValidationIssue]) -> None:
 class GraphValidationRule(Protocol):
     """Protocol that individual validation rules must satisfy."""
 
+    @abc.abstractmethod
     def validate(self, graph: Graph) -> Sequence[GraphValidationIssue]:
         """Validate the provided graph and return any discovered issues."""
         ...

src/graphon/graph_engine/command_channels/in_memory_channel.py

@@ -5,7 +5,7 @@
 """
 
 from queue import Empty, Queue
-from typing import final
+from typing import TYPE_CHECKING, final
 
 from ..entities.commands import GraphEngineCommand
 
@@ -49,3 +49,13 @@ def send_command(self, command: GraphEngineCommand) -> None:
 
         """
         self._queue.put(command)
+
+
+if TYPE_CHECKING:
+    from .protocol import CommandChannel
+
+    # static assertion to ensure InMemoryChannel implements CommandChannel.
+    def _assert_command_channel(
+        channel: InMemoryChannel,
+    ) -> CommandChannel:  # pyright: ignore[reportUnusedFunction]
+        return channel

src/graphon/graph_engine/command_channels/protocol.py

@@ -4,6 +4,7 @@
 to/from a GraphEngine instance, supporting both local and distributed scenarios.
 """
 
+import abc
 from typing import Protocol
 
 from ..entities.commands import GraphEngineCommand
@@ -16,6 +17,7 @@ class CommandChannel(Protocol):
     this channel is dedicated to that single execution.
     """
 
+    @abc.abstractmethod
     def fetch_commands(self) -> list[GraphEngineCommand]:
         """Fetch pending commands for this GraphEngine instance.
 
@@ -27,6 +29,7 @@ def fetch_commands(self) -> list[GraphEngineCommand]:
         """
         ...
 
+    @abc.abstractmethod
     def send_command(self, command: GraphEngineCommand) -> None:
         """Send a command to be processed by this GraphEngine instance.
 

src/graphon/graph_engine/command_channels/redis_channel.py

@@ -5,9 +5,10 @@
 Each instance uses a unique key for its command queue.
 """
 
+import abc
 import json
 from contextlib import AbstractContextManager
-from typing import Any, Protocol, final
+from typing import TYPE_CHECKING, Any, Protocol, final
 
 from ..entities.commands import (
     AbortCommand,
@@ -27,18 +28,26 @@
 class RedisPipelineProtocol(Protocol):
     """Minimal Redis pipeline contract used by the command channel."""
 
+    @abc.abstractmethod
     def lrange(self, name: str, start: int, end: int) -> Any: ...
+    @abc.abstractmethod
     def delete(self, *names: str) -> Any: ...
+    @abc.abstractmethod
     def execute(self) -> list[Any]: ...
+    @abc.abstractmethod
     def rpush(self, name: str, *values: str) -> Any: ...
+    @abc.abstractmethod
     def expire(self, name: str, time: int) -> Any: ...
+    @abc.abstractmethod
     def set(self, name: str, value: str, ex: int | None = None) -> Any: ...
+    @abc.abstractmethod
     def get(self, name: str) -> Any: ...
 
 
 class RedisClientProtocol(Protocol):
     """Redis client contract required by the command channel."""
 
+    @abc.abstractmethod
     def pipeline(self) -> AbstractContextManager[RedisPipelineProtocol]: ...
 
 
@@ -160,3 +169,13 @@ def _has_pending_commands(self) -> bool:
             pending_value, _ = pipe.execute()
 
         return pending_value is not None
+
+
+if TYPE_CHECKING:
+    from .protocol import CommandChannel
+
+    # static assertion to ensure RedisChannel implements CommandChannel.
+    def _assert_command_channel(
+        channel: RedisChannel,
+    ) -> CommandChannel:  # pyright: ignore[reportUnusedFunction]
+        return channel

src/graphon/graph_engine/command_processing/command_handlers.py

@@ -1,5 +1,5 @@
 import logging
-from typing import final, override
+from typing import TYPE_CHECKING, final, override
 
 from graphon.entities.pause_reason import SchedulingPause
 from graphon.runtime.graph_runtime_state import GraphExecutionProtocol
@@ -69,3 +69,21 @@ def handle(
                     execution.workflow_id,
                     exc,
                 )
+
+
+if TYPE_CHECKING:
+    # static assertions to ensure command handlers implement CommandHandler.
+    def _assert_abort_handler_protocol(
+        handler: AbortCommandHandler,
+    ) -> CommandHandler[AbortCommand]:  # pyright: ignore[reportUnusedFunction]
+        return handler
+
+    def _assert_pause_handler_protocol(
+        handler: PauseCommandHandler,
+    ) -> CommandHandler[PauseCommand]:  # pyright: ignore[reportUnusedFunction]
+        return handler
+
+    def _assert_update_variables_handler_protocol(
+        handler: UpdateVariablesCommandHandler,
+    ) -> CommandHandler[UpdateVariablesCommand]:  # pyright: ignore[reportUnusedFunction]
+        return handler

src/graphon/graph_engine/command_processing/command_processor.py

@@ -1,5 +1,6 @@
 """Main command processor for handling external commands."""
 
+import abc
 import logging
 from collections.abc import Callable
 from typing import Protocol, final
@@ -15,6 +16,7 @@
 class CommandHandler[CommandT: GraphEngineCommand](Protocol):
     """Protocol for command handlers."""
 
+    @abc.abstractmethod
     def handle(
         self,
         command: CommandT,

src/graphon/graph_engine/domain/node_execution.py

@@ -1,6 +1,7 @@
 """NodeExecution entity representing a node's execution state."""
 
 from dataclasses import dataclass
+from typing import TYPE_CHECKING
 
 from graphon.enums import NodeState
 
@@ -40,3 +41,13 @@ def mark_skipped(self) -> None:
     def increment_retry(self) -> None:
         """Increment the retry count for this node."""
         self.retry_count += 1
+
+
+if TYPE_CHECKING:
+    from graphon.runtime.graph_runtime_state import NodeExecutionProtocol
+
+    # static assertion to ensure NodeExecution implements NodeExecutionProtocol.
+    def _assert_node_execution_protocol(
+        execution: NodeExecution,
+    ) -> NodeExecutionProtocol:  # pyright: ignore[reportUnusedFunction]
+        return execution

src/graphon/graph_engine/ready_queue/in_memory.py

@@ -5,7 +5,7 @@
 """
 
 import queue
-from typing import final
+from typing import TYPE_CHECKING, final
 
 from .protocol import ReadyQueue, ReadyQueueState
 
@@ -137,3 +137,9 @@ def loads(self, data: str) -> None:
         # Restore items
         for item in state.items:
             self._queue.put(item)
+
+
+if TYPE_CHECKING:
+    # static assertion to ensure InMemoryReadyQueue implements ReadyQueue.
+    def _assert_ready_queue(queue_impl: InMemoryReadyQueue) -> ReadyQueue:  # pyright: ignore[reportUnusedFunction]
+        return queue_impl

src/graphon/graph_engine/ready_queue/protocol.py

@@ -4,6 +4,7 @@
 for execution, supporting both in-memory and persistent storage scenarios.
 """
 
+import abc
 from collections.abc import Sequence
 from typing import Protocol
 
@@ -35,6 +36,7 @@ class ReadyQueue(Protocol):
     that can be serialized for state storage.
     """
 
+    @abc.abstractmethod
     def put(self, item: str) -> None:
         """Add a node ID to the ready queue.
 
@@ -44,6 +46,7 @@ def put(self, item: str) -> None:
         """
         ...
 
+    @abc.abstractmethod
     def get(self, timeout: float | None = None) -> str:
         """Retrieve and remove a node ID from the queue.
 
@@ -56,6 +59,7 @@ def get(self, timeout: float | None = None) -> str:
         """
         ...
 
+    @abc.abstractmethod
     def task_done(self) -> None:
         """Indicate that a previously retrieved task is complete.
 
@@ -64,6 +68,7 @@ def task_done(self) -> None:
         """
         ...
 
+    @abc.abstractmethod
     def empty(self) -> bool:
         """Check if the queue is empty.
 
@@ -73,6 +78,7 @@ def empty(self) -> bool:
         """
         ...
 
+    @abc.abstractmethod
     def qsize(self) -> int:
         """Get the approximate size of the queue.
 
@@ -82,6 +88,7 @@ def qsize(self) -> int:
         """
         ...
 
+    @abc.abstractmethod
     def dumps(self) -> str:
         """Serialize the queue state to a JSON string for storage.
 
@@ -92,6 +99,7 @@ def dumps(self) -> str:
         """
         ...
 
+    @abc.abstractmethod
     def loads(self, data: str) -> None:
         """Restore the queue state from a JSON string.