feat: rename and reorgnaize metadata

Author: brian14708

README.md

@@ -22,8 +22,8 @@ pip install git+https://github.com/brian14708/duron.git
 
 Duron defines two kinds of functions:
 
-- `@duron.fn` — deterministic orchestration. It replays from logs, ensuring that control flow only advances when every prior step is known.
-- `@duron.op` — side effects. Wrap anything that touches the outside world (APIs, databases, file I/O). Duron records its return value so it runs once per unique input.
+- `@duron.durable` — deterministic orchestration. It replays from logs, ensuring that control flow only advances when every prior step is known.
+- `@duron.effect` — side effects. Wrap anything that touches the outside world (APIs, databases, file I/O). Duron records its return value so it runs once per unique input.
 
 ```python
 import asyncio
@@ -34,22 +34,22 @@ import duron
 from duron.contrib.storage import FileLogStorage
 
 
-@duron.op
+@duron.effect
 async def work(name: str) -> str:
     print("⚡ Preparing to greet...")
     await asyncio.sleep(2)  # Simulate I/O
     print("⚡ Greeting...")
     return f"Hello, {name}!"
 
 
-@duron.op
+@duron.effect
 async def generate_lucky_number() -> int:
     print("⚡ Generating lucky number...")
     await asyncio.sleep(1)  # Simulate I/O
     return random.randint(1, 100)
 
 
-@duron.fn
+@duron.durable
 async def greeting_flow(ctx: duron.Context, name: str) -> str:
     message, lucky_number = await asyncio.gather(
         ctx.run(work, name), ctx.run(generate_lucky_number)

examples/agent.py

@@ -20,7 +20,7 @@
 from rich.console import Console
 
 import duron
-from duron import Defer, Signal, SignalInterrupt, Stream, StreamWriter
+from duron import Provided, Signal, SignalInterrupt, Stream, StreamWriter
 from duron.codec import Codec
 from duron.contrib.storage import FileLogStorage
 from duron.tracing import Tracer, span
@@ -51,12 +51,12 @@ def decode_json(self, encoded: JSONValue, expected_type: TypeHint[Any]) -> objec
         return cast("object", TypeAdapter(expected_type).validate_python(encoded))
 
 
-@duron.fn(codec=PydanticCodec())
+@duron.durable(codec=PydanticCodec())
 async def agent_fn(
     ctx: duron.Context,
-    input_: Stream[str] = Defer,
-    signal: Signal[None] = Defer,
-    output: StreamWriter[tuple[str, str]] = Defer,
+    input_: Stream[str] = Provided,
+    signal: Signal[None] = Provided,
+    output: StreamWriter[tuple[str, str]] = Provided,
 ) -> None:
     history: list[ChatCompletionMessageParam] = [
         {
@@ -133,7 +133,7 @@ async def agent_fn(
                         break
 
 
-@duron.op
+@duron.effect
 async def call_tool(params: ChatCompletionMessageToolCallUnion) -> tuple[str, str]:  # noqa: RUF029
     if params.type != "function" or not params.function.name:
         return params.id, '{"status": "error", "message": "Invalid tool call"}'
@@ -207,9 +207,7 @@ async def completion(
     ctx: duron.Context,
     messages: list[ChatCompletionMessageParam],
 ) -> ChatCompletion:
-    @duron.op(
-        metadata={"type": "chat.completions.create"},
-    )
+    @duron.effect
     async def _completion(
         messages: list[ChatCompletionMessageParam],
     ) -> ChatCompletion:

examples/hello_world.py

@@ -14,15 +14,15 @@
 logger = logging.getLogger(__name__)
 
 
-@duron.op
+@duron.effect
 async def work(name: str) -> str:
     print("⚡ Preparing to greet...")
     await asyncio.sleep(2)
     print("⚡ Greeting...")
     return f"Hello, {name}!"
 
 
-@duron.op
+@duron.effect
 async def generate_lucky_number() -> int:
     print("⚡ Generating lucky number...")
     await asyncio.sleep(1)
@@ -31,17 +31,17 @@ async def generate_lucky_number() -> int:
     return random.randint(1, 100)
 
 
-@duron.op(checkpoint=True, initial=lambda: 0, reducer=lambda a, _b: a + 10)
-async def count_up(count: int, target: int) -> AsyncGenerator[None, int]:
+@duron.effect(checkpoint=True, initial=lambda: 0, reducer=int.__add__)
+async def count_up(count: int, target: int) -> AsyncGenerator[int, int]:
     print("⚡ Counting...")
     await asyncio.sleep(0.5)
     while count < target:
-        count = yield
+        count = yield 10
         print(f"⚡ Current count: {count}")
         await asyncio.sleep(0.05)
 
 
-@duron.fn
+@duron.durable
 async def greeting_flow(ctx: duron.Context, name: str) -> str:
     message, lucky_number = await asyncio.gather(
         ctx.run(work, name), ctx.run(generate_lucky_number)

pyproject.toml

@@ -67,7 +67,7 @@ extra-standard-library = ["typing_extensions"]
 
 [tool.ruff.lint.flake8-type-checking]
 runtime-evaluated-base-classes = ["typing_extensions.TypedDict"]
-runtime-evaluated-decorators = ["duron.fn", "duron.op"]
+runtime-evaluated-decorators = ["duron.durable", "duron.effect"]
 exempt-modules = ["typing", "typing_extensions"]
 
 [tool.ruff.lint.flake8-tidy-imports]

src/duron/__init__.py

@@ -6,6 +6,6 @@
 from duron._core.stream import Stream as Stream
 from duron._core.stream import StreamClosed as StreamClosed
 from duron._core.stream import StreamWriter as StreamWriter
-from duron._decorator.fn import fn as fn
-from duron._decorator.op import op as op
-from duron.typing import Defer as Defer
+from duron._decorator.durable import durable as durable
+from duron._decorator.effect import effect as effect
+from duron.typing import Provided as Provided

src/duron/_core/context.py

@@ -4,7 +4,6 @@
 import contextvars
 from contextlib import contextmanager
 from contextvars import ContextVar
-from dataclasses import dataclass
 from random import Random
 from typing import TYPE_CHECKING, cast
 from typing_extensions import (
@@ -16,11 +15,16 @@
     overload,
 )
 
-from duron._core.ops import Barrier, ExternalPromiseCreate, FnCall, create_op
+from duron._core.ops import (
+    Barrier,
+    ExternalPromiseCreate,
+    FnCall,
+    OpAnnotations,
+    create_op,
+)
 from duron._core.signal import create_signal
 from duron._core.stream import create_stream, run_stream
-from duron._decorator.op import CheckpointOp, Op
-from duron._util.linked_dict import LinkedDict
+from duron._decorator.effect import CheckpointFn, EffectFn
 from duron.typing import inspect_function
 
 if TYPE_CHECKING:
@@ -30,7 +34,7 @@
 
     from duron._core.signal import Signal, SignalWriter
     from duron._core.stream import Stream, StreamWriter
-    from duron._decorator.fn import Fn
+    from duron._decorator.durable import DurableFn
     from duron._loop import EventLoop
     from duron.codec import JSONValue
     from duron.typing import TypeHint
@@ -40,16 +44,7 @@
     _P = ParamSpec("_P")
 
 _context: ContextVar[Context | None] = ContextVar("duron.context", default=None)
-
-
-@final
-@dataclass(slots=True)
-class Annotation:
-    metadata: LinkedDict[str, JSONValue]
-    labels: LinkedDict[str, str]
-
-
-_annotation: ContextVar[Annotation | None] = ContextVar(
+_annotation: ContextVar[OpAnnotations | None] = ContextVar(
     "duron.context.annotation", default=None
 )
 
@@ -58,7 +53,7 @@ class Annotation:
 class Context:
     __slots__ = ("_fn", "_loop", "_token")
 
-    def __init__(self, task: Fn[..., object], loop: EventLoop) -> None:
+    def __init__(self, task: DurableFn[..., object], loop: EventLoop) -> None:
         self._loop: EventLoop = loop
         self._fn = task
         self._token: Token[Context | None] | None = None
@@ -89,24 +84,24 @@ def current() -> Context:
     @overload
     async def run(
         self,
-        fn: Callable[_P, Coroutine[Any, Any, _T]] | Op[_P, _T],
+        fn: Callable[_P, Coroutine[Any, Any, _T]] | EffectFn[_P, _T],
         /,
         *args: _P.args,
         **kwargs: _P.kwargs,
     ) -> _T: ...
     @overload
     async def run(
         self,
-        fn: Callable[_P, _T] | CheckpointOp[_P, _T, Any],
+        fn: Callable[_P, _T] | CheckpointFn[_P, _T, Any],
         /,
         *args: _P.args,
         **kwargs: _P.kwargs,
     ) -> _T: ...
     async def run(
         self,
         fn: Callable[_P, Coroutine[Any, Any, _T] | _T]
-        | Op[_P, _T]
-        | CheckpointOp[_P, _T, Any],
+        | EffectFn[_P, _T]
+        | CheckpointFn[_P, _T, Any],
         /,
         *args: _P.args,
         **kwargs: _P.kwargs,
@@ -115,39 +110,38 @@ async def run(
             msg = "Context time can only be used in the context loop"
             raise RuntimeError(msg)
 
-        if isinstance(fn, CheckpointOp):
+        if isinstance(fn, CheckpointFn):
             async with self.run_stream(
-                cast("CheckpointOp[_P, _T, Any]", fn), *args, **kwargs
+                cast("CheckpointFn[_P, _T, Any]", fn), *args, **kwargs
             ) as stream:
                 await stream.discard()
                 return await stream
 
-        if isinstance(fn, Op):
+        if isinstance(fn, EffectFn):
             return_type = fn.return_type
-            metadata = fn.metadata
         else:
             return_type = inspect_function(fn).return_type
-            metadata = None
 
-        callable_ = fn.fn if isinstance(fn, Op) else fn
+        callable_ = fn.fn if isinstance(fn, EffectFn) else fn
         op = create_op(
             self._loop,
             FnCall(
                 callable=callable_,
-                name=cast("str", getattr(callable_, "__name__", repr(callable_))),
                 args=args,
                 kwargs=kwargs,
                 return_type=return_type,
                 context=contextvars.copy_context(),
-                metadata=self._get_metadata(metadata),
-                labels=self._get_labels(None),
+                annotations=OpAnnotations.extend(
+                    _annotation.get(),
+                    name=cast("str", getattr(callable_, "__name__", repr(callable_))),
+                ),
             ),
         )
         return cast("_T", await op)
 
     def run_stream(
         self,
-        fn: CheckpointOp[_P, _T, _S],
+        fn: CheckpointFn[_P, _T, _S],
         /,
         *args: _P.args,
         **kwargs: _P.kwargs,
@@ -180,8 +174,11 @@ async def create_stream(
             self._loop,
             dtype,
             external=external,
-            metadata=self._get_metadata(None),
-            labels=self._get_labels({"name": name} if name else None),
+            annotations=OpAnnotations.extend(
+                _annotation.get(),
+                name=name,
+                labels={"name": name} if name else None,
+            ),
         )
 
     async def create_signal(
@@ -193,23 +190,27 @@ async def create_signal(
         return await create_signal(
             self._loop,
             dtype,
-            metadata=self._get_metadata(None),
-            labels=self._get_labels({"name": name} if name else None),
+            annotations=OpAnnotations.extend(
+                _annotation.get(),
+                labels={"name": name} if name else None,
+            ),
         )
 
     async def create_promise(
-        self,
-        dtype: type[_T],
+        self, dtype: type[_T], /, *, name: str | None = None
     ) -> tuple[str, asyncio.Future[_T]]:
         if asyncio.get_running_loop() is not self._loop:
             msg = "Context time can only be used in the context loop"
             raise RuntimeError(msg)
         fut = create_op(
             self._loop,
             ExternalPromiseCreate(
-                metadata=self._get_metadata(None),
                 return_type=dtype,
-                labels=self._get_labels(None),
+                annotations=OpAnnotations.extend(
+                    _annotation.get(),
+                    name=name,
+                    labels={"name": name} if name else None,
+                ),
             ),
         )
         return (
@@ -257,40 +258,9 @@ def annotate(
 
         current = _annotation.get()
         token = _annotation.set(
-            Annotation(
-                metadata=current.metadata.extend(metadata)
-                if current
-                else LinkedDict(metadata),
-                labels=current.labels.extend(labels) if current else LinkedDict(labels),
-            )
+            OpAnnotations.extend(current, metadata=metadata, labels=labels)
         )
         try:
             yield
         finally:
             _annotation.reset(token)
-
-    @staticmethod
-    def _get_metadata(
-        merge: dict[str, JSONValue] | None,
-    ) -> dict[str, JSONValue] | None:
-        anno = _annotation.get()
-        current = anno.metadata if anno else None
-        if current is None:
-            return merge
-        if merge:
-            return current.extend(merge).materialize()
-        return current.materialize()
-
-    @staticmethod
-    def _get_labels(
-        merge: dict[str, str] | None,
-    ) -> dict[str, str] | None:
-        anno = _annotation.get()
-        current = anno.labels if anno else None
-        if merge:
-            if current is None:
-                return merge
-            return current.extend(merge).materialize()
-        if current is None:
-            return None
-        return current.materialize()