Documentation
¶
Overview ¶
Package assemble is Harbor's assembly entry point: the ONE exported, error-returning config→stack fan-out that turns a validated *config.Config into a running runtime stack.
What this package replaces ¶
Previously the dependency-ordered composition — stores → bus → llm → memory → skills → tasks → catalog (builtins + OAuth + approval + MCP attach) → sessions → planner → run loop, with reverse-order closers and partial-failure cleanup — existed in exactly two places: `cmd/harbor/cmd_dev.go::bootDevStack` (package main) and `harbortest/devstack`'s unexported `tryAssemble` (gated behind a `*testing.T` wrapper). The two copies had already drifted (the devstack MCP attach silently dropped the ToolPolicy projection; the devstack assembly never constructed cfg-declared OAuth providers). Both callers are now thin wrappers over `Assemble`; there is no second ordering left to drift. one model, no legacy "before" mode).
Scope ¶
Assemble ends where the network surface begins. Protocol surfaces, transports, auth validators, listeners, CORS, the Console, draft stores, and the per-task run-loop *driver* (the `task.spawned` subscriber) stay with the caller — `cmd/harbor` keeps its production driver, `harbortest/devstack` its test-kit mirror, and a headless embedder drives `Stack.RunLoop.Run` directly (see docs/recipes/embed-harbor-headless.md).
Lifecycle contract ¶
On error, Assemble returns the PARTIAL *Stack alongside the error so the caller's deferred Close drains every subsystem that opened before the failure. On success the caller owns the stack and MUST call Close (reverse dependency order, idempotent).
Concurrent reuse ¶
The returned *Stack is a compiled artifact: every field is set once during Assemble and never mutated afterwards (Close flips an internal once). The composed subsystems carry their own concurrent-reuse concurrent-reuse guarantees; per-run state lives in ctx + planner.RunContext, never on the stack.
Index ¶
Constants ¶
This section is empty.
Variables ¶
var DefaultMCPIdentity = identity.Identity{TenantID: "dev", UserID: "dev", SessionID: "dev"}
DefaultMCPIdentity is the fallback identity stamped on MCP server-pushed events that arrive without an inflight call when the caller supplies no Options.MCPDefaultIdentity. It matches the `harbor dev` single-operator triple.
var ( // ErrNotRunnable fires when RunOnce is called on a stack assembled // without a planner / run loop (no LLM driver, or SkipSteering / // SkipRunLoop). The wrapped detail names the missing component. ErrNotRunnable = errors.New("assemble: stack is not runnable") )
Sentinel errors RunOnce returns when the stack is not runnable. Callers compare via errors.Is.
Functions ¶
This section is empty.
Types ¶
type Options ¶
type Options struct {
// Logger receives the pre-telemetry bootstrap warnings (the window
// before the redactor + bus exist). Once the telemetry Logger is
// constructed, the assembly threads its Slog() bridge into every
// subsequently-built subsystem, so subsystem log lines flow
// through the canonical pipeline.
// Defaults to slog.Default().
Logger *slog.Logger
// LLMSnapshot, when non-nil, overrides the snapshot Assemble would
// otherwise project from cfg.LLM via llm.SnapshotFromConfig (
// ). `harbor dev` uses this for the mock-LLM
// escape hatch; tests flip the driver without rewriting yaml.
LLMSnapshot *llm.ConfigSnapshot
// PlannerOverride, when non-nil, replaces the registry-resolved
// planner concrete. Tests inject stub / scripted / pausing
// planners; production never sets it.
PlannerOverride planner.Planner
// SkillStore, when non-nil, wins over the cfg-opened store. The
// caller owns its lifecycle (it is NOT added to the closer chain).
SkillStore skills.SkillStore
// Embedder, when non-nil, wins over the cfg-opened embedding
// client. The caller owns its lifecycle (it is NOT added to the
// closer chain). Tests inject deterministic embedders; headless
// embedders inject a pre-built client.
Embedder embeddings.Embedder
// OAuthProviders pre-populates / overrides entries in the provider
// map the catalog Builder consults. Entries here win over
// same-named cfg-built providers; the caller owns their lifecycle.
OAuthProviders map[string]toolauth.OAuthProvider
// PreRegisterTools is registered on the catalog BEFORE the builtin
// registration and the Builder apply, so operator config
// in cfg.Tools.Entries can wrap in-process fixtures.
PreRegisterTools []tools.ToolDescriptor
// MCPDefaultIdentity is the transport-event fallback identity for
// attached MCP servers. Zero value falls
// back to DefaultMCPIdentity.
MCPDefaultIdentity identity.Identity
// MetricsOptions is threaded into telemetry.NewMetricsRegistry.
// The devstack injects an in-process manual reader so the kit does
// not require a metrics-exporter driver per test binary.
MetricsOptions []telemetry.MetricsOption
// TelemetryOptions is threaded into telemetry.New.
// Tests inject telemetry.WithWriter to observe emitted
// records; production callers pass nothing (stdout handler).
TelemetryOptions []telemetry.Option
// TracerOptions is threaded into telemetry.NewTracer.
// Tests inject telemetry.WithSpanExporter with an
// in-memory recorder so derived spans are observable without a
// collector; production callers pass nothing (exporter selection
// follows cfg.Telemetry.OTelEndpoint — noop without a collector).
TracerOptions []telemetry.TracerOption
// ApprovalAuthorizer overrides the resolve-privilege seam threaded
// into every catalog-built ApprovalGate. Nil
// defaults to the runtime-vocabulary
// `approval.NewIdentityAuthorizer()` (originating identity /
// control scope). The serving binary and the devstack inject the
// Protocol-side `server.NewProtocolScopeAuthorizer` so wire-driven
// resolution keeps today\'s admin / console:fleet acceptance.
ApprovalAuthorizer toolapproval.ResolveAuthorizer
// SkipCatalog disables the tool-catalog band (search cache,
// builtins, pause Coordinator, OAuth providers, approval gates, MCP
// attach, executor). Catalog / Coordinator / Gates / OAuthProviders
// / MCPRegistry / Executor are nil. Test-kit convenience; the
// production binary never sets it.
SkipCatalog bool
// SkipSteering disables the steering Registry + planner + RunLoop
// band. Steering / Planner / RunLoop are nil. Test-kit convenience.
SkipSteering bool
// SkipRunLoop disables only the RunLoop construction (the steering
// Registry and planner still build). Test-kit convenience.
SkipRunLoop bool
}
Options carries the injection points the two existing callers need (CLAUDE.md §4.3 / the assembly plan's options-surface rule: the union of what cmd/harbor and harbortest/devstack consume today, not a speculative embedder wishlist).
type RunOption ¶ added in v1.8.0
type RunOption func(*runOnceConfig)
RunOption configures a RunOnce invocation. The functional-option shape keeps RunOnce's signature stable as new per-run knobs land (a streaming sink is a sibling addition, not a signature change).
func WithInputArtifacts ¶ added in v1.8.0
WithInputArtifacts pre-resolves operator-uploaded artifact IDs into the run's first-turn multimodal inputs.
func WithRunID ¶ added in v1.8.0
WithRunID pins the run's RunID instead of synthesising a fresh ULID. Useful for correlating a run with externally-minted identifiers.
func WithStream ¶ added in v1.8.0
func WithStream(sink func(StreamEvent)) RunOption
WithStream registers a per-run streaming sink on a RunOnce invocation. RunOnce still BLOCKS and returns the terminal planner.AnswerEnvelope; the sink receives StreamEvents — token deltas, planner-step boundaries, and tool dispatches — as they occur. Because the underlying callbacks fire synchronously on the run goroutine, every StreamEvent is delivered before RunOnce returns: the ordering is deterministic.
The sink is called on the run goroutine, so it must not block — a blocking sink stalls the run; push to a buffered channel for fan-out. A nil sink is a no-op. Each RunOnce call captures its own sink, so concurrent runs against one shared Stack never cross chunks (the concurrent-reuse contract, CLAUDE.md §5).
WithStream composes with any bus-backed streaming the stack already wires: the sink is additive, so a Console attached to the same run's event bus and an embed sink both observe the run's chunks.
type Stack ¶
type Stack struct {
// Cfg is the *config.Config the stack was assembled from.
Cfg *config.Config
// Redactor / State / Bus / Metrics / Artifacts / Tasks / Sessions /
// Agents are always non-nil after a successful Assemble — the
// runtime's load-bearing core.
Redactor audit.Redactor
State state.StateStore
Bus events.EventBus
Metrics *telemetry.MetricsRegistry
Artifacts artifacts.ArtifactStore
Tasks tasks.TaskRegistry
Sessions *sessions.Registry
Agents *agentregistry.Registry
// Telemetry is the canonical redactor-mandatory structured Logger
// (RFC §6.14), constructed with the bus-paired emitter so
// Logger.Error emits the slog record AND a `runtime.error` bus
// event. Always non-nil after a successful Assemble.
// The Options.Logger slog logger remains the BOOT logger
// for the pre-redactor bootstrap window only; the subsystems the
// assembly constructs after this Logger exists receive its Slog()
// bridge, and request-scoped emission
// goes through this Logger directly.
Telemetry *telemetry.Logger
// Tracer is the canonical OTel tracer (RFC §6.14). Spans are a
// derivation of the event bus: the assembly starts
// telemetry.BridgeBusToTracer alongside the metrics bridge.
// Always non-nil after a successful Assemble; with no collector
// configured the noop exporter drops the spans (in-process
// propagation still works).
Tracer *telemetry.Tracer
// RunErrorHandler is the production engine run-error handler:
// it routes the structured engine.RunError
// through Telemetry.Error so a terminal node failure emits the
// paired `runtime.error` bus event. Flow composition forwards it
// via `flow.WithRunErrorHandler(stack.RunErrorHandler)`.
RunErrorHandler engine.RunErrorHandler
// LLM / LLMSnapshot are populated when cfg.LLM.Driver (or
// Options.LLMSnapshot) names a driver. LLMSnapshot is the resolved
// snapshot the client was opened with — posture surfaces project it.
LLM llm.LLMClient
LLMSnapshot llm.ConfigSnapshot
// KeyRotator is the admin-write seam for Console-driven LLM key
// rotation — it swaps the SAME atomic key holder the opened driver
// reads per call. Populated alongside LLM (nil when no LLM driver is
// configured). The serving binary wires it into the
// `governance.rotate_key` service.
KeyRotator *llm.ProviderKeyRotator
// Embedder is populated when the cfg carries a non-zero
// `embeddings` block (or Options.Embedder is set — caller-owned
// lifecycle). The semantic retrieval modes in Memory / Skills
// consume it via their Deps; it is also directly usable for
// à-la-carte embedding (docs/recipes/embed-and-retrieve.md).
Embedder embeddings.Embedder
// Memory / Skills follow cfg.Memory.Driver / cfg.Skills.Driver
// (Options.SkillStore wins for Skills).
Memory memory.MemoryStore
Skills skills.SkillStore
// Catalog band (nil under SkipCatalog).
Catalog tools.ToolCatalog
Coordinator pauseresume.Coordinator
Gates map[string]*toolapproval.ApprovalGate
OAuthProviders map[string]toolauth.OAuthProvider
MCPRegistry *mcpdrv.Registry
Executor steering.ToolExecutor
// MCPToolContext is the MCP Apps tool-context store the MCP providers
// capture through (input + lowered result behind a declared `ui://`
// app) and the host reads back from for `mcp.apps.tool_context`. Nil
// under SkipCatalog (no MCP providers attach). Wired into every
// Provider so a planner-path app tool call's context is captured.
MCPToolContext *mcpconsole.ToolContextStore
// Steering band (nil under SkipSteering; RunLoop additionally nil
// under SkipRunLoop or when no planner could be resolved).
Steering *steering.Registry
Planner planner.Planner
RunLoop *steering.RunLoop
// Compression is the trajectory-compression runner:
// planner.NewCompressionRunner over the LLM-backed
// summarizer.NewTrajectorySummariser. Non-nil only when
// cfg.Planner.TokenBudget > 0 (and the steering band ran) — the
// per-task run-loop drivers project it onto RunSpec.Compression
// alongside Base.Budget.TokenBudget. Nil = compression off.
Compression *planner.CompressionRunner
// contains filtered or unexported fields
}
Stack is the composed runtime bundle Assemble returns. Fields are nil when the corresponding layer was skipped via Options or not implied by the cfg (LLM / Memory / Skills follow their cfg blocks).
func Assemble ¶
Assemble composes the runtime stack from a validated *config.Config. See the package doc for scope, the lifecycle contract (partial stack on error), and what stays with the caller.
func (*Stack) Close ¶
Close runs every subsystem's Close in reverse dependency order and joins the errors. Idempotent: a second Close is a no-op. Safe on a partial stack returned alongside an Assemble error.
func (*Stack) RunOnce ¶ added in v1.8.0
func (s *Stack) RunOnce( ctx context.Context, goal string, id identity.Identity, opts ...RunOption, ) (planner.AnswerEnvelope, error)
RunOnce drives goal through the assembled planner/run-loop under the mandatory identity triple and returns the terminal answer envelope. It BLOCKS until the run reaches a terminal Finish (or errors). One call replaces the hand-built RunContext + RunSpec + RunLoop.Run + envelope-extraction ceremony an embedder previously wrote per run.
The RunContext is built through runctx.NewRunContext, so the run sees the SAME memory / skills / artifact / streaming projections the dev run-loop drivers compose. Identity is mandatory and fails loud: an incomplete triple returns a wrapped identity error before any work.
On a goal-satisfying Finish the session's memory (when wired) records the turn — best-effort, mirroring the drivers: a writeback error is not fatal to the returned answer. A non-goal Finish is returned with its FinishReason verbatim (the caller maps it via planner.TaskErrorCodeForFinish); a RunLoop error is wrapped and returned.
Streaming: pass WithStream(sink) to observe token deltas, planner-step boundaries, and tool dispatches as they occur. RunOnce still blocks and returns the same envelope; the sink fires synchronously on the run goroutine, so every StreamEvent arrives before RunOnce returns.
Concurrent reuse: RunOnce is safe to call from N concurrent goroutines against a single shared Stack — every invocation builds its own per-run RunContext (counters, trajectory, projections) and reads run-specific data only from its arguments, never from the Stack. The compiled artifacts (RunLoop, Planner, Executor, Catalog, stores) are immutable after Assemble and internally synchronised.
type StreamEvent ¶ added in v1.8.0
type StreamEvent struct {
// Kind discriminates the event (token / tool_dispatched / step).
Kind StreamEventKind
// Text is the incremental output delta for a StreamToken event;
// empty for StreamStep and StreamToolDispatched.
Text string
// Reasoning is true when a StreamToken / StreamStep event rides the
// reasoning ("thinking") channel rather than the answer channel.
Reasoning bool
}
StreamEvent is one observation a WithStream sink receives while RunOnce drives a run. It is intentionally minimal: the streaming surface carries progress signals (token deltas, step + tool boundaries), never raw tool arguments or results.
type StreamEventKind ¶ added in v1.8.0
type StreamEventKind string
StreamEventKind enumerates the streaming-sink event kinds a WithStream sink observes. It is a sealed string enum.
const ( // StreamToken is an incremental output delta from the LLM provider — // the run's answer (or its reasoning trace) as it is generated. Text // carries the delta; Reasoning is true when the delta rides the // reasoning ("thinking") channel rather than the answer channel. StreamToken StreamEventKind = "token" // StreamToolDispatched marks a tool the run dispatched and the // executor returned for WITHOUT error. Text is empty: the dispatch is // the signal, and raw tool arguments/results are never streamed (they // route through the audit redactor — CLAUDE.md §7). StreamToolDispatched StreamEventKind = "tool_dispatched" // StreamStep marks a planner-step boundary — one LLM call finished // streaming. Reasoning distinguishes the reasoning-channel terminal // from the answer-channel terminal. StreamStep StreamEventKind = "step" )