junk

package
v1.2.3 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 15, 2026 License: AGPL-3.0 Imports: 21 Imported by: 0

Documentation

Overview

Package junk holds session-orchestration primitives that are pending extraction into better-defined packages. The intent is to drive the surface area of this package to zero over time.

Package session provides the Stream and Manager primitives that orchestrate per-session inference and session lifecycle in the ore framework.

Stream is a per-session primitive that owns the loop.Step, Thread, TurnProcessor, and provider for a single active conversation. It provides ingress (Submit, Process) and egress (Subscribe) plus lifecycle controls (Cancel, Close).

Submit enqueues an event into the stream's internal FIFO queue and returns immediately. A single worker goroutine processes events one at a time in order, so concurrent submissions are naturally serialized without dropping messages.

The worker goroutine starts lazily on the first Submit or Process call via sync.Once, avoiding goroutine overhead for idle streams. Submit returns an error only if the stream has been closed; it never returns ErrSessionBusy (which has been removed).

Process is a convenience wrapper around Submit that blocks until the worker has finished processing the enqueued event. It is useful when the caller needs to know when a turn is complete (e.g. HTTP handlers that must close an NDJSON response stream).

Manager is a factory/registry for Stream handles. It creates and manages active streams, each pairing a persistent Thread with an ephemeral loop.Step. Applications configure a Manager with a provider, step factory, and cognitive pattern (TurnProcessor). Conduits obtain a *Stream from the Manager (via Create, Attach, or Get) and invoke Submit, Process, Subscribe, Cancel, and Close on that handle, never touching loop.Step directly.

Migration note: the Session interface has been removed. Use *junk.Stream directly. Event types (Event, UserMessageEvent, InterruptEvent) have moved from the conduit package to junk. All event types carry an optional context.Context for routing metadata (e.g., provenance). Set it when constructing the event: 

UserMessageEvent{Content: "hello", Ctx: loop.WithProvenance(context.Background(), "slack-123")}

Lifecycle events: 

LifecycleEvent signals phase transitions for the inference pipeline:
"submitted" (message accepted), "streaming" (first artifact arrived),
and "done" (turn or pipeline complete). Conduits should subscribe to
it to drive UI state without inferring lifecycle from data events.

State persistence is handled automatically by the Manager via loop.WithState, which binds the thread's state to the Step so that TurnCompleteEvent is appended after every turn. Applications only need a stepFactory when adding custom transforms, handlers, or other loop.Step options: 

store := NewMemoryStore()
prov, _ := openai.New(openai.WithAPIKey(apiKey), openai.WithModel(model))
stepFactory := func(stream *Stream) ([]loop.Option, error) {
	// Build transforms that use stream.Metadata, stream.ID, etc.
	return nil, nil  // use default step with auto-persistence
}
mgr := junk.NewManager(store, prov, stepFactory, cognitive.NewTurnProcessor(cognitive.ReActFactory, tracer))

The factory receives *Stream so it can bind per-session runtime ledger. For example, a factory can close over the stream to inject a dynamic system prompt that reads from stream.GetMetadata("persona"): 

stepFactory := func(stream *Stream) ([]loop.Option, error) {
	sp, _ := systemprompt.New(systemprompt.WithContentFunc(func() string {
		if p, ok := stream.GetMetadata("persona"); ok {
			return "You are a " + p + "."
		}
		return "You are a helpful assistant."
	}))
	return []loop.Option{loop.WithTransforms(sp)}, nil
}

// Obtain a *Stream from the manager.
stream, _ := mgr.Create()

// Subscribe to output events via the Stream handle.
ch := stream.Subscribe("text_delta", "turn_complete")

// Submit an event via the Stream handle (non-blocking).
_ = stream.Submit(UserMessageEvent{Content: "hello"})

// HTTP conduit composes with the Manager. UI is enabled by default.
c, _ := httpc.New(mgr, httpc.WithAddr(":8080"))
_ = c.Start(ctx)

// TUI conduit composes with the Manager.
tuiConduit, _ := tui.New(mgr)
_ = tuiConduit.Start(ctx)

// Retrieve the conversation history without exposing the internal Thread.
turns := stream.Turns()

// Emit custom output events (e.g. status updates) into the stream's FanOut.
_ = stream.Emit(ctx, loop.PropertiesEvent{Operations: []loop.PropertyOperation{{Op: loop.PropertyOpSet, Key: "thread_id", Value: stream.ID()}}})

Index

Constants

View Source 
const (
	MetadataKeyModelName            = "ore.model.name"
	MetadataKeyModelThinkingLevel   = "ore.model.thinking_level"
	MetadataKeyModelTemperature     = "ore.model.temperature"
	MetadataKeyModelMaxOutputTokens = "ore.model.max_output_tokens"
)

Model metadata keys. These are the framework contract keys that the slash command in x/tool/set_model (and any other session-level model configurator) writes into Thread.Metadata. The session package owns the canonical form; downstream tools import the constants when they need to set a value, or write the literal string. The "ore.model." prefix reserves the namespace for framework-level model configuration.

Variables

View Source 
var ErrThreadCorrupt = errors.New("thread file corrupt")

ErrThreadCorrupt is the sentinel wrapped by Store.Get and Store.GetBy when a thread file exists but cannot be parsed. The underlying error is the actual unmarshal failure; callers can recover it via errors.Unwrap or errors.Is.

Use errors.Is(err, ErrThreadCorrupt) to detect this case. The wrapped error itself is constructed with fmt.Errorf("...: %w: %w", ErrThreadCorrupt, underlyingErr) so that errors.Is succeeds against either target.

View Source 
var ErrThreadNotFound = errors.New("thread not found")

ErrThreadNotFound is returned by Store.Get and Store.GetBy when no thread matches the requested identifier (UUID for Get, metadata key-value pair for GetBy). Callers can distinguish this from a parse failure with errors.Is.

Functions

This section is empty.

Types

type Event 

type Event interface {
	Kind() string
	Context() context.Context
}

Event is the base interface for all ingress events to a session Stream. Custom event types can be defined in other packages by implementing the public Kind() and Context() methods.

type InterceptResult 

type InterceptResult struct {
	// Event is the replacement event to continue processing. If nil, the
	// original event is consumed and no further processing occurs.
	Event Event
	// Notice is the list of ephemeral, user-visible messages emitted as
	// loop.NoticeEvent after Intercept returns.
	Notice []loop.Notice
}

InterceptResult is the result of an interceptor processing an event.

Notice carries ephemeral, user-visible messages that are emitted as loop.NoticeEvent and never persisted to ledger. Each Notice carries a Severity that conduits use to pick a rendering style (Success, Info, Warn, Error). A nil slice means no notices were produced.

type Interceptor 

type Interceptor interface {
	Intercept(ctx context.Context, event Event, stream *Stream, emitter loop.Emitter) (InterceptResult, error)
}

Interceptor processes events before they enter the LLM pipeline. It receives a junk.Event, the active *junk.Stream, and a loop.Emitter for signaling activity, and can either:

  • Return a non-nil Event in the result to rewrite the event
  • Return a nil Event in the result to consume the event (no further processing)
  • Return an error to abort processing

Notice messages are ephemeral UI messages that are not persisted to ledger. Each Notice carries a Severity so conduits can pick a rendering style.

The *junk.Stream parameter is the stream that owns the in-flight event, so interceptors can call stream-scoped methods like SetMetadata. It is never nil for a configured interceptor — stream.processOne passes the active stream unconditionally.

type InterceptorFunc 

type InterceptorFunc func(ctx context.Context, event Event, stream *Stream, emitter loop.Emitter) (InterceptResult, error)

InterceptorFunc is a function type that implements Interceptor.

func (InterceptorFunc) Intercept 

func (f InterceptorFunc) Intercept(ctx context.Context, event Event, stream *Stream, emitter loop.Emitter) (InterceptResult, error)

Intercept delegates to the function.

type InterruptEvent 

type InterruptEvent struct {
	// Ctx carries the provenance/context metadata for the interrupt event.
	Ctx context.Context
}

InterruptEvent represents the user interrupting the current operation.

func (InterruptEvent) Context 

func (e InterruptEvent) Context() context.Context

Context returns the event's context.Context metadata.

func (InterruptEvent) Kind 

func (e InterruptEvent) Kind() string

Kind returns the event kind identifier.

type JSONStore 

type JSONStore struct {
	// contains filtered or unexported fields
}

JSONStore persists threads as individual JSON files in a directory.

func NewJSONStore 

func NewJSONStore(dir string) (*JSONStore, error)

NewJSONStore creates a new JSONStore backed by the given directory. The directory is created if it does not exist. Thread data is loaded lazily on first access via Get, List, or GetBy.

Malformed or unreadable .json files are silently skipped during access.

func (*JSONStore) Create 

func (s *JSONStore) Create() (*Thread, error)

Create generates a new Thread with a random ID and persists it.

func (*JSONStore) Delete 

func (s *JSONStore) Delete(id string) bool

Delete removes a thread from the cache and deletes its file.

func (*JSONStore) Get 

func (s *JSONStore) Get(id string) (*Thread, error)

Get retrieves a thread by ID.

Return values:

  • (thread, nil): thread found and loaded successfully.
  • (nil, ErrThreadNotFound): no thread with that ID exists.
  • (nil, ErrThreadCorrupt wrapping cause): a thread file exists at the expected path but cannot be parsed. The wrapped error is the underlying unmarshal failure so callers can introspect it.

The previous (thread, bool) signature silently returned (nil, false) for both "not found" and "corrupt", which made any parse failure indistinguishable from a missing file. Callers could not log or report the real cause. See issue #453.

func (*JSONStore) GetBy 

func (s *JSONStore) GetBy(key, value string) (*Thread, error)

GetBy retrieves a thread by a metadata key-value pair. It scans all thread files on disk and returns the first match.

If the scan encounters a thread file that exists but cannot be parsed, the parse error is returned wrapped in ErrThreadCorrupt. A missing thread file is silently skipped (matching the previous behavior; a single missing file in the scan is not the same as "the requested thread does not exist"). If the scan completes with no match, ErrThreadNotFound is returned.

func (*JSONStore) List 

func (s *JSONStore) List() ([]*Thread, error)

List returns all threads in the store.

Corrupt files (files that exist but cannot be parsed) are silently skipped, matching the long-standing behavior of this method. Direct lookups via Get surface corruption as ErrThreadCorrupt; bulk lookups via List intentionally do not, because a single corrupt file in the directory shouldn't prevent the caller from seeing the rest of the threads. Callers that need to find every corrupt file must walk the directory themselves.

func (*JSONStore) Save 

func (s *JSONStore) Save(thread *Thread) error

Save writes the thread to disk atomically (via a temporary file and os.Rename) and updates the in-memory cache. The thread's UpdatedAt timestamp is also advanced.

type Manager 

type Manager struct {
	// contains filtered or unexported fields
}

Manager owns the Thread↔Step binding and acts as a factory/registry for Stream handles.

func NewManager 

func NewManager(store Store, prov provider.Provider, newStep func(*Stream) ([]loop.Option, error), processor TurnProcessor, opts ...ManagerOption) *Manager

NewManager creates a new Manager with the given dependencies.

func (*Manager) Attach 

func (m *Manager) Attach(threadID string) (*Stream, error)

Attach gets or creates an active stream for an existing thread.

Returns an error if the thread does not exist in the store, if the stored thread file is corrupt (Store returned ErrThreadCorrupt), or if the step factory fails. The "thread not found" case is the most common; the caller (typically a conduit's Start path) surfaces the wrapped error to the user with the appropriate context.

func (*Manager) Close 

func (m *Manager) Close(sessionID string) error

Close closes a stream and removes it from the active map. The underlying thread is NOT deleted from the store.

func (*Manager) Create 

func (m *Manager) Create() (*Stream, error)

Create creates a new thread and an active stream backed by it. Returns an error if the underlying thread store cannot create a thread or if the step factory fails.

func (*Manager) CreateWithID 

func (m *Manager) CreateWithID(id string) (*Stream, error)

CreateWithID creates a new thread with the given ID and an active stream backed by it.

func (*Manager) Get 

func (m *Manager) Get(sessionID string) (*Stream, error)

Get returns the active Stream for a given session ID. An error is returned if the stream does not exist.

func (*Manager) GetBy 

func (m *Manager) GetBy(key, value string) (*Thread, error)

GetBy retrieves a thread by a metadata key-value pair. The returned error is the Store's error (ErrThreadNotFound, ErrThreadCorrupt wrapping cause, or any other implementation-specific error). The previous (Thread, bool) signature conflated these cases; see issue #453.

func (*Manager) GetThread 

func (m *Manager) GetThread(id string) (*Thread, error)

GetThread retrieves a thread by ID. The returned error is the Store's error; see GetBy.

func (*Manager) List 

func (m *Manager) List() []*Stream

List returns handles for all active streams.

func (*Manager) ListThreads 

func (m *Manager) ListThreads() ([]*Thread, error)

ListThreads returns all stored threads.

func (*Manager) RegisterSink 

func (m *Manager) RegisterSink(kinds []string, fn SinkFunc) func()

RegisterSink registers a callback that receives OutputEvents from all active and future streams matching the given kinds. An empty kinds slice means all event kinds. Registration also triggers forwarding for any streams that already exist at the time of registration.

It returns a function that removes the sink from the manager. Calling the returned function more than once is safe and idempotent.

type ManagerOption 

type ManagerOption func(*Manager)

ManagerOption configures a Manager.

func WithDefaultMetadata 

func WithDefaultMetadata(fn func(*Stream) map[string]string) ManagerOption

WithDefaultMetadata sets a function that provides default metadata for every stream at creation/attachment time. The function receives the stream so it can include the thread ID or other stream-specific values.

func WithInterceptor 

func WithInterceptor(interceptor Interceptor) ManagerOption

WithInterceptor sets an interceptor that processes UserMessageEvent events before they enter the LLM inference pipeline for every stream managed by this Manager.

type MemoryStore 

type MemoryStore struct {
	// contains filtered or unexported fields
}

MemoryStore is an in-memory Store implementation.

func NewMemoryStore 

func NewMemoryStore() *MemoryStore

NewMemoryStore creates a new empty in-memory store.

func (*MemoryStore) Create 

func (s *MemoryStore) Create() (*Thread, error)

Create generates a new Thread with a random ID and stores it.

func (*MemoryStore) Delete 

func (s *MemoryStore) Delete(id string) bool

Delete removes a thread from the store.

func (*MemoryStore) Get 

func (s *MemoryStore) Get(id string) (*Thread, error)

Get retrieves a thread by ID.

Returns ErrThreadNotFound if no thread with the given ID is stored. The previous (Thread, bool) signature has been replaced so callers can distinguish a miss from a corruption; see issue #453.

func (*MemoryStore) GetBy 

func (s *MemoryStore) GetBy(key, value string) (*Thread, error)

GetBy retrieves a thread by a metadata key-value pair. It performs a linear scan over all threads and returns the first match.

Returns ErrThreadNotFound if no thread matches the key-value pair. In-memory corruption is not possible (no disk reads), so this implementation never returns ErrThreadCorrupt.

func (*MemoryStore) List 

func (s *MemoryStore) List() ([]*Thread, error)

List returns all threads in the store.

func (*MemoryStore) Save 

func (s *MemoryStore) Save(thread *Thread) error

Save updates the thread's UpdatedAt and stores it.

type SessionSwitchEvent 

type SessionSwitchEvent struct {
	SessionID string
	Ctx       context.Context
}

SessionSwitchEvent signals a cross-session navigation.

func (SessionSwitchEvent) Context 

func (e SessionSwitchEvent) Context() context.Context

Context returns the event's context.Context metadata.

func (SessionSwitchEvent) Kind 

func (e SessionSwitchEvent) Kind() string

Kind returns the event kind identifier.

func (SessionSwitchEvent) MarshalJSON 

func (e SessionSwitchEvent) MarshalJSON() ([]byte, error)

MarshalJSON serializes the event to JSON. It includes the session_id and an optional context envelope with provenance and traceparent.

type SinkFunc 

type SinkFunc func(streamID string, event loop.OutputEvent)

SinkFunc receives OutputEvents from a specific stream, together with the originating stream ID. It may be invoked concurrently for multiple streams.

type Store 

type Store interface {
	// Create generates a new Thread with a random UUID and stores it.
	Create() (*Thread, error)
	// Get retrieves a Thread by ID. Returns ErrThreadNotFound if no
	// such thread exists, or an error wrapping ErrThreadCorrupt if
	// the thread file is present but unparseable.
	Get(id string) (*Thread, error)
	// GetBy retrieves a Thread by a metadata key-value pair.
	// Returns ErrThreadNotFound if no thread matches, or an error
	// wrapping ErrThreadCorrupt if any thread file encountered during
	// the scan is unparseable.
	GetBy(key, value string) (*Thread, error)
	// Save persists the given Thread, updating its UpdatedAt timestamp.
	Save(thread *Thread) error
	// Delete removes a Thread by ID and returns true if it existed.
	Delete(id string) bool
	// List returns all stored Threads.
	List() ([]*Thread, error)
}

Store abstracts persistence for Thread instances. Implementations must be safe for concurrent use.

Get and GetBy return errors so callers can distinguish a missing thread from a corrupt one. The previous (Thread, bool) signature conflated the two cases, which made any parse failure indistinguishable from "no such thread" and prevented useful diagnostics. Implementations return ErrThreadNotFound for the former and wrap their cause in ErrThreadCorrupt for the latter. See issue #453.

type Stream 

type Stream struct {
	// contains filtered or unexported fields
}

Stream is a per-session primitive that owns the loop.Step, Thread, TurnProcessor, and provider for a single active conversation. It provides ingress (Process, Submit) and egress (Subscribe) for the session, plus lifecycle controls (Cancel, Close).

Events submitted via Submit() are enqueued in an unbounded FIFO queue and processed serially by a single internal worker goroutine. Process() also enqueues but blocks until the event has been fully processed.

func (*Stream) AllMetadata 

func (s *Stream) AllMetadata() map[string]string

AllMetadata returns a defensive copy of the thread's metadata map. Mutating the returned map does not affect the underlying thread. The stream mutex is held for the duration of the copy so a concurrent SetMetadata cannot race with the read. Conduits that need to seed their view at Start time (e.g. the TUI status bar) should call AllMetadata() before stream.Subscribe, since Subscribe is live-only and does not replay historical PropertiesEvents.

func (*Stream) AppendTurn 

func (s *Stream) AppendTurn(ctx context.Context, role ledger.Role, artifacts ...artifact.Artifact) error

AppendTurn records a turn into the thread state and broadcasts a TurnCompleteEvent to all subscribers of the stream. It is the canonical entry point for external producers (slash handlers, compaction, future tool-side producers) that need to inject turns without going through step.Turn / step.Submit.

Behavior:

  • The TurnCompleteEvent is emitted via s.step.Emit. The stream's default OnEmit (registered by the Manager) appends the turn to thread.State synchronously, before the event reaches the async FanOut. Subscribers see the event with the canonical state already updated, so live consumers (TUI, telemetry) do not need to call ReloadHistory to observe the new turn.
  • The thread is NOT auto-saved. Callers that want the change to persist to the store should call Save() explicitly, or rely on the stream's normal save flow after the next Process() call. This matches the rest of the Stream's contract: state mutations are persisted by the inference pipeline, not by ad-hoc producers.

Errors:

  • "session %s is closed" if the stream has been closed.

func (*Stream) Cancel 

func (s *Stream) Cancel() error

Cancel aborts an ongoing turn by cancelling its context.

func (*Stream) Close 

func (s *Stream) Close() error

Close closes the stream's Step and marks it as closed. The underlying thread is NOT deleted from the store.

Close cancels any in-flight turn, waits for the worker goroutine to exit, and then closes the step. This ensures all pending Process calls receive errors and no goroutine leak occurs.

func (*Stream) DeleteMetadata added in v1.2.2 

func (s *Stream) DeleteMetadata(key string)

DeleteMetadata removes a metadata key from the underlying thread and emits a PropertiesEvent carrying a single PropertyOpDelete operation. Deleting a non-present key is a no-op for both the thread state (Go's delete() on a missing key returns the zero value) and the receiving state in conduits (they apply deletes via the same path the TUI uses in Task 4).

DeleteMetadata exists to fulfill issue #531: prior to this method, callers could not represent key removal through the property event protocol without collapsing "absent" onto "present-with-empty-value".

func (*Stream) Emit 

func (s *Stream) Emit(ctx context.Context, event loop.OutputEvent) error

Emit injects a custom output event into the stream's FanOut, allowing handlers, interceptors, and application logic to emit meta-events that are delivered to all subscribers alongside standard artifact and turn-complete events.

The stream must not be closed.

Errors:

  • "session %s is closed" if the stream has been closed

func (*Stream) GetMetadata 

func (s *Stream) GetMetadata(key string) (string, bool)

GetMetadata retrieves a metadata value from the underlying thread.

func (*Stream) ID 

func (s *Stream) ID() string

ID returns the stream's unique identifier (same as the thread ID).

func (*Stream) LoadTurns 

func (s *Stream) LoadTurns(turns []ledger.Turn)

LoadTurns replaces the thread's turn state with the provided slice. It acquires the stream's mutex to ensure thread-safe state mutation.

This is the legacy replacement path; the new Thread is tree-backed, so callers migrating from a linear Buffer-backed state should construct a fresh ledger.Thread with each turn's ParentID set explicitly.

func (*Stream) MarkBoundary 

func (s *Stream) MarkBoundary(summaryTurnID, info string) error

MarkBoundary records a compaction boundary by stamping the summary turn with ledger.ControlStop. The walk (ledger.Thread.ResolveActivePath) then terminates at the summary, hiding everything that came before it from the LLM-facing view.

summaryTurnID is the ID of the summary turn (already appended to the thread's state via AppendTurn or equivalent). info is the JSON-encoded BoundaryInfo produced by [Summarize]; it is persisted to Thread.Metadata under "ore.compaction.boundary.info" so TUI/audit consumers can read it without re-deriving from the LLM-facing summary text.

MarkBoundary is the canonical way to record a compaction. The caller is responsible for appending the summary turn (via AppendTurn or ledger.Thread.Append) before calling MarkBoundary; the function does not mutate the turn list.

MarkBoundary holds the stream mutex for the duration of the metadata and control writes. The serial-pipeline-only contract applies.

func (*Stream) Process 

func (s *Stream) Process(ctx context.Context, event Event) error

Process enqueues the event and blocks until the worker has finished processing it. Context cancellation aborts the waiting, not the in-flight turn; use Cancel() to abort a running turn.

After the TurnProcessor returns (including all tool-call loops), Process emits a LifecycleEvent{Phase: "done"} to signal pipeline completion before performing save cleanup. Subscribers can use this event for lifecycle signalling (audio notifications, UI state finalization).

Errors:

  • "session %s is closed" if the stream has been closed
  • "unsupported event kind" for unknown event types
  • "process event: ..." wrapping any TurnProcessor or save error

func (*Stream) Save 

func (s *Stream) Save() error

func (*Stream) SetMetadata 

func (s *Stream) SetMetadata(key, value string)

SetMetadata sets a metadata value on the underlying thread.

func (*Stream) Spec 

func (s *Stream) Spec() (models.Spec, bool)

Spec derives a models.Spec from the thread's metadata. The bool result is false when no recognized metadata keys are set; the caller should use the loop's default in that case. The framework does not merge metadata into a base Spec; the session is responsible for producing the effective Spec for each turn.

Recognized keys (all private to this package; documented here as the framework contract): 

"ore.model.name"              → Spec.Name
"ore.model.thinking_level"    → Spec.ThinkingLevel
"ore.model.temperature"       → Spec.Temperature (parsed float)
"ore.model.max_output_tokens" → Spec.MaxOutputTokens (parsed int)

Unknown keys are ignored. The slash command in x/tool/set_model is the canonical writer; see MetadataKeyModelName for the "set the model" entry point.

func (*Stream) State 

func (s *Stream) State() ledger.State

State returns the thread's mutable conversation ledger. The handle is the same State the loop pipeline uses (via loop.WithState); reads observe the current turn history, and writes through the returned Meta propagate to subsequent reads.

As with the rest of the State interface, the returned handle is not safe for concurrent use; the stream serializes access to its own turns and metadata, but the State object itself shares the Buffer's "serial pipeline only" contract.

func (*Stream) Submit 

func (s *Stream) Submit(event Event) error

Submit enqueues the event in the stream's unbounded FIFO queue and returns immediately. A single internal worker goroutine drains the queue and processes events serially, one at a time.

InterruptEvent clears all pending events from the queue before being enqueued itself, and cancels any in-flight turn via Cancel().

Errors:

  • "session %s is closed" if the stream has been closed

func (*Stream) Subscribe 

func (s *Stream) Subscribe(kinds ...string) <-chan loop.OutputEvent

Subscribe returns a filtered output event channel for the stream's loop.Step FanOut. If no kinds are provided, the channel receives all events regardless of kind. If the stream is closed, the returned channel is immediately closed.

Subscribe is live-only: it delivers events from the point of subscription onward and does not replay historical events. Conduits that need historical state should fetch it via Turns() before subscribing, or load it via LoadTurns() after external mutations (e.g. compaction).

The returned channel is closed when the session is closed. Callers should range over the channel and handle closure: 

ch := stream.Subscribe("text_delta", "turn_complete")
for event := range ch {
    // process event
}

func (*Stream) Turns 

func (s *Stream) Turns() []ledger.Turn

Turns returns a defensive (shallow) copy of the thread's turn history. The slice of Turns is copied, but each Turn's Artifacts slice is shared. Callers should treat the returned artifacts as immutable.

type Thread 

type Thread struct {
	// ID is the unique identifier for this thread (random UUID).
	ID string
	// State holds the mutable thread turn history.
	State *ledger.Thread
	// Metadata holds arbitrary key-value pairs for conduit-specific
	// thread mapping (e.g., external system identifiers).
	Metadata map[string]string
}

Thread represents a persistent thread with identity, state, and metadata.

State is the tree-backed ledger.Thread that owns the turn history. The Metadata field is conduit-specific (e.g., mapping to external system identifiers) and is independent of the per-conversation ledger.Meta().

ID is the address by which the Store locates this thread (e.g., the on-disk filename). It is assigned by the Store at creation time and is not stored on the inner ledger.Thread (per the "ID is external" principle). CreatedAt and UpdatedAt are no longer persisted; turn timestamps are the conversation's temporal data.

func (*Thread) GetMetadata 

func (c *Thread) GetMetadata(key string) (string, bool)

GetMetadata retrieves a metadata value from the thread.

func (*Thread) MarshalJSON 

func (c *Thread) MarshalJSON() ([]byte, error)

MarshalJSON serializes the thread to JSON. The format is the threadJSON shape above.

func (*Thread) SetMetadata 

func (c *Thread) SetMetadata(key, value string)

SetMetadata sets a metadata value on the thread.

func (*Thread) UnmarshalJSON 

func (c *Thread) UnmarshalJSON(data []byte) error

UnmarshalJSON deserializes the thread from JSON. The State is reconstructed as a fresh ledger.Thread with all turns populated and the CurrentTip set. CreatedAt/UpdatedAt are not stored in the new format and are left at their zero values.

type TurnProcessor 

type TurnProcessor func(ctx context.Context, step *loop.Step, st ledger.State, prov provider.Provider, spec models.Spec) (ledger.State, error)

TurnProcessor runs the full inference pipeline for a single turn after the user event has been submitted to ledger. It is called with the stream's loop.Step, state, provider, and a models.Spec derived from the session's metadata (or the empty zero value when no metadata is set, in which case the step's configured default spec applies).

type UserMessageEvent 

type UserMessageEvent struct {
	// Content holds the raw user message text.
	Content string

	// Ctx carries the provenance/context metadata for the user message.
	Ctx context.Context
}

UserMessageEvent represents the user submitting a text message.

func (UserMessageEvent) Context 

func (e UserMessageEvent) Context() context.Context

Context returns the event's context.Context metadata.

func (UserMessageEvent) Kind 

func (e UserMessageEvent) Kind() string

Kind returns the event kind identifier.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.