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
└── Process.store(Daily)— lineage durable tail → resource journal
Process / Queue 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 Process.store(Daily) are registered, the same published line can appear in the node journal and the resource 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 { Logs } from "hyperlink-ts"
import { import EffectEffect, import FiberFiber, import StreamStream } from "effect"
const const program: Effect.Effect<
any,
unknown,
unknown
>
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: <Effect.Effect<Fiber.Fiber<unknown[], unknown>, never, unknown> | Effect.Effect<unknown[], unknown, never> | Effect.Effect<void, never, never>, any>(f: () => Generator<Effect.Effect<Fiber.Fiber<unknown[], unknown>, never, unknown> | Effect.Effect<unknown[], unknown, never> | Effect.Effect<void, never, never>, any, never>) => Effect.Effect<any, unknown, unknown> (+1 overload)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<
unknown[],
unknown
>
const collector: {
id: number;
currentOpCount: number;
getRef: <A>(ref: Context.Reference<A>) => A;
context: Context.Context<never>;
setContext: (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: <Effect.Effect<unknown[], unknown, unknown>>(effectOrOptions?: Effect.Effect<unknown[], unknown, unknown> | undefined, options?: {
readonly startImmediately?: boolean | undefined;
readonly uninterruptible?: boolean | "inherit" | undefined;
} | undefined) => Effect.Effect<Fiber.Fiber<unknown[], unknown>, never, unknown>
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: <
unknown,
unknown,
unknown
>(
self: Stream.Stream<unknown, unknown, unknown>
) => Effect.Effect<unknown[], unknown, unknown>
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: <unknown, unknown, unknown>(self: Stream.Stream<unknown, unknown, unknown>, n: number) => Stream.Stream<unknown, unknown, unknown> (+1 overload)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(Logs.stream, 1)),
)
yield* import EffectEffect.const logInfo: (
...message: ReadonlyArray<any>
) => Effect.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: unknownconst entry: {
date: string;
level: LogLevel;
message: string;
cause: string;
annotations: Readonly<Record<string, string>>;
spans: ReadonlyArray<string>;
}
var Array: ArrayConstructorArray.ArrayConstructor.from<unknown>(iterable: Iterable<unknown> | ArrayLike<unknown>): unknown[] (+3 overloads)Creates an array from an iterable object.
from(yield* import FiberFiber.const join: <unknown[], unknown>(self: Fiber.Fiber<unknown[], unknown>) => Effect.Effect<unknown[], unknown, never>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<
unknown[],
unknown
>
const collector: {
id: number;
currentOpCount: number;
getRef: <A>(ref: Context.Reference<A>) => A;
context: Context.Context<never>;
setContext: (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: unknownconst entry: {
date: string;
level: LogLevel;
message: string;
cause: string;
annotations: Readonly<Record<string, string>>;
spans: ReadonlyArray<string>;
}
const runnable: Effect.Effect<
any,
any,
any
>
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<
any,
unknown,
unknown
>
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<any, unknown, unknown>, Effect.Effect<any, any, any>, Effect.Effect<any, any, any>>(this: Effect.Effect<any, unknown, unknown>, ab: (_: Effect.Effect<any, unknown, unknown>) => Effect.Effect<any, any, any>, bc: (_: Effect.Effect<any, any, any>) => Effect.Effect<any, any, any>): Effect.Effect<any, any, any> (+21 overloads)pipe(import EffectEffect.const provide: <any>(layers: any, options?: {
readonly local?: boolean | undefined;
} | undefined) => <A, E, R>(self: Effect.Effect<A, E, R>) => Effect.Effect<A, any, any> (+5 overloads)
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(Logs.layer), import EffectEffect.const scoped: <A, E, R>(
self: Effect.Effect<A, E, R>
) => Effect.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 — Process.store(tag), QueueHyperlink.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 { Logs, Process, import HyperlinkHyperlink, Store } 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>;
};
}
import NodeNode.Tag<class BillingNodeclass BillingNode {
key: Identifier;
Service: {
protocol: Context.Service.Shape<typeof RpcClient.Protocol>;
};
}
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>;
};
}
Process$1.Tag<Daily>(): ProcessTagBuild<Daily>
export Process$1.Tag
Define a managed process as a toolkit resource. 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 Process.Tag<Health>()("app/Health") {}
class Prices extends Process.Tag<Prices>()("app/Prices", PriceSchema) {}
class PricesE extends Process.Tag<PricesE>()("app/Prices", PriceSchema, FetchErr) {}
class PricesCfg extends Process.Tag<PricesCfg>()("app/Prices", {
success: PriceSchema,
error: FetchErr,
}) {}
Pass options.node to bind the process to a
Resource.Node
.
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;
}
Store$1.Service<AppStore>(id: string): <Args>(...args: Args) => StoreServiceClass<AppStore, string, RegsOfStoreInput<[...Args] extends infer T ? T extends T & [...Args] ? T extends readonly [infer Only] ? Only : T : never : never>>
export Store$1.Service
Declare an aggregate store bundle — class extends with
layerMemory
/
layer
.
layerMemory uses in-memory refs. layer({ filename }) persists to SQLite; omit filename for memory.
Service<class AppStoreclass AppStore {
Service: Service;
key: Identifier;
}
class BillingNodeclass BillingNode {
key: Identifier;
Service: {
protocol: Context.Service.Shape<typeof RpcClient.Protocol>;
};
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;
}
Process$1.store<typeof Daily>(tag: typeof Daily): RegisteredWithContract<string, ProcessStoreAnalyticsContract<Tag>, ProcessStoreAnalyticsContract<Tag>, typeof Daily> (+1 overload)
export Process$1.store
Register this process on an app
Store.Service
— built-in execution analytics with an
optional bare spec object merged in:
Process.store(Daily)
Process.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>;
};
groupId: string;
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: any
resourceRows: any
fromExport: any
},
unknown,
unknown
>
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: <any, {
nodeRows: any;
resourceRows: any;
fromExport: any;
}>(f: () => Generator<any, {
nodeRows: any;
resourceRows: any;
fromExport: any;
}, never>) => Effect.Effect<{
nodeRows: any;
resourceRows: any;
fromExport: any;
}, unknown, unknown> (+1 overload)
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: anynodeRows = yield* Logs.byNode(class BillingNodeclass BillingNode {
key: Identifier;
Service: {
protocol: Context.Service.Shape<typeof RpcClient.Protocol>;
};
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;
}
limit: numberlimit: 200 })
// Hyperlink journal — that registration's scope (same key as Daily.key).
const const resourceRows: anyresourceRows = yield* Logs.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>;
};
groupId: string;
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;
}
limit: numberlimit: 100 })
// Preferred: live + durable product export for the tag.
const { const query: anyquery } = yield* import HyperlinkHyperlink.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>;
};
groupId: string;
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: anyfromExport = yield* const query: anyquery({ limit: numberlimit: 100 })
return { nodeRows: anynodeRows, resourceRows: anyresourceRows, fromExport: anyfromExport }
})Layer order
Toolkit layer / serve soft-default in-memory Storage (R fulfilled). Provide your Store.Service into the resource Layer so Soft unwrap captures that store — especially before queue workers fork at Layer build:
Effect.provide(
program,
QueueHyperlink.layer(MyQueue, { effect: worker }).pipe(
Layer.provideMerge(AppStore.layerMemory),
),
)Bare QueueHyperlink.layer / Process.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 Queue, Process, 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 // resource 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 — Process and Queue 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(resourceKey)(entry) // key anywhere in the path
LogEntry.atRoot(segment)(entry) // lineage[0] === segment
LogEntry.atLeaf(resourceKey)(entry) // last segment === resourceKeyLegacy processId / queueId annotations are gone — writers stamp lineage only via Logs.withScope. Hyperlink kind is Hyperlink.kindOf(tag), not an annotation field.
Per-resource export
Prefer Hyperlink.logs(tag) for Handle-shaped access — live Stream plus durable query:
import { LogEntry, Process, 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>;
};
}
Process$1.Tag<Daily>(): ProcessTagBuild<Daily>
export Process$1.Tag
Define a managed process as a toolkit resource. 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 Process.Tag<Health>()("app/Health") {}
class Prices extends Process.Tag<Prices>()("app/Prices", PriceSchema) {}
class PricesE extends Process.Tag<PricesE>()("app/Prices", PriceSchema, FetchErr) {}
class PricesCfg extends Process.Tag<PricesCfg>()("app/Prices", {
success: PriceSchema,
error: FetchErr,
}) {}
Pass options.node to bind the process to a
Resource.Node
.
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: any
mine: any
},
unknown,
unknown
>
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: <any, {
stream: any;
mine: any;
}>(f: () => Generator<any, {
stream: any;
mine: any;
}, never>) => Effect.Effect<{
stream: any;
mine: any;
}, unknown, unknown> (+1 overload)
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: anyconst 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: anyquery } = yield* import HyperlinkHyperlink.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>;
};
groupId: string;
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: anyhistory = yield* const query: anyquery({ limit: numberlimit: 50 })
const const mine: anymine = const history: anyhistory.filter(LogEntry.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>;
};
groupId: string;
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, ServiceOf<ProcessInstanceSpec<Void, Never>, Daily>>.key: stringkey))
return { stream: any(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: anymine }
})Pipe Hyperlink.withLogExport onto a Tag when you want yield* Tag.logs as a member:
class MailQueue extends QueueHyperlink.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(Process.store(QuietProc)),
BillingNode.logs,
) {}
// Tag-side live floor (Hyperlink.logs stream)
class QuietProc extends Process.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-resource rows come from that Node’s journal — typically NodeStatus.logs.query — filtered by resource key. Locally, Hyperlink.logs(tag).query prefers registration Storage and falls back to NodeStatus when Storage isn’t there.
import { LogEntry, NodeStatus } from "hyperlink-ts"
import { Stream } from "effect"
const resourceKey = LiveScorePoller.key
NodeStatus.logs.stream.pipe(Stream.filter(LogEntry.hasKey(resourceKey)))
const rows = yield* NodeStatus.logs.query({ limit: 300 })
const scoped = rows.filter(LogEntry.hasKey(resourceKey))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 NodeStatus.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