diff --git a/.changeset/slimy-moons-share.md b/.changeset/slimy-moons-share.md
new file mode 100644
index 0000000..1828047
--- /dev/null
+++ b/.changeset/slimy-moons-share.md
@@ -0,0 +1,5 @@
+---
+"@effect-rx/rx": patch
+---
+
+api cleanup
diff --git a/docs/rx/Result.ts.md b/docs/rx/Result.ts.md
index 51ece9a..51cdb22 100644
--- a/docs/rx/Result.ts.md
+++ b/docs/rx/Result.ts.md
@@ -147,7 +147,7 @@ Added in v1.0.0
 **Signature**
 
 ```ts
-export declare const waiting: <E, A>(previous: Option.Option<Success<E, A> | Failure<E, A>>) => Waiting<E, A>
+export declare const waiting: <E, A>(previous: Initial<E, A> | Success<E, A> | Failure<E, A>) => Waiting<E, A>
 ```
 
 Added in v1.0.0
@@ -248,7 +248,7 @@ Added in v1.0.0
 ```ts
 export interface Waiting<E, A> extends Result.Variance<E, A> {
   readonly _tag: 'Waiting'
-  readonly previous: Option.Option<Success<E, A> | Failure<E, A>>
+  readonly previous: Initial<E, A> | Success<E, A> | Failure<E, A>
 }
 ```
 
diff --git a/docs/rx/Rx.ts.md b/docs/rx/Rx.ts.md
index 2a6a082..8a197b9 100644
--- a/docs/rx/Rx.ts.md
+++ b/docs/rx/Rx.ts.md
@@ -37,10 +37,12 @@ Added in v1.0.0
     - [Get (type alias)](#get-type-alias)
     - [GetResult (type alias)](#getresult-type-alias)
     - [Mount (type alias)](#mount-type-alias)
+    - [Read (type alias)](#read-type-alias)
     - [Refresh (type alias)](#refresh-type-alias)
     - [Set (type alias)](#set-type-alias)
     - [Subscribe (type alias)](#subscribe-type-alias)
     - [SubscribeGetter (type alias)](#subscribegetter-type-alias)
+    - [Write (type alias)](#write-type-alias)
   - [RxResultFn (interface)](#rxresultfn-interface)
   - [RxRuntime (interface)](#rxruntime-interface)
   - [Writable (interface)](#writable-interface)
@@ -95,9 +97,9 @@ Added in v1.0.0
 
 ```ts
 export declare const effect: {
-  <E, A>(create: (get: Rx.Get, ctx: Context) => Effect.Effect<never, E, A>): Rx<Result.Result<E, A>>
+  <E, A>(create: Rx.Read<Effect.Effect<never, E, A>>): Rx<Result.Result<E, A>>
   <RR, R extends RR, E, A, RE>(
-    create: (get: Rx.Get, ctx: Context) => Effect.Effect<R, E, A>,
+    create: Rx.Read<Effect.Effect<R, E, A>>,
     options: { readonly runtime: RxRuntime<RE, RR> }
   ): Rx<Result.Result<E | RE, A>>
 }
@@ -136,10 +138,7 @@ Added in v1.0.0
 **Signature**
 
 ```ts
-export declare const readable: <A>(
-  read: (get: Rx.Get, ctx: Context) => A,
-  refresh?: (f: <A>(rx: Rx<A>) => void) => void
-) => Rx<A>
+export declare const readable: <A>(read: Rx.Read<A>, refresh?: (f: <A>(rx: Rx<A>) => void) => void) => Rx<A>
 ```
 
 Added in v1.0.0
@@ -163,9 +162,9 @@ Added in v1.0.0
 
 ```ts
 export declare const scoped: {
-  <E, A>(create: (get: Rx.Get, ctx: Context) => Effect.Effect<Scope.Scope, E, A>): Rx<Result.Result<E, A>>
+  <E, A>(create: Rx.Read<Effect.Effect<Scope.Scope, E, A>>): Rx<Result.Result<E, A>>
   <RR, R extends RR | Scope.Scope, E, A, RE>(
-    create: (get: Rx.Get, ctx: Context) => Effect.Effect<R, E, A>,
+    create: Rx.Read<Effect.Effect<R, E, A>>,
     options: { readonly runtime: RxRuntime<RE, RR> }
   ): Rx<Result.Result<E | RE, A>>
 }
@@ -205,12 +204,10 @@ Added in v1.0.0
 
 ```ts
 export declare const stream: {
-  <E, A>(create: (get: Rx.Get, ctx: Context) => Stream.Stream<never, E, A>): Rx<
-    Result.Result<E | NoSuchElementException, A>
-  >
+  <E, A>(create: Rx.Read<Stream.Stream<never, E, A>>): Rx<Result.Result<E | NoSuchElementException, A>>
   <RR, R extends RR, E, A, RE>(
-    create: (get: Rx.Get, ctx: Context) => Stream.Stream<R, E, A>,
-    runtime: RxRuntime<RE, RR>
+    create: Rx.Read<Stream.Stream<R, E, A>>,
+    options: { readonly runtime: RxRuntime<RE, RR> }
   ): Rx<Result.Result<E | RE | NoSuchElementException, A>>
 }
 ```
@@ -223,12 +220,12 @@ Added in v1.0.0
 
 ```ts
 export declare const streamPull: {
-  <E, A>(
-    create: (get: Rx.Get, ctx: Context) => Stream.Stream<never, E, A>,
-    options?: { readonly disableAccumulation?: boolean }
-  ): Writable<Result.Result<NoSuchElementException | E, A[]>, void>
+  <E, A>(create: Rx.Read<Stream.Stream<never, E, A>>, options?: { readonly disableAccumulation?: boolean }): Writable<
+    Result.Result<NoSuchElementException | E, A[]>,
+    void
+  >
   <RR, R extends RR, E, A, RE>(
-    create: (get: Rx.Get, ctx: Context) => Stream.Stream<R, E, A>,
+    create: Rx.Read<Stream.Stream<R, E, A>>,
     options: { readonly runtime: RxRuntime<RE, RR>; readonly disableAccumulation?: boolean | undefined }
   ): Writable<Result.Result<NoSuchElementException | E | RE, A[]>, void>
 }
@@ -242,8 +239,8 @@ Added in v1.0.0
 
 ```ts
 export declare const writable: <R, W>(
-  read: (get: Rx.Get, ctx: Context) => R,
-  write: (get: Rx.Get, set: Rx.Set, setSelf: (_: R) => void, value: W) => void,
+  read: Rx.Read<R>,
+  write: Rx.Write<R, W>,
   refresh?: (f: <A>(rx: Rx<A>) => void) => void
 ) => Writable<R, W>
 ```
@@ -301,7 +298,7 @@ Added in v1.0.0
 export interface Rx<A> extends Pipeable, Inspectable.Inspectable {
   readonly [TypeId]: TypeId
   readonly keepAlive: boolean
-  readonly read: (get: Rx.Get, ctx: Context) => A
+  readonly read: Rx.Read<A>
   readonly refresh: (f: <A>(rx: Rx<A>) => void) => void
   readonly label?: readonly [name: string, stack: string]
 }
@@ -343,6 +340,16 @@ export type Mount = <A>(rx: Rx<A>) => () => void
 
 Added in v1.0.0
 
+### Read (type alias)
+
+**Signature**
+
+```ts
+export type Read<A> = (get: Rx.Get, ctx: Context) => A
+```
+
+Added in v1.0.0
+
 ### Refresh (type alias)
 
 **Signature**
@@ -389,6 +396,16 @@ export type SubscribeGetter = <A>(rx: Rx<A>, f: () => void) => readonly [get: ()
 
 Added in v1.0.0
 
+### Write (type alias)
+
+**Signature**
+
+```ts
+export type Write<R, W> = (get: Rx.Get, set: Rx.Set, setSelf: (_: R) => void, refreshSelf: () => void, value: W) => void
+```
+
+Added in v1.0.0
+
 ## RxResultFn (interface)
 
 **Signature**
@@ -416,7 +433,7 @@ Added in v1.0.0
 ```ts
 export interface Writable<R, W> extends Rx<R> {
   readonly [WritableTypeId]: WritableTypeId
-  readonly write: (get: Rx.Get, set: Rx.Set, setSelf: (_: R) => void, value: W) => void
+  readonly write: (get: Rx.Get, set: Rx.Set, setSelf: (_: R) => void, refreshSelf: () => void, value: W) => void
 }
 ```
 
diff --git a/packages/rx/src/Result.ts b/packages/rx/src/Result.ts
index ec471c9..925f37d 100644
--- a/packages/rx/src/Result.ts
+++ b/packages/rx/src/Result.ts
@@ -76,17 +76,16 @@ export const fromExit = <E, A>(exit: Exit.Exit<E, A>): Success<E, A> | Failure<E
  */
 export const waitingFrom = <E, A>(previous: Option.Option<Result<E, A>>): Waiting<E, A> => {
   if (previous._tag === "None") {
-    return waiting(Option.none())
+    return waiting(constInitial)
   }
 
   switch (previous.value._tag) {
-    case "Initial":
-      return waiting(Option.none())
     case "Waiting":
       return previous.value
+    case "Initial":
     case "Success":
     case "Failure":
-      return waiting(Option.some(previous.value))
+      return waiting(previous.value)
   }
 }
 
@@ -111,7 +110,7 @@ const constInitial: Initial<never, never> = Object.assign(Object.create(ResultPr
  */
 export interface Waiting<E, A> extends Result.Variance<E, A> {
   readonly _tag: "Waiting"
-  readonly previous: Option.Option<Success<E, A> | Failure<E, A>>
+  readonly previous: Initial<E, A> | Success<E, A> | Failure<E, A>
 }
 
 /**
@@ -124,7 +123,7 @@ export const isWaiting = <E, A>(result: Result<E, A>): result is Waiting<E, A> =
  * @since 1.0.0
  * @category constructors
  */
-export const waiting = <E, A>(previous: Option.Option<Success<E, A> | Failure<E, A>>): Waiting<E, A> => {
+export const waiting = <E, A>(previous: Initial<E, A> | Success<E, A> | Failure<E, A>): Waiting<E, A> => {
   const result = Object.create(ResultProto)
   result._tag = "Waiting"
   result.previous = previous
@@ -200,7 +199,7 @@ export const noWaiting = <E, A>(result: Result<E, A>): NoWaiting<E, A> => {
     case "Failure":
       return result
     case "Waiting":
-      return result.previous._tag === "None" ? constInitial : result.previous.value
+      return result.previous
   }
 }
 
diff --git a/packages/rx/src/Rx.ts b/packages/rx/src/Rx.ts
index e136e74..0e051e6 100644
--- a/packages/rx/src/Rx.ts
+++ b/packages/rx/src/Rx.ts
@@ -35,7 +35,7 @@ export type TypeId = typeof TypeId
 export interface Rx<A> extends Pipeable, Inspectable.Inspectable {
   readonly [TypeId]: TypeId
   readonly keepAlive: boolean
-  readonly read: (get: Rx.Get, ctx: Context) => A
+  readonly read: Rx.Read<A>
   readonly refresh: (f: <A>(rx: Rx<A>) => void) => void
   readonly label?: readonly [name: string, stack: string]
 }
@@ -45,6 +45,27 @@ export interface Rx<A> extends Pipeable, Inspectable.Inspectable {
  * @category models
  */
 export declare namespace Rx {
+  /**
+   * @since 1.0.0
+   * @category models
+   */
+  export type Read<A> = (
+    get: Rx.Get,
+    ctx: Context
+  ) => A
+
+  /**
+   * @since 1.0.0
+   * @category models
+   */
+  export type Write<R, W> = (
+    get: Rx.Get,
+    set: Rx.Set,
+    setSelf: (_: R) => void,
+    refreshSelf: () => void,
+    value: W
+  ) => void
+
   /**
    * @since 1.0.0
    * @category models
@@ -128,7 +149,7 @@ export type WritableTypeId = typeof WritableTypeId
  */
 export interface Writable<R, W> extends Rx<R> {
   readonly [WritableTypeId]: WritableTypeId
-  readonly write: (get: Rx.Get, set: Rx.Set, setSelf: (_: R) => void, value: W) => void
+  readonly write: (get: Rx.Get, set: Rx.Set, setSelf: (_: R) => void, refreshSelf: () => void, value: W) => void
 }
 
 /**
@@ -185,7 +206,7 @@ function defaultRefresh(this: Rx<any>, f: any) {
  * @category constructors
  */
 export const readable = <A>(
-  read: (get: Rx.Get, ctx: Context) => A,
+  read: Rx.Read<A>,
   refresh: (f: <A>(rx: Rx<A>) => void) => void = defaultRefresh
 ): Rx<A> => {
   const rx = Object.create(RxProto)
@@ -200,8 +221,8 @@ export const readable = <A>(
  * @category constructors
  */
 export const writable = <R, W>(
-  read: (get: Rx.Get, ctx: Context) => R,
-  write: (get: Rx.Get, set: Rx.Set, setSelf: (_: R) => void, value: W) => void,
+  read: Rx.Read<R>,
+  write: Rx.Write<R, W>,
   refresh: (f: <A>(rx: Rx<A>) => void) => void = defaultRefresh
 ): Writable<R, W> => {
   const rx = Object.create(WritableProto)
@@ -223,7 +244,7 @@ export const state = <A>(
     function(_ctx) {
       return initialValue
     },
-    function(_get, _set, setSelf, value) {
+    function(_get, _set, setSelf, _refreshSelf, value) {
       setSelf(value)
     }
   )
@@ -246,7 +267,7 @@ function makeEffect<E, A>(
   if (previous._tag === "Some") {
     return Result.waitingFrom(previous)
   }
-  return Result.waiting(Option.none())
+  return Result.waiting(Result.initial())
 }
 
 function makeEffectRuntime<R, E, A, RE>(
@@ -272,13 +293,13 @@ function makeEffectRuntime<R, E, A, RE>(
  * @category constructors
  */
 export const effect: {
-  <E, A>(create: (get: Rx.Get, ctx: Context) => Effect.Effect<never, E, A>): Rx<Result.Result<E, A>>
+  <E, A>(create: Rx.Read<Effect.Effect<never, E, A>>): Rx<Result.Result<E, A>>
   <RR, R extends RR, E, A, RE>(
-    create: (get: Rx.Get, ctx: Context) => Effect.Effect<R, E, A>,
+    create: Rx.Read<Effect.Effect<R, E, A>>,
     options: { readonly runtime: RxRuntime<RE, RR> }
   ): Rx<Result.Result<RE | E, A>>
 } = <R, E, A, RE>(
-  create: (get: Rx.Get, ctx: Context) => Effect.Effect<R, E, A>,
+  create: Rx.Read<Effect.Effect<R, E, A>>,
   options?: { readonly runtime?: RxRuntime<RE, R> }
 ) =>
   readable<Result.Result<E, A>>(function(get, ctx) {
@@ -293,13 +314,13 @@ export const effect: {
  * @category constructors
  */
 export const scoped: {
-  <E, A>(create: (get: Rx.Get, ctx: Context) => Effect.Effect<Scope.Scope, E, A>): Rx<Result.Result<E, A>>
+  <E, A>(create: Rx.Read<Effect.Effect<Scope.Scope, E, A>>): Rx<Result.Result<E, A>>
   <RR, R extends (RR | Scope.Scope), E, A, RE>(
-    create: (get: Rx.Get, ctx: Context) => Effect.Effect<R, E, A>,
+    create: Rx.Read<Effect.Effect<R, E, A>>,
     options: { readonly runtime: RxRuntime<RE, RR> }
   ): Rx<Result.Result<RE | E, A>>
 } = <R, E, A, RE>(
-  create: (get: Rx.Get, ctx: Context) => Effect.Effect<R, E, A>,
+  create: Rx.Read<Effect.Effect<R, E, A>>,
   options?: { readonly runtime?: RxRuntime<RE, R> }
 ) =>
   readable<Result.Result<E, A>>(function(get, ctx) {
@@ -355,7 +376,7 @@ export const effectFn: {
     return options?.runtime
       ? makeEffectRuntime(ctx, effect, options.runtime)
       : makeEffect(ctx, effect as Effect.Effect<never, E, A>)
-  }, function(get, set, _setSelf, arg) {
+  }, function(get, set, _setSelf, _refreshSelf, arg) {
     set(argRx, [get(argRx)[0] + 1, arg])
   })
 }
@@ -397,7 +418,7 @@ export const scopedFn: {
     return options?.runtime
       ? makeEffectRuntime(ctx, scopedEffect, options.runtime)
       : makeEffect(ctx, scopedEffect as Effect.Effect<never, E, A>)
-  }, function(get, set, _setSelf, arg) {
+  }, function(get, set, _setSelf, _refreshSelf, arg) {
     set(argRx, [get(argRx)[0] + 1, arg])
   })
 }
@@ -437,7 +458,7 @@ function makeStream<E, A>(
   const cancel = runCallback(
     Stream.runForEach(
       stream,
-      (a) => Effect.sync(() => ctx.setSelf(Result.waiting(Option.some(Result.success(a)))))
+      (a) => Effect.sync(() => ctx.setSelf(Result.waiting(Result.success(a))))
     ),
     (exit) => {
       if (exit._tag === "Failure") {
@@ -486,20 +507,20 @@ function makeStreamRuntime<R, E, A, RE>(
  */
 export const stream: {
   <E, A>(
-    create: (get: Rx.Get, ctx: Context) => Stream.Stream<never, E, A>
+    create: Rx.Read<Stream.Stream<never, E, A>>
   ): Rx<Result.Result<E | NoSuchElementException, A>>
   <RR, R extends RR, E, A, RE>(
-    create: (get: Rx.Get, ctx: Context) => Stream.Stream<R, E, A>,
-    runtime: RxRuntime<RE, RR>
+    create: Rx.Read<Stream.Stream<R, E, A>>,
+    options: { readonly runtime: RxRuntime<RE, RR> }
   ): Rx<Result.Result<RE | E | NoSuchElementException, A>>
 } = <R, E, A, RE>(
-  create: (get: Rx.Get, ctx: Context) => Stream.Stream<R, E, A>,
-  runtime?: RxRuntime<RE, R>
+  create: Rx.Read<Stream.Stream<R, E, A>>,
+  options?: { readonly runtime?: RxRuntime<RE, R> }
 ) =>
   readable<Result.Result<E | NoSuchElementException, A>>(function(get, ctx) {
     const stream = create(get, ctx)
-    if (runtime !== undefined) {
-      return makeStreamRuntime(ctx, stream, runtime)
+    if (options?.runtime !== undefined) {
+      return makeStreamRuntime(ctx, stream, options.runtime)
     }
     return makeStream(ctx, stream as Stream.Stream<never, E, A>)
   })
@@ -509,18 +530,18 @@ export const stream: {
  * @category constructors
  */
 export const streamPull: {
-  <E, A>(create: (get: Rx.Get, ctx: Context) => Stream.Stream<never, E, A>, options?: {
+  <E, A>(create: Rx.Read<Stream.Stream<never, E, A>>, options?: {
     readonly disableAccumulation?: boolean
   }): Writable<Result.Result<E | NoSuchElementException, Array<A>>, void>
   <RR, R extends RR, E, A, RE>(
-    create: (get: Rx.Get, ctx: Context) => Stream.Stream<R, E, A>,
+    create: Rx.Read<Stream.Stream<R, E, A>>,
     options: {
       readonly runtime: RxRuntime<RE, RR>
       readonly disableAccumulation?: boolean
     }
   ): Writable<Result.Result<RE | E | NoSuchElementException, Array<A>>, void>
 } = <R, E, A>(
-  create: (get: Rx.Get, ctx: Context) => Stream.Stream<R, E, A>,
+  create: Rx.Read<Stream.Stream<R, E, A>>,
   options?: {
     readonly runtime?: RxRuntime<unknown, R>
     readonly disableAccumulation?: boolean
@@ -538,7 +559,6 @@ export const streamPull: {
     )
     return Stream.toPull(accStream)
   }, options?.runtime as any)
-  const counter = state(0)
 
   return writable<Result.Result<E | NoSuchElementException, Array<A>>, void>(function(get, ctx) {
     const previous = ctx.self<Result.Result<E | NoSuchElementException, Array<A>>>()
@@ -549,7 +569,6 @@ export const streamPull: {
       }
       return pullResult as any
     }
-    get(counter)
     const pull = pipe(
       pullResult.value,
       Effect.map((_) => Chunk.toReadonlyArray(_) as Array<A>),
@@ -571,8 +590,8 @@ export const streamPull: {
     return options?.runtime
       ? makeEffectRuntime(ctx, pull, options.runtime)
       : makeEffect(ctx, pull as any)
-  }, function(get, set, _setSelf, _) {
-    set(counter, get(counter) + 1)
+  }, function(_get, _set, _setSelf, refreshSelf, _) {
+    refreshSelf()
   }, function(refresh) {
     refresh(pullRx)
   })
diff --git a/packages/rx/src/internal/registry.ts b/packages/rx/src/internal/registry.ts
index e460aac..17954a7 100644
--- a/packages/rx/src/internal/registry.ts
+++ b/packages/rx/src/internal/registry.ts
@@ -28,19 +28,21 @@ class RegistryImpl implements Registry.Registry {
   }
 
   set = <R, W>(rx: Rx.Writable<R, W>, value: W): void => {
+    const node = this.ensureNode(rx)
     rx.write(
       this.get,
       this.set,
-      this.ensureNode(rx).setValue,
+      node.setValue,
+      node.invalidate,
       value
     )
   }
 
-  refresh = <A>(rx: Rx.Rx<A> & Rx.Refreshable): void => {
-    rx.refresh((rx) => this.ensureNode(rx).invalidate())
+  refresh<A>(rx: Rx.Rx<A> & Rx.Refreshable): void {
+    rx.refresh(this.invalidateRx)
   }
 
-  subscribe: Rx.Rx.Subscribe = (rx, f, options) => {
+  subscribe<A>(rx: Rx.Rx<A>, f: (_: A) => void, options?: { readonly immediate?: boolean }): () => void {
     const node = this.ensureNode(rx)
     if (options?.immediate) {
       f(node.value())
@@ -56,7 +58,7 @@ class RegistryImpl implements Registry.Registry {
     }
   }
 
-  subscribeGetter = <A>(rx: Rx.Rx<A>, f: () => void): readonly [get: () => A, unmount: () => void] => {
+  subscribeGetter<A>(rx: Rx.Rx<A>, f: () => void): readonly [get: () => A, unmount: () => void] {
     const node = this.ensureNode(rx)
     function get() {
       return node.value()
@@ -91,6 +93,10 @@ class RegistryImpl implements Registry.Registry {
     return new Node(this, rx)
   }
 
+  invalidateRx = <A>(rx: Rx.Rx<A>): void => {
+    this.ensureNode(rx).invalidate()
+  }
+
   scheduleRxRemoval(rx: Rx.Rx<any>): void {
     queueMicrotask(() => {
       const node = this.nodes.get(rx)
@@ -226,7 +232,7 @@ class Node<A> {
     }
   }
 
-  invalidate(): void {
+  invalidate = (): void => {
     if (this.state === NodeState.valid) {
       this.state = NodeState.stale
       this.disposeLifetime()
diff --git a/packages/rx/test/Rx.test.ts b/packages/rx/test/Rx.test.ts
index 7df10b3..d1c3190 100644
--- a/packages/rx/test/Rx.test.ts
+++ b/packages/rx/test/Rx.test.ts
@@ -78,7 +78,7 @@ describe("Rx", () => {
       { runtime: counterRuntime }
     )
     const timesTwo = Rx.effect(
-      (get, ctx) =>
+      (_get, ctx) =>
         Effect.gen(function*(_) {
           const counter = yield* _(Counter)
           const multiplier = yield* _(Multiplier)
@@ -159,9 +159,8 @@ describe("Rx", () => {
     await vi.advanceTimersByTimeAsync(50)
     result = r.get(count)
     assert(Result.isWaiting(result))
-    assert(Option.isSome(result.previous))
-    assert(Result.isSuccess(result.previous.value))
-    assert.deepEqual(result.previous.value.value, 0)
+    assert(Result.isSuccess(result.previous))
+    assert.deepEqual(result.previous.value, 0)
 
     await vi.advanceTimersByTimeAsync(50)
     result = r.get(count)