application

package
v0.21.0 Latest Latest
Warning

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

 Go to latest Published: Oct 5, 2026 License: AGPL-3.0 Imports: 94 Imported by: 0

Documentation

Overview

Package application orchestrates agent turns, including context assembly, persistence, memory, compaction, approvals, and runtime invocation.

Index

Constants

View Source 
const UserMessageKindSkillActivation = turn.UserMessageKindSkillActivation

Variables

View Source 
var ErrDirectModelUnavailable = errors.New("direct runtime model unavailable")
View Source 
var ErrExternalAgentWorkspaceTargetUnsupported = errors.New("workspace_target_id is not supported for external agent sessions")
View Source 
var ErrExternalForkAnchorMissing = errors.New("fork source turn has no runtime turn anchor")

ErrExternalForkAnchorMissing reports a fork anchored at a turn that recorded no runtime turn id, so the runtime-side cut cannot match the visible history.

View Source 
var ErrExternalForkUnsupported = errors.New("runtime does not support forking")

ErrExternalForkUnsupported reports a fork request against an external runtime that cannot fork.

View Source 
var ErrModelPreferenceConflict = errors.New("session model preference changed")

ErrModelPreferenceConflict means a picker read predates a newer write.

View Source 
var ErrQueueInputIncomplete = errors.New("queue: input is missing team, bot, or session identity")

ErrQueueInputIncomplete reports an ingress that admitted queue input without the identity the continuation needs. It is a wiring error at the ingress, not a rejected user request, so handlers surface it as unavailable.

View Source 
var ErrQueueItemNotEditable = errors.New("queue: only the sender can edit a queued item")

ErrQueueItemNotEditable reports an edit by anyone other than the sender the item recorded. A queued command runs as that sender, with their attribution, credentials, and memory, so rewriting its text is theirs alone; others with queue access can still cancel or reorder it.

View Source 
var ErrWorkspaceTargetWorkdirConflict = errors.New(
	"this session is bound to a workdir; its workspace target cannot be changed")

ErrWorkspaceTargetWorkdirConflict is returned when a request tries to move a workdir-bound session onto a different workspace target. The binding is immutable: the workdir's directory only exists on the workdir's target, so honoring the switch would produce a session whose working directory does not exist.

Functions

func AutoCompactionThreshold added in v0.20.0 

func AutoCompactionThreshold(userThreshold, contextTokenBudget int) int

AutoCompactionThreshold is the async trigger level, exported so read-side surfaces label the same level the turn path acts on. A zero return leaves automatic compaction off when no usable model window is available. The synchronous backstop is deliberately not exposed: it only runs on the history path, which a reader cannot observe.

func ExternalAgentError added in v0.21.0 

func ExternalAgentError(err error) error

ExternalAgentError translates an External Agent failure the user can act on into its public error: a runtime failure ExternalRuntimeError translates, an agent that is unknown, disabled or not set up, a runtime without its owner, a missing workspace dependency, a workspace that is not a container, or input the agent cannot take.

An error that already carries a catalog code, and an error that is none of these, is returned unchanged.

func ExternalRuntimeError added in v0.21.0 

func ExternalRuntimeError(err error) error

ExternalRuntimeError translates the failures a runtime driver reports as an external.Failure or an ACP PromptError into their public errors. Any other error is returned unchanged, so a caller that gives the remaining failures its own code (a device login, a credential purge) applies it to exactly the errors it applied it to before.

func QueuePayloadText added in v0.20.0 

func QueuePayloadText(payload []byte) string

QueuePayloadText renders only user-visible text. Invalid/empty payloads never fall back to the raw envelope, which can contain a deferred command credential.

func RuntimeControlError added in v0.21.0 

func RuntimeControlError(err error) error

RuntimeControlError translates a runtime control failure into its public error. It is the single sentinel table for every transport that serves runtime controls: the service methods apply it before returning, and HTTP paths that reach a driver without the service apply it themselves.

An error that already carries a catalog code is returned unchanged, so a second translation never masks the first one. An External Agent failure keeps the code ExternalAgentError gives it.

func SubagentRunHandleFromContext 

func SubagentRunHandleFromContext(ctx context.Context) (sessionruntime.RunHandle, bool)

SubagentRunHandleFromContext returns the run handle AdmitSubagentRun stored on the context, if the context belongs to an admitted subagent run.

func WithRunOutcome added in v0.21.0 

func WithRunOutcome(ctx context.Context, runID string, outcome RunOutcome) context.Context

WithRunOutcome carries the owner's outcome of run runID to the run's terminal observer. The owner passes the returned context to its terminal write, so the run's result record can report the cause the owner held; the terminal state itself only carries a code. Detaching the context from its cancellation keeps the value, so a durable finish retry still reports it.

The value names its run because contexts outlive runs: a follow-up run started from a terminal observation inherits the finished run's context.

Types

type AssistantOutput 

type AssistantOutput = turn.AssistantOutput

The canonical turn value objects are re-exported inside application so the orchestration code can use them without maintaining a second DTO family.

type BackgroundTaskNotifications added in v0.20.0 

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

BackgroundTaskNotifications persists dependency lifecycle feedback in the explicitly selected conversation and delivers it to that conversation's channel. Script output stays in the Manage-only operation log.

func NewBackgroundTaskNotifications added in v0.20.0 

func NewBackgroundTaskNotifications(
	messages messagepkg.Writer,
	sessions backgroundNotificationSessions,
	deliver func(context.Context, session.Thread, string) error,
) *BackgroundTaskNotifications

func (*BackgroundTaskNotifications) Handle added in v0.20.0 

func (n *BackgroundTaskNotifications) Handle(ctx context.Context, evt background.TaskEvent) error

Handle ignores unscoped tasks: installing a bot dependency must never notify every historical conversation belonging to the bot. Callers authorize the optional session at operation admission; delivery verifies its bot again.

type ChatAttachment 

type ChatAttachment = turn.Attachment

The canonical turn value objects are re-exported inside application so the orchestration code can use them without maintaining a second DTO family.

type ChatRequest 

type ChatRequest struct {
	// ShutdownResume rebuilds saved history without the transient channel pipeline.
	ShutdownResume bool `json:"-"`
	// OnModelPreferenceSettled releases subsequent picker writes once this
	// turn can no longer overwrite them. It does not acknowledge generation.
	OnModelPreferenceSettled func() `json:"-"`

	BotID    string `json:"-"`
	ChatID   string `json:"-"`
	ThreadID string `json:"-"`
	// RunID is the server-minted identity of this turn's run. Downstream state
	// that must agree on "which turn is this" — the compaction barrier, the ACP
	// session, interactive tool headers — keys on it.
	RunID string `json:"-"`
	// RunHandle is the server-owned execution capability for durable step and
	// queue commits. Transport callers cannot supply it.
	RunHandle sessionruntime.RunHandle `json:"-"`
	// TurnID and TurnPosition are the turn admission already allocated for this
	// run. They travel with the request so the persisted user turn lands under
	// the id the client was handed at run_accepted, instead of the history layer
	// minting a second one (SR-TURN-001). Empty means no admission decided the
	// turn — channel inbound and schedules — and history allocates it.
	TurnID                    string                `json:"-"`
	TurnPosition              *int64                `json:"-"`
	Token                     string                `json:"-"`
	UserID                    string                `json:"-"`
	SourceChannelIdentityID   string                `json:"-"`
	DisplayName               string                `json:"-"`
	RouteID                   string                `json:"-"`
	ChatToken                 string                `json:"-"`
	ExternalMessageID         string                `json:"-"`
	ReplyTarget               string                `json:"-"`
	ConversationType          string                `json:"-"`
	ConversationName          string                `json:"-"`
	SourceReplyToMessageID    string                `json:"-"`
	ReplySender               string                `json:"-"`
	ReplyPreview              string                `json:"-"`
	ReplyAttachments          []turn.Attachment     `json:"-"`
	MentionsBot               bool                  `json:"-"`
	RepliesToBot              bool                  `json:"-"`
	ForwardMessageID          string                `json:"-"`
	ForwardFromUserID         string                `json:"-"`
	ForwardFromConversationID string                `json:"-"`
	ForwardSender             string                `json:"-"`
	ForwardDate               int64                 `json:"-"`
	UserMessagePersisted      bool                  `json:"-"`
	PersistedUserMessageID    string                `json:"-"`
	ReusePersistedUserMessage bool                  `json:"-"`
	EventID                   string                `json:"-"`
	RawQuery                  string                `json:"-"`
	ModelQuery                string                `json:"-"`
	UserMessageKind           string                `json:"-"`
	UserVisibleText           string                `json:"-"`
	SkillActivation           *turn.SkillActivation `json:"-"`
	ToolHTTPURL               string                `json:"-"`
	SessionType               string                `json:"-"`
	RuntimeType               string                `json:"-"`
	SkipMemoryExtraction      bool                  `json:"-"`
	SkipHistoryTurn           bool                  `json:"-"`
	// TurnReplacement is set only for an admitted retry/edit run. Its step
	// output stays hidden until the queue coordinator reaches the true final
	// boundary and publishes this replacement in its own transaction.
	TurnReplacement     *messagepkg.TurnReplacement `json:"-"`
	SkipTitleGeneration bool                        `json:"-"`
	ForceFreshRuntime   bool                        `json:"-"`
	// AgentCommand is the exact agent-command selector the Web admission layer
	// matched against a live ACP runtime. The session pool re-validates it
	// against the final session at prompt time; it never crosses the turn
	// transport (Web admission runs in-process).
	AgentCommand                 string           `json:"-"`
	HistoryCutoffBeforeMessageID string           `json:"-"`
	RequiredHistoryMessageID     string           `json:"-"`
	WorkspaceTarget              *WorkspaceTarget `json:"-"`

	// OutboundAssetCollector returns asset refs accumulated during outbound
	// streaming. It is never serialized across the turn transport.
	OutboundAssetCollector func() []turn.OutboundAssetRef `json:"-"`

	// InjectCh receives user messages between tool rounds. Remote transports
	// use turn.RunHandle.Inject instead.
	InjectCh <-chan turn.InjectMessage `json:"-"`
	// QueueSteerEnabled enables the fenced native queue consumer.
	QueueSteerEnabled bool `json:"-"`
	StepIndexOffset   int  `json:"-"`
	// PublishRuntimeEvents is set for server-owned continuations, which do not
	// have a client runHandle pump to publish native events into the session
	// runtime projection.
	PublishRuntimeEvents bool `json:"-"`

	Query             string                       `json:"query"`
	Model             string                       `json:"model,omitempty"`
	Provider          string                       `json:"provider,omitempty"`
	ReasoningEffort   string                       `json:"reasoning_effort,omitempty"`
	WorkspaceTargetID string                       `json:"workspace_target_id,omitempty"`
	Channels          []string                     `json:"channels,omitempty"`
	CurrentChannel    string                       `json:"current_channel,omitempty"`
	Messages          []turn.ModelMessage          `json:"messages,omitempty"`
	Attachments       []turn.Attachment            `json:"attachments,omitempty"`
	RequestedSkills   []turn.RequestedSkillContext `json:"-"`
}

ChatRequest is the application-layer input used while orchestrating a chat turn. Transport callers should prefer turn.StartTurnCommand; the additional channel and function fields below are strictly in-process runtime state.

type ChatResponse 

type ChatResponse struct {
	Messages []turn.ModelMessage `json:"messages"`
	Model    string              `json:"model,omitempty"`
	Provider string              `json:"provider,omitempty"`
}

ChatResponse is the output of a non-streaming application call.

type CommittedToolApprovalResponse 

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

type CommittedUserInputResponse 

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

CommittedUserInputResponse is the durable half of an ask_user response. Keeping it separate from the continuation lets the runtime acknowledge the user's click as soon as the decision commits, without imposing the command acknowledgement deadline on the following model call.

type ContentPart 

type ContentPart = turn.ContentPart

The canonical turn value objects are re-exported inside application so the orchestration code can use them without maintaining a second DTO family.

type EditLatestMessageInput 

type EditLatestMessageInput struct {
	BotID        string
	SessionID    string
	RunID        string
	TurnID       string
	TurnPosition *int64
	// TargetTurnID names the round being replaced. See RetryLatestMessageInput.
	TargetTurnID           string
	Text                   string
	Attachments            []ChatAttachment
	ActorChannelIdentityID string
	ActorUserID            string
	ChatToken              string
	Model                  string
	ReasoningEffort        string
	WorkspaceTargetID      string
	ToolHTTPURL            string
	RunHandle              sessionruntime.RunHandle
	InjectCh               chan turnpkg.InjectMessage
	// OnModelPreferenceSettled: see RetryLatestMessageInput.
	OnModelPreferenceSettled func()
}

type InjectMessage 

type InjectMessage = turn.InjectMessage

The canonical turn value objects are re-exported inside application so the orchestration code can use them without maintaining a second DTO family.

type InjectedMessageRecord 

type InjectedMessageRecord struct {
	HeaderifiedText string
	InsertAfter     int
}

InjectedMessageRecord records where an injected message belongs in the persisted model-message sequence.

type ModelMessage 

type ModelMessage = turn.ModelMessage

The canonical turn value objects are re-exported inside application so the orchestration code can use them without maintaining a second DTO family.

type OutboundAssetRef 

type OutboundAssetRef = turn.OutboundAssetRef

The canonical turn value objects are re-exported inside application so the orchestration code can use them without maintaining a second DTO family.

type PlatformIdentity 

type PlatformIdentity struct {
	ID               string
	Platform         string
	ExternalIdentity string
	SelfIdentity     map[string]any
}

PlatformIdentity is the Agent-owned projection of a connected platform account used while assembling the system prompt.

type PlatformIdentitySource 

type PlatformIdentitySource interface {
	ListPlatformIdentities(ctx context.Context, botID string) ([]PlatformIdentity, error)
}

PlatformIdentitySource supplies connected platform identities without exposing Channel configuration types to the application layer.

type QueueAdmissionFailure added in v0.21.0 

type QueueAdmissionFailure int

QueueAdmissionFailure names why queue admission refused an input. It is the transport-neutral half of the queue error translation: every ingress classifies with QueueAdmissionFailureOf and renders the failure in its own published vocabulary, which differs between HTTP and channels and does not change here. 

const (
	// QueueAdmissionNotRefused is the zero value: the error is nil or is not
	// an admission refusal, and the caller handles it as an ordinary error.
	QueueAdmissionNotRefused QueueAdmissionFailure = iota
	QueueAdmissionSteerUnsupported
	QueueAdmissionNoActiveRun
	QueueAdmissionInvocationConflict
	QueueAdmissionOverloaded
	QueueAdmissionCapacityExceeded
	QueueAdmissionInvalidReference
	// QueueAdmissionUnavailable is a server wiring fault at the ingress, so
	// senders see it as unavailable rather than as a bad request.
	QueueAdmissionUnavailable
)

func QueueAdmissionFailureOf added in v0.21.0 

func QueueAdmissionFailureOf(err error) QueueAdmissionFailure

QueueAdmissionFailureOf classifies an error returned by EnqueueSteer or EnqueueFollowUp.

type QueueInput added in v0.21.0 

type QueueInput struct {
	TeamID       string
	BotID        string
	SessionID    string
	InvocationID string
	// UserID and SourceChannelIdentityID name the sender the way the
	// ingress's ordinary turns do. Empty values are allowed for a channel
	// that has no platform account for the sender.
	UserID                  string
	SourceChannelIdentityID string
	Text                    string
}

QueueInput is one text input an ingress admits into a session queue. It carries the identity the ingress authenticated, because the continuation that eventually runs the input happens after the request is gone and, in a hosted deployment, possibly on another instance: nothing at that point can recover which team or user the text belonged to.

type RequestedSkillContext 

type RequestedSkillContext = turn.RequestedSkillContext

The canonical turn value objects are re-exported inside application so the orchestration code can use them without maintaining a second DTO family.

type ResolveRunConfigResult 

type ResolveRunConfigResult struct {
	RunConfig              native.RunConfig
	ModelID                string // database UUID of the selected model
	RuntimeType            string
	ContextBudgetMaxTokens int
}

ResolveRunConfigResult holds a fully resolved run configuration for one agent request, together with the selected model and session runtime. Produced by application.Service.ResolveRunConfig and consumed by the turn runtime adapters.

type RetryLatestMessageInput 

type RetryLatestMessageInput struct {
	BotID        string
	SessionID    string
	RunID        string
	TurnID       string
	TurnPosition *int64
	// TargetTurnID names the round being replaced. It is distinct from TurnID,
	// which names the new turn this operation admits.
	TargetTurnID           string
	ActorChannelIdentityID string
	ActorUserID            string
	ChatToken              string
	Model                  string
	ReasoningEffort        string
	WorkspaceTargetID      string
	ToolHTTPURL            string
	// RunHandle and InjectCh are server-owned admission capabilities. They are
	// populated only by the in-process Web runtime, never by client JSON.
	RunHandle sessionruntime.RunHandle
	InjectCh  chan turnpkg.InjectMessage
	// OnModelPreferenceSettled releases subsequent picker writes once this
	// turn's preference write-back has finished (issue #879). Same contract
	// as ChatRequest.OnModelPreferenceSettled.
	OnModelPreferenceSettled func()
}

type RunOutcome added in v0.21.0 

type RunOutcome struct {
	Status string
	Cause  error
}

RunOutcome is how the owner of a run reports its end to the terminal write. Status is a session runtime run status, or empty to leave the outcome to the session runtime, which derives it from what the run recorded. Cause is the failure the owner holds, if any.

func (RunOutcome) ErrorCode added in v0.21.0 

func (o RunOutcome) ErrorCode() string

ErrorCode is the code the terminal write records for the run: the catalogued code the cause carries, or empty to let the session runtime name it. The code travels in its own column; it is never written as the run's error message.

type Runner 

type Runner interface {
	Chat(ctx context.Context, req ChatRequest) (ChatResponse, error)
	StreamChat(ctx context.Context, req ChatRequest) (<-chan StreamChunk, <-chan error)
	TriggerSchedule(ctx context.Context, botID string, payload schedule.TriggerPayload, token string) (schedule.TriggerResult, error)
}

Runner defines conversation execution behavior for sync, stream, and scheduled flows.

type RuntimeControlRequest added in v0.20.0 

type RuntimeControlRequest = turn.RuntimeControlRequest

RuntimeControlRequest is scoped to a thread and its current actor. Drivers may query their runtime process for current control state.

type RuntimeSessionExecutionInfo added in v0.20.0 

type RuntimeSessionExecutionInfo struct {
	RequiresWorkspaceExec bool
	BotID                 string
	RuntimeOwnerAccountID string
}

type ScheduleGateway 

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

ScheduleGateway adapts schedule trigger calls to the chat Service.

func NewScheduleGateway 

func NewScheduleGateway(service *Service) *ScheduleGateway

NewScheduleGateway creates a ScheduleGateway backed by the given Service.

func (*ScheduleGateway) TriggerSchedule 

func (g *ScheduleGateway) TriggerSchedule(ctx context.Context, botID string, payload schedule.TriggerPayload, token string) (schedule.TriggerResult, error)

TriggerSchedule delegates a schedule trigger to the chat Service.

type Service 

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

Service orchestrates chat with the internal agent.

func NewService 

func NewService(
	log *slog.Logger,
	modelsService *models.Service,
	queries dbstore.Queries,
	messageService messagepkg.Service,
	settingsService *settings.Service,
	accountService *accounts.Service,
	a *native.Agent,
	clockLocation *time.Location,
	timeout time.Duration,
) *Service

NewService creates an application service backed by the native agent.

func (*Service) AbortRuntimeRun 

func (s *Service) AbortRuntimeRun(
	ctx context.Context,
	botID, sessionID, runID, controlID string,
) (bool, error)

AbortRuntimeRun routes an abort through the durable runtime and then reconciles its lifecycle asynchronously. The acknowledgement remains owned by AbortControl; audit persistence failures are counted and logged only.

func (*Service) AbortSessionRuns added in v0.20.0 

func (s *Service) AbortSessionRuns(ctx context.Context, botID, sessionID string) error

AbortSessionRuns interrupts the session's active run, if any, and waits — bounded — until it actually stops. Deleting or resetting a session on a direct external runtime must not leave its turn executing (and mutating the workspace) behind the deletion; cancel alone only signals the driver, so callers that destroy the session next need the settle, not the signal. A run that cannot settle inside the window fails the call rather than letting the caller proceed over a still-running turn.

func (*Service) AdmitSubagentRun 

func (s *Service) AdmitSubagentRun(
	ctx context.Context,
	botID, threadID, invocationID string,
	submission []byte,
) (context.Context, tools.SubagentAdmission, func(tools.SubagentTerminal), error)

AdmitSubagentRun puts a spawned agent's turn through the same durable admission every other turn takes, and answers in the vocabulary of the turn port so the tool layer never sees a runtime type.

The slot a subagent takes is its own thread's, not its parent's: a parent may have several agents working at once, and each of those threads still runs one turn at a time. Busy therefore means *this agent* is already working — a fact the parent model can act on — rather than a failure to report.

func (*Service) AdvancePlainTextUserInput 

func (s *Service) AdvancePlainTextUserInput(ctx context.Context, input userinput.AdvanceTextInput) (userinput.AdvanceTextResult, error)

func (*Service) ApplyUserMessageHookAndPersistUserTurn 

func (s *Service) ApplyUserMessageHookAndPersistUserTurn(ctx context.Context, req ChatRequest) (ChatRequest, messagepkg.Message, error)

ApplyUserMessageHookAndPersistUserTurn applies the normal user-message hook before writing a user turn ahead of agent execution. Web skill activations use this so a denied hook cannot leave a persisted user-only special turn.

func (*Service) CancelFollowUp added in v0.20.0 

func (s *Service) CancelFollowUp(ctx context.Context, botID, sessionID, itemID string) error

func (*Service) CancelSteer added in v0.20.0 

func (s *Service) CancelSteer(ctx context.Context, botID, sessionID, itemID string) error

func (*Service) Chat 

func (s *Service) Chat(ctx context.Context, req ChatRequest) (ChatResponse, error)

Chat sends a synchronous chat request and stores the result.

func (*Service) CommitToolApprovalResponse 

func (s *Service) CommitToolApprovalResponse(ctx context.Context, input ToolApprovalResponseInput) (CommittedToolApprovalResponse, error)

func (*Service) CommitUserInputResponse 

func (s *Service) CommitUserInputResponse(ctx context.Context, input UserInputResponseInput) (CommittedUserInputResponse, error)

func (*Service) ContinueCommittedToolApprovalResponse 

func (s *Service) ContinueCommittedToolApprovalResponse(ctx context.Context, committed CommittedToolApprovalResponse, eventCh chan<- WSStreamEvent) error

func (*Service) ContinueCommittedUserInputResponse 

func (s *Service) ContinueCommittedUserInputResponse(ctx context.Context, committed CommittedUserInputResponse, eventCh chan<- WSStreamEvent) error

func (*Service) ControlRuntimeGoal added in v0.20.0 

func (s *Service) ControlRuntimeGoal(ctx context.Context, request RuntimeControlRequest, action string) (err error)

func (*Service) DeferSessionCompaction 

func (s *Service) DeferSessionCompaction(botID, sessionID, runID string) func()

func (*Service) DrainActiveTurns added in v0.21.0 

func (s *Service) DrainActiveTurns(ctx context.Context) error

DrainActiveTurns runs after transports cancel their requests and before the PostgreSQL pool closes. It is bounded by the caller's shutdown context.

func (*Service) EditLatestMessageWS 

func (s *Service) EditLatestMessageWS(ctx context.Context, input EditLatestMessageInput, eventCh chan<- WSStreamEvent, abortCh <-chan struct{}) (RunOutcome, error)

func (*Service) EnqueueDeferredTurn added in v0.20.0 

func (s *Service) EnqueueDeferredTurn(ctx context.Context, cmd turn.StartTurnCommand) error

EnqueueDeferredTurn places a complete user turn that met a busy session into the session's follow-up queue. The command is stored intact, so the continuation keeps the channel route, attachments, and reply metadata of the original message. The caller receives the same admission errors as any other follow-up: in particular ErrNoActiveRun means the run ended between the busy admission result and this call, and the caller should retry admission.

func (*Service) EnqueueFollowUp added in v0.20.0 

func (s *Service) EnqueueFollowUp(ctx context.Context, input QueueInput) (sessionruntime.FollowUpItem, error)

func (*Service) EnqueueSteer added in v0.20.0 

func (s *Service) EnqueueSteer(ctx context.Context, input QueueInput) (sessionruntime.SteerItem, error)

func (*Service) EnsureTerminalContextLifecycle 

func (s *Service) EnsureTerminalContextLifecycle(
	ctx context.Context,
	runID, botID, sessionID string,
	cause error,
)

EnsureTerminalContextLifecycle records a content-light fallback for runs that fail before native context assembly creates a snapshot. A terminal writer with an authoritative holder always wins this read-before-create race.

func (*Service) ExecuteRuntimeCommand added in v0.20.0 

func (s *Service) ExecuteRuntimeCommand(ctx context.Context, request RuntimeControlRequest) (result turn.RuntimeCommandResult, resultErr error)

ExecuteRuntimeCommand handles commands which do not create a conversation turn. Turn commands go through the existing chat admission/persistence path.

func (*Service) InlineImageAttachments 

func (s *Service) InlineImageAttachments(ctx context.Context, botID string, refs []timeline.ImageAttachmentRef) []sdk.ImagePart

InlineImageAttachments resolves image content hashes to sdk.ImagePart values using the configured asset loader. Intended for the discuss driver to inline images from new RC segments before calling the LLM.

func (*Service) LinkOutboundAssets 

func (s *Service) LinkOutboundAssets(ctx context.Context, botID, sessionID string, assets []messagepkg.AssetRef)

LinkOutboundAssets links bot-generated assets to the assistant message that produced them. Assets carrying a `tool_call_id` metadata entry are anchored to the assistant message containing that tool call, so a history rebuild keeps the live ordering (e.g. a generated image stays above the closing text). Assets without one fall back to the latest assistant message. When sessionID is provided, the search is scoped to that session; otherwise it falls back to a bot-wide search. Used by the WebSocket path where attachment ingestion happens after message persistence.

func (*Service) ListSessionQueues added in v0.20.0 

func (s *Service) ListSessionQueues(ctx context.Context, botID, sessionID string) (SessionQueues, error)

func (*Service) PatchSessionModelPreference added in v0.20.0 

func (s *Service) PatchSessionModelPreference(ctx context.Context, botID, sessionID string, modelRef, effort *string, expectedRevision string) error

PatchSessionModelPreference handles the picker PATCH (issue #879, spec §3.3 / §5-2): reconcile the pair and write it. Reconcile = the effort must be legal for the target model; an illegal or empty tier silently resolves to the model default, so the DB never stores an illegal pair.

Target model resolution when the request omits it: existing session preference, then bot default. Effort-only patches therefore reconcile against the model the session would actually use. An explicit but unresolvable/empty model reference is an error (the FK would reject it anyway; fail loudly instead of degrading).

expectedRevision is the revision the picker read; "" means the session has none yet. Every PATCH is compare-and-set: a picker may never overwrite a send or a newer picker operation it did not see.

func (*Service) PendingRuntimeDecisions added in v0.20.0 

func (s *Service) PendingRuntimeDecisions(ctx context.Context, runID string) ([]sessionruntime.DecisionTarget, error)

PendingRuntimeDecisions resolves every durable decision that parked runID. It is used only by expired-owner recovery, where preserving the exact rows is required before advancing the run's fencing token — a turn can park on several approvals and user inputs at once, and dropping any of them here would supersede a decision the user can still answer.

func (*Service) Pipeline 

func (s *Service) Pipeline() *timeline.Pipeline

Pipeline returns the configured pipeline, or nil.

func (*Service) PrepareEditLatestTurnOperation 

func (s *Service) PrepareEditLatestTurnOperation(ctx context.Context, sessionID, turnID string) (string, error)

PrepareEditLatestTurnOperation validates the replacement target before Session Runtime publishes it to subscribers and returns the exact persisted user message where the old turn begins.

func (*Service) PrepareExternalFork added in v0.20.0 

func (s *Service) PrepareExternalFork(ctx context.Context, botID, sessionID, turnID string) (map[string]any, error)

PrepareExternalFork forks the runtime-side conversation behind an external session and returns the complete runtime metadata the forked Memoh session must carry. The anchor turn's recorded runtime turn id bounds the runtime-side fork to the same cut as the visible history.

func (*Service) PrepareRetryLatestTurnOperation 

func (s *Service) PrepareRetryLatestTurnOperation(ctx context.Context, sessionID, turnID string) (string, error)

PrepareRetryLatestTurnOperation validates the replacement target before Session Runtime publishes it to subscribers and returns the exact persisted message where the old assistant tail begins.

func (*Service) PromoteFollowUpToSteer added in v0.20.0 

func (s *Service) PromoteFollowUpToSteer(ctx context.Context, botID, sessionID string, followUp sessionruntime.FollowUpPendingRef) (sessionruntime.PromoteFollowUpResult, error)

func (*Service) ReconcileSessionModelPreference added in v0.20.0 

func (s *Service) ReconcileSessionModelPreference(ctx context.Context, botID, modelRef, effort string) (string, string, error)

ReconcileSessionModelPreference validates and normalizes a candidate pair (issue #879, spec v2 §3.3): the model must exist on an enabled provider (empty ref falls back to the bot default); an illegal or empty effort silently resolves to the model's default tier, so the DB never stores an illegal pair (S6). Returns the model's UUID and the normalized effort. Shared by the picker PATCH and the first-send INSERT so both write points reconcile identically.

func (*Service) RecordSubagentResume added in v0.21.0 

func (s *Service) RecordSubagentResume(ctx context.Context, session tools.SessionContext, sessionID, query, model string) error

RecordSubagentResume is called after admission and before the first child tool.

func (*Service) ReorderFollowUp added in v0.20.0 

func (s *Service) ReorderFollowUp(ctx context.Context, botID, sessionID string, item, before sessionruntime.FollowUpPendingRef) ([]sessionruntime.FollowUpItem, error)

func (*Service) ReorderSteer added in v0.20.0 

func (s *Service) ReorderSteer(ctx context.Context, botID, sessionID string, item, before sessionruntime.SteerPendingRef) ([]sessionruntime.SteerItem, error)

func (*Service) ResolveRunConfig 

func (s *Service) ResolveRunConfig(ctx context.Context, botID, sessionID, channelIdentityID, currentPlatform, replyTarget, conversationType, chatToken string) (ResolveRunConfigResult, error)

ResolveRunConfig builds a complete RunConfig (model, system prompt, tools, identity) for a bot+session without loading messages or requiring a query. The caller is responsible for filling RunConfig.Messages. Used by discuss turns to reuse the service's model, tools, and prompt pipeline.

func (*Service) ResolveRuntimeDecision 

func (s *Service) ResolveRuntimeDecision(ctx context.Context, commandType, decisionID string) (sessionruntime.DecisionTarget, error)

ResolveRuntimeDecision reads the authoritative decision row before any live owner lookup. Terminal rows are returned too: the router needs to distinguish a known, already-decided request from an unfenced ACP/MCP request.

func (*Service) ResolveTurnIDForMessage 

func (s *Service) ResolveTurnIDForMessage(ctx context.Context, sessionID, messageID string) (string, error)

ResolveTurnIDForMessage maps a stored message id onto the round that contains it. It backs the compatibility shim for the pre-turn `message_id` spelling and has no other caller: a client holds a turn id from admission onward and names the round directly, while needing a stored message id is exactly what used to force it to wait for the round to persist.

Remove it, and the shim, once the compatibility window for `message_id` closes. (Not marked Deprecated in the godoc sense — the shim is its intended caller, so flagging that call adds noise rather than signal.)

func (*Service) RespondToolApproval 

func (s *Service) RespondToolApproval(ctx context.Context, input turn.ToolApprovalResponse, eventCh chan<- json.RawMessage) error

RespondToolApproval resumes a turn deferred on tool approval.

func (*Service) RespondUserInput 

func (s *Service) RespondUserInput(ctx context.Context, input turn.UserInputResponse, eventCh chan<- json.RawMessage) error

RespondUserInput resumes a turn deferred on ask_user.

func (*Service) RetryLatestMessageWS 

func (s *Service) RetryLatestMessageWS(ctx context.Context, input RetryLatestMessageInput, eventCh chan<- WSStreamEvent, abortCh <-chan struct{}) (RunOutcome, error)

func (*Service) RuntimeCommands added in v0.20.0 

func (s *Service) RuntimeCommands(ctx context.Context, request RuntimeControlRequest) (commands []external.Command, err error)

RuntimeCommands reads declarations without fetching unrelated runtime state. Catalog discovery needs chat access, not ownership of the runtime controls. Executing a declared command still uses the existing control/admission gates.

func (*Service) RuntimeControls added in v0.20.0 

func (s *Service) RuntimeControls(ctx context.Context, request RuntimeControlRequest) (out external.Controls, resultErr error)

func (*Service) RuntimeGoal added in v0.20.0 

func (s *Service) RuntimeGoal(ctx context.Context, request RuntimeControlRequest) (goal *external.Goal, err error)

func (*Service) RuntimeSessionExecutionInfo added in v0.20.0 

func (s *Service) RuntimeSessionExecutionInfo(ctx context.Context, sessionID string) (RuntimeSessionExecutionInfo, error)

func (*Service) SetACPSessionPool 

func (s *Service) SetACPSessionPool(pool acpPrompter)

func (*Service) SetAllowedTeam 

func (s *Service) SetAllowedTeam(teamID string)

SetAllowedTeam restricts the service to a single team. The in-process runtime's database pool is session-bound to one team GUC, so commands for any other team must fail closed (turn.ErrTeamNotServed) instead of silently operating on the bound team's data. The composition root injects the self-hosted singleton team; a hosted multi-team runtime replaces this with request-scoped team binding.

func (*Service) SetBackgroundManager 

func (s *Service) SetBackgroundManager(m *background.Manager)

SetBackgroundManager configures the background task manager used for task summaries and background status tooling.

func (*Service) SetBotPermissionChecker 

func (s *Service) SetBotPermissionChecker(checker botPermissionChecker)

func (*Service) SetCompactionService 

func (s *Service) SetCompactionService(service *compaction.Service)

SetCompactionService configures the compaction service for context compaction.

func (*Service) SetContextAbsoluteMaxTokens 

func (s *Service) SetContextAbsoluteMaxTokens(v int)

SetContextAbsoluteMaxTokens sets the server-wide context admission cap (CM-ADM-001). Zero keeps the shared default; the cap is never disabled.

func (*Service) SetEventPublisher 

func (s *Service) SetEventPublisher(p messageevent.Publisher)

SetEventPublisher configures the event publisher for broadcasting events such as session title updates.

func (*Service) SetExternalRuntimes added in v0.20.0 

func (s *Service) SetExternalRuntimes(drivers ...external.Driver)

SetExternalRuntimes registers out-of-process runtime drivers (codex, claude-code, the ACP pool's adapter). Sessions whose runtime type has no registered driver fail with a stable "runtime unavailable" error instead of silently degrading to the built-in model runtime.

func (*Service) SetGatewayAssetLoader 

func (s *Service) SetGatewayAssetLoader(loader gatewayAssetLoader)

SetGatewayAssetLoader configures optional asset loading used to inline attachments before calling the agent gateway.

func (*Service) SetHookService 

func (s *Service) SetHookService(service *hooks.Service)

func (*Service) SetMemoryRegistry 

func (s *Service) SetMemoryRegistry(registry *memprovider.Registry)

SetMemoryRegistry sets the provider registry for memory operations.

func (*Service) SetPipeline 

func (s *Service) SetPipeline(p *timeline.Pipeline)

SetPipeline configures the DCP pipeline for RC-based context assembly. When set, resolve() will use RC from the pipeline instead of loading history from bot_history_messages for sessions that have pipeline data.

func (*Service) SetPlatformIdentitySource 

func (s *Service) SetPlatformIdentitySource(source PlatformIdentitySource)

SetPlatformIdentitySource configures the neutral source used to load platform identity metadata for system prompt generation.

func (*Service) SetResumeReadiness added in v0.21.0 

func (s *Service) SetResumeReadiness(check func(context.Context, string, string) error)

SetResumeReadiness checks workspace transport before claiming a continuation.

func (*Service) SetResumeSecret added in v0.21.0 

func (s *Service) SetResumeSecret(secret string)

func (*Service) SetRuntimeMode added in v0.20.0 

func (s *Service) SetRuntimeMode(ctx context.Context, request RuntimeControlRequest) (out external.ModeState, err error)

func (*Service) SetSessionResumeScopeProvider added in v0.21.0 

func (s *Service) SetSessionResumeScopeProvider(provider SessionResumeScopeProvider)

SetSessionResumeScopeProvider installs hosted tenancy before startup. The OSS default names only its singleton team; it never enumerates all session rows.

func (*Service) SetSessionRuntime 

func (s *Service) SetSessionRuntime(manager *sessionruntime.Manager)

SetSessionRuntime injects the durable admission gate. Setter injection rather than a constructor argument because the manager and this service are wired into the same fx graph and each is reachable from the other's dependencies.

func (*Service) SetSessionService 

func (s *Service) SetSessionService(service SessionService)

SetSessionService configures the session service used for auto title generation.

func (*Service) SetSkillLoader 

func (s *Service) SetSkillLoader(sl SkillLoader)

SetSkillLoader sets the skill loader used to populate usable skills in gateway requests.

func (*Service) SetSyncCompactionMode added in v0.20.0 

func (s *Service) SetSyncCompactionMode(mode string)

SetSyncCompactionMode sets the rollout mode for the pre-turn synchronous compaction backstop on the discuss and pipeline-chat paths.

func (*Service) SetToolApprovalService 

func (s *Service) SetToolApprovalService(service *toolapproval.Service)

func (*Service) SetUserInputService 

func (s *Service) SetUserInputService(service *userinput.Service)

func (*Service) SetWorkdirResolver 

func (s *Service) SetWorkdirResolver(resolver sessionWorkdirResolver)

SetWorkdirResolver configures resolution of session workdir bindings.

func (*Service) SetWorkspaceTargetResolver 

func (s *Service) SetWorkspaceTargetResolver(resolver workspaceTargetResolver)

SetWorkspaceTargetResolver configures request-scoped Computer resolution.

func (*Service) StartSessionResume added in v0.21.0 

func (s *Service) StartSessionResume(ctx context.Context) error

StartSessionResume probes the scope catalog before launching a bounded worker. The worker retains a cursor per current tenant instead of querying tenant data from the startup context. Run contexts retain the same binding after detaching.

func (*Service) StartTurn 

func (s *Service) StartTurn(ctx context.Context, cmd turn.StartTurnCommand) (turn.RunHandle, error)

StartTurn validates the command, admits it durably, starts the underlying stream, and returns a handle whose Events/Errs mirror the application stream.

func (*Service) StopSessionResume added in v0.21.0 

func (s *Service) StopSessionResume()

StopSessionResume stops discovery only. Active runs are interrupted by the manager after it persists their marker, not by cancellation of this worker.

func (*Service) StopTurn added in v0.20.0 

func (s *Service) StopTurn(ctx context.Context, cmd turn.StopCommand) (bool, error)

StopTurn routes cancellation to the durable owner even after the channel stream has closed, using the same application entry point as the web Stop.

func (*Service) StoreRound 

func (s *Service) StoreRound(ctx context.Context, botID, sessionID, channelIdentityID, currentPlatform string, sdkMessages []sdk.Message, modelID string) error

StoreRound persists SDK messages as a complete round (assistant + tool output) into bot_history_messages with full metadata, usage tracking, and memory extraction. Used by the discuss driver so it shares the same persistence quality as chat mode.

func (*Service) StreamChat 

func (s *Service) StreamChat(ctx context.Context, req ChatRequest) (<-chan StreamChunk, <-chan error)

StreamChat runs a streaming chat via the internal agent.

func (*Service) StreamChatWS 

func (s *Service) StreamChatWS(
	ctx context.Context,
	req ChatRequest,
	eventCh chan<- WSStreamEvent,
	abortCh <-chan struct{},
) (RunOutcome, error)

StreamChatWS resolves the agent context and streams agent events. Events are sent on eventCh. When abortCh is closed, the context is cancelled. The returned error is a failure the caller still has to report; the outcome names a failure the stream already delivered, for the run's terminal write.

func (*Service) SubagentFailure added in v0.21.0 

func (*Service) SubagentFailure(event native.StreamEvent) error

SubagentFailure is the error a spawned attempt that ended with event reports to the spawn provider: the run's public failure, named as for any native run, with the event's cause in its chain.

func (*Service) SubagentRunObserver 

func (s *Service) SubagentRunObserver(ctx context.Context) native.SpawnRunObserver

SubagentRunObserver returns a per-event publisher that feeds one spawned agent run's stream into the session runtime, or nil when there is nothing to publish to — no runtime configured, or a context without an admitted handle.

The returned function mirrors forwardWSStreamEvents' publishing discipline: events leave as publicAgentStreamEvent makes them, so a failed spawned run is named like any other native run; they are published on a context that survives the run's cancellation (an aborted run's final events are exactly the ones a subscriber must see); and a lost ownership stops publishing outright because every later event would fail identically.

func (*Service) SubagentStepCommit 

func (s *Service) SubagentStepCommit(
	ctx context.Context,
	botID, sessionID, modelID, turnRequestMessageID string,
	contextLifecycle *contextfrag.LifecycleHolder,
	onPersisted func(),
) (
	func(context.Context, int, *step.Record) error,
	func(context.Context, int, *step.Record) error,
)

SubagentStepCommit returns the per-step persistence callback for one spawned agent run, or nil when incremental persistence is unavailable — no admitted handle on the context, no fence, no persisted request row to bind steps to, or a message service that cannot write fenced steps. A nil return means the spawn path keeps its terminal-snapshot persistence, so this is a capability probe as much as a constructor.

The callbacks persist exactly what the legacy terminal path persists — every non-user message of the step, marshalled whole — only earlier: each complete step or safe interrupted checkpoint lands in one fenced transaction instead of the whole run landing at the end. onPersisted fires after either kind of durable write; the spawn provider uses it to stop retrying an attempt whose output is already part of history.

func (*Service) TriggerSchedule 

func (s *Service) TriggerSchedule(ctx context.Context, botID string, payload schedule.TriggerPayload, token string) (triggerResult schedule.TriggerResult, err error)

TriggerSchedule executes a scheduled command via the internal agent.

func (*Service) UpdateFollowUp added in v0.20.0 

func (s *Service) UpdateFollowUp(ctx context.Context, botID, sessionID, itemID, editorUserID, text string) (sessionruntime.FollowUpItem, error)

UpdateFollowUp replaces the text of an accepted follow-up. See UpdateSteer.

func (*Service) UpdateSteer added in v0.20.0 

func (s *Service) UpdateSteer(ctx context.Context, botID, sessionID, itemID, editorUserID, text string) (sessionruntime.SteerItem, error)

UpdateSteer replaces the text of an accepted steer on behalf of editorUserID, who must be the sender the item recorded. Sender, routing, and attachment metadata are immutable for a queued command; only the text is rewritten, and the backend rechecks accepted status atomically with the edit.

func (*Service) ValidateWorkspaceTarget 

func (s *Service) ValidateWorkspaceTarget(ctx context.Context, botID, targetID string) error

ValidateWorkspaceTarget validates a user-selected Computer without changing the Bot's Primary target. It is used by handlers before creating a session.

type SessionQueues added in v0.20.0 

type SessionQueues struct {
	SteerSupported bool
	Steer          []sessionruntime.SteerItem
	FollowUp       []sessionruntime.FollowUpItem
}

SessionQueues is the application surface for user-facing queue operations. Items are transient and live in the configured memory or Redis runtime.

type SessionResumeScopePage added in v0.21.0 

type SessionResumeScopePage struct {
	Scopes     []string
	NextCursor string
	Complete   bool
}

SessionResumeScopePage contains opaque tenant IDs, never cross-tenant rows. Hosted deployments can reuse their runtime reaper's scope catalog.

type SessionResumeScopeProvider added in v0.21.0 

type SessionResumeScopeProvider interface {
	CurrentScope(context.Context) string
	ListScopePage(context.Context, string, int32) (SessionResumeScopePage, error)
	BindScope(context.Context, string) context.Context
}

SessionResumeScopeProvider binds every resume operation to its database scope. Implementations must enumerate through a restricted catalog, then use ordinary tenant-bound credentials for run queries, authorization and execution.

type SessionService 

type SessionService interface {
	Get(ctx context.Context, sessionID string) (session.Thread, error)
	UpdateTitle(ctx context.Context, sessionID, title string) (session.Thread, error)
	UpdateMetadata(ctx context.Context, sessionID string, metadata map[string]any) (session.Thread, error)
	MergeRuntimeMetadata(ctx context.Context, sessionID, runtimeType string, delta map[string]any) (session.Thread, error)
}

SessionService is the interface the application service uses for session metadata updates.

type SkillActivation 

type SkillActivation = turn.SkillActivation

The canonical turn value objects are re-exported inside application so the orchestration code can use them without maintaining a second DTO family.

type SkillActivationSkill 

type SkillActivationSkill = turn.SkillActivationSkill

The canonical turn value objects are re-exported inside application so the orchestration code can use them without maintaining a second DTO family.

type SkillEntry 

type SkillEntry struct {
	Name        string
	Description string
	Content     string
	Path        string
	Metadata    map[string]any
}

SkillEntry represents a skill loaded from the container.

type SkillLoader 

type SkillLoader interface {
	LoadSkills(ctx context.Context, botID string) ([]SkillEntry, error)
}

SkillLoader loads skills for a given bot from its container.

type StreamChunk 

type StreamChunk = json.RawMessage

StreamChunk is one raw event emitted by the application stream.

type ToolApprovalResponseInput 

type ToolApprovalResponseInput struct {
	ControlID              string
	BotID                  string
	ThreadID               string
	ActorChannelIdentityID string
	ActorUserID            string
	ApprovalID             string
	ExplicitID             string
	ReplyExternalMessageID string
	Decision               string
	// OptionID names the agent-provided permission option the decider picked;
	// empty means the plain binary decision.
	OptionID                   string
	Reason                     string
	ChatToken                  string
	SuppressActivePromptAttach bool
}

type ToolCall 

type ToolCall = turn.ToolCall

The canonical turn value objects are re-exported inside application so the orchestration code can use them without maintaining a second DTO family.

type ToolCallFunction 

type ToolCallFunction = turn.ToolCallFunction

The canonical turn value objects are re-exported inside application so the orchestration code can use them without maintaining a second DTO family.

type UserInputResponseInput 

type UserInputResponseInput struct {
	ControlID                  string
	BotID                      string
	ThreadID                   string
	ActorChannelIdentityID     string
	ActorUserID                string
	UserInputID                string
	ExplicitID                 string
	ReplyExternalMessageID     string
	Answers                    []userinput.QuestionAnswer
	UILanguage                 string
	TextAnswer                 string
	Canceled                   bool
	Reason                     string
	ChatToken                  string
	SuppressActivePromptAttach bool
}

type WSStreamEvent 

type WSStreamEvent = json.RawMessage

WSStreamEvent represents a raw JSON event forwarded from the agent.

type WorkspaceTarget 

type WorkspaceTarget struct {
	TargetID string `json:"target_id"`
	Kind     string `json:"kind"`
	Name     string `json:"name"`
}

WorkspaceTarget is the immutable execution-location snapshot resolved for one application request.

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.