Logs
Logs in hyperlink-ts are one pipeline: every Effect.log on a Node lands on a single live bus, and — when you register journals on your Store — durable followers persist those lines into scoped history. You consume the same lines live (Logs.stream, Hyperlink.logs) or from Storage (Logs.byNode, Logs.byHyperlink).
There is no separate “process log API” and “queue log API.” Capture is central. Scopes are how you carve the bus into journals and Handle-facing exports.
This chapter is the narrative guide. Deep tables and fixture indexes remain in docs/LOGS.md when you need a lookup.
What you get
A finished Node stack looks like this:
Your Node (key: billing/scores)
Store.Service
└── Logs.layer — one capture Logger + Logs.Relay bus
└── BillingNode.logs — match-all durable tail → node journal
└── Daemon.store(Daily)— lineage durable tail → HyperService journal
Daemon / WorkPool layers
└── Logs.withScope(tag) — appends tag.key onto the fiber lineage pathCapture — exactly one merged capture Logger per Node (
Logs.layer, baked intoStore.Service).Bus — one
Logs.Relay: PubSub plus a bounded in-memory snapshot for late subscribers.Durable tails — one Stream follower per store registration: level gate → match → append to that registration’s private
_logsjournal (Effect-style underscore field — not on public handle types).Lineage — a JSON array of segment keys on each line (
Logs.withScope), so filters can select by Hyperlink or by ancestry.Export —
Hyperlink.logs(tag)for{ stream, query }on a Hyperlink Handle surface. Prefer this for app reads; apps may freely declare their own Store shape namedlog.
Two copies are intentional. If both Node.logs and Daemon.store(Daily) are registered, the same published line can appear in the node journal and the HyperService journal. Dedup is per scope, not global.
Your first live bus
Provide Logs.layer and every log on that fiber context reaches the bus:
import { import LogsLogs } from "hyperlink-ts"
import { import EffectEffect, import FiberFiber, import StreamStream } from "effect"
const const program: Effect.Effect<
string,
never,
LogRelay
>
const program: {
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
toString: () => string;
toJSON: () => unknown;
}
import EffectEffect.const gen: {
<Eff extends Effect<any, any, any>, AEff>(
f: () => Generator<Eff, AEff, never>
): Effect<
AEff,
[Eff] extends [never]
? never
: [Eff] extends [
Effect<infer _A, infer E, infer _R>
]
? E
: never,
[Eff] extends [never]
? never
: [Eff] extends [
Effect<infer _A, infer _E, infer R>
]
? R
: never
>
<Self, Eff extends Effect<any, any, any>, AEff>(
options: { readonly self: Self },
f: (this: Self) => Generator<Eff, AEff, never>
): Effect<
AEff,
[Eff] extends [never]
? never
: [Eff] extends [
Effect<infer _A, infer E, infer _R>
]
? E
: never,
[Eff] extends [never]
? never
: [Eff] extends [
Effect<infer _A, infer _E, infer R>
]
? R
: never
>
}
Provides a way to write effectful code using generator functions, simplifying
control flow and error handling.
When to use
Use when you want to write effectful code that looks and behaves like
synchronous code, while still handling asynchronous tasks, errors, and complex
control flow such as loops and conditions.
Generator functions work similarly to async/await but keep errors,
requirements, and interruption in the Effect type. You can yield* values
from effects and return the final result at the end.
Example (Sequencing effects with generators)
import { Data, Effect } from "effect"
class DiscountRateError extends Data.TaggedError("DiscountRateError")<{}> {}
const addServiceCharge = (amount: number) => amount + 1
const applyDiscount = (
total: number,
discountRate: number
): Effect.Effect<number, DiscountRateError> =>
discountRate === 0
? Effect.fail(new DiscountRateError())
: Effect.succeed(total - (total * discountRate) / 100)
const fetchTransactionAmount = Effect.promise(() => Promise.resolve(100))
const fetchDiscountRate = Effect.promise(() => Promise.resolve(5))
export const program = Effect.gen(function*() {
const transactionAmount = yield* fetchTransactionAmount
const discountRate = yield* fetchDiscountRate
const discountedAmount = yield* applyDiscount(
transactionAmount,
discountRate
)
const finalAmount = addServiceCharge(discountedAmount)
return `Final amount to charge: ${finalAmount}`
})
gen(function* () {
// Subscribe before you log if you need the first lines (live fan-out).
const const collector: Fiber.Fiber<
Array<LogEntry>,
never
>
const collector: {
id: number;
currentOpCount: number;
getRef: <A>(ref: Context.Reference<A>) => A;
context: Context.Context<never>;
setContext: (context: Context.Context<never>) => void;
currentScheduler: Scheduler;
currentDispatcher: SchedulerDispatcher;
currentSpan: AnySpan | undefined;
currentLogLevel: LogLevel;
minimumLogLevel: LogLevel;
currentStackFrame: StackFrame | undefined;
maxOpsBeforeYield: number;
currentPreventYield: boolean;
addObserver: (cb: (exit: Exit<A, E>) => void) => () => void;
interruptUnsafe: (fiberId?: number | undefined, annotations?: Context.Context<never> | undefined) => void;
pollUnsafe: () => Exit<A, E> | undefined;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
import EffectEffect.const forkChild: <
Arg extends
| Effect<any, any, any>
| {
readonly startImmediately?:
| boolean
| undefined
readonly uninterruptible?:
| boolean
| "inherit"
| undefined
}
| undefined = {
readonly startImmediately?:
| boolean
| undefined
readonly uninterruptible?:
| boolean
| "inherit"
| undefined
}
>(
effectOrOptions?: Arg,
options?:
| {
readonly startImmediately?:
| boolean
| undefined
readonly uninterruptible?:
| boolean
| "inherit"
| undefined
}
| undefined
) => [Arg] extends [
Effect<infer _A, infer _E, infer _R>
]
? Effect<Fiber<_A, _E>, never, _R>
: <A, E, R>(
self: Effect<A, E, R>
) => Effect<Fiber<A, E>, never, R>
Returns an effect that forks this effect into its own separate fiber,
returning the fiber immediately, without waiting for it to begin executing
the effect.
Details
You can use the forkChild method whenever you want to execute an effect in a
new fiber, concurrently and without "blocking" the fiber executing other
effects. Using fibers can be tricky, so instead of using this method
directly, consider other higher-level methods, such as raceWith,
zipPar, and so forth.
The fiber returned by this method has methods to interrupt the fiber and to
wait for it to finish executing the effect. See Fiber for more
information.
Whenever you use this method to launch a new fiber, the new fiber is
attached to the parent fiber's scope. This means when the parent fiber
terminates, the child fiber will be terminated as well, ensuring that no
fibers leak. This behavior is called "auto supervision", and if this
behavior is not desired, you may use the forkDetach or forkIn methods.
Example (Forking a child fiber)
import { Effect, Fiber } from "effect"
const longRunningTask = Effect.gen(function*() {
yield* Effect.sleep("2 seconds")
yield* Effect.log("Task completed")
return "result"
})
const program = Effect.gen(function*() {
const fiber = yield* longRunningTask.pipe(Effect.forkChild)
// or fork a fiber that starts immediately:
yield* longRunningTask.pipe(Effect.forkChild({ startImmediately: true }))
yield* Effect.log("Task forked, continuing...")
const result = yield* Fiber.join(fiber)
return result
})
forkChild(
import StreamStream.const runCollect: <A, E, R>(
self: Stream<A, E, R>
) => Effect.Effect<Array<A>, E, R>
Runs the stream and collects all elements into an array.
Example (Collecting stream values)
import { Console, Effect, Stream } from "effect"
const stream = Stream.make(1, 2, 3, 4, 5)
const program = Effect.gen(function*() {
const collected = yield* Stream.runCollect(stream)
yield* Console.log(collected)
})
Effect.runPromise(program)
// [1, 2, 3, 4, 5]
runCollect(import StreamStream.const take: {
(n: number): <A, E, R>(
self: Stream<A, E, R>
) => Stream<A, E, R>
<A, E, R>(
self: Stream<A, E, R>,
n: number
): Stream<A, E, R>
}
Takes the first n elements from this stream, returning Stream.empty when n < 1.
Example (Taking values from the left)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
const values = yield* Stream.make(1, 2, 3, 4, 5).pipe(
Stream.take(3),
Stream.runCollect
)
yield* Console.log(values)
})
Effect.runPromise(program)
// Output: [ 1, 2, 3 ]
take(import LogsLogs.const stream: Stream.Stream<
LogEntry,
never,
LogRelay
>
const stream: {
channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
Unfiltered live bus (snapshot prefix + PubSub).
stream, 1)),
)
yield* import EffectEffect.const logInfo: (
...message: ReadonlyArray<any>
) => Effect<void>
Logs one or more messages at the INFO level.
Example (Logging information)
import { Effect } from "effect"
const program = Effect.gen(function*() {
yield* Effect.logInfo("Application starting up")
yield* Effect.logInfo("Config loaded:", "production", "Port:", 3000)
// Useful for general information
const version = "1.2.3"
yield* Effect.logInfo("Application version:", version)
})
Effect.runPromise(program)
// Output:
// timestamp=2023-... level=INFO message="Application starting up"
// timestamp=2023-... level=INFO message="Config loaded: production Port: 3000"
// timestamp=2023-... level=INFO message="Application version: 1.2.3"
logInfo("hello from the bus")
const [const entry: LogEntryconst entry: {
date: string;
level: LogLevel;
message: string;
cause: string;
annotations: Readonly<Record<string, string>>;
spans: ReadonlyArray<string>;
}
var Array: ArrayConstructorArray.ArrayConstructor.from<LogEntry>(iterable: Iterable<LogEntry> | ArrayLike<LogEntry>): LogEntry[] (+3 overloads)Creates an array from an iterable object.
from(yield* import FiberFiber.const join: <A, E>(
self: Fiber<A, E>
) => Effect<A, E>
Joins a fiber, blocking until it completes. If the fiber succeeds,
returns its value. If it fails, the error is propagated.
When to use
Use when you need a forked fiber's failure to fail the current Effect because
that fiber is part of the current workflow.
Gotchas
Joining a failed fiber propagates the fiber's Cause. Use
await
when
you need to inspect the Exit instead of failing.
Example (Joining a fiber)
import { Effect, Fiber } from "effect"
const program = Effect.gen(function*() {
const fiber = yield* Effect.forkChild(Effect.succeed(42))
const result = yield* Fiber.join(fiber)
console.log(result) // 42
})
join(const collector: Fiber.Fiber<
Array<LogEntry>,
never
>
const collector: {
id: number;
currentOpCount: number;
getRef: <A>(ref: Context.Reference<A>) => A;
context: Context.Context<never>;
setContext: (context: Context.Context<never>) => void;
currentScheduler: Scheduler;
currentDispatcher: SchedulerDispatcher;
currentSpan: AnySpan | undefined;
currentLogLevel: LogLevel;
minimumLogLevel: LogLevel;
currentStackFrame: StackFrame | undefined;
maxOpsBeforeYield: number;
currentPreventYield: boolean;
addObserver: (cb: (exit: Exit<A, E>) => void) => () => void;
interruptUnsafe: (fiberId?: number | undefined, annotations?: Context.Context<never> | undefined) => void;
pollUnsafe: () => Exit<A, E> | undefined;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
const entry: LogEntryconst entry: {
date: string;
level: LogLevel;
message: string;
cause: string;
annotations: Readonly<Record<string, string>>;
spans: ReadonlyArray<string>;
}
LogEntry.message: stringmessage
})
// Provide Logs.layer at the Node root (or rely on Store.Service, which bakes it in).
const const runnable: Effect.Effect<
string,
never,
never
>
const runnable: {
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
toString: () => string;
toJSON: () => unknown;
}
const program: Effect.Effect<
string,
never,
LogRelay
>
const program: {
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
toString: () => string;
toJSON: () => unknown;
}
Pipeable.pipe<Effect.Effect<string, never, LogRelay>, Effect.Effect<string, never, never>, Effect.Effect<string, never, never>>(this: Effect.Effect<string, never, LogRelay>, ab: (_: Effect.Effect<string, never, LogRelay>) => Effect.Effect<string, never, never>, bc: (_: Effect.Effect<string, never, never>) => Effect.Effect<string, never, never>): Effect.Effect<string, never, never> (+21 overloads)pipe(import EffectEffect.const provide: {
<
Layers extends [
Layer.Any,
...Array<Layer.Any>
]
>(
layers: Layers,
options?:
| { readonly local?: boolean | undefined }
| undefined
): <A, E, R>(
self: Effect<A, E, R>
) => Effect<
A,
E | Layer.Error<Layers[number]>,
| Layer.Services<Layers[number]>
| Exclude<R, Layer.Success<Layers[number]>>
>
<ROut, E2, RIn>(
layer: Layer.Layer<ROut, E2, RIn>,
options?:
| { readonly local?: boolean | undefined }
| undefined
): <A, E, R>(
self: Effect<A, E, R>
) => Effect<A, E | E2, RIn | Exclude<R, ROut>>
<R2>(context: Context.Context<R2>): <A, E, R>(
self: Effect<A, E, R>
) => Effect<A, E, Exclude<R, R2>>
<
A,
E,
R,
Layers extends [
Layer.Any,
...Array<Layer.Any>
]
>(
self: Effect<A, E, R>,
layers: Layers,
options?:
| { readonly local?: boolean | undefined }
| undefined
): Effect<
A,
E | Layer.Error<Layers[number]>,
| Layer.Services<Layers[number]>
| Exclude<R, Layer.Success<Layers[number]>>
>
<A, E, R, ROut, E2, RIn>(
self: Effect<A, E, R>,
layer: Layer.Layer<ROut, E2, RIn>,
options?:
| { readonly local?: boolean | undefined }
| undefined
): Effect<A, E | E2, RIn | Exclude<R, ROut>>
<A, E, R, R2>(
self: Effect<A, E, R>,
context: Context.Context<R2>
): Effect<A, E, Exclude<R, R2>>
}
Provides dependencies to an effect using layers or a context. Use options.local
to build the layer every time; by default, layers are shared between provide
calls.
Example (Providing dependencies with a layer)
import { Context, Effect, Layer } from "effect"
interface Database {
readonly query: (sql: string) => Effect.Effect<string>
}
const Database = Context.Service<Database>("Database")
const DatabaseLive = Layer.succeed(Database)({
query: Effect.fn("Database.query")((sql: string) => Effect.succeed(`Result for: ${sql}`))
})
const program = Effect.gen(function*() {
const db = yield* Database
return yield* db.query("SELECT * FROM users")
})
const provided = Effect.provide(program, DatabaseLive)
Effect.runPromise(provided).then(console.log)
// Output: "Result for: SELECT * FROM users"
provide(import LogsLogs.const layer: Layer<LogRelay, never, never>const layer: {
build: (memoMap: MemoMap, scope: Scope.Scope) => Effect<Context.Context<LogRelay>, never, never>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
Node root: relay + one merged capture logger.
layer), import EffectEffect.const scoped: <A, E, R>(
self: Effect<A, E, R>
) => Effect<A, E, Exclude<R, Scope>>
Runs an effect with a scope that closes when the effect completes.
When to use
Use to acquire scoped resources for the duration of a single workflow.
Details
Finalizers for resources acquired inside the workflow run as soon as the
workflow completes, whether by success, failure, or interruption.
Example (Running a scoped acquisition)
import { Console, Effect } from "effect"
const resource = Effect.acquireRelease(
Console.log("Acquiring resource").pipe(Effect.as("resource")),
() => Console.log("Releasing resource")
)
const program = Effect.scoped(
Effect.gen(function*() {
const res = yield* resource
yield* Console.log(`Using ${res}`)
return res
})
)
Effect.runFork(program)
// Output: "Acquiring resource"
// Output: "Using resource"
// Output: "Releasing resource"
scoped)Logs.snapshot is the bounded tail already held on the relay — useful for “what just happened” without opening a Stream. Logs.replay re-emits a captured LogEntry through the ambient Logger.
Durable journals
Live-only is enough for ephemeral UIs. History needs Storage.
Register journals on a Store.Service. Node-wide history uses Node.logs (or Hyperlink.store(Node)). Per-Hyperlink history uses the toolkit store registration — Daemon.store(tag), WorkPool.store(tag), and friends — which carry a private _logs journal. Read durable history with Logs.byNode / Logs.byHyperlink / Hyperlink.logs(tag).query (not a public handle.log surface).
import { import LogsLogs, import DaemonDaemon, import HyperlinkHyperlink, import StoreStore } from "hyperlink-ts"
import * as import NodeNode from "hyperlink-ts/Node"
import { import EffectEffect } from "effect"
class class BillingNodeclass BillingNode {
key: Identifier;
Service: {
protocol: Context.Service.Shape<typeof RpcClient.Protocol>;
ping: Effect.Effect<number, NodeUnreachable>;
status: { readonly get: Effect.Effect<NodeStatusSnapshot, NodeUnreachable>; readonly changes: Stream.Stream<NodeStatusSnapshot, NodeUnreachable> };
logs: { readonly stream: Stream.Stream<LogEntry, NodeUnreachable>; readonly query: (options: { readonly limit: number }) => Effect.Effect<ReadonlyArray<LogEntry>, NodeUnreachable> };
};
}
import NodeNode.Tag<Self, ROut = never>(): { (key: string): NodeTagClass<Self, ROut, BareAddress>; (key: string, target: { readonly path: string; readonly kind?: 'IpcSocket'; readonly onConflict?: OnConflict }): NodeTagClass<Self, ROut, IpcAddress>; (key: string, target: number | `:${number}`): NodeTagClass<Self, ROut, HttpAddress>; (key: string, target: `ws://${string}` | `wss://${string}`): NodeTagClass<Self, ROut, WsAddress>; (key: string, target: `http://${string}` | `https://${string}`): NodeTagClass<Self, ROut, HttpAddress>; (key: string, target: { readonly url: `ws://${string}` | `wss://${string}`; readonly kind?: 'WebSocket'; readonly onConflict?: OnConflict }): NodeTagClass<Self, ROut, WsAddress>; (key: string, target: { readonly url: string; readonly kind: 'WebSocket'; readonly onConflict?: OnConflict }): NodeTagClass<Self, ROut, WsAddress>; (key: string, target: { readonly url: string; readonly kind: 'Http'; readonly onConflict?: OnConflict }): NodeTagClass<Self, ROut, HttpAddress>; <T extends ShorthandTarget>(key: string, target: T): NodeTagClass<Self, ROut, MultiAddress<KindsOf<T>>>; (key: string, target: string | { readonly url: string; readonly kind?: ProtocolKind; readonly onConflict?: OnConflict }): NodeTagClass<Self, ROut, UrlAddressLoose>; (key: string, target?: LooseNodeTarget): NodeTagClass<Self, ROut, BareAddress | IpcAddress | HttpAddress | WsAddress | UrlAddressLoose | MultiAddress<ProtocolKind>> }Declare a node — a named transport endpoint a HyperService connects to. Two-stage and keyed by
a string, mirroring Effect's Context.Service<Self, Shape>()(key) (a node is a Context.Key,
resolved by its key in the Context map) and every sibling factory (Hyperlink.Tag<Self>(), …).
The second call infers the target shape, so the { http, ws } shorthand types its
ProtocolKind
set precisely. Optional catalog type param ROut (C2) — prefer import type
for those handles (C4). Templates (no address until cloned) live on
Node
.Prototype:
class EdgeNode extends Node.Tag<EdgeNode>()("edge") {} // no address yet
class Worker extends Node.Tag<Worker>()("worker", 3001) {} // → http://localhost:3001/rpc, kind "Http"
class Mail extends Node.Tag<Mail>()("mail", "https://mail.internal/rpc") {} // full url, as-is, kind "Http"
class Live extends Node.Tag<Live>()("live", { url: "wss://live/rpc" }) {} // kind "WebSocket" (inferred from ws url)
class Push extends Node.Tag<Push>()("push", { url: "/rpc", kind: "WebSocket" }) {} // same-origin path, explicit kind
class Local extends Node.Tag<Local>()("local", { path: "/tmp/local.sock" }) {} // kind "IpcSocket" (Unix domain)
class Droplet extends Node.Tag<Droplet>()("droplet", { http: "http://d/rpc", ws: "ws://d/rpc" }) {} // multi-protocol
import type { Jobs, Emails } from "@app/contracts"
class AppWorker extends Node.Tag<AppWorker, Jobs | Emails>()("app/Worker", { path: "/tmp/w.sock" }) {}
class MailWorker extends Node.Prototype<MailWorker, Mail>("app/MailWorker") {}
The key is the service key. The optional address matches a dial target: a port
(3001 or ":3001" → http://localhost:3001/rpc), a full url (used as-is), { url, kind } for
an explicit endpoint, { path } for a Unix-domain socket (kind: "IpcSocket"), or the
{ http, ws, ipc } multi-protocol shorthand. The node carries
ProtocolKind
so the topology
is self-describing about where AND how:
connect
(node) derives the transport with no
protocol argument.
Dialable targets return an
AddressedNode
(kind: ProtocolKind) so
Hyperlink.client(Tag, Worker) can auto-wire
connect
. Bare Node.Tag()("x")
stays address-less (kind: undefined) — still needs explicit connect / lookup.
Tag<class BillingNodeclass BillingNode {
key: Identifier;
Service: {
protocol: Context.Service.Shape<typeof RpcClient.Protocol>;
ping: Effect.Effect<number, NodeUnreachable>;
status: { readonly get: Effect.Effect<NodeStatusSnapshot, NodeUnreachable>; readonly changes: Stream.Stream<NodeStatusSnapshot, NodeUnreachable> };
logs: { readonly stream: Stream.Stream<LogEntry, NodeUnreachable>; readonly query: (options: { readonly limit: number }) => Effect.Effect<ReadonlyArray<LogEntry>, NodeUnreachable> };
};
}
class Dailyclass Daily {
key: Identifier;
Service: {
status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsStarted: number; readonly runsSucceeded: number; readonly runsFailed: number; readonly nextTriggerRun?: Utc | u…;
start: Effect.Effect<void, never, never>;
stop: Effect.Effect<void, never, never>;
wake: Effect.Effect<void, never, never>;
resetCadence: Effect.Effect<void, never, never>;
events: Stream<Struct.ReadonlySide<{ readonly _tag: tag<'Started'>; readonly key: String; readonly scheduleKey: NullOr<String>; readonly startedAt: Number; readonly isStartupRun: Boolean }, 'Type'> | Struct.ReadonlySide<{ readonly _tag: tag<'Compl…;
run: Effect.Effect<void, never, never>;
};
}
import DaemonDaemon.const Tag: <
Self
>() => DaemonTagBuild<Self>
Define a managed daemon as a toolkit HyperService. Self is given explicitly (Effect's ()
two-stage form). The base tag carries observation + lifecycle; add a schedule with
.pipe(
schedule
(…)). Declare value/error wire schemas on the tag:
class Health extends Daemon.Tag<Health>()("app/Health") {}
class Prices extends Daemon.Tag<Prices>()("app/Prices", PriceSchema) {}
class PricesE extends Daemon.Tag<PricesE>()("app/Prices", PriceSchema, FetchErr) {}
class PricesCfg extends Daemon.Tag<PricesCfg>()("app/Prices", {
success: PriceSchema,
error: FetchErr,
}) {}
Pass options.node to bind the daemon to a
Node.Tag
.
Tag<class Dailyclass Daily {
key: Identifier;
Service: {
status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsStarted: number; readonly runsSucceeded: number; readonly runsFailed: number; readonly nextTriggerRun?: Utc | u…;
start: Effect.Effect<void, never, never>;
stop: Effect.Effect<void, never, never>;
wake: Effect.Effect<void, never, never>;
resetCadence: Effect.Effect<void, never, never>;
events: Stream<Struct.ReadonlySide<{ readonly _tag: tag<'Started'>; readonly key: String; readonly scheduleKey: NullOr<String>; readonly startedAt: Number; readonly isStartupRun: Boolean }, 'Type'> | Struct.ReadonlySide<{ readonly _tag: tag<'Compl…;
run: Effect.Effect<void, never, never>;
};
}
class AppStoreclass AppStore {
Service: Service;
key: Identifier;
}
import StoreStore.const Service: <AppStore>(
id: string
) => <const Args extends ReadonlyArray<unknown>>(
...args: Args
) => IsSingleStoreInput<Input> extends true
? Store.SingleStoreServiceClass<
AppStore,
string,
ContractForSingleInput<Input>
>
: Store.StoreServiceClass<
AppStore,
string,
RegsOfStoreInput<Input>
>
Declare an app store — class extends with
layerMemory
/
layer
.
Three input shapes:
- Single store — bare registration:
WorkPool.store(Mail) → yield* MailStore
- Tag-keyed multi — array:
[WorkPool.store(Mail), …] → yield* AppStore.at(Mail)
- Custom-keyed — object:
{ mail: WorkPool.store(Mail), … } → yield* AppStore.at("mail")
layerMemory uses in-memory refs. layer({ filename }) persists to SQLite (filename required).
Service<class AppStoreclass AppStore {
Service: Service;
key: Identifier;
}
class BillingNodeclass BillingNode {
key: Identifier;
Service: {
protocol: Context.Service.Shape<typeof RpcClient.Protocol>;
ping: Effect.Effect<number, NodeUnreachable>;
status: { readonly get: Effect.Effect<NodeStatusSnapshot, NodeUnreachable>; readonly changes: Stream.Stream<NodeStatusSnapshot, NodeUnreachable> };
logs: { readonly stream: Stream.Stream<LogEntry, NodeUnreachable>; readonly query: (options: { readonly limit: number }) => Effect.Effect<ReadonlyArray<LogEntry>, NodeUnreachable> };
};
of: (this: void, self: NodeProtocol) => NodeProtocol;
context: (self: NodeProtocol) => Context<BillingNode>;
use: (f: (service: NodeProtocol) => Effect.Effect<A, E, R>) => Effect.Effect<A, E, BillingNode | R>;
useSync: (f: (service: NodeProtocol) => A) => Effect.Effect<A, never, BillingNode>;
Identifier: Identifier;
stack: string | undefined;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
toString: () => string;
toJSON: () => unknown;
url: undefined;
path: undefined;
kind: undefined;
logs: unknown;
onConflict: OnConflict;
}
logs: unknownlogs,
import DaemonDaemon.function store<typeof Daily>(tag: typeof Daily): RegisteredWithContract<string, Readonly<Record<string, StoreSpecEntry>>, DaemonStoreAnalyticsContract<typeof Daily>, typeof Daily> (+1 overload)Register this daemon on an app
Store.Service
— built-in execution analytics with an
optional bare spec object merged in:
Daemon.store(Daily)
Daemon.store(Daily, {
audit: auditSchema,
}, ({ audit, event }) => ({
appendAudit: audit.append,
}))
store(class Dailyclass Daily {
key: Identifier;
Service: {
status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsStarted: number; readonly runsSucceeded: number; readonly runsFailed: number; readonly nextTriggerRun?: Utc | u…;
start: Effect.Effect<void, never, never>;
stop: Effect.Effect<void, never, never>;
wake: Effect.Effect<void, never, never>;
resetCadence: Effect.Effect<void, never, never>;
events: Stream<Struct.ReadonlySide<{ readonly _tag: tag<'Started'>; readonly key: String; readonly scheduleKey: NullOr<String>; readonly startedAt: Number; readonly isStartupRun: Boolean }, 'Type'> | Struct.ReadonlySide<{ readonly _tag: tag<'Compl…;
run: Effect.Effect<void, never, never>;
};
description: string | undefined;
of: (this: void, self: { readonly [x: string]: Effect.Effect<unknown, never, Hyperlink.Local<Daily>>; readonly status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly …;
context: (self: { readonly [x: string]: Effect.Effect<unknown, never, Hyperlink.Local<Daily>>; readonly status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsStarted:…;
use: (f: (service: { readonly [x: string]: Effect.Effect<unknown, never, Hyperlink.Local<Daily>>; readonly status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsS…;
useSync: (f: (service: { readonly [x: string]: Effect.Effect<unknown, never, Hyperlink.Local<Daily>>; readonly status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsS…;
Identifier: Identifier;
stack: string | undefined;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
toString: () => string;
toJSON: () => unknown;
}
const program: Effect.Effect<
{
nodeRows: ReadonlyArray<LogEntry>
resourceRows: ReadonlyArray<LogEntry>
fromExport: ReadonlyArray<LogEntry>
},
never,
never
>
const program: {
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
toString: () => string;
toJSON: () => unknown;
}
import EffectEffect.const gen: {
<Eff extends Effect<any, any, any>, AEff>(
f: () => Generator<Eff, AEff, never>
): Effect<
AEff,
[Eff] extends [never]
? never
: [Eff] extends [
Effect<infer _A, infer E, infer _R>
]
? E
: never,
[Eff] extends [never]
? never
: [Eff] extends [
Effect<infer _A, infer _E, infer R>
]
? R
: never
>
<Self, Eff extends Effect<any, any, any>, AEff>(
options: { readonly self: Self },
f: (this: Self) => Generator<Eff, AEff, never>
): Effect<
AEff,
[Eff] extends [never]
? never
: [Eff] extends [
Effect<infer _A, infer E, infer _R>
]
? E
: never,
[Eff] extends [never]
? never
: [Eff] extends [
Effect<infer _A, infer _E, infer R>
]
? R
: never
>
}
Provides a way to write effectful code using generator functions, simplifying
control flow and error handling.
When to use
Use when you want to write effectful code that looks and behaves like
synchronous code, while still handling asynchronous tasks, errors, and complex
control flow such as loops and conditions.
Generator functions work similarly to async/await but keep errors,
requirements, and interruption in the Effect type. You can yield* values
from effects and return the final result at the end.
Example (Sequencing effects with generators)
import { Data, Effect } from "effect"
class DiscountRateError extends Data.TaggedError("DiscountRateError")<{}> {}
const addServiceCharge = (amount: number) => amount + 1
const applyDiscount = (
total: number,
discountRate: number
): Effect.Effect<number, DiscountRateError> =>
discountRate === 0
? Effect.fail(new DiscountRateError())
: Effect.succeed(total - (total * discountRate) / 100)
const fetchTransactionAmount = Effect.promise(() => Promise.resolve(100))
const fetchDiscountRate = Effect.promise(() => Promise.resolve(5))
export const program = Effect.gen(function*() {
const transactionAmount = yield* fetchTransactionAmount
const discountRate = yield* fetchDiscountRate
const discountedAmount = yield* applyDiscount(
transactionAmount,
discountRate
)
const finalAmount = addServiceCharge(discountedAmount)
return `Final amount to charge: ${finalAmount}`
})
gen(function* () {
// Node journal — every line the match-all follower persisted for this Node.
const const nodeRows: ReadonlyArray<LogEntry>nodeRows = yield* import LogsLogs.const byNode: (
node: NodeLogKey | NodeLogKeySource,
options?: LogReadOptions
) => Effect.Effect<ReadonlyArray<LogEntry>>
Read durable logs for a whole node (every HyperService on that node).
Needs Node.logs / Hyperlink.store(Node) on an app
Store.Service
(Soft-override the
toolkit layer — see docs/guides/stores.md). Soft-default Memory alone is engine observability
only — no Logs platform / durable _logs tails.
byNode(class BillingNodeclass BillingNode {
key: Identifier;
Service: {
protocol: Context.Service.Shape<typeof RpcClient.Protocol>;
ping: Effect.Effect<number, NodeUnreachable>;
status: { readonly get: Effect.Effect<NodeStatusSnapshot, NodeUnreachable>; readonly changes: Stream.Stream<NodeStatusSnapshot, NodeUnreachable> };
logs: { readonly stream: Stream.Stream<LogEntry, NodeUnreachable>; readonly query: (options: { readonly limit: number }) => Effect.Effect<ReadonlyArray<LogEntry>, NodeUnreachable> };
};
of: (this: void, self: NodeProtocol) => NodeProtocol;
context: (self: NodeProtocol) => Context<BillingNode>;
use: (f: (service: NodeProtocol) => Effect.Effect<A, E, R>) => Effect.Effect<A, E, BillingNode | R>;
useSync: (f: (service: NodeProtocol) => A) => Effect.Effect<A, never, BillingNode>;
Identifier: Identifier;
stack: string | undefined;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
toString: () => string;
toJSON: () => unknown;
url: undefined;
path: undefined;
kind: undefined;
logs: unknown;
onConflict: OnConflict;
}
LogReadOptions.limit?: number | undefinedlimit: 200 })
// Hyperlink journal — that registration's scope (same key as Daily.key).
const const resourceRows: ReadonlyArray<LogEntry>resourceRows = yield* import LogsLogs.const byHyperlink: (
serviceKey:
| HyperlinkLogKey
| HyperlinkLogKeySource,
options?: LogReadOptions
) => Effect.Effect<ReadonlyArray<LogEntry>>
Read durable logs for a specific HyperService by full key (same string as
Hyperlink.logs
(tag) / store registration / lineage segment).
Pass a scope tag (Daemon.Tag / WorkPool.Tag / …) or its .key string.
Hyperlink kind is
Hyperlink.kindOf
on the tag — not a separate query argument.
Requires that HyperService's store registration (Daemon.store / WorkPool.store, …) on the
ambient
Store.Storage
. Missing registration fails via
Store.resolveOrDie
(StoreScopeNotRegistered) — empty success is not used as a silent signal for “wrong key.”
byHyperlink(class Dailyclass Daily {
key: Identifier;
Service: {
status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsStarted: number; readonly runsSucceeded: number; readonly runsFailed: number; readonly nextTriggerRun?: Utc | u…;
start: Effect.Effect<void, never, never>;
stop: Effect.Effect<void, never, never>;
wake: Effect.Effect<void, never, never>;
resetCadence: Effect.Effect<void, never, never>;
events: Stream<Struct.ReadonlySide<{ readonly _tag: tag<'Started'>; readonly key: String; readonly scheduleKey: NullOr<String>; readonly startedAt: Number; readonly isStartupRun: Boolean }, 'Type'> | Struct.ReadonlySide<{ readonly _tag: tag<'Compl…;
run: Effect.Effect<void, never, never>;
};
description: string | undefined;
of: (this: void, self: { readonly [x: string]: Effect.Effect<unknown, never, Hyperlink.Local<Daily>>; readonly status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly …;
context: (self: { readonly [x: string]: Effect.Effect<unknown, never, Hyperlink.Local<Daily>>; readonly status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsStarted:…;
use: (f: (service: { readonly [x: string]: Effect.Effect<unknown, never, Hyperlink.Local<Daily>>; readonly status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsS…;
useSync: (f: (service: { readonly [x: string]: Effect.Effect<unknown, never, Hyperlink.Local<Daily>>; readonly status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsS…;
Identifier: Identifier;
stack: string | undefined;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
toString: () => string;
toJSON: () => unknown;
}
LogReadOptions.limit?: number | undefinedlimit: 100 })
// Preferred: live + durable product export for the tag.
const { const query: (options?: {
readonly limit?: number
}) => Effect.Effect<ReadonlyArray<LogEntry>>
import HyperlinkHyperlink.logs<Tag extends StoreScopeTag>(tag: Tag): Effect.Effect<Hyperlink.LogsExportHandle, never, never>Live stream: lineage + optional
Hyperlink.logStreamLevel
gate.
Local
LogRelay
when present; otherwise NodeStatus.logs (remote client).
Durable query prefers registration Storage, else NodeStatus (remote).
logs(class Dailyclass Daily {
key: Identifier;
Service: {
status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsStarted: number; readonly runsSucceeded: number; readonly runsFailed: number; readonly nextTriggerRun?: Utc | u…;
start: Effect.Effect<void, never, never>;
stop: Effect.Effect<void, never, never>;
wake: Effect.Effect<void, never, never>;
resetCadence: Effect.Effect<void, never, never>;
events: Stream<Struct.ReadonlySide<{ readonly _tag: tag<'Started'>; readonly key: String; readonly scheduleKey: NullOr<String>; readonly startedAt: Number; readonly isStartupRun: Boolean }, 'Type'> | Struct.ReadonlySide<{ readonly _tag: tag<'Compl…;
run: Effect.Effect<void, never, never>;
};
description: string | undefined;
of: (this: void, self: { readonly [x: string]: Effect.Effect<unknown, never, Hyperlink.Local<Daily>>; readonly status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly …;
context: (self: { readonly [x: string]: Effect.Effect<unknown, never, Hyperlink.Local<Daily>>; readonly status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsStarted:…;
use: (f: (service: { readonly [x: string]: Effect.Effect<unknown, never, Hyperlink.Local<Daily>>; readonly status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsS…;
useSync: (f: (service: { readonly [x: string]: Effect.Effect<unknown, never, Hyperlink.Local<Daily>>; readonly status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsS…;
Identifier: Identifier;
stack: string | undefined;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
toString: () => string;
toJSON: () => unknown;
}
const fromExport: ReadonlyArray<LogEntry>fromExport = yield* const query: (options?: {
readonly limit?: number
}) => Effect.Effect<ReadonlyArray<LogEntry>>
limit?: number | undefinedlimit: 100 })
return { nodeRows: ReadonlyArray<LogEntry>nodeRows, resourceRows: ReadonlyArray<LogEntry>resourceRows, fromExport: ReadonlyArray<LogEntry>fromExport }
})Layer order
Toolkit layer / serve soft-default in-memory Storage (R fulfilled). Provide your Store.Service into the HyperService Layer so Soft unwrap captures that store — especially before queue workers fork at Layer build:
Effect.provide(
program,
WorkPool.layer(MyQueue, { effect: worker }).pipe(
Layer.provideMerge(AppStore.layerMemory),
),
)Bare WorkPool.layer / Daemon.layer (or *Memory aliases) work without an AppStore; durable logs still need Store.Service.layer*. Recipe SSOT: docs/guides/stores.md.
Keys
Say the identifier kind out loud. Mixing them is the common failure mode.
| Kind | Identifies | Declared as | Used for |
|---|---|---|---|
| Node log key | One OS process / runtime host | Node.Tag(…) → .key | Node.logs, Logs.byNode, annotations.node |
| Hyperlink key | One WorkPool, Daemon, Gate, or custom Tag | Tag(…) → .key | store scope, lineage segments, byHyperlink |
| Lineage segment | One hop in ancestry | element of the lineage JSON array | LogEntry.hasKey / atRoot / atLeaf |
| Annotation key | Field name on LogEntry.annotations | LogAnnotationKeys.* | metadata keys, not buckets |
BillingNode.key // node log key — "billing/scores"
Daily.key // service key — "app/Daily"
LogAnnotationKeys.node // annotation key — "node" (holds a node log key value)
LogAnnotationKeys.lineage // annotation key — JSON array of lineage segment keysNode log key rules
It must equal that process’s
Node.Tagkey —BillingNode.key, not an invented"my-node".Prefer slash-separated paths (
domain/role):"billing/scores","wnba/live".Every node-journal line is stamped with
annotations.node= that key.Query with
Logs.byNode(BillingNode)(or the string key, if unknown statically).
// ❌ drifts from Node.Tag
Logs.byNode("wnba") // WnbaNode.key is "wnba/scores"
Logs.byNode("my-node")Hyperlink keys
Hyperlink identity is tag.key (may contain /; some metrics Tags use an @ prefix). Hyperlink journals, lineage filters, and Hyperlink.logs all key off that string.
Lineage
Each log line may carry a lineage path: an ordered list of segment keys under LogAnnotationKeys.lineage. Engines stamp it with Logs.withScope(tag) when they materialize work — Daemon and WorkPool do this for you.
withScope is append-only. Nested scopes combine:
effect
.pipe(Logs.withScope(Child))
.pipe(Logs.withScope(Parent))
// fiber lineage → ["parent/Key", "child/Key"]Re-entering the same leaf key is idempotent (no duplicate last segment). A Node root is not auto-injected into lineage; the node journal uses annotations.node instead.
Predicates
import { LogEntry } from "hyperlink-ts"
LogEntry.lineage(entry) // ReadonlyArray<string>
LogEntry.hasKey(serviceKey)(entry) // key anywhere in the path
LogEntry.atRoot(segment)(entry) // lineage[0] === segment
LogEntry.atLeaf(serviceKey)(entry) // last segment === serviceKeyLegacy processId / queueId annotations are gone — writers stamp lineage only via Logs.withScope. Hyperlink kind is Hyperlink.kindOf(tag), not an annotation field.
Per-hyperlink export
Prefer Hyperlink.logs(tag) for Handle-shaped access — live Stream plus durable query:
import { import LogEntryLogEntry, import DaemonDaemon, import HyperlinkHyperlink } from "hyperlink-ts"
import { import EffectEffect } from "effect"
class class Dailyclass Daily {
key: Identifier;
Service: {
status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsStarted: number; readonly runsSucceeded: number; readonly runsFailed: number; readonly nextTriggerRun?: Utc | u…;
start: Effect.Effect<void, never, never>;
stop: Effect.Effect<void, never, never>;
wake: Effect.Effect<void, never, never>;
resetCadence: Effect.Effect<void, never, never>;
events: Stream<Struct.ReadonlySide<{ readonly _tag: tag<'Started'>; readonly key: String; readonly scheduleKey: NullOr<String>; readonly startedAt: Number; readonly isStartupRun: Boolean }, 'Type'> | Struct.ReadonlySide<{ readonly _tag: tag<'Compl…;
run: Effect.Effect<void, never, never>;
};
}
import DaemonDaemon.const Tag: <
Self
>() => DaemonTagBuild<Self>
Define a managed daemon as a toolkit HyperService. Self is given explicitly (Effect's ()
two-stage form). The base tag carries observation + lifecycle; add a schedule with
.pipe(
schedule
(…)). Declare value/error wire schemas on the tag:
class Health extends Daemon.Tag<Health>()("app/Health") {}
class Prices extends Daemon.Tag<Prices>()("app/Prices", PriceSchema) {}
class PricesE extends Daemon.Tag<PricesE>()("app/Prices", PriceSchema, FetchErr) {}
class PricesCfg extends Daemon.Tag<PricesCfg>()("app/Prices", {
success: PriceSchema,
error: FetchErr,
}) {}
Pass options.node to bind the daemon to a
Node.Tag
.
Tag<class Dailyclass Daily {
key: Identifier;
Service: {
status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsStarted: number; readonly runsSucceeded: number; readonly runsFailed: number; readonly nextTriggerRun?: Utc | u…;
start: Effect.Effect<void, never, never>;
stop: Effect.Effect<void, never, never>;
wake: Effect.Effect<void, never, never>;
resetCadence: Effect.Effect<void, never, never>;
events: Stream<Struct.ReadonlySide<{ readonly _tag: tag<'Started'>; readonly key: String; readonly scheduleKey: NullOr<String>; readonly startedAt: Number; readonly isStartupRun: Boolean }, 'Type'> | Struct.ReadonlySide<{ readonly _tag: tag<'Compl…;
run: Effect.Effect<void, never, never>;
};
}
const program: Effect.Effect<
{
stream: Stream<
LogEntry.LogEntry,
never,
never
>
mine: Array<LogEntry.LogEntry>
},
never,
never
>
const program: {
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
toString: () => string;
toJSON: () => unknown;
}
import EffectEffect.const gen: {
<Eff extends Effect<any, any, any>, AEff>(
f: () => Generator<Eff, AEff, never>
): Effect<
AEff,
[Eff] extends [never]
? never
: [Eff] extends [
Effect<infer _A, infer E, infer _R>
]
? E
: never,
[Eff] extends [never]
? never
: [Eff] extends [
Effect<infer _A, infer _E, infer R>
]
? R
: never
>
<Self, Eff extends Effect<any, any, any>, AEff>(
options: { readonly self: Self },
f: (this: Self) => Generator<Eff, AEff, never>
): Effect<
AEff,
[Eff] extends [never]
? never
: [Eff] extends [
Effect<infer _A, infer E, infer _R>
]
? E
: never,
[Eff] extends [never]
? never
: [Eff] extends [
Effect<infer _A, infer _E, infer R>
]
? R
: never
>
}
Provides a way to write effectful code using generator functions, simplifying
control flow and error handling.
When to use
Use when you want to write effectful code that looks and behaves like
synchronous code, while still handling asynchronous tasks, errors, and complex
control flow such as loops and conditions.
Generator functions work similarly to async/await but keep errors,
requirements, and interruption in the Effect type. You can yield* values
from effects and return the final result at the end.
Example (Sequencing effects with generators)
import { Data, Effect } from "effect"
class DiscountRateError extends Data.TaggedError("DiscountRateError")<{}> {}
const addServiceCharge = (amount: number) => amount + 1
const applyDiscount = (
total: number,
discountRate: number
): Effect.Effect<number, DiscountRateError> =>
discountRate === 0
? Effect.fail(new DiscountRateError())
: Effect.succeed(total - (total * discountRate) / 100)
const fetchTransactionAmount = Effect.promise(() => Promise.resolve(100))
const fetchDiscountRate = Effect.promise(() => Promise.resolve(5))
export const program = Effect.gen(function*() {
const transactionAmount = yield* fetchTransactionAmount
const discountRate = yield* fetchDiscountRate
const discountedAmount = yield* applyDiscount(
transactionAmount,
discountRate
)
const finalAmount = addServiceCharge(discountedAmount)
return `Final amount to charge: ${finalAmount}`
})
gen(function* () {
const { const stream: Stream<
LogEntry.LogEntry,
never,
never
>
const stream: {
channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
const query: (options?: {
readonly limit?: number
}) => Effect.Effect<ReadonlyArray<LogEntry>>
import HyperlinkHyperlink.logs<Tag extends StoreScopeTag>(tag: Tag): Effect.Effect<Hyperlink.LogsExportHandle, never, never>Live stream: lineage + optional
Hyperlink.logStreamLevel
gate.
Local
LogRelay
when present; otherwise NodeStatus.logs (remote client).
Durable query prefers registration Storage, else NodeStatus (remote).
logs(class Dailyclass Daily {
key: Identifier;
Service: {
status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsStarted: number; readonly runsSucceeded: number; readonly runsFailed: number; readonly nextTriggerRun?: Utc | u…;
start: Effect.Effect<void, never, never>;
stop: Effect.Effect<void, never, never>;
wake: Effect.Effect<void, never, never>;
resetCadence: Effect.Effect<void, never, never>;
events: Stream<Struct.ReadonlySide<{ readonly _tag: tag<'Started'>; readonly key: String; readonly scheduleKey: NullOr<String>; readonly startedAt: Number; readonly isStartupRun: Boolean }, 'Type'> | Struct.ReadonlySide<{ readonly _tag: tag<'Compl…;
run: Effect.Effect<void, never, never>;
};
description: string | undefined;
of: (this: void, self: { readonly [x: string]: Effect.Effect<unknown, never, Hyperlink.Local<Daily>>; readonly status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly …;
context: (self: { readonly [x: string]: Effect.Effect<unknown, never, Hyperlink.Local<Daily>>; readonly status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsStarted:…;
use: (f: (service: { readonly [x: string]: Effect.Effect<unknown, never, Hyperlink.Local<Daily>>; readonly status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsS…;
useSync: (f: (service: { readonly [x: string]: Effect.Effect<unknown, never, Hyperlink.Local<Daily>>; readonly status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsS…;
Identifier: Identifier;
stack: string | undefined;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
toString: () => string;
toJSON: () => unknown;
}
const history: ReadonlyArray<LogEntry.LogEntry>history = yield* const query: (options?: {
readonly limit?: number
}) => Effect.Effect<ReadonlyArray<LogEntry>>
limit?: number | undefinedlimit: 50 })
const const mine: Array<LogEntry.LogEntry>mine = const history: ReadonlyArray<LogEntry.LogEntry>history.ReadonlyArray<LogEntry>.filter(predicate: (value: LogEntry.LogEntry, index: number, array: readonly LogEntry.LogEntry[]) => unknown, thisArg?: any): LogEntry.LogEntry[] (+1 overload)Returns the elements of an array that meet the condition specified in a callback function.
filter(import LogEntryLogEntry.const hasKey: (
key: string
) => Predicate.Predicate<LogEntry>
true when the lineage segment key appears anywhere in
lineage
.
hasKey(class Dailyclass Daily {
key: Identifier;
Service: {
status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsStarted: number; readonly runsSucceeded: number; readonly runsFailed: number; readonly nextTriggerRun?: Utc | u…;
start: Effect.Effect<void, never, never>;
stop: Effect.Effect<void, never, never>;
wake: Effect.Effect<void, never, never>;
resetCadence: Effect.Effect<void, never, never>;
events: Stream<Struct.ReadonlySide<{ readonly _tag: tag<'Started'>; readonly key: String; readonly scheduleKey: NullOr<String>; readonly startedAt: Number; readonly isStartupRun: Boolean }, 'Type'> | Struct.ReadonlySide<{ readonly _tag: tag<'Compl…;
run: Effect.Effect<void, never, never>;
};
description: string | undefined;
of: (this: void, self: { readonly [x: string]: Effect.Effect<unknown, never, Hyperlink.Local<Daily>>; readonly status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly …;
context: (self: { readonly [x: string]: Effect.Effect<unknown, never, Hyperlink.Local<Daily>>; readonly status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsStarted:…;
use: (f: (service: { readonly [x: string]: Effect.Effect<unknown, never, Hyperlink.Local<Daily>>; readonly status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsS…;
useSync: (f: (service: { readonly [x: string]: Effect.Effect<unknown, never, Hyperlink.Local<Daily>>; readonly status: Hyperlink.Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; readonly runsS…;
Identifier: Identifier;
stack: string | undefined;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
toString: () => string;
toJSON: () => unknown;
}
ServiceClass<Daily, string, { readonly [x: string]: Effect<unknown, never, Local<Daily>>; readonly status: Subscribable<{ readonly supervising: boolean; readonly armed: boolean; readonly activeInstances: number; ... 7 more ...; readonly lastRunDurationMillis?: number | undefined; }>; ... 5 more ...; readonly run: Effect<...>; }>.key: stringkey))
return { stream: Stream<LogEntry.LogEntry, never, never>(property) stream: {
channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
mine: Array<LogEntry.LogEntry>mine }
})Pipe Hyperlink.withLogExport onto a Tag when you want yield* Tag.logs as a member:
class MailQueue extends WorkPool.Tag<MailQueue>()("app/Mail", MailJob).pipe(
Hyperlink.withLogExport,
) {}
const { stream, query } = yield* MailQueue.logsOn the raw bus, filter yourself:
Logs.stream.pipe(Stream.filter(LogEntry.hasKey(Daily.key)))Levels
Two knobs, two jobs:
| API | Affects |
|---|---|
Store.logLevel* / registration log level | What the durable tail persists for that scope |
Store.streamLevel* on a registration (or Hyperlink.logStreamLevel* on a Tag) | What Hyperlink.logs(tag).stream emits live |
class AppStore extends Store.Service<AppStore>("@app/Store")(
Store.streamLevelWarn(Daemon.store(QuietProc)),
BillingNode.logs,
) {}
// Tag-side live floor (Hyperlink.logs stream)
class QuietProc extends Daemon.Tag<QuietProc>()("app/Quiet").pipe(
Hyperlink.logStreamLevelWarn,
) {}"All" means no floor. "None" drops everything for that surface.
Remote clients
When the dashboard (or any client) reaches a Node over RPC, durable per-hyperlink rows come from that node’s journal — (yield* MyNode).logs — filtered by service key. Locally, Hyperlink.logs(tag).query prefers registration Storage and falls back to the node-handle logs path when Storage isn’t there.
import * as LogEntry from "hyperlink-ts/LogEntry"
import { Stream } from "effect"
const serviceKey = LiveScorePoller.key
const n = yield* LiveNode // connected node handle
n.logs.stream.pipe(Stream.filter(LogEntry.hasKey(serviceKey)))
const rows = yield* n.logs.query({ limit: 300 })
const scoped = rows.filter(LogEntry.hasKey(serviceKey))The server must still provide a Store.Service with Node.logs (and any toolkit stores you care about) on the Node stack. httpServer infers the node log key from served Tags’ bound Node for the handle’s logs.query.
Modules
| Concern | Package | Role |
|---|---|---|
| Platform | hyperlink-ts/Logs | Layer, bus, withScope, byNode / byHyperlink |
| Entry + predicates | hyperlink-ts/LogEntry | Wire shape, hasKey / atRoot / atLeaf |
| Annotation keys | hyperlink-ts/LogContext | LogAnnotationKeys |
| Export | hyperlink-ts/Hyperlink | logs, withLogExport, logStreamLevel* |
| Journals | hyperlink-ts/Store | Store.Service, Node.logs, toolkit .store |
Migration (removed surfaces)
| Old | Use instead |
|---|---|
Logs.persistLayer + store/Log | Node.logs + toolkit .store on Store.Service |
NodeLogs.* | Logs.* |
LogRelay / replayLogEntry / *RelayLayer flat aliases | Logs.Relay / Logs.replay / Logs.layer |
Engine captureLogs / handle .logs | Logs.layer + Logs.withScope + Hyperlink.logs |
HistoryStore `${tag.key}/logs` | Registration _logs + Hyperlink.logs / Logs.by* |
See also
Stores — registering journals and reading Storage
docs/LOGS.md— key catalog and fixture mapQueues / Processes — engines that stamp lineage at materialize