Typed Sinks

Project sinks consume one declared record type. Lifecycle and command-output streams remain separate even when they share a provider and destination.

Lifecycle sink

sinks/publish.py

from providers.destination_client import DestinationClient
from sqlbuild.sinks import (
    LifecycleEvent,
    LifecycleEventKind,
    lifecycle_event_sink,
    lifecycle_event_to_json,
)


@lifecycle_event_sink(
    event_kinds={
        LifecycleEventKind.INVOCATION,
        LifecycleEventKind.RUN,
        LifecycleEventKind.RESOURCE,
        LifecycleEventKind.AUDIT,
    }
)
def publish_lifecycle(event: LifecycleEvent, destination_client: DestinationClient) -> None:
    destination_client.publish(
        route="sqlbuild.lifecycle.v2",
        key=event.invocation_id,
        payload=lifecycle_event_to_json(event).encode("utf-8"),
    )

Using invocation_id as a partition key keeps one invocation on one Kafka partition. Keep event_id in the payload as the deduplication identity. Confirm the destination topic is not compacted when complete lifecycle history must remain replayable.

Command-output sink

from sqlbuild.sinks import CommandOutputRecord, command_output_sink, command_output_to_json


@command_output_sink(streams={"stdout", "stderr"})
def publish_output(record: CommandOutputRecord, destination_client: DestinationClient) -> None:
    destination_client.publish(
        route="sqlbuild.command_output.v1",
        key=record.record_id,
        payload=command_output_to_json(record).encode("utf-8"),
    )

Command output groups adjacent text into bounded byte chunks and flushes on elapsed time, stream change, size, or close. A loss record reports bounded-queue drops. This stream is useful for remote transcripts but is not lifecycle evidence.

A sink with a parameter annotated sqlbuild.recall.RecallWriter is not an ordinary sink: it decides what Recall stores for its stream.

Configuration

Runtime configuration can narrow declaration filters:

[sinks.lifecycle]
event_kinds = ["run", "resource", "operation", "statement", "audit"]
min_severity = "info"

[sinks.lifecycle.named.publish_lifecycle]
event_kinds = ["resource", "statement"]
min_severity = "warning"

shutdown_timeout sets how long a command waits at exit for lifecycle sinks to drain queued events. It is a fixed duration from "0s" to "10m", defaults to "2s", and is accepted only under [sinks.lifecycle], not per named sink. "0s" drops anything still queued at exit. The final quarter of the budget is reserved for run and invocation outcomes: when it begins, other queued events are dropped and counted so the outcomes still arrive. The setting does not change the one-second timeout for each sink call. A sink call still running when the budget ends gets up to that one second to return before providers are torn down; teardown always runs on the command’s own thread.

[sinks.lifecycle]
shutdown_timeout = "30s"

SQLBuild owns envelope validation, local queueing, filtering, and bounded dispatch. The project owns destination credentials, serialization, routes/topics, acknowledgements, retries, retention, and durability. SQLBuild core does not provide a Kafka or ClickHouse implementation.

Delivery behavior

• Each sink receives events in order on its own long-lived delivery thread.
• Lifecycle dispatch prioritizes failures and terminal facts, remaining FIFO within equal priority; invocation_completed and invocation_failed come after every other queued event.
• Queue overflow can displace or drop lower-priority records and is reflected in sink accounting. Run and invocation outcomes (run_started, run_completed, run_failed, invocation_started, invocation_completed, invocation_failed) are always admitted and never displaced.
• A failing or timed-out sink is isolated from command correctness.
• Each sink call has one second to return. When a call takes longer, the command prints Warning: lifecycle sink '<name>' did not return from one call within 1s; its events are dropped until that call returns. That call counts as failed and the sink’s events are dropped and counted while it runs. Run and invocation outcomes are kept instead and delivered, with every later event, once the call returns. Time SQLBuild itself spends waiting for Python’s GIL does not count against the call.
• audit_completed facts are published as each audit finishes (model audits in sqb build when their model finishes), not in one burst at the end of the run.
• When lifecycle events are dropped or fail, the command prints one warning to stderr with the counts and affected sinks, for example Warning: command completed successfully, but lifecycle event export was incomplete: 140 of 248 events dropped, 0 failed (sink 'orders_events'). Check sink health: a sink call slower than 1s drops that sink's events until it returns; events still queued at exit need a longer sinks.lifecycle.shutdown_timeout. A successful command then ends with sqb audit completed successfully; only its lifecycle event export is affected. The exit code and stdout, including --json output, are unchanged.
• Provider setup and declaration errors fail before execution because the project configuration is invalid.
• Providers shared by multiple sinks are set up and torn down once per command.

Use lifecycle facts—not command-output records—to build execution state and timelines.