service

package
v0.9.0 Latest Latest
Warning

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

 Go to latest Published: Oct 29, 2026 License: Apache-2.0, MIT Imports: 130 Imported by: 0

Documentation

Overview

Package service - artifact collector.

ArtifactCollector drains a session sandbox's output directory into the tenant file service after a skill turn finishes. It is intentionally kept stateless and testable: a narrow SandboxArtifactSource interface abstracts the sandbox side so unit tests can stub in a fake filesystem, and the file service is passed in so blobs land in the same storage backend the tenant already uses for uploaded attachments.

Contract:

  • Never delete files from the sandbox — skills can share files across turns; deletion would break that ("不清空输出目录").
  • Never lazy-create a sandbox: the collector reads from an already-live sandbox and returns an empty slice when none exists.
  • Best-effort: individual errors are logged and skipped, never returned, so a stray unreadable file cannot block the assistant reply.
  • De-duplication by (SourcePath, ModTime), with a content-hash check when mtime changed: git checkout refreshes mtime on unchanged bytes and must not re-attach; a same-size in-place rewrite must.

Package service provides business logic implementations for WeKnora application This package contains service layer implementations that coordinate between repositories and handlers, applying business rules and transaction management

Package service - forked-session sandbox bootstrap.

ForkBootstrapper implements sandbox.SessionBootstrapper for forked sessions: it points the first create at the fork snapshot, then rolls /workspace back to the fork point with git. Git tracks the full tree, including input/ and output/; object storage is not rewritten into the sandbox.

The provider snapshot is the live sandbox at fork *time*, so reset must also drop later-turn git objects. Otherwise `git checkout` of a post-fork commit would restore files the working tree just rolled back. Packages and other paths outside /workspace stay at the fork moment; that is intentional.

It is all-or-nothing for the filesystem: handing back a sandbox that sits at the fork MOMENT rather than the fork POINT would look normal while silently working from the wrong baseline. Snapshot deletion is not part of that contract — Cube refuses to delete a snapshot while the source and child sandboxes still hold runtime refs, and failing the bootstrap for that would destroy a correctly rolled-back sandbox.

Package service - orphan fork snapshot collection.

A fork takes a provider snapshot immediately but provisions the sandbox lazily. A branch the user never opens therefore leaves a snapshot behind forever. Persist can also fail after the snapshot already exists, or the process can crash before the session row is committed; those IDs live in fork_snapshot_leases so this reaper can still find them.

The happy path tries to delete at consume time. Cube often refuses while the source and child sandboxes still hold runtime refs, so this reaper retries leftover snapshot IDs after those refs drop, and still collects forks that were never opened.

Package service: knowledge housekeeping.

HousekeepingService periodically scans for knowledge rows that have been stuck in "processing" longer than any reasonable execution window and marks them as failed. This is the safety net that catches anything the other defences (asynq retry, dead-letter callback, image_multimodal finalize-on-last-attempt) miss — for example:

  • Worker process killed mid-handler before any defer could run.
  • DocReader call genuinely exceeding DocReaderCallTimeout AND the worker subsequently being lost before retry kicks in.
  • Multimodal Redis counter set to N but ALL N image tasks failing in ways that bypass finalize (extremely rare; defence-in-depth here).

Without this sweep, a single unlucky failure mode can leave a knowledge row in "processing" forever — invisible to users except as a permanent spinner. With this sweep the worst-case latency from stall to user- visible failure is bounded to ~1 stale-threshold + 1 sweep interval.

Package service: span tracker.

SpanTracker is the pipeline-facing facade for recording per-attempt progress trees. It mirrors Langfuse's vocabulary (root / span / generation) so the UI's mental model matches what operators already use for LLM call observability.

Lifecycle: 

attempt := tracker.OpenAttempt(ctx, knowledgeID, langfuseTraceID)
  // creates the root span; every subsequent Begin* call uses this attempt

stage := tracker.BeginStage(ctx, knowledgeID, attempt, types.StageDocReader, input)
  // ...do work...
tracker.EndSpan(ctx, stage, output)         // success
tracker.FailSpan(ctx, stage, code, msg, err) // error
tracker.SkipSpan(ctx, stage, reason)        // intentionally not run

sub := tracker.BeginSubSpan(ctx, parentSpan, "multimodal.image[0]", types.SpanKindGeneration, input)
  // ...

All operations are best-effort: a DB error is logged and swallowed so a tracker hiccup never breaks the parsing pipeline. Knowledge.parse_status remains the authoritative source of truth for completion.

Package service provides lazy startup and dialling for the sandbox graphical desktop.

The desktop is not started with the sandbox: E2B runs envd as init and never executes the image CMD, and Cube's ENTRYPOINT is already envd's. One idempotent script driven over Exec covers both, heals a sandbox after pause/resume, and costs nothing for sessions that never open the tab.

Package service provides one-shot handshake tickets for the sandbox desktop WebSocket.

Unlike the terminal's ticket (sandbox_terminal_ticket.go) this is an opaque random string in a store, not a signed JWT. The desktop ticket must be single-use, and once a store is needed to enforce that, the stateless JWT's only advantage is gone — while its drawback (an HS256 blob sitting in gateway access logs and browser history for its whole TTL) remains. The terminal's implementation is deliberately left untouched.

Package service: session sandbox terminal.

The interactive terminal is a lookup-only entry point into the sandbox bound to a session: it resolves the session's pinned config exactly like artifact collection does (never "whatever the agent points at today"), refuses to provision when no live sandbox exists, and hands a provider- neutral PTY handle to the WebSocket layer.

Package service - session fork.

Fork copies a session's history through a chosen user or assistant message into a brand new session, and arranges for that new session's first sandbox to boot from a snapshot of the source sandbox rolled back to the fork point.

The whole operation is cheap and synchronous: it writes the database and takes one provider snapshot. No sandbox is created here — provisioning stays lazy, exactly as it is for ordinary sessions.

Package service - in-place session rewind.

Rewind truncates the current session at a chosen user or assistant message and, when the live sandbox still holds that turn's git checkpoint, rolls /workspace back to it. Unlike fork it does not snapshot, does not create a session, and refuses the whole operation if git reset fails — a half-applied rewind (conversation gone, files still ahead) is harder to recover from than a 500 the user can retry.

Unopened forks are the exception to "no sandbox means conversation only": they still carry ForkBootstrap for the first provision. Rewind retargets that SHA to the kept history, or drops the bootstrap when nothing remains to reset to, so the lazy sandbox cannot boot ahead of the transcript.

Package service: session -> sandbox config pin.

A session's remote sandbox is long-lived (created with onTimeout=pause, so its TTL pauses rather than destroys it), while the operations around it - skill execution, attachment upload, artifact collection, teardown - each resolve a backend independently at different points in time.

If those points read "whichever config the agent points at right now", an admin re-pointing an agent mid-conversation would make artifact collection query the wrong account (artifacts silently vanish) and teardown fail (a paused sandbox nobody knows the ID of, billing forever).

The pin records which config the session's CURRENT sandbox lives on, and which workspace owns that config. Both halves are needed: configs are keyed by (tenant, id), so an id alone does not name a backend. It is deliberately ephemeral: sessions outlive sandboxes by months, so a permanent pin would make "nothing references this config" never become true.

Package service: sandbox backend config management.

The load-bearing rule: some fields decide whether a config can still OPERATE the sandboxes it already created. Overwriting them in place while sandboxes are alive breaks those sandboxes in one of two ways, and both are bad enough to refuse the save.

Losing the control plane (provider, API URL, API key) means the new credentials have no authority over the old sandboxes, so they can no longer be listed or deleted. Session sandboxes are created with onTimeout=pause, so the provider TTL never reclaims them either - the leak is permanent and it bills. A paused sandbox is also what a session expects to RESUME, so conversations break on top.

Losing the data plane (E2B sandbox domain; Cube proxy URL and sandbox domain) keeps cleanup possible, but every envd request now goes to the wrong host. Every live session on this config fails at once - skill execution, attachment staging, artifact collection - while the sandboxes stay alive and keep billing until their sessions are deleted.

Hence both groups are refused while sandboxes exist. The admin ends the owning sessions or creates a second config and re-points agents; both keep the old values intact, which is what keeps cleanup possible.

Package service - per-tenant sandbox resolution helpers.

The sandbox package must not depend on repositories, so the config lookup is adapted here and injected as sandbox.TenantSandboxConfigLoader.

Package service - per-turn /workspace git checkpoints.

WorkspaceCheckpointer commits the session sandbox's /workspace at the end of every agent turn so session fork and rewind can later roll a sandbox back to a specific turn. Git metadata lives at sandbox.SessionGitDir, outside /workspace, so wiping the work tree cannot drop checkpoint history. It mirrors ArtifactCollector's shape deliberately: a narrow injectable interface for the sandbox side, and strict best-effort semantics so an auxiliary feature can never break a reply.

Contract:

  • Never returns an error. A failed checkpoint yields nil, the message simply carries no checkpoint, and that fork point degrades.
  • Never lazy-creates a sandbox: an empty sandboxID short-circuits.
  • Idempotent inside the sandbox: the script initialises the repository only when absent, so pause/resume and sandbox rebuilds self-heal.

Package service - rolling a session sandbox's /workspace back to a commit.

Two callers share this: ForkBootstrapper rolls a forked session's first sandbox back to the fork point, and SessionRewindService rolls the session's own live sandbox back to a rewind point. Both need identical semantics, so the script and the exec wrapper live here rather than in either one.

Index

Constants

View Source 
const (
	// DefaultLLMTemperature Use low temperature for more deterministic results
	DefaultLLMTemperature = 0.1

	// PMIWeight Proportion of PMI in calculating relationship weight
	PMIWeight = 0.6

	// StrengthWeight Proportion of relationship strength in calculating relationship weight
	StrengthWeight = 0.4

	// IndirectRelationWeightDecay Decay coefficient for indirect relationship weights
	IndirectRelationWeightDecay = 0.5

	// MaxConcurrentEntityExtractions Maximum concurrency for entity extraction
	MaxConcurrentEntityExtractions = 4

	// MaxConcurrentRelationExtractions Maximum concurrency for relationship extraction
	MaxConcurrentRelationExtractions = 4

	// DefaultRelationBatchSize Default batch size for relationship extraction
	DefaultRelationBatchSize = 5

	// MinEntitiesForRelation Minimum number of entities required for relationship extraction
	MinEntitiesForRelation = 2

	// MinWeightValue Minimum weight value to avoid division by zero
	MinWeightValue = 1.0

	// WeightScaleFactor Weight scaling factor to normalize weights to 1-10 range
	WeightScaleFactor = 9.0
)
View Source 
const (
	ImageCaptionSwitchKey = captionFieldKeyCaptionEnabled
	ImageOCRSwitchKey     = captionFieldKeyOCREnabled
)

ImageCaptionSwitchKey and ImageOCRSwitchKey are this pipeline's two action switches, under the names it declares them. They are exported because the knowledge-base PATCH handler maps the API's image_caption_enabled / image_ocr_enabled onto them: a second spelling of the key at that call site is exactly how a rename comes to write a setting nothing reads.

View Source 
const (

	// MaxEnvValueBytes bounds one stored value. It is generous enough for a
	// PEM-encoded key and far short of anything that belongs in a file.
	MaxEnvValueBytes = 8 * 1024

	// MaxUserEnvVarsPerScope bounds how many variables one principal may keep
	// in one scope: a sandbox config, or one skill on it.
	MaxUserEnvVarsPerScope = 50
)
View Source 
const (
	DetailInvalidOldPassword = "invalid_old_password"
	DetailPasswordPolicy     = "password_policy"
	DetailSamePassword       = "same_password"
)

Machine-readable change-password failure reasons for HTTP details fields.

View Source 
const (
	EnvSourceUnset     = "unset"
	EnvSourceWorkspace = "workspace"
	EnvSourceUser      = "user"
)

Where the value a skill would actually receive comes from. The three states are distinct on purpose: "the workspace filled this in for everybody" and "I filled this in for myself" lead to different actions, and neither is "nobody has filled this in".

View Source 
const (
	WikiOpIngest  = "ingest"
	WikiOpRetract = "retract"
)
View Source 
const DefaultForkSnapshotLeaseGrace = 15 * time.Minute

DefaultForkSnapshotLeaseGrace is how long a snapshot lease lives before the reaper treats it as abandoned. It must outlast CreateForked (milliseconds) and a brief process restart, but not a week — there is no session for the user to come back to.

View Source 
const DefaultForkSnapshotRetention = 7 * 24 * time.Hour

DefaultForkSnapshotRetention is how long an unopened fork keeps its snapshot. A week is long enough that "I'll get back to that branch on Monday" works, and short enough that abandoned forks do not accumulate.

View Source 
const DefaultInviteCodeValidityDays = 7

Default invite code validity in days; allowed values: 0 (never), 1, 7, 30

View Source 
const DefaultMemberLimit = 200

DefaultMemberLimit is the default max tenant-members per organization (0 = unlimited)

View Source 
const DefaultSandboxDesktopLastTTL = 7 * 24 * time.Hour

DefaultSandboxDesktopLastTTL is long enough to outlive a paused sandbox and a skill-install rebuild so a reconnect on another replica still sees the previous sandbox ID.

View Source 
const DefaultSandboxDesktopTicketTTL = 2 * time.Minute

DefaultSandboxDesktopTicketTTL is long enough for the WebSocket handshake, short enough that a leaked query string is not a session.

View Source 
const (

	// DefaultSandboxTerminalTicketTTL is long enough for the WebSocket
	// handshake, short enough that a leaked query string is not a 24h
	// session. After upgrade the ticket itself is discarded; the open PTY
	// rechecks the bound access-token id instead.
	DefaultSandboxTerminalTicketTTL = 2 * time.Minute
)
View Source 
const MAX_ITERATIONS = 100 // Max iterations for agent execution
View Source 
const MaxCatalogOverlayBytes = 1024 * 1024

MaxCatalogOverlayBytes bounds a console overlay document.

Variables

View Source 
var (
	ErrAgentShareNotFound      = errors.New("agent share not found")
	ErrAgentSharePermission    = errors.New("permission denied for this share operation")
	ErrAgentNotFoundForShare   = errors.New("agent not found")
	ErrNotAgentOwner           = errors.New("only agent owner can share")
	ErrOrgRoleCannotShareAgent = errors.New("only editors and admins can share agents to this organization")
	// ErrBuiltinAgentNotShareable is returned for built-in agents: every
	// workspace has its own copy under the same ID, so a shared built-in would
	// be indistinguishable from the receiver's own agent.
	ErrBuiltinAgentNotShareable = errors.New("built-in agents cannot be shared")
	ErrAgentNotConfigured       = errors.New(
		"agent is not fully configured (missing required chat model, or rerank model when the search_knowledge " +
			"tool is enabled)",
	)
)
View Source 
var (
	ErrAgentNotFound       = errors.New("agent not found")
	ErrCannotModifyBuiltin = errors.New("cannot modify built-in agent basic info")
	ErrCannotDeleteBuiltin = errors.New("cannot delete built-in agent")
	ErrAgentNameRequired   = errors.New("agent name is required")
)

Custom agent related errors

View Source 
var (
	ErrEmbedChannelNotFound = errors.New("embed channel not found")
	ErrEmbedTokenInvalid    = errors.New("embed publish token is invalid")
	ErrEmbedChannelDisabled = errors.New("embed channel is disabled")
	ErrEmbedChunkNotFound   = errors.New("embed chunk not found")
	ErrEmbedChunkForbidden  = errors.New("embed chunk not accessible")
)
View Source 
var (
	ErrShareNotFound         = errors.New("share not found")
	ErrSharePermissionDenied = errors.New("permission denied for this share operation")
	ErrKBNotFound            = errors.New("knowledge base not found")
	ErrNotKBOwner            = errors.New("only knowledge base owner can share")
	// ErrOrgRoleCannotShare: only editors and admins (in tenant's org role) may share KBs to that org; viewers cannot
	ErrOrgRoleCannotShare = errors.New("only editors and admins can share knowledge bases to this organization")
)
View Source 
var (
	// ErrInvalidFileType is returned when an unsupported file type is provided
	ErrInvalidFileType = errors.New("unsupported file type")
	// ErrInvalidURL is returned when an invalid URL is provided
	ErrInvalidURL = errors.New("invalid URL")
	// ErrChunkNotFound is returned when a requested chunk cannot be found.
	// Aliases the repository sentinel so a chunk-not-found from the repo
	// errors.Is-matches at the service and middleware layers (a single
	// identity instead of two string-equal-but-distinct errors).
	ErrChunkNotFound = repository.ErrChunkNotFound
	// ErrDuplicateFile is returned when trying to add a file that already exists
	ErrDuplicateFile = errors.New("file already exists")
	// ErrDuplicateURL is returned when trying to add a URL that already exists
	ErrDuplicateURL = errors.New("URL already exists")
	// ErrImageNotParse is returned when trying to update image information without enabling multimodel
	ErrImageNotParse = errors.New("image not parse without enable multimodel")
)

Error definitions for knowledge service operations

View Source 
var (
	// ErrKnowledgeBaseProfileUnsupported aliases the types sentinel.
	ErrKnowledgeBaseProfileUnsupported = types.ErrKnowledgeBaseProfileUnsupported
	// ErrKnowledgeBaseProfileModelNotConfigured aliases the types sentinel.
	ErrKnowledgeBaseProfileModelNotConfigured = types.ErrKnowledgeBaseProfileModelNotConfigured
)
View Source 
var (
	// ErrMCPEndpointNotFound is returned when an endpoint does not exist or
	// belongs to another workspace; the two cases are deliberately
	// indistinguishable to callers.
	ErrMCPEndpointNotFound = errors.New("mcp endpoint not found")
	// ErrMCPEndpointDisabled is returned when an otherwise valid token targets
	// an endpoint that has been switched off.
	ErrMCPEndpointDisabled = errors.New("mcp endpoint disabled")
	// ErrMCPEndpointTokenInvalid is returned when the bearer token does not
	// resolve to the addressed endpoint.
	ErrMCPEndpointTokenInvalid = errors.New("mcp endpoint token invalid")
)
View Source 
var (
	ErrOrgNotFound           = errors.New("organization not found")
	ErrOrgPermissionDenied   = errors.New("permission denied for this organization")
	ErrCannotRemoveOwner     = errors.New("cannot remove organization owner tenant")
	ErrCannotChangeOwnerRole = errors.New("cannot change organization owner tenant role")
	ErrTenantNotInOrg        = errors.New("tenant is not a member of this organization")
	ErrInvalidRole           = errors.New("invalid role")
	ErrInviteCodeExpired     = errors.New("invite code has expired")
	ErrInvalidValidityDays   = errors.New("invite_code_validity_days must be 0, 1, 7, or 30")
	ErrOrgMemberLimitReached = errors.New("organization member limit reached")
	ErrOrgMemberLimitTooLow  = errors.New("member limit cannot be lower than current member count")
)
View Source 
var (
	ErrPendingRequestExists    = errors.New("pending request already exists")
	ErrJoinRequestNotFound     = errors.New("join request not found")
	ErrCannotUpgradeToSameRole = errors.New("cannot request upgrade to same or lower role")
	ErrAlreadyAdmin            = errors.New("tenant is already an admin")
)
View Source 
var (
	// ErrDesktopUnsupported means this session can never show a desktop as
	// currently configured: either the backend cannot relay one, or the
	// config's image has no start-desktop.sh.
	ErrDesktopUnsupported = errors.New("sandbox desktop is not supported")

	// ErrDesktopStartFailed means the image is right but the desktop did not
	// come up. Retryable.
	ErrDesktopStartFailed = errors.New("sandbox desktop failed to start")
)
View Source 
var (
	// ErrForkSourceBusy reports that the source session has an agent turn in
	// flight. Snapshotting pauses the source sandbox on every backend, which
	// would interrupt a tool mid-execution, so the fork is refused rather than
	// queued. HTTP 409.
	ErrForkSourceBusy = errors.New("session fork: source session has an active turn")

	// ErrForkSessionNotFound covers a missing session and a session the caller
	// does not own. Both map to HTTP 404 so ownership is not enumerable.
	ErrForkSessionNotFound = errors.New("session fork: source session not found")

	// ErrForkMessageNotFound covers a missing message and a message that
	// belongs to a different session. HTTP 404.
	ErrForkMessageNotFound = errors.New("session fork: fork point message not found")

	// ErrForkMessageNotUser is returned when the fork point exists but is
	// neither a user nor an assistant message. HTTP 400.
	ErrForkMessageNotUser = errors.New("session fork: fork point must be a user or assistant message")
)
View Source 
var (
	// ErrRewindSourceBusy reports that the session has an agent turn in
	// flight, or the rewind point is an unfinished assistant message. HTTP 409.
	ErrRewindSourceBusy = errors.New("session rewind: session has an active turn")

	// ErrRewindSessionNotFound covers a missing session and a session the
	// caller does not own. Both map to HTTP 404 so ownership is not enumerable.
	ErrRewindSessionNotFound = errors.New("session rewind: session not found")

	// ErrRewindMessageNotFound covers a missing message and a message that
	// belongs to a different session. HTTP 404.
	ErrRewindMessageNotFound = errors.New("session rewind: rewind point message not found")

	// ErrRewindMessageRole is returned when the rewind point exists but is
	// neither a user nor an assistant message. HTTP 400.
	ErrRewindMessageRole = errors.New("session rewind: rewind point must be a user or assistant message")

	// ErrRewindNoCheckpoint is returned when the session has a live sandbox
	// and kept history includes an assistant turn, but no reachable git SHA.
	// Truncating would leave files ahead of the conversation. HTTP 409.
	ErrRewindNoCheckpoint = errors.New("session rewind: no reachable workspace checkpoint")

	// ErrRewindSandboxReplaced is returned when the kept checkpoint belongs
	// to a sandbox the session no longer uses. Truncating would leave the
	// live workspace ahead of the conversation. HTTP 409.
	ErrRewindSandboxReplaced = errors.New("session rewind: sandbox was replaced")
)
View Source 
var (
	// ErrInvitationNotFound is returned when no invitation row matches
	// the supplied id.
	ErrInvitationNotFound = errors.New("invitation not found")

	// ErrPendingInvitationExists is returned by Create when a pending
	// invitation for (tenant, invitee) is already in flight. The
	// handler maps this to 409.
	ErrPendingInvitationExists = errors.New("a pending invitation for this user already exists")

	// ErrAlreadyMember is returned by Create when the invitee is
	// already an active member of the tenant; sending an invite would
	// be a no-op at best and a confusing UX at worst.
	ErrAlreadyMember = errors.New("user is already an active member of the tenant")

	// ErrInvitationNotPending is returned by Accept / Decline / Revoke
	// when the row exists but has already been finalised. Maps to 409.
	ErrInvitationNotPending = errors.New("invitation is no longer pending")

	// ErrInvitationExpired is returned by Accept / Decline when the
	// pending row has aged past its expires_at. Maps to 410.
	ErrInvitationExpired = errors.New("invitation has expired")

	// ErrInvitationForbidden is returned by Accept / Decline when the
	// caller is not the invitee. Owner-driven Revoke is gated at the
	// route layer so this error only surfaces on the /me/ paths.
	ErrInvitationForbidden = errors.New("only the invitee can accept or decline this invitation")

	// ErrInvitationTokenInvalid is returned by LookupByToken /
	// AcceptByToken when the supplied plaintext token does not match
	// any active share-link row. The handler maps this to 410 Gone.
	// We deliberately collapse "unknown" / "expired" / "revoked" into
	// a single sentinel so an attacker can't probe which slots used
	// to exist.
	ErrInvitationTokenInvalid = errors.New("invitation token is invalid or has been revoked")
)

Sentinel errors returned by tenantInvitationService. Callers compare with errors.Is to render the appropriate HTTP responses.

View Source 
var (
	// ErrMembershipNotFound is returned when no active membership row
	// matches the (user, tenant) pair the caller requested.
	ErrMembershipNotFound = errors.New("tenant membership not found")

	// ErrMembershipAlreadyExists is returned by AddMember when the
	// (user, tenant) pair already has an active membership.
	ErrMembershipAlreadyExists = errors.New("tenant membership already exists")

	// ErrInvalidTenantRole is returned when the caller passes a role
	// value that is not one of the four defined TenantRole constants.
	ErrInvalidTenantRole = errors.New("invalid tenant role")

	// ErrAPIKeyCannotAssignOwner is returned when an API-key principal
	// attempts to persist the Owner role through member or invitation
	// management. manage_members deliberately excludes ownership transfer:
	// a machine principal may manage lower roles, but must never mint a
	// durable human Owner who could subsequently manage API keys or delete
	// the tenant.
	ErrAPIKeyCannotAssignOwner = errors.New("API keys cannot assign the owner role")

	// ErrLastOwner is returned when an operation would leave the tenant
	// without an active Owner. Demoting the last Owner or removing them
	// is forbidden; an explicit ownership transfer must happen first.
	ErrLastOwner = errors.New("cannot demote or remove the last active owner of the tenant")
)

Sentinel errors returned by tenantMemberService. Callers compare with errors.Is to render appropriate HTTP responses (404 / 409 / 403).

View Source 
var (

	// ErrUserEmailExists is returned by Register when the target entity's
	// email already exists.
	ErrUserEmailExists = errors.New("user with this email already exists")

	// ErrUserUsernameExists is returned by Register when the target entity's
	// username already exists.
	ErrUserUsernameExists = errors.New("user with this username already exists")

	// ErrUserIdentityConflict is returned by AdminCreateUser when only part
	// of the requested identity (email or username) collides with an existing
	// user, so a blind idempotent retry would return the wrong account.
	ErrUserIdentityConflict = errors.New("email and username refer to conflicting existing identities")

	// ErrPasswordPolicy is returned when a newly chosen password does not
	// meet the product's public 8-32 character, letter-and-number contract.
	// It is exported so HTTP handlers can translate the failure to a 400
	// without exposing bcrypt or persistence errors.
	ErrPasswordPolicy = errors.New(
		"password must be 8-32 characters and contain at least one letter and one number")
	// ErrComplexPasswordPolicy is returned when the runtime complex-password
	// switch is on and the new password is missing a required character class.
	ErrComplexPasswordPolicy = errors.New(
		"password must be 8-32 characters and must contain uppercase and lowercase " +
			"letters, numbers, and special characters")

	// ErrInvalidOldPassword is returned by ChangePassword when the supplied
	// current password does not match the stored hash. Handlers map this to
	// a 400 so callers can prompt the user without treating it as a 500.
	ErrInvalidOldPassword = errors.New("invalid old password")

	// ErrSamePassword is returned when the new password equals the current
	// one so callers can reject no-op rotations that would still revoke
	// every session.
	ErrSamePassword = errors.New("new password must differ from current password")
)
View Source 
var (
	ErrCannotDisableSelf      = apprepo.ErrCannotDisableSelf
	ErrCannotDisableLastAdmin = apprepo.ErrCannotDisableLastAdmin
	ErrUserNotFound           = apprepo.ErrUserNotFound
)

Account-status errors share repository identity for errors.Is at the HTTP boundary.

View Source 
var (
	ErrFavoriteInvalidType = errors.New("invalid favorite resource type")
	ErrFavoriteEmptyID     = errors.New("favorite resource id is required")
)

Sentinel errors so the handler can map cleanly to HTTP status codes without leaking GORM internals.

View Source 
var DefaultAutomationConfig = AutomationConfig{
	Tick:                15 * time.Second,
	Lease:               2 * time.Minute,
	MaxConcurrent:       4,
	MaxPerTenant:        100,
	Retention:           200,
	MinInterval:         automation.DefaultMinInterval,
	DefaultMaxDuration:  30 * time.Minute,
	FailuresBeforePause: 3,
}

DefaultAutomationConfig is used for every zero field.

View Source 
var ErrAgentKBScopeNotShareable = errors.New(
	"the agent's knowledge bases include ones you cannot share; only their creators or workspace admins can",
)

ErrAgentKBScopeNotShareable is returned when an agent share would expose knowledge bases its sharer could not share directly.

View Source 
var ErrArtifactNotFound = errors.NewNotFoundError("artifact not found")

ErrArtifactNotFound is returned when the addressed artifact does not exist, belongs to a message that is gone, or was already deleted. Handlers map it to 404 so a caller cannot probe which of the three it was.

View Source 
var ErrChunkRevisionConflict = repository.ErrChunkRevisionConflict
View Source 
var ErrDesktopTicketInvalid = errors.New("invalid or expired desktop ticket")

ErrDesktopTicketInvalid covers unknown, expired, and already-consumed tickets. They are one error on purpose: telling them apart would let a caller probe which tickets ever existed.

View Source 
var ErrEmbedSessionUnavailable = errors.New("embed session tokens unavailable")
View Source 
var ErrEmbedWebhookURLInvalid = errors.New("invalid embed webhook URL")

ErrEmbedWebhookURLInvalid is returned when a webhook URL fails format or SSRF checks.

View Source 
var ErrInvalidCatalog = errors.New("invalid model catalog")

ErrInvalidCatalog marks overlays rejected by validation or compilation.

View Source 
var ErrInvalidTenantID = errors.New("invalid tenant ID")

ErrInvalidTenantID represents an error for invalid tenant ID

View Source 
var ErrModelNotFound = errors.New("model not found")

ErrModelNotFound is returned when a model cannot be found in the repository

View Source 
var ErrNamedSandboxBackendUnsupported = stderrors.New(
	"named sandbox configs only support cube, e2b and docker backends",
)

ErrNamedSandboxBackendUnsupported marks a sandbox type that cannot be stored as a user-facing named backend config.

View Source 
var ErrSandboxConfigCreatesInFlight = stderrors.New("sandbox config still has sandboxes being created")

ErrSandboxConfigCreatesInFlight is returned while a sandbox of this config is still being created; retry the deletion once it finishes.

View Source 
var ErrSandboxConfigNameRequired = stderrors.New("sandbox config name is required")

ErrSandboxConfigNameRequired is a sentinel so transports can classify this as bad input without matching on the message text.

View Source 
var ErrSandboxInventoryUnverifiable = stderrors.New(
	"cannot verify whether the sandbox config still owns sandboxes")

ErrSandboxInventoryUnverifiable is returned when the provider cannot be reached to answer "does this config still own sandboxes?".

It is deliberately distinct from ErrSandboxesStillLive: one means "there ARE sandboxes, deal with them", the other means "we cannot tell". The second is the only case a force delete may override, because an endpoint whose DNS record is gone would otherwise make its config permanently undeletable.

View Source 
var ErrSandboxesStillLive = stderrors.New("sandbox config still owns live sandboxes")

ErrSandboxesStillLive is returned when an identity change or deletion is refused because the current credentials still own provider resources.

View Source 
var ErrSkillBundleInvalid = errors.New("skill bundle is invalid")

ErrSkillBundleInvalid marks every rejection of an uploaded archive, so the handler can map the whole class to 400 without matching on message text.

View Source 
var ErrSkillSnapshotBlocksTemplateChange = stderrors.New(
	"sandbox connection cannot change while this config has a skill snapshot")

ErrSkillSnapshotBlocksTemplateChange is returned when this config already has a skill snapshot and the caller tries to retarget its connection, DNS, or spawn template, or to rebuild the cluster's WeKnora template. Sessions boot the snapshot, so those edits would not reach them. The way out is a second config.

View Source 
var ErrSkillSnapshotReleaseFailed = stderrors.New("failed to release skill snapshots")

ErrSkillSnapshotReleaseFailed is returned when deleting a sandbox config cannot destroy every provider snapshot on that config's ledger. Leaving the row is recoverable; leaking a billable snapshot is not.

View Source 
var ErrSkillSourceInvalid = errors.New("skill source is invalid")

ErrSkillSourceInvalid marks every rejection of a registry / git / URL source so the handler can map the class to 400 without matching on text.

View Source 
var ErrSummaryRefreshStale = errors.New("summary refresh superseded")

ErrSummaryRefreshStale means the source chunks or metadata changed while a summary refresh was running. The caller should discard the result and leave the current summary_status untouched so a newer refresh can finish.

View Source 
var ErrTerminalAuthDenied = errors.New("terminal authorization no longer valid")

ErrTerminalAuthDenied is a definite authorization failure for an open sandbox terminal: the minting access token is gone or revoked, the user is inactive or deleted, membership was removed, or the session is no longer owned.

Every other error is a lookup failure and is returned as-is. The distinction matters: this runs on a timer against every open terminal, so treating a database hiccup as "denied" would disconnect the whole fleet at once and then have it all reconnect together. The middleware takes the same stance (see resolveTenantRole's membership lookup).

View Source 
var ErrTerminalUnsupported = sandbox.ErrTerminalUnsupported

ErrTerminalUnsupported reports that the session's sandbox backend cannot serve interactive terminals. Same sentinel as sandbox.ErrTerminalUnsupported so errors.Is matches either name.

View Source 
var ErrUserSandboxBusy = stderrors.New("user sandbox is busy")

ErrUserSandboxBusy is returned while a terminal, desktop or command is still running on the machine; the admin is asked to retry once the user has closed it.

View Source 
var ErrWikiIngestConcurrent = errors.New("concurrent wiki task active")

ErrWikiIngestConcurrent is returned by the wiki ingest handler in Lite mode when another batch is already running for the same KB (the in-process liteLocks guard is held). The asynq/sync executor's RetryDelayFunc uses errors.Is on this sentinel to apply a short, fixed retry delay instead of exponential backoff, so the deferred batch retries promptly once the active one releases. Standard (Redis) mode no longer takes an exclusive per-KB lock (Phase 3) and never returns this.

View Source 
var ErrWikiRevertToCurrentVersion = errors.New("cannot revert to the current version")

ErrWikiRevertToCurrentVersion is returned when a revert targets the version the page is already on — a client-side mistake (usually a stale history list), not a server fault, so handlers map it to 400.

View Source 
var ErrWikiWriteDroppedTableRows = errors.New("wiki page write dropped table rows the stored page still has")

ErrWikiWriteDroppedTableRows is the explicit refusal signal a machine write gets when the incoming body no longer carries table rows the stored page still has. The stored page is left untouched.

It is returned for both guarded policies — see UpdatePage — and it exists because an unchanged `version` cannot stand in for it: UpdatePage leaves the version alone for a successful write that changes nothing user-visible, so callers that must retry a refused contribution (wiki ingest's reduce) need a signal of its own. See wikiWriteMissingRowIdentities.

View Source 
var ValidInviteCodeValidityDays = map[int]bool{0: true, 1: true, 7: true, 30: true}

ValidInviteCodeValidityDays are the allowed values for invite_code_validity_days

Functions

func ApplyKnowledgeProcessOverrides 

func ApplyKnowledgeProcessOverrides(
	ctx context.Context,
	kb *types.KnowledgeBase,
	knowledge *types.Knowledge,
	processOverrides *types.KnowledgeProcessOverrides,
	fileTypes []string,
	enableMultimodel *bool,
) (types.EffectiveProcessConfig, error)

ApplyKnowledgeProcessOverrides validates optional overrides, persists them on the knowledge record, and returns the effective config for task enqueue.

func AssertAccessTokenNotRevoked 

func AssertAccessTokenNotRevoked(token *types.AuthToken, userID string) error

AssertAccessTokenNotRevoked reports whether a stored access-token row is still the minting credential for userID (right owner and type, not revoked). Expiry is checked separately: a live PTY must survive silent access-token rotation, while minting a new ticket still requires a current token.

func AssertAccessTokenStillActive 

func AssertAccessTokenStillActive(token *types.AuthToken, userID string, now time.Time) error

AssertAccessTokenStillActive is the minting check: not revoked and not expired. The handshake ticket POST uses this so a stale Bearer token cannot open a new PTY.

func BrokenImagePipelineRules added in v0.9.0 

func BrokenImagePipelineRules(
	pipelineID types.ImagePipelineID,
	params map[string]any,
) []*types.ImagePipelineValidationError

BrokenImagePipelineRules lists every rule of the named pipeline that the stored tunables break. Unlike the single error ValidateImagePipelineParams returns, this keeps each violation separate so that the settings panel can show all of them at once — one rule per line, each pointing at the control that has to change. A pipeline that declares no rules yields none, which is how a pipeline whose actions run by themselves is never told it is wrong.

func CheckSandboxTerminalAuth 

func CheckSandboxTerminalAuth(
	ctx context.Context,
	users interfaces.UserService,
	members interfaces.TenantMemberService,
	sessions interfaces.SessionService,
	claims SandboxTerminalTicketClaims,
	rbacEnforced bool,
	rejectExpired bool,
) (*types.User, error)

CheckSandboxTerminalAuth re-validates the identity that opened a PTY and returns the resolved user, so the handshake can install the auth session without looking the user up a second time.

Call it before the upgrade and periodically on the live bridge; using the one function for both is deliberate, so a live terminal is held to exactly the checks that admitted it — except token expiry. Handshake minting and the upgrade still require a non-expired access token; the live recheck does not, because axios silently rotates access tokens without revoking the row the ticket was bound to. Logout / password reset still revoke every row and tear the PTY down.

rejectExpired is true for the handshake and false for the open-PTY timer.

func ConvertWebSearchResults 

func ConvertWebSearchResults(webResults []*types.WebSearchResult) []*types.SearchResult

ConvertWebSearchResults converts WebSearchResult to SearchResult

func DecideOCR 

func DecideOCR(attrs types.ImageAttrs, actions types.ImageActionsConfig, onUnobserved bool) bool

DecideOCR is a pure function: no ctx, no repo, no model. Input the observed attributes, the policy table and the caller's clause for an unobserved image; output whether OCR runs. It is the single place the "need OCR" decision lives, so it can be unit-tested and replayed.

The default policy encodes, for every attribute the table reads: 

ocr = (contain.text == "block")
    || (contain.data_visual == true)
    || the attribute was not observed (conservative: onUnobserved)

where "not observed" covers both a line the model never wrote and a value that was not one of the allowed ones — the parser leaves such an attribute absent rather than inventing a default. An incomplete observation therefore keeps extracting, so a missed block of text costs one extra call instead of being lost.

The clause arrives as an argument rather than inside the table because it is not an attribute match: it decides whether OCR is scheduled at all, which is the scheduling pipeline's decision (its ocr_on_unobserved setting).

The result is equivalent to the old 7-class table: decorative / logo / photo (text reliably absent or captured by the description) skip OCR; text_screenshot / table / chart / other run OCR.

func DefaultDataset 

func DefaultDataset() dataset

DefaultDataset loads and initializes the default dataset from parquet files

func DispatchEmbedWebhook 

func DispatchEmbedWebhook(ch *types.EmbedChannel, eventType, sessionID string, payload map[string]any)

DispatchEmbedWebhook POSTs an event to the channel webhook URL (best-effort, async).

func EmbedSessionDescription 

func EmbedSessionDescription(channelID string) string

EmbedSessionDescription returns the marker stored on embed-created sessions.

func EnqueueWikiIngest 

func EnqueueWikiIngest(
	ctx context.Context,
	task interfaces.TaskEnqueuer,
	pendingRepo interfaces.TaskPendingOpsRepository,
	tenantID uint64,
	kbID, knowledgeID string,
) (bool, error)

func EnqueueWikiRetract 

func EnqueueWikiRetract(ctx context.Context, task interfaces.TaskEnqueuer,
	pendingRepo interfaces.TaskPendingOpsRepository, payload WikiRetractPayload,
)

EnqueueWikiRetract queues a wiki retraction op (a delete cleanup). Identical persistence model as EnqueueWikiIngest — the op rides in task_pending_ops and an asynq trigger fires shortly after to process the batch. Retracts use a slightly shorter ProcessIn delay because there is no "user upload arriving in waves" pattern to debounce against — a deletion fires once and we want the cleanup to land promptly.

func EvaluateImagePipelineRules added in v0.9.0 

func EvaluateImagePipelineRules(
	pipelineID types.ImagePipelineID,
	rules []types.ImagePipelineRules,
	fields []types.ImageFieldDef,
	params map[string]any,
) []*types.ImagePipelineValidationError

EvaluateImagePipelineRules walks a pipeline's rules and reports every one its tunables break, in declaration order. Reporting all of them rather than the first is the point: each rule carries its own wording and its own field, so a panel that shows three broken rules tells the user exactly what to change instead of making them fix one at a time and re-read the page. An empty result is a configuration the pipeline will run.

A rule names the fields it is about rather than a condition to compute: "at least one of these is on" is what a set of action switches can mean, and a pipeline that offers none to switch off declares no rule at all rather than one that always holds.

A field the parameters do not mention reads its declared default, the same way the run does: a knowledge base saved before a field existed is not told that it switched something off by never mentioning it.

func HashMCPEndpointToken 

func HashMCPEndpointToken(token string) string

HashMCPEndpointToken returns the storage digest of a plaintext token.

func IsAudioType 

func IsAudioType(fileType string) bool

IsAudioType checks if a file type is an audio format

func IsEmbedSessionToken 

func IsEmbedSessionToken(token string) bool

IsEmbedSessionToken reports whether token is a session token (ems_ prefix).

func IsImageType 

func IsImageType(fileType string) bool

func IsPasswordPolicyError 

func IsPasswordPolicyError(err error) bool

IsPasswordPolicyError reports whether err is a documented password-policy failure. Handlers map both the simple and complex sentinels to the same machine-readable details token so the SPA can localize them.

func IsVideoType 

func IsVideoType(fileType string) bool

IsVideoType checks if a file type is a video format

func IsWeKnoraCloudDocReaderAddr 

func IsWeKnoraCloudDocReaderAddr(addr string) bool

func IssueSandboxTerminalTicket 

func IssueSandboxTerminalTicket(userID string, tenantID uint64, sessionID, tokenID string, ttl time.Duration) (string, error)

IssueSandboxTerminalTicket mints a short-lived JWT for the browser WebSocket handshake. It is not an access token: ValidateToken rejects it.

func ListImagePipelines added in v0.9.0 

func ListImagePipelines() []types.ImagePipelineSpec

ListImagePipelines renders every registered pipeline as the settings panel sees it. Sorted by id so the panel's option order is stable and does not follow package init order.

func LoadAgentHistory 

func LoadAgentHistory(
	ctx context.Context,
	messageRepo interfaces.MessageRepository,
	sessionID string,
	tokenBudget int,
	retainRetrievalHistory bool,
) ([]chat.Message, float64, error)

LoadAgentHistory rebuilds the multi-turn LLM context for an Agent-mode session directly from the persistent messages table. The result is a chronologically ordered list of chat.Message entries suitable for prepending to the current turn (without system prompt; the engine adds that itself).

For each historical turn it emits:

  1. A user message (RenderedContent if present, else Content, plus any image captions appended).
  2. For each AgentStep with non-terminal tool calls (i.e. excluding final_answer), an assistant message carrying the step's thought and tool_calls, followed by one tool message per tool result.
  3. A final assistant message with the canonical answer (msg.Content with <think> blocks stripped).

Turns lacking either user or assistant content are skipped. Turns are returned in chronological order.

History is sized in tokens, not turns. When a turn carries a context checkpoint (a compaction summary persisted by an earlier run), history starts from it: the summary replaces that turn and everything before it. The newest turns after that are replayed while they fit tokenBudget. A turn left out is gone (neither replayed nor summarized), so the budget is the whole window (agent.HistoryTokenBudget), past the compaction threshold: overflow reaches the first round's compaction, which summarizes it into a new checkpoint. Compaction, not a turn count, keeps a long session in the window. Every replayed message is tagged with its turn's assistant message ID so this run's compaction can persist that checkpoint.

Rows are read backwards a page at a time and stop at the checkpoint or once the budget is full, so a long session is never read whole. Turns are priced as the engine sends them, so retainRetrievalHistory must be the run's setting, and in the provider's tokens: estimates are scaled by the newest turn's measured ContextTokenScale, which is also returned so the engine's first compaction check uses the same scale. Budget and trigger must agree, or turns the loader drops can sit below a trigger that never fires.

DB is treated as the single source of truth — there is no Redis/in-memory cache layer above this function. Callers are expected to invoke it once per turn before handing the messages to the agent engine.

func MergeParserEngineOverrides 

func MergeParserEngineOverrides(tenantOverrides map[string]string, uploadOverrides map[string]string) map[string]string

MergeParserEngineOverrides merges upload overrides on top of tenant overrides safely.

func NewAgentService 

func NewAgentService(
	cfg *config.Config,
	modelService interfaces.ModelService,
	knowledgeBaseService interfaces.KnowledgeBaseService,
	knowledgeService interfaces.KnowledgeService,
	fileService interfaces.FileService,
	chunkService interfaces.ChunkService,
	mcpServiceService interfaces.MCPServiceService,
	mcpManager *mcp.MCPManager,
	eventBus *event.EventBus,
	db *gorm.DB,
	webSearchService interfaces.WebSearchService,
	duckdb *sql.DB,
	wikiPageService interfaces.WikiPageService,
	tenantService interfaces.TenantService,
	messageService interfaces.MessageService,
	memoryService interfaces.MemoryService,
	storageResolver interfaces.StorageBackendResolver,
	toolApprovalGate approval.MCPApproval,
	sandboxMgr sandbox.Manager,
	sandboxResolver sandbox.TenantSandboxResolver,
	sandboxPinner *SessionSandboxPinner,
	sandboxPolicy WorkspaceSandboxPolicy,
	hostSandbox HostSandboxManager,
	browserSkill *browserskill.Manager,
	userRepo interfaces.UserRepository,
	graphRepo interfaces.RetrieveGraphRepository,
	backgroundWork *BackgroundWork,
	shellJobs *ShellJobService,
	operationDispatcher *operations.HandlerDispatcher,
	customAgentService interfaces.CustomAgentService,
	collaboration *AgentCollaboration,
) interfaces.AgentService

NewAgentService creates a new agent service

func NewAgentShareService 

func NewAgentShareService(
	shareRepo interfaces.AgentShareRepository,
	disabledRepo interfaces.TenantDisabledSharedAgentRepository,
	orgRepo interfaces.OrganizationRepository,
	agentRepo interfaces.CustomAgentRepository,
	userRepo interfaces.UserRepository,
	webSearchProviderRepo interfaces.WebSearchProviderRepository,
	kbRepo interfaces.KnowledgeBaseRepository,
) interfaces.AgentShareService

NewAgentShareService creates a new agent share service

func NewAuditLogService 

func NewAuditLogService(repo interfaces.AuditLogRepository) interfaces.AuditLogService

NewAuditLogService constructs the production service.

func NewChunkExtractService 

func NewChunkExtractService(
	config *config.Config,
	modelService interfaces.ModelService,
	knowledgeBaseRepo interfaces.KnowledgeBaseRepository,
	knowledgeRepo interfaces.KnowledgeRepository,
	chunkRepo interfaces.ChunkRepository,
	graphEngine interfaces.RetrieveGraphRepository,
	spanTracker SpanTracker,
) interfaces.TaskHandler

NewChunkExtractService creates a new chunk extract service

func NewChunkExtractTask 

func NewChunkExtractTask(
	ctx context.Context,
	client interfaces.TaskEnqueuer,
	tenantID uint64,
	chunkID string,
	modelID string,
	knowledgeID string,
	attempt int,
	chunkIndex int,
) (bool, error)

NewChunkExtractTask creates a new chunk extract task. It returns (enqueued, err): enqueued is true only when a task was actually placed on the queue. When NEO4J is disabled the call is a no-op and returns (false, nil) — callers that seeded a pending-subtask counter for this chunk MUST release that slot, otherwise the parent knowledge stays stuck in "finalizing" forever (the graph subtask it's waiting on was never enqueued).

func NewChunkService 

func NewChunkService(
	chunkRepository interfaces.ChunkRepository,
	knowledgeRepo interfaces.KnowledgeRepository,
	kbRepository interfaces.KnowledgeBaseRepository,
	modelService interfaces.ModelService,
	retrieveEngine interfaces.RetrieveEngineRegistry,
	ownership retriever.TenantStoreOwnership,
	task interfaces.TaskEnqueuer,
	spanTracker SpanTracker,
) interfaces.ChunkService

NewChunkService creates a new chunk service It initializes a service with the provided chunk repository Parameters:

  • chunkRepository: Repository for chunk operations

Returns:

  • interfaces.ChunkService: Initialized chunk service implementation

func NewCustomAgentService 

func NewCustomAgentService(
	repo interfaces.CustomAgentRepository,
	chunkRepo interfaces.ChunkRepository,
	kbService interfaces.KnowledgeBaseService,
	kbShareService interfaces.KBShareService,
	wikiPageRepo interfaces.WikiPageRepository,
	tagRepo interfaces.KnowledgeTagRepository,
	knowledgeRepo interfaces.KnowledgeRepository,
	agentShareRepo interfaces.AgentShareRepository,
) interfaces.CustomAgentService

NewCustomAgentService creates a new custom agent service

func NewDataSourceService 

func NewDataSourceService(
	dsRepo interfaces.DataSourceRepository,
	syncLogRepo interfaces.SyncLogRepository,
	knowledgeService interfaces.KnowledgeService,
	kbService interfaces.KnowledgeBaseService,
	taskEnqueuer interfaces.TaskEnqueuer,
	connectorRegistry *datasource.ConnectorRegistry,
	scheduler *datasource.Scheduler,
	tenantRepo interfaces.TenantRepository,
	tagService interfaces.KnowledgeTagService,
	audit interfaces.AuditLogService,
) interfaces.DataSourceService

NewDataSourceService creates a new data source service

func NewDataTableSummaryService 

func NewDataTableSummaryService(
	modelService interfaces.ModelService,
	knowledgeBaseService interfaces.KnowledgeBaseService,
	knowledgeService interfaces.KnowledgeService,
	fileService interfaces.FileService,
	chunkService interfaces.ChunkService,
	tenantService interfaces.TenantService,
	retrieveEngine interfaces.RetrieveEngineRegistry,
	ownership retriever.TenantStoreOwnership,
	sqlDB *sql.DB,
	storageResolver interfaces.StorageBackendResolver,
) interfaces.TaskHandler

NewDataTableSummaryService creates a new DataTableSummaryService

func NewDataTableSummaryTask 

func NewDataTableSummaryTask(
	ctx context.Context,
	client interfaces.TaskEnqueuer,
	tenantID uint64,
	knowledgeID string,
	summaryModel string,
	embeddingModel string,
) error

NewTableExtractTask creates a new table extract task

func NewDatasetService 

func NewDatasetService() interfaces.DatasetService

NewDatasetService creates a new DatasetService instance

func NewEmbedChannelService 

func NewEmbedChannelService(
	repo interfaces.EmbedChannelRepository,
	agentService interfaces.CustomAgentService,
	chunkService interfaces.ChunkService,
	redisClient *redis.Client,
) interfaces.EmbedChannelService

func NewEvaluationService 

func NewEvaluationService(
	config *config.Config,
	dataset interfaces.DatasetService,
	knowledgeBaseService interfaces.KnowledgeBaseService,
	knowledgeService interfaces.KnowledgeService,
	sessionService interfaces.SessionService,
	modelService interfaces.ModelService,
) interfaces.EvaluationService

func NewGraphBuilder 

func NewGraphBuilder(config *config.Config, chatModel chat.Chat) types.GraphBuilder

NewGraphBuilder creates a new graph builder

func NewGraphSearchService added in v0.9.0 

func NewGraphSearchService(graph interfaces.GraphSearchRepository,
	evidence interfaces.GraphSearchEvidenceRepository,
) interfaces.GraphSearchService

NewGraphSearchService constructs the public-query provenance boundary.

func NewImageMultimodalService 

func NewImageMultimodalService(
	chunkService interfaces.ChunkService,
	modelService interfaces.ModelService,
	kbService interfaces.KnowledgeBaseService,
	knowledgeRepo interfaces.KnowledgeRepository,
	tenantRepo interfaces.TenantRepository,
	retrieveEngine interfaces.RetrieveEngineRegistry,
	ownership retriever.TenantStoreOwnership,
	ollamaService *ollama.OllamaService,
	taskEnqueuer interfaces.TaskEnqueuer,
	redisClient *redis.Client,
	fileSvc interfaces.FileService,
	storageResolver interfaces.StorageBackendResolver,
	resourceCatalog interfaces.ResourceCatalog,
	spanTracker SpanTracker,
	docReader interfaces.DocumentReader,
) interfaces.TaskHandler

func NewKBShareService 

func NewKBShareService(
	shareRepo interfaces.KBShareRepository,
	orgRepo interfaces.OrganizationRepository,
	kbRepo interfaces.KnowledgeBaseRepository,
	kgRepo interfaces.KnowledgeRepository,
	chunkRepo interfaces.ChunkRepository,
	audit interfaces.AuditLogService,
) interfaces.KBShareService

NewKBShareService creates a new knowledge base share service

func NewKnowledgeAutoTagService 

func NewKnowledgeAutoTagService(
	knowledgeRepo interfaces.KnowledgeRepository,
	tagRepo interfaces.KnowledgeTagRepository,
	kbService interfaces.KnowledgeBaseService,
	chunkRepo interfaces.ChunkRepository,
	modelService interfaces.ModelService,
	spanTracker SpanTracker,
) interfaces.TaskHandler

NewKnowledgeAutoTagService creates the asynchronous automatic-tagging task handler.

func NewKnowledgeBaseService 

func NewKnowledgeBaseService(repo interfaces.KnowledgeBaseRepository,
	kgRepo interfaces.KnowledgeRepository,
	chunkRepo interfaces.ChunkRepository,
	shareRepo interfaces.KBShareRepository,
	kbShareService interfaces.KBShareService,
	modelService interfaces.ModelService,
	retrieveEngine interfaces.RetrieveEngineRegistry,
	ownership retriever.TenantStoreOwnership,
	tenantRepo interfaces.TenantRepository,
	fileSvc interfaces.FileService,
	storageResolver interfaces.StorageBackendResolver,
	graphEngine interfaces.RetrieveGraphRepository,
	asynqClient interfaces.TaskEnqueuer,
	taskInspector interfaces.TaskInspector,
	taskPendingRepo interfaces.TaskPendingOpsRepository,
	dsRepo interfaces.DataSourceRepository,
	syncLogRepo interfaces.SyncLogRepository,
	dsScheduler *datasource.Scheduler,
	audit interfaces.AuditLogService,
	resourceCatalog interfaces.ResourceCatalog,
	wikiRepo interfaces.WikiPageRepository,
) interfaces.KnowledgeBaseService

NewKnowledgeBaseService creates a new knowledge base service

func NewKnowledgePostProcessService 

func NewKnowledgePostProcessService(
	knowledgeRepo interfaces.KnowledgeRepository,
	kbService interfaces.KnowledgeBaseService,
	chunkService interfaces.ChunkService,
	chunkRepo interfaces.ChunkRepository,
	taskEnqueuer interfaces.TaskEnqueuer,
	pendingRepo interfaces.TaskPendingOpsRepository,
	redisClient *redis.Client,
	spanTracker SpanTracker,
) interfaces.TaskHandler

func NewKnowledgeService 

func NewKnowledgeService(
	config *config.Config,
	repo interfaces.KnowledgeRepository,
	documentReader interfaces.DocumentReader,
	kbService interfaces.KnowledgeBaseService,
	tenantRepo interfaces.TenantRepository,
	tenantService interfaces.TenantService,
	chunkService interfaces.ChunkService,
	chunkRepo interfaces.ChunkRepository,
	tagRepo interfaces.KnowledgeTagRepository,
	tagService interfaces.KnowledgeTagService,
	fileSvc interfaces.FileService,
	storageResolver interfaces.StorageBackendResolver,
	resourceCatalog interfaces.ResourceCatalog,
	modelService interfaces.ModelService,
	task interfaces.TaskEnqueuer,
	taskInspector interfaces.TaskInspector,
	graphEngine interfaces.RetrieveGraphRepository,
	retrieveEngine interfaces.RetrieveEngineRegistry,
	ownership retriever.TenantStoreOwnership,
	redisClient *redis.Client,
	kbShareService interfaces.KBShareService,
	imageResolver *docparser.ImageResolver,
	wikiRepo interfaces.WikiPageRepository,
	wikiService interfaces.WikiPageService,
	taskPendingRepo interfaces.TaskPendingOpsRepository,
	spanTracker SpanTracker,
	audit interfaces.AuditLogService,
) (interfaces.KnowledgeService, error)

NewKnowledgeService creates a new knowledge service instance

func NewKnowledgeTagService 

func NewKnowledgeTagService(
	kbService interfaces.KnowledgeBaseService,
	repo interfaces.KnowledgeTagRepository,
	knowledgeRepo interfaces.KnowledgeRepository,
	chunkRepo interfaces.ChunkRepository,
	retrieveEngine interfaces.RetrieveEngineRegistry,
	ownership retriever.TenantStoreOwnership,
	modelService interfaces.ModelService,
	task interfaces.TaskEnqueuer,
	kbShareService interfaces.KBShareService,
	tenantRepo interfaces.TenantRepository,
	audit interfaces.AuditLogService,
) (interfaces.KnowledgeTagService, error)

NewKnowledgeTagService creates a new tag service.

func NewMCPEndpointService 

func NewMCPEndpointService(
	repo interfaces.MCPEndpointRepository,
	kbService interfaces.KnowledgeBaseService,
	kbShareService interfaces.KBShareService,
	agentService interfaces.CustomAgentService,
) interfaces.MCPEndpointService

NewMCPEndpointService creates the workspace MCP endpoint service.

func NewMCPServiceService 

func NewMCPServiceService(
	mcpServiceRepo interfaces.MCPServiceRepository,
	mcpManager *mcp.MCPManager,
	oauthRepo interfaces.MCPOAuthRepository,
) interfaces.MCPServiceService

NewMCPServiceService creates a new MCP service service

func NewMCPToolApprovalService 

func NewMCPToolApprovalService(
	repo interfaces.MCPToolApprovalRepository,
	mcpRepo interfaces.MCPServiceRepository,
) interfaces.MCPToolApprovalService

NewMCPToolApprovalService constructs the MCP tool approval service.

func NewMessageService 

func NewMessageService(messageRepo interfaces.MessageRepository,
	sessionRepo interfaces.SessionRepository,
	tenantService interfaces.TenantService,
	kbService interfaces.KnowledgeBaseService,
	knowService interfaces.KnowledgeService,
	modelService interfaces.ModelService,
	suggestionRepo interfaces.MessageSuggestionRepository,
) interfaces.MessageService

NewMessageService creates a new message service instance with the required repositories

func NewMessageSuggestionService 

func NewMessageSuggestionService(
	repo interfaces.MessageSuggestionRepository,
	messageService interfaces.MessageService,
	modelService interfaces.ModelService,
	customAgentService interfaces.CustomAgentService,
) interfaces.MessageSuggestionService

func NewModelService 

func NewModelService(repo interfaces.ModelRepository,
	kbRepo interfaces.KnowledgeBaseRepository,
	agentRepo interfaces.CustomAgentRepository,
	ollamaService *ollama.OllamaService,
	pooler embedding.EmbedderPooler,
	tenantService interfaces.TenantService,
) interfaces.ModelService

NewModelService creates a new model service instance

func NewOrganizationService 

func NewOrganizationService(
	orgRepo interfaces.OrganizationRepository,
	userRepo interfaces.UserRepository,
	shareRepo interfaces.KBShareRepository,
	agentShareRepo interfaces.AgentShareRepository,
) interfaces.OrganizationService

NewOrganizationService creates a new organization service

func NewResourceCatalog 

func NewResourceCatalog(repo interfaces.ResourceRepository) interfaces.ResourceCatalog

NewResourceCatalog creates the stable resource-reference domain service.

func NewResourceCatalogWithStorage added in v0.9.0 

func NewResourceCatalogWithStorage(
	repo interfaces.ResourceRepository,
	storage interfaces.StorageBackendRepository,
) interfaces.ResourceCatalog

NewResourceCatalogWithStorage snapshots the same environment as initRawFileService. A later workspace-default change must not change the destination of global writes.

func NewSessionService 

func NewSessionService(cfg *config.Config,
	sessionRepo interfaces.SessionRepository,
	messageRepo interfaces.MessageRepository,
	knowledgeBaseService interfaces.KnowledgeBaseService,
	knowledgeService interfaces.KnowledgeService,
	chunkService interfaces.ChunkService,
	modelService interfaces.ModelService,
	tenantService interfaces.TenantService,
	eventManager *chatpipeline.EventManager,
	agentService interfaces.AgentService,
	webSearchStateRepo interfaces.WebSearchStateService,
	webSearchProviderRepo interfaces.WebSearchProviderRepository,
	kbShareService interfaces.KBShareService,
	suggestionRepo interfaces.MessageSuggestionRepository,
	sandboxMgr sandbox.Manager,
	sandboxResolver sandbox.TenantSandboxResolver,
	sandboxPinner *SessionSandboxPinner,
	sandboxPolicy WorkspaceSandboxPolicy,
	hostSandbox HostSandboxManager,
	memoryService interfaces.MemoryService,
	sandboxConfigRepo repository.TenantSandboxConfigRepository,
	tenantSkillRepo repository.TenantSkillRepository,
	busyGate *SessionBusyGate,
	backgroundWork *BackgroundWork,
) interfaces.SessionService

NewSessionService creates a new session service instance with all required dependencies

func NewSystemSettingService 

func NewSystemSettingService(
	repo interfaces.SystemSettingRepository,
	audit interfaces.AuditLogService,
	rdb *redis.Client,
	cfg *config.Config,
) interfaces.SystemSettingService

NewSystemSettingService is the dig provider. audit may be nil (matches the tenantMemberService convention — tests that don't care about audit can pass nil and emitAudit no-ops). rdb may also be nil when REDIS_ADDR is unset — the service still uses its local cache, but skips cross-replica pubsub invalidation.

func NewTemporaryDocumentService 

func NewTemporaryDocumentService(
	repo interfaces.TemporaryDocumentRepository,
	fileService interfaces.FileService,
	resourceCatalog interfaces.ResourceCatalog,
	documentReader interfaces.DocumentReader,
	imageResolver *docparser.ImageResolver,
	modelService interfaces.ModelService,
	tenantService interfaces.TenantService,
	taskEnqueuer interfaces.TaskEnqueuer,
	messages interfaces.MessageRepository,
) interfaces.TemporaryDocumentService

func NewTenantAPIKeyService 

func NewTenantAPIKeyService(repo interfaces.TenantAPIKeyRepository) interfaces.TenantAPIKeyService

func NewTenantInvitationService 

func NewTenantInvitationService(
	repo interfaces.TenantInvitationRepository,
	memberSvc interfaces.TenantMemberService,
	audit interfaces.AuditLogService,
) interfaces.TenantInvitationService

NewTenantInvitationService wires the dependencies. memberSvc is required (Accept must create the tenant_members row); audit is optional and matches the same nil-safe pattern tenantMemberService uses.

func NewTenantMemberService 

func NewTenantMemberService(
	repo interfaces.TenantMemberRepository,
	audit interfaces.AuditLogService,
	userRepo interfaces.UserRepository,
	tokenRepo interfaces.AuthTokenRepository,
) interfaces.TenantMemberService

NewTenantMemberService constructs the service. Wired up via the DI container alongside the other application services. The auditService is optional — passing nil disables durable audit but keeps the underlying mutations working, so a container reshuffle that constructs tenant_member before audit_log won't crash and tests don't need to stub the dependency unless they care about audit behaviour.

userRepo / tokenRepo are also optional and only used by RemoveMember cleanup: after a membership is soft-deleted we best-effort clear a dangling users.tenant_id / LastActiveTenantID and revoke outstanding sessions so the removed user cannot keep a JWT scoped to a workspace they no longer belong to. Passing nil keeps unit tests that only exercise membership invariants free of extra stubs.

func NewTenantSandboxConfigLoader 

func NewTenantSandboxConfigLoader(
	repo repository.TenantSandboxConfigRepository,
) sandbox.TenantSandboxConfigLoader

NewTenantSandboxConfigLoader adapts the config repository onto the sandbox package's loader contract.

func NewTenantService 

func NewTenantService(
	repo interfaces.TenantRepository,
	storageRepo interfaces.StorageBackendRepository,
	settings interfaces.SystemSettingService,
) interfaces.TenantService

NewTenantService creates a new tenant service instance

func NewUserEnvResolver 

func NewUserEnvResolver(
	rows []*types.TenantSkillEntity,
	userEnvs userEnvReader,
	containerEnv map[string]string,
	tenantID uint64,
	configID string,
) *userEnvResolver

NewUserEnvResolver builds the resolver for one agent run. rows are this run's installed skills, already carrying their declarations and admin values; they may be empty, in which case only config-wide variables are resolved.

func NewUserResourceFavoriteService 

func NewUserResourceFavoriteService(repo interfaces.UserResourceFavoriteRepository) interfaces.UserResourceFavoriteService

NewUserResourceFavoriteService wraps the repository with input validation (allowlist of resource types, non-empty resource id). We keep service thin on purpose — favoriting is a non-business action that doesn't need audit logging or cross-aggregate side effects.

func NewUserService 

func NewUserService(
	configInfo *config.Config,
	userRepo interfaces.UserRepository,
	tokenRepo interfaces.AuthTokenRepository,
	tenantService interfaces.TenantService,
	memberService interfaces.TenantMemberService,
	systemSettingSvc interfaces.SystemSettingService,
) interfaces.UserService

NewUserService creates a new user service instance

func NewVectorStoreService 

func NewVectorStoreService(
	repo interfaces.VectorStoreRepository,
	kbRepo interfaces.KnowledgeBaseRepository,
	storeRegistry interfaces.StoreRegistry,
	factory interfaces.EngineFactory,
	db *gorm.DB,
) interfaces.VectorStoreService

NewVectorStoreService creates a new vector store service.

kbRepo and db are required by the delete guard, which counts bound KBs inside a transaction. storeRegistry and factory are optional in tests (passing nil disables dynamic registration / unregistration).

func NewWeKnoraCloudService 

func NewWeKnoraCloudService(
	repo interfaces.ModelRepository,
	tenantRepo interfaces.TenantRepository,
) interfaces.WeKnoraCloudService

NewWeKnoraCloudService 构造 WeKnoraCloudService

func NewWebSearchProviderService 

func NewWebSearchProviderService(repo interfaces.WebSearchProviderRepository) interfaces.WebSearchProviderService

NewWebSearchProviderService creates a new web search provider service

func NewWebSearchService 

func NewWebSearchService(
	cfg *config.Config,
	registry *infra_web_search.Registry,
	providerRepo interfaces.WebSearchProviderRepository,
) (interfaces.WebSearchService, error)

NewWebSearchService creates a new web search service. The registry holds provider type factories; the providerRepo loads tenant-specific configurations.

func NewWebSearchStateService 

func NewWebSearchStateService(
	redisClient *redis.Client,
	knowledgeService interfaces.KnowledgeService,
	knowledgeBaseService interfaces.KnowledgeBaseService,
) interfaces.WebSearchStateService

NewWebSearchStateService creates a new web search state service instance

func NewWikiIngestService 

func NewWikiIngestService(
	wikiService interfaces.WikiPageService,
	kbService interfaces.KnowledgeBaseService,
	knowledgeSvc interfaces.KnowledgeService,
	knowledgeRepo interfaces.KnowledgeRepository,
	chunkRepo interfaces.ChunkRepository,
	modelService interfaces.ModelService,
	task interfaces.TaskEnqueuer,
	audit interfaces.AuditLogService,
	pendingRepo interfaces.TaskPendingOpsRepository,
	deadLetterRepo interfaces.TaskDeadLetterRepository,
	redisClient *redis.Client,
	spanTracker SpanTracker,
) interfaces.TaskHandler

NewWikiIngestService creates a new wiki ingest service

func NewWikiPageService 

func NewWikiPageService(
	repo interfaces.WikiPageRepository,
	chunkRepo interfaces.ChunkRepository,
	kbService interfaces.KnowledgeBaseService,
	taskPendingRepo interfaces.TaskPendingOpsRepository,
	redisClient *redis.Client,
	kbShareService interfaces.KBShareService,
) interfaces.WikiPageService

NewWikiPageService creates a new wiki page service

func ParseShellJobNoticeState added in v0.9.0 

func ParseShellJobNoticeState(text string) string

ParseShellJobNoticeState reads the terminal state out of a shell-job notice. The text is "Background shell <id> <state>: <title>. …".

func RecordWikiContentActivity 

func RecordWikiContentActivity(
	ctx context.Context,
	audit interfaces.AuditLogService,
	tenantID uint64,
	kbID string,
	actions map[string]int,
)

RecordWikiContentActivity writes the bounded Wiki mutation summary shown in the knowledge-base activity feed. Wiki mutations used to reach this feed as a side effect of inserting rows into the dedicated wiki_log_entries table; keeping the projection explicit lets the activity feed remain authoritative without maintaining a second, Wiki-only event store.

func ResolveComplexPasswordEnabled 

func ResolveComplexPasswordEnabled(
	ctx context.Context,
	cfg *config.Config,
	settings interfaces.SystemSettingService,
) bool

ResolveComplexPasswordEnabled is the shared runtime resolver for the complex-password switch. Priority matches other auth tunables: DB system_settings > ENV > cfg.Auth (startup YAML/env) > false.

func ResolveDefaultTenantStorageQuota added in v0.9.0 

func ResolveDefaultTenantStorageQuota(ctx context.Context, settings interfaces.SystemSettingService) int64

ResolveDefaultTenantStorageQuota returns the effective default in bytes. Each setting uses DB > ENV > default resolution. A positive MB setting takes precedence over the legacy GB setting; zero or negative MB falls back to GB. Invalid GB falls back to 10 GiB. Overflow must never turn a configured quota into a non-positive (unlimited) value.

func ResolveProcessConfig 

func ResolveProcessConfig(kb *types.KnowledgeBase, overrides *types.KnowledgeProcessOverrides) types.EffectiveProcessConfig

ResolveProcessConfig merges KB defaults with per-upload overrides for the parse pipeline.

func SandboxIdentityChanged 

func SandboxIdentityChanged(oldCfg, newCfg *types.TenantSandboxConfig) bool

SandboxIdentityChanged reports whether sandboxes already created under oldCfg would stop being operable under newCfg. Names, templates, TTLs, HTTP timeouts and env vars deliberately do not count: they only shape FUTURE sandboxes.

newCfg must already have been merged with the stored config (see types.MergeSandboxConfigForUpdate). The API masks secrets on read, so comparing a raw payload would report a key rotation on every single save and leave the config permanently uneditable.

The comparison needs no deployment baseline: named configs inherit nothing, so the stored fields are the whole identity. It is also deliberately free of validation — see sandbox.IdentityOf for why an unreachable old endpoint must still be judgeable.

func SanitizeSandboxConfig 

func SanitizeSandboxConfig(
	incoming, existing *types.TenantSandboxConfig,
) (*types.TenantSandboxConfig, error)

SanitizeSandboxConfig resolves redacted secrets and validates the payload before it can be persisted.

func ShellJobResult added in v0.9.0 

func ShellJobResult(snap sandbox.ShellJobSnapshot, lite bool) interfaces.WorkResult

ShellJobResult maps a directory snapshot into the shared work result. lite selects the host-session wording of the running and lost notes.

func SignEmbedSessionHandle 

func SignEmbedSessionHandle(ch *types.EmbedChannel, sessionID string) string

SignEmbedSessionHandle binds a chat session id to its embed channel with an HMAC keyed by a server-only deployment secret. The handle is handed to the widget at session-creation time and must be presented on every history load / chat call. Because the session id travels in the request path (and can land in access logs), this signature — sent in a header, never logged — is the real authorization secret: a leaked session id is useless without it. Rotating the channel token invalidates outstanding handles, which is acceptable.

func SignEmbedWebhookBody 

func SignEmbedWebhookBody(secret string, raw []byte) string

SignEmbedWebhookBody returns the hex HMAC signature for tests.

func SummarizeSystemSettings added in v0.9.0 

func SummarizeSystemSettings(rows []*types.SystemSetting, key string) []types.SystemSettingSummary

List returns all known settings for the management UI. Persisted rows are enriched with registry metadata. Registry keys without a saved DB override are returned as virtual rows using the effective fallback value (ENV/config/default), so merely migrating the schema never changes runtime behaviour. SummarizeSystemSettings excludes unknown rows and secrets. Registry metadata is authoritative, so a stray database row cannot introduce arbitrary config fields into the assistant's catalog. Values remain JSON to preserve type.

func ValidateEmbedWebhookURL 

func ValidateEmbedWebhookURL(raw string) error

ValidateEmbedWebhookURL checks an optional outbound webhook URL. Empty is allowed.

func ValidateImagePipelineParams added in v0.9.0 

func ValidateImagePipelineParams(pipelineID types.ImagePipelineID, params map[string]any) error

ValidateImagePipelineParams asks the pipeline an id names whether the tunables a knowledge base is about to store can be run. The id is resolved exactly as a running task would resolve it, so the answer cannot disagree with the pipeline that would actually handle the images.

func ValidateJSONUploadContent 

func ValidateJSONUploadContent(fileName string, file *multipart.FileHeader) error

ValidateJSONUploadContent rejects malformed .json uploads before they are stored and queued. Shares the same validity gate as the simple JSON reader. Call this from HTTP upload / replace handlers only — CreateKnowledgeFromFile itself must not, so datasource sync and IM can still create a knowledge row that fails in async parse (delete-then-create must not leave a gap).

func ValidatePasswordPolicy 

func ValidatePasswordPolicy(password string, complexPasswordEnabled bool) error

ValidatePasswordPolicy keeps registration, self-service rotation and administrative password resets aligned with the SPA form. Letter and digit classes are ASCII so they match the frontend regexes; special characters are the documented whitelist. Password bytes are never logged or included in the returned error.

func ValidateProcessOverrides 

func ValidateProcessOverrides(
	ctx context.Context,
	kb *types.KnowledgeBase,
	overrides *types.KnowledgeProcessOverrides,
	fileTypes []string,
) error

ValidateProcessOverrides validates batch overrides against file types in the upload.

func VerifyEmbedSessionHandle 

func VerifyEmbedSessionHandle(ch *types.EmbedChannel, sessionID, sig string) bool

VerifyEmbedSessionHandle reports whether sig is a valid handle for sessionID on channel ch, using a constant-time comparison.

func WikiDeletedTombstoneKey 

func WikiDeletedTombstoneKey(kbID, knowledgeID string) string

WikiDeletedTombstoneKey returns the Redis key used to mark a knowledge as recently deleted, so wiki_ingest tasks in flight can short-circuit. Exposed so knowledgeService.cleanupWikiOnKnowledgeDelete can write the same key without duplicating the format string.

func WikiPendingLanguage 

func WikiPendingLanguage(ctx context.Context, db *gorm.DB, tenantID uint64, kbID string) string

WikiPendingLanguage returns the language recorded on the KB's newest queued wiki ingest op, or "" if none. A trigger re-armed without a request context must carry it: batch-level taxonomy planning reads the trigger's language, and would otherwise fall back to the server default.

Types

type AgentCollaboration added in v0.9.0 

type AgentCollaboration struct {
	Runtime *collaboration.Runtime

	ParentSteer  func(context.Context, *types.Session, *types.Message) types.SteerSink
	ParentEvents func(context.Context, *types.Session, *types.Message, *event.EventBus) func(context.Context) error
	ChildEvents  func(context.Context, *types.Session, *types.Message, *event.EventBus) func(context.Context) error
	// contains filtered or unexported fields
}

AgentCollaboration owns durable work and binds SessionService after DI has resolved its dependency on AgentService. No request context or credentials survive in the work record; workers reauthorize the user and membership.

func NewAgentCollaboration added in v0.9.0 

func NewAgentCollaboration(
	db *gorm.DB,
	agents interfaces.CustomAgentService,
	messages interfaces.MessageService,
	users interfaces.UserService,
	members interfaces.TenantMemberService,
	tenants interfaces.TenantService,
) *AgentCollaboration

NewAgentCollaboration constructs the service and its durable execution runtime.

func (*AgentCollaboration) Bind added in v0.9.0 

func (s *AgentCollaboration) Bind(sessions interfaces.SessionService)

Bind attaches the session executor used by child runs.

func (*AgentCollaboration) BindRecovery added in v0.9.0 

func (s *AgentCollaboration) BindRecovery(
	broker *hitl.Broker,
	gate approval.BuiltinApproval,
	streams interfaces.StreamManager,
)

BindRecovery connects approval recovery and live stream persistence.

func (*AgentCollaboration) HandlesPrompt added in v0.9.0 

func (s *AgentCollaboration) HandlesPrompt(ctx context.Context, p *types.AgentPrompt) bool

HandlesPrompt is used by the HTTP resumer to leave delegated continuations to the durable queue. The generic HTTP path must not bypass its run fence.

func (*AgentCollaboration) HistoryStatuses added in v0.9.0 

func (s *AgentCollaboration) HistoryStatuses(
	ctx context.Context,
	sessionID string,
	messageIDs []string,
) (map[string]json.RawMessage, error)

HistoryStatuses restores UI state independently of Redis and live execution. The handler has authorized session access; this additional scope prevents a shared session read from exposing another user's delegated work.

func (*AgentCollaboration) PendingPrompts added in v0.9.0 

func (s *AgentCollaboration) PendingPrompts(
	ctx context.Context,
	sessionID string,
) (map[string][]*types.PendingPromptView, error)

PendingPrompts projects all children onto the parent without changing the prompt's actual owner/session. Parallel children may each need approval.

func (*AgentCollaboration) RecoverAnswered added in v0.9.0 

func (s *AgentCollaboration) RecoverAnswered(ctx context.Context, p *types.AgentPrompt) (bool, error)

RecoverAnswered retains answered prompts in the durable scheduler across restart. Executing prompts remain non-replayable because their effects are uncertain.

func (*AgentCollaboration) Stop added in v0.9.0 

func (s *AgentCollaboration) Stop(ctx context.Context, sessionID, messageID string) error

Stop cancels the scoped collaboration for a session.

func (*AgentCollaboration) StopOwned added in v0.9.0 

func (s *AgentCollaboration) StopOwned(
	ctx context.Context,
	tenantID uint64,
	userID, sessionID, messageID string,
) error

StopOwned is called only after the HTTP owner check or with a broker-owned prompt identity. Persist stop even when the parent process is gone.

func (*AgentCollaboration) StopPrompt added in v0.9.0 

func (s *AgentCollaboration) StopPrompt(ctx context.Context, p *types.AgentPrompt) error

StopPrompt settles the run that owns a closed prompt. A parent prompt cancels its collaboration. A child prompt lives in the child session, so it is matched through the parent group and only that child is closed.

type ArtifactCollector 

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

ArtifactCollector drains skill-generated files from the sandbox when a turn completes.

func NewArtifactCollector 

func NewArtifactCollector(
	source SandboxArtifactSource,
	fileService interfaces.FileService,
	store SessionArtifactStore,
	catalog interfaces.ResourceCatalog,
	config ArtifactCollectorConfig,
) *ArtifactCollector

NewArtifactCollector wires up an ArtifactCollector. Callers keep a single instance per process; the collector holds no per-turn state. catalog may be nil (resource registry disabled); binding is then skipped.

func NewArtifactCollectorFromSandboxManager 

func NewArtifactCollectorFromSandboxManager(
	sandboxMgr sandbox.Manager,
	sandboxResolver sandbox.TenantSandboxResolver,
	pinner *SessionSandboxPinner,
	host *HostSessionResolver,
	fileService interfaces.FileService,
	repo interfaces.MessageRepository,
	catalog interfaces.ResourceCatalog,
) *ArtifactCollector

NewArtifactCollectorFromSandboxManager is the DI-friendly constructor that promotes a sandbox.Manager to SandboxArtifactSource only when the underlying implementation actually supports per-session file inspection (currently *sandbox.SessionBoundManager). For any other backend the returned *ArtifactCollector is nil, which the AgentStreamHandler treats as "no artifacts to attach" — matching the graceful-degradation contract documented in the design spec. The resolver is consulted per turn so a workspace whose own backend supports artifacts still gets them even when the process-wide default does not. The collector is therefore built whenever either side could supply a source, and Collect degrades to "nothing to attach" when neither does.

func (*ArtifactCollector) BindArtifactsToMessage 

func (c *ArtifactCollector) BindArtifactsToMessage(
	ctx context.Context, messageID string, artifacts types.MessageArtifacts,
)

BindArtifactsToMessage makes messageID an owner of each artifact's resource handle as well. Deleting the message that originally produced the file then cannot invalidate a reference made from a later one. Best-effort: a binding failure never discards the artifact, which is already stored and downloadable through the /artifacts endpoint.

func (*ArtifactCollector) Collect 

func (c *ArtifactCollector) Collect(
	ctx context.Context,
	sessionID string,
	messageID string,
	tenantID uint64,
	outputDir string,
) (types.MessageArtifacts, error)

Collect scans the session sandbox's output directory and persists any newly-created or newly-modified files to the tenant file service.

The returned slice contains one MessageArtifact per persisted file. When no sandbox is bound to the session, or when the source is nil (skill backend disabled), Collect returns nil, nil — the caller is expected to treat both cases as "nothing to attach" and NOT set message.Artifacts.

Errors returned by Collect are limited to internal invariants (nil dependencies). Per-file errors (unreadable, too-large, upload failure) are logged and skipped so a single misbehaving artifact never breaks the turn.

func (*ArtifactCollector) CollectTarget 

func (c *ArtifactCollector) CollectTarget(ctx context.Context, sessionID string) (string, bool)

CollectTarget reports the directory Collect should scan for this session, and whether collection must be skipped entirely.

skip is true when collecting would scan the user's project: the backend advertised a workspace with no separate output tree, OutputDir is Root, or the layout provider failed. A nil source is also skip: there is nothing to drain, and skip=false would let the caller fill /workspace/output. An empty dir with skip false means the backend advertises no layout at all — the caller's remote default applies.

The directory and the skip decision come from one lookup on purpose. Asking twice re-resolved the session's sandbox (a pin read plus a manager resolve) and let the two answers disagree: a second lookup that failed after the first succeeded returned no directory, sending a host session's collection back to the remote /workspace/output.

func (*ArtifactCollector) CollectWithNotify 

func (c *ArtifactCollector) CollectWithNotify(
	ctx context.Context,
	sessionID string,
	messageID string,
	tenantID uint64,
	outputDir string,
	notify func(pending int),
) (types.MessageArtifacts, error)

CollectWithNotify is Collect plus a progress hook fired after files have been hashed and we know they will be uploaded. The frontend uses this to show a toolbar placeholder while object-storage uploads (often a few seconds for HTML charts) finish. notify is skipped when nothing will be persisted, so a fork restore that only refreshes mtime stays quiet.

func (*ArtifactCollector) SessionArtifacts 

func (c *ArtifactCollector) SessionArtifacts(ctx context.Context, sessionID string) types.MessageArtifacts

SessionArtifacts returns every artifact already recorded against the session. Empty on error, so the caller degrades to "nothing to resolve" rather than dropping the turn.

Callers need this to resolve an answer's file references: a turn can name a file it did not (re)generate — the first turn after a session fork points the sandbox back at an earlier commit, so every unchanged file is de-duplicated out of that turn's own artifact list — and the reference still has to be bound to that file's stable handle.

Files the user deleted are filtered out: they are still in the store as tombstones so loadKnownSet will not re-collect them, but an answer must not resolve a name to a file whose bytes are gone.

type ArtifactCollectorConfig 

type ArtifactCollectorConfig struct {
	// OutputDir is the absolute path inside the sandbox to scan. Empty
	// falls back to skills.ArtifactOutputDir() at call time.
	OutputDir string

	// MaxFileBytes is the largest single file that will be persisted.
	// Files larger than this are logged and skipped so a runaway skill
	// cannot exhaust the WeKnora process's memory.
	MaxFileBytes int64
}

ArtifactCollectorConfig bounds the collector's I/O and storage footprint. All fields have safe zero-value defaults applied by newBoundedConfig.

type AuditLogRetentionRunner 

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

AuditLogRetentionRunner sweeps audit_logs once a day, deleting rows older than `retentionDays`. It is a small, self-contained background goroutine — no robfig/cron, no asynq — because retention has no wall-clock alignment requirement: we just need "approximately daily, eventually". A bare time.Ticker keeps the dependency surface minimal.

retentionDays <= 0 makes Start a no-op; this is the configured way to disable retention entirely. Validation happens at config-load time so by the time we're here a non-positive value is intentional.

func NewAuditLogRetentionRunner 

func NewAuditLogRetentionRunner(
	cfg *config.Config, svc interfaces.AuditLogService,
) *AuditLogRetentionRunner

NewAuditLogRetentionRunner constructs the runner with production defaults. retention_days is read from the config; passing the full *config.Config (rather than just an int) keeps the dig wiring trivial — it's the same shape as every other config-aware constructor in the container. The constructor only validates inputs; nothing fires until Start is called.

func (*AuditLogRetentionRunner) Start 

func (r *AuditLogRetentionRunner) Start(ctx context.Context)

Start spins up the background goroutine. Calling it more than once is a no-op (sync.Once), so container wiring that mistakenly invokes us twice doesn't double-purge. retentionDays <= 0 means the runner stays dormant — Stop will still complete cleanly.

func (*AuditLogRetentionRunner) Stop 

func (r *AuditLogRetentionRunner) Stop()

Stop signals the loop to exit and blocks until it returns. Idempotent. If Start was never called, Stop returns immediately (no doneCh to wait on — see the `started` flag in the struct comment).

func (*AuditLogRetentionRunner) StopWithin 

func (r *AuditLogRetentionRunner) StopWithin(timeout time.Duration)

StopWithin is Stop with a bound. timeout <= 0 waits until the loop returns. A positive timeout continues shutdown while a purge is still inside its own database deadline, instead of holding every later hook.

type AutomationConfig added in v0.9.0 

type AutomationConfig struct {
	// Tick is how often each instance looks for due work.
	Tick time.Duration
	// Lease is how long a running run stays owned without a heartbeat.
	Lease time.Duration
	// MaxConcurrent is how many runs one instance executes at once.
	MaxConcurrent int
	// MaxPerTenant caps the automations of one workspace.
	MaxPerTenant int64
	// Retention is how many runs each automation keeps.
	Retention int
	// MinInterval is the shortest gap allowed between two runs.
	MinInterval time.Duration
	// DefaultMaxDuration caps a run that sets no limit of its own.
	DefaultMaxDuration time.Duration
	// FailuresBeforePause pauses an automation after that many failed runs
	// in a row.
	FailuresBeforePause int
}

AutomationConfig bounds the scheduler and the automations it runs.

type AutomationGrant added in v0.9.0 

type AutomationGrant struct {
	*types.AgentGrant
	Lapse string `json:"lapse,omitempty"`
}

AutomationGrant is a standing grant of an automation, with why it no longer applies when it does not.

type AutomationGrantCandidate added in v0.9.0 

type AutomationGrantCandidate struct {
	PromptID    string    `json:"prompt_id"`
	GrantKey    string    `json:"grant_key"`
	Label       string    `json:"label"`
	Risk        string    `json:"risk"`
	Tool        string    `json:"tool"`
	ServiceName string    `json:"service_name,omitempty"`
	AskedAt     time.Time `json:"asked_at"`
}

AutomationGrantCandidate is a call an automation's runs asked approval for that its owner may allow for future runs.

type AutomationService added in v0.9.0 

type AutomationService struct {

	// Events publishes a run to its message's event stream; see
	// AgentCollaboration.ChildEvents.
	Events func(context.Context, *types.Session, *types.Message, *event.EventBus) func(context.Context) error
	// StopDelegation fences children and discards recoverable parent waits.
	StopDelegation func(context.Context, uint64, string, string, string) error
	// contains filtered or unexported fields
}

AutomationService manages automations and runs them on schedule as their owners. No credential is stored: each run reloads and rechecks the owner.

func NewAutomationService added in v0.9.0 

func NewAutomationService(
	db *gorm.DB,
	agents interfaces.CustomAgentService,
	users interfaces.UserService,
	members interfaces.TenantMemberService,
	tenants interfaces.TenantService,
	messages interfaces.MessageService,
	grants interfaces.AgentGrantRepository,
	notices *NotificationService,
	models interfaces.ModelService,
) *AutomationService

NewAutomationService builds the service. Bind it before Start.

func (*AutomationService) AddGrant added in v0.9.0 

func (s *AutomationService) AddGrant(ctx context.Context, id, promptID string) (*types.AgentGrant, error)

AddGrant lets future runs make the call an earlier run asked approval for in prompt promptID without asking. The grant stops applying when the agent's configuration or the owner's role changes.

func (*AutomationService) Bind added in v0.9.0 

func (s *AutomationService) Bind(
	sessions interfaces.SessionService, broker *hitl.Broker,
	runs *SuspendedRunService, streams interfaces.StreamManager,
)

Bind attaches what runs need once DI has built it.

func (*AutomationService) CancelRun added in v0.9.0 

func (s *AutomationService) CancelRun(ctx context.Context, runID string) (*types.AutomationRun, error)

CancelRun stops one of the caller's runs.

func (*AutomationService) Configure added in v0.9.0 

func (s *AutomationService) Configure(cfg AutomationConfig)

Configure overrides the defaults with cfg's non-zero fields.

func (*AutomationService) Create added in v0.9.0 

func (s *AutomationService) Create(ctx context.Context, p *types.AutomationPatch) (*types.Automation, error)

Create adds an automation owned by the caller.

func (*AutomationService) Delete added in v0.9.0 

func (s *AutomationService) Delete(ctx context.Context, id string) error

Delete stops an automation's runs and removes it with its conversations and standing grants.

func (*AutomationService) FindBySession added in v0.9.0 

func (s *AutomationService) FindBySession(ctx context.Context, sessionID string) (*types.Automation, error)

FindBySession returns the caller's automation whose run, or persistent conversation, is sessionID.

func (*AutomationService) Get added in v0.9.0 

func (s *AutomationService) Get(ctx context.Context, id string) (*types.Automation, error)

Get returns one of the caller's automations.

func (*AutomationService) GetRun added in v0.9.0 

func (s *AutomationService) GetRun(ctx context.Context, runID string) (*types.AutomationRun, error)

GetRun returns one of the caller's runs.

func (*AutomationService) GrantCandidates added in v0.9.0 

func (s *AutomationService) GrantCandidates(ctx context.Context, id string) ([]AutomationGrantCandidate, error)

GrantCandidates lists the calls the automation's runs asked approval for that are not allowed for future runs yet, newest first.

func (*AutomationService) Grants added in v0.9.0 

func (s *AutomationService) Grants(ctx context.Context, id string) ([]AutomationGrant, error)

Grants lists an automation's standing grants.

func (*AutomationService) HandlesPrompt added in v0.9.0 

func (s *AutomationService) HandlesPrompt(ctx context.Context, p *types.AgentPrompt) bool

HandlesPrompt claims the prompts automation runs raise, so answering one continues the run under the automation instead of as an ordinary chat. It fails closed: when it cannot tell, the prompt is left to the scheduler.

func (*AutomationService) List added in v0.9.0 

func (s *AutomationService) List(ctx context.Context) ([]*types.Automation, error)

List returns the caller's automations, newest first.

func (*AutomationService) ListRuns added in v0.9.0 

func (s *AutomationService) ListRuns(
	ctx context.Context, id string, limit, offset int,
) ([]*types.AutomationRun, int64, error)

ListRuns returns a page of an automation's runs, newest first.

func (*AutomationService) Pause added in v0.9.0 

func (s *AutomationService) Pause(ctx context.Context, id string) (*types.Automation, error)

Pause stops scheduling; runs already going continue.

func (*AutomationService) RecoverAnswered added in v0.9.0 

func (s *AutomationService) RecoverAnswered(ctx context.Context, p *types.AgentPrompt) (bool, error)

RecoverAnswered keeps an answered prompt of an automation run alive until the scheduler resumes the run.

func (*AutomationService) Resume added in v0.9.0 

func (s *AutomationService) Resume(ctx context.Context, id string) (*types.Automation, error)

Resume schedules it again from now, clearing a pause and its failures.

func (*AutomationService) RunNow added in v0.9.0 

func (s *AutomationService) RunNow(ctx context.Context, id string) (*types.AutomationRun, error)

RunNow queues a manual run. It is refused while another run is queued, running or waiting.

func (*AutomationService) Start added in v0.9.0 

func (s *AutomationService) Start(ctx context.Context)

Start schedules automations until ctx ends. Every instance runs it: due automations and queued runs are claimed by compare-and-set in the database, so each run executes once however many instances look.

func (*AutomationService) Update added in v0.9.0 

func (s *AutomationService) Update(
	ctx context.Context, id string, p *types.AutomationPatch,
) (*types.Automation, error)

Update changes the fields p carries.

func (*AutomationService) ValidateTrigger added in v0.9.0 

func (s *AutomationService) ValidateTrigger(t types.AutomationTrigger, n int) ([]time.Time, error)

ValidateTrigger checks t and returns its next n firing times.

func (*AutomationService) Wake added in v0.9.0 

func (s *AutomationService) Wake()

Wake makes the scheduler look for work now.

type BackgroundEventEmitter added in v0.9.0 

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

BackgroundEventEmitter publishes the background work events of shared design §7. Features call it; they never pick the stream themselves.

func NewBackgroundEventEmitter added in v0.9.0 

func NewBackgroundEventEmitter(streams interfaces.StreamManager) *BackgroundEventEmitter

NewBackgroundEventEmitter returns an emitter that publishes background work events on the stream selected by the shared live-run rules.

func (*BackgroundEventEmitter) EmitCommandOutput added in v0.9.0 

func (e *BackgroundEventEmitter) EmitCommandOutput(
	ctx context.Context, s interfaces.SessionRef, parentMessageID, workID string,
	data event.CommandOutputData,
) error

EmitCommandOutput previews running output on the parent's stream. When the parent run is no longer live the event is dropped: the process can be replayed from the feature's own storage, and appending after complete would corrupt replay.

func (*BackgroundEventEmitter) EmitWorkFinished added in v0.9.0 

func (e *BackgroundEventEmitter) EmitWorkFinished(
	ctx context.Context, s interfaces.SessionRef, parentMessageID string,
	r interfaces.WorkResult,
) error

EmitWorkFinished publishes work_finished. The parent stream is used when that run is live; otherwise the session's current live run is used.

func (*BackgroundEventEmitter) EmitWorkStarted added in v0.9.0 

func (e *BackgroundEventEmitter) EmitWorkStarted(
	ctx context.Context, s interfaces.SessionRef, parentMessageID string,
	a interfaces.BackgroundActivity, toolCallID string,
) error

EmitWorkStarted routes to the parent stream when live, else to the session's current live run, else nowhere.

func (*BackgroundEventEmitter) ParentLive added in v0.9.0 

func (e *BackgroundEventEmitter) ParentLive(
	ctx context.Context, s interfaces.SessionRef, parentMessageID string,
) (bool, error)

ParentLive reports whether parentMessageID is the session's live run.

type BackgroundKeepAliver added in v0.9.0 

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

BackgroundKeepAliver keeps the sandbox of a session with live background work from being reclaimed by its provider. Callers may invoke it on every poll; it reaches the provider only once the interval the last successful refresh reported has passed.

func NewBackgroundKeepAliver added in v0.9.0 

func NewBackgroundKeepAliver(pinned *PinnedSessionSandbox) *BackgroundKeepAliver

NewBackgroundKeepAliver returns a keep-aliver over the session's pinned sandbox.

func (*BackgroundKeepAliver) KeepAlive added in v0.9.0 

func (k *BackgroundKeepAliver) KeepAlive(ctx context.Context, s interfaces.SessionRef) error

KeepAlive implements interfaces.SessionKeepAliver.

type BackgroundWork added in v0.9.0 

type BackgroundWork struct {
	Registry  interfaces.ActivityRegistry
	Inbox     interfaces.NoticeInbox
	Listeners *background.Listeners
	Handler   interfaces.WorkHandler
}

BackgroundWork bundles the shared background-work parts the agent, the session handlers and features use. Handler is nil until a feature that submits background work is wired; until then Enabled is false and nothing background-related reaches a run.

func NewBackgroundWork added in v0.9.0 

func NewBackgroundWork(registry interfaces.ActivityRegistry, inbox interfaces.NoticeInbox) *BackgroundWork

NewBackgroundWork returns the bundle with an empty listener set.

func (*BackgroundWork) Enabled added in v0.9.0 

func (b *BackgroundWork) Enabled() bool

Enabled reports whether background work can exist at all.

type CatalogInstallResult 

type CatalogInstallResult struct {
	Installs map[string]string `json:"installs"`
	Errors   map[string]string `json:"errors,omitempty"`
}

CatalogInstallResult is the per-sandbox outcome of one catalog install call. Partial success is a successful HTTP accept with Errors filled in, not a silent 202: the caller has to be able to tell which configs never started.

type CatalogProviderView 

type CatalogProviderView struct {
	ID         string             `json:"id"`
	Name       string             `json:"name"`
	API        string             `json:"api"`
	ModelTypes []types.ModelType  `json:"model_types"`
	Settings   map[string]any     `json:"settings"`
	Models     []models.ModelSpec `json:"models"`
	// ModelThinkingLevels lists, per chat model id, the levels offered once
	// thinking is enabled, resolved against the vendor map and protocol so
	// the console does not re-implement level inheritance.
	ModelThinkingLevels map[string][]api.ReasoningEffort `json:"model_thinking_levels"`
	// VendorThinkingLevels is what a chat model without its own map offers.
	VendorThinkingLevels []api.ReasoningEffort `json:"vendor_thinking_levels"`
}

CatalogProviderView deliberately omits provider credentials and headers.

type CatalogRevision 

type CatalogRevision struct {
	Version   uint64          `json:"version"`
	Overlay   json.RawMessage `json:"overlay" swaggertype:"object"`
	UpdatedBy string          `json:"updated_by"`
	UpdatedAt time.Time       `json:"updated_at"`
}

CatalogRevision is one earlier published overlay.

type CatalogState 

type CatalogState struct {
	SyncError       string                `json:"sync_error,omitempty"`
	AppliedVersion  uint64                `json:"applied_version"`
	Version         uint64                `json:"version"`
	Baseline        string                `json:"baseline"`
	Overlay         json.RawMessage       `json:"overlay" swaggertype:"object"`
	History         []CatalogRevision     `json:"history"`
	Builtin         []CatalogProviderView `json:"builtin"`
	Deployment      []CatalogProviderView `json:"deployment"`
	Effective       []CatalogProviderView `json:"effective"`
	DeploymentError string                `json:"deployment_error,omitempty"`
}

CatalogState is what the console reads: every layer plus the overlay.

type CatalogUpdate 

type CatalogUpdate struct {
	Version  uint64          `json:"version"`
	Baseline string          `json:"baseline"`
	Overlay  json.RawMessage `json:"overlay" swaggertype:"object"`
}

CatalogUpdate is a preview or publish request.

type ChunkExtractService 

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

ChunkExtractService is a service for extracting chunks

func (*ChunkExtractService) Handle 

func (s *ChunkExtractService) Handle(ctx context.Context, t *asynq.Task) error

Handle handles the chunk extraction task

type ChunkRelation 

type ChunkRelation struct {
	// Weight relationship weight, calculated based on PMI and strength
	Weight float64

	// Degree total degree of related entities
	Degree int
}

ChunkRelation represents a relationship between two Chunks

type ConfigEnvGroup 

type ConfigEnvGroup struct {
	SandboxConfigID   string          `json:"sandbox_config_id"`
	SandboxConfigName string          `json:"sandbox_config_name"`
	Description       string          `json:"description,omitempty"`
	Vars              []EnvVarView    `json:"vars"`
	Skills            []SkillEnvGroup `json:"skills"`
}

ConfigEnvGroup is one sandbox config: the caller's own config-wide variables plus the declared credentials of the skills installed on it.

type CreateSandboxConfigInput 

type CreateSandboxConfigInput struct {
	Name        string
	Description string
	Config      *types.TenantSandboxConfig
}

CreateSandboxConfigInput is the create payload.

type DataSourceService 

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

DataSourceService implements the DataSourceService interface

func (*DataSourceService) ClearDataSourceCredentials 

func (s *DataSourceService) ClearDataSourceCredentials(ctx context.Context, id string) error

ClearDataSourceCredentials wipes the connector credential map without touching any other config field. Idempotent.

func (*DataSourceService) CreateDataSource 

func (s *DataSourceService) CreateDataSource(ctx context.Context, ds *types.DataSource) (*types.DataSource, error)

CreateDataSource creates a new data source configuration

func (*DataSourceService) DeleteDataSource 

func (s *DataSourceService) DeleteDataSource(ctx context.Context, id string) error

DeleteDataSource deletes a data source (soft delete)

func (*DataSourceService) GetDataSource 

func (s *DataSourceService) GetDataSource(ctx context.Context, id string) (*types.DataSource, error)

GetDataSource retrieves a data source by ID

func (*DataSourceService) GetSyncLog 

func (s *DataSourceService) GetSyncLog(ctx context.Context, syncLogID string) (*types.SyncLog, error)

GetSyncLog retrieves a specific sync log entry

func (*DataSourceService) GetSyncLogs 

func (s *DataSourceService) GetSyncLogs(ctx context.Context, dsID string, limit int, offset int) ([]*types.SyncLog, error)

GetSyncLogs retrieves sync history for a data source

func (*DataSourceService) ListAvailableResources 

func (s *DataSourceService) ListAvailableResources(
	ctx context.Context, dsID string, parentID string,
) ([]types.Resource, error)

ListAvailableResources lists resources available for sync in the external system. parentID enables lazy (on-demand) loading of hierarchical resources: pass "" to list the top level, or a resource's ExternalID to list only its direct children.

func (*DataSourceService) ListDataSources 

func (s *DataSourceService) ListDataSources(ctx context.Context, kbID string) ([]*types.DataSource, error)

ListDataSources lists all data sources for a knowledge base

func (*DataSourceService) ManualSync 

func (s *DataSourceService) ManualSync(ctx context.Context, dsID string) (*types.SyncLog, error)

ManualSync triggers an immediate sync for a data source

func (*DataSourceService) PatchDataSourceSyncSettings added in v0.9.0 

func (s *DataSourceService) PatchDataSourceSyncSettings(
	ctx context.Context, before, after *types.DataSource, fields []string,
) (*types.DataSource, error)

PatchDataSourceSyncSettings preserves worker state and settings omitted by the caller.

func (*DataSourceService) PauseDataSource 

func (s *DataSourceService) PauseDataSource(ctx context.Context, id string) error

PauseDataSource pauses a data source's scheduled syncs

func (*DataSourceService) ProcessSync 

func (s *DataSourceService) ProcessSync(ctx context.Context, task *asynq.Task) error

ProcessSync handles the actual sync operation (called by asynq task)

func (*DataSourceService) ResolveResourceAncestors 

func (s *DataSourceService) ResolveResourceAncestors(
	ctx context.Context, dsID string, resourceIDs []string,
) ([]string, error)

ResolveResourceAncestors resolves the ancestor ExternalIDs needed to reveal the given resources in a lazily-loaded picker (see the connector method for details).

func (*DataSourceService) ResumeDataSource 

func (s *DataSourceService) ResumeDataSource(ctx context.Context, id string) error

ResumeDataSource resumes a paused data source

func (*DataSourceService) UpdateDataSource 

func (s *DataSourceService) UpdateDataSource(ctx context.Context, ds *types.DataSource) (*types.DataSource, error)

UpdateDataSource updates an existing data source

func (*DataSourceService) UpdateDataSourceCredentials 

func (s *DataSourceService) UpdateDataSourceCredentials(
	ctx context.Context, id string, credentials map[string]interface{},
) (*types.DataSource, error)

UpdateDataSourceCredentials replaces the connector credential map. This is a single atomic write; the previous credential set is discarded entirely (callers cannot patch individual keys because half-configured connector auth is meaningless). After persisting, the live connection is validated so the caller learns immediately if the new credentials are wrong.

func (*DataSourceService) ValidateConnection 

func (s *DataSourceService) ValidateConnection(ctx context.Context, dsID string) error

ValidateConnection tests the connection to an external data source

func (*DataSourceService) ValidateCredentials 

func (s *DataSourceService) ValidateCredentials(ctx context.Context, connectorType string, credentials map[string]interface{}) error

ValidateCredentials tests connectivity using raw credentials without persisting anything.

type DataTableSummaryPayload 

type DataTableSummaryPayload struct {
	types.TracingContext
	TenantID       uint64 `json:"tenant_id"`
	KnowledgeID    string `json:"knowledge_id"`
	SummaryModel   string `json:"summary_model"`
	EmbeddingModel string `json:"embedding_model"`
}

DataTableExtractPayload represents the table extract task payload

type DataTableSummaryService 

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

DataTableSummaryService is a service for extracting tables

func (*DataTableSummaryService) Handle 

func (s *DataTableSummaryService) Handle(ctx context.Context, t *asynq.Task) error

Handle implements the TaskHandler interface for table extraction 整体流程：初始化 -> 准备资源 -> 加载数据 -> 生成摘要 -> 创建索引

type DatasetService 

type DatasetService struct{}

DatasetService provides operations for working with datasets

func (*DatasetService) GetDatasetByID 

func (d *DatasetService) GetDatasetByID(ctx context.Context, datasetID string) ([]*types.QAPair, error)

GetDatasetByID retrieves QA pairs from dataset by ID

type DesktopTarget added in v0.9.0 

type DesktopTarget struct {
	SessionID string
	MachineID string
	SandboxID string
	// contains filtered or unexported fields
}

DesktopTarget is a located desktop. The caller holds its operation, fixes the machine, claims the machine's slot, then starts it.

func (*DesktopTarget) FixMachine added in v0.9.0 

func (t *DesktopTarget) FixMachine(ctx context.Context) error

FixMachine records which machine and sandbox this desktop is for. Managers that cannot name the machine fall back to the session id (one machine per session), with no sandbox pin.

func (*DesktopTarget) HoldOperation added in v0.9.0 

func (t *DesktopTarget) HoldOperation(ctx context.Context) (context.Context, func(), error)

HoldOperation registers the desktop relay so a rebuild or reclaim of a shared machine waits for it. The returned context ends if the hold is lost. A no-op for per-session machines.

type EnvVarView 

type EnvVarView struct {
	Name        string `json:"name"`
	Description string `json:"description,omitempty"`
	Required    bool   `json:"required,omitempty"`
	Source      string `json:"source"`
	// UpdatedAt is set only for the caller's own value; a workspace value has
	// no per-variable timestamp to report.
	UpdatedAt *time.Time `json:"updated_at,omitempty"`
}

EnvVarView is one variable as its owner may see it. There is deliberately no value field: this view exists so a member can tell what is still missing, and a stored value is never read back out.

type EvaluationService 

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

EvaluationService handles evaluation tasks for knowledge base and chat models

func (*EvaluationService) EvalDataset 

func (e *EvaluationService) EvalDataset(ctx context.Context, detail *types.EvaluationDetail, knowledgeBaseID string) error

EvalDataset performs the actual evaluation of a dataset Processes each QA pair in parallel and records metrics

func (*EvaluationService) Evaluation 

func (e *EvaluationService) Evaluation(ctx context.Context,
	datasetID string, knowledgeBaseID string, chatModelID string, rerankModelID string,
) (*types.EvaluationDetail, error)

Evaluation starts a new evaluation task with given parameters datasetID: ID of the dataset to evaluate against knowledgeBaseID: ID of the knowledge base to use (empty to create new) chatModelID: ID of the chat model to evaluate rerankModelID: ID of the rerank model to evaluate

func (*EvaluationService) EvaluationResult 

func (e *EvaluationService) EvaluationResult(ctx context.Context, taskID string) (*types.EvaluationDetail, error)

type ForkBootstrapper 

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

ForkBootstrapper provisions forked sessions' first sandbox.

func NewForkBootstrapper 

func NewForkBootstrapper(
	sessions forkBootstrapSessionStore,
	messages forkBootstrapMessageStore,
	runner SandboxShellRunner,
	snapshots ForkSnapshotDeleter,
) *ForkBootstrapper

NewForkBootstrapper wires the bootstrapper.

func NewForkBootstrapperFromRepos 

func NewForkBootstrapperFromRepos(
	sessions interfaces.SessionRepository,
	messages interfaces.MessageRepository,
) *ForkBootstrapper

NewForkBootstrapperFromRepos is the exported DI constructor. The sandbox client is bound later via WithClient.

func (*ForkBootstrapper) AfterCreate 

func (b *ForkBootstrapper) AfterCreate(
	ctx context.Context, key sandbox.SessionSandboxKey, handle sandbox.RemoteSandboxHandle,
) error

AfterCreate rolls the new sandbox back to the fork point.

func (*ForkBootstrapper) OnCreateFailed 

func (b *ForkBootstrapper) OnCreateFailed(
	ctx context.Context, key sandbox.SessionSandboxKey, createErr error,
)

OnCreateFailed retires a fork snapshot that Create could not boot.

AfterCreate never runs on this path, so git-reset abandon cannot fire. NotFound / invalid template mean the snapshot will not start next time either; leave transient errors (timeout, capacity, outage) on the row so a later Resolve can retry the same snapshot.

func (*ForkBootstrapper) TemplateOverride 

func (b *ForkBootstrapper) TemplateOverride(
	ctx context.Context, key sandbox.SessionSandboxKey,
) (string, error)

TemplateOverride returns the fork snapshot for a session whose bootstrap is still pending.

func (*ForkBootstrapper) WithClient 

func (b *ForkBootstrapper) WithClient(client sandbox.RemoteSandboxClient) sandbox.SessionBootstrapper

WithClient returns a copy that executes AfterCreate against the client that created the handle. AfterCreate runs under the session lifecycle lock and must not call Resolve.

type ForkDegradeReason 

type ForkDegradeReason string

ForkDegradeReason explains why a fork could not carry the source sandbox's state over. A degraded fork still succeeds — it just starts from a brand new sandbox — because the alternative (refusing) would block the common cases: Docker kills idle containers outright, and plain chat sessions never had a sandbox to begin with. 

const (
	// ForkDegradeNoCheckpoint means the turn preceding the fork point never
	// produced a git checkpoint (commit failed, or it ran without a sandbox).
	ForkDegradeNoCheckpoint ForkDegradeReason = "NO_CHECKPOINT"

	// ForkDegradeSandboxReplaced means the checkpoint belongs to a sandbox the
	// session no longer uses. In-sandbox git history restarts from zero on
	// every rebuild, so that SHA is unreachable from the current sandbox.
	ForkDegradeSandboxReplaced ForkDegradeReason = "SANDBOX_REPLACED"

	// ForkDegradeSandboxGone means the source session has no live sandbox to
	// snapshot.
	ForkDegradeSandboxGone ForkDegradeReason = "SANDBOX_GONE"

	// ForkDegradeSnapshotUnsupported means snapshotting failed or the backend
	// does not support it.
	ForkDegradeSnapshotUnsupported ForkDegradeReason = "SNAPSHOT_UNSUPPORTED"

	// ForkDegradeCopyFailed means the shared machine could not copy the source
	// session's checkpoint into the new session's directory.
	ForkDegradeCopyFailed ForkDegradeReason = "COPY_FAILED"
)

type ForkResult 

type ForkResult struct {
	SessionID string            `json:"session_id"`
	Degraded  bool              `json:"degraded"`
	Reason    ForkDegradeReason `json:"reason,omitempty"`
}

ForkResult is what the HTTP layer renders.

type ForkSnapshotDeleter 

type ForkSnapshotDeleter interface {
	DeleteSnapshot(ctx context.Context, snapshotID string) error
}

ForkSnapshotDeleter retires a consumed or abandoned fork snapshot.

func NewResolverForkSnapshotDeleter 

func NewResolverForkSnapshotDeleter(
	resolver sandbox.TenantSandboxResolver, fallback sandbox.Manager,
) ForkSnapshotDeleter

NewResolverForkSnapshotDeleter deletes fork snapshots through the tenant's sandbox manager. Cube and E2B credentials are per-tenant, so an ID-only delete cannot reach them.

type ForkSnapshotReaper 

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

ForkSnapshotReaper deletes snapshots of forks that were never opened.

func NewForkSnapshotReaper 

func NewForkSnapshotReaper(
	sessions reaperSessionStore, snapshots ForkSnapshotDeleter, retention time.Duration,
) *ForkSnapshotReaper

NewForkSnapshotReaper wires the reaper.

func NewForkSnapshotReaperFromRepos 

func NewForkSnapshotReaperFromRepos(
	sessions interfaces.SessionRepository, snapshots ForkSnapshotDeleter,
) *ForkSnapshotReaper

NewForkSnapshotReaperFromRepos is the exported DI constructor. It uses the default retention; tests inject a shorter window through NewForkSnapshotReaper.

func (*ForkSnapshotReaper) ReapOnce 

func (r *ForkSnapshotReaper) ReapOnce(ctx context.Context) (int, error)

ReapOnce runs a single collection pass and reports how many snapshots it retired.

A per-snapshot delete failure is logged and skipped rather than returned: one unreachable snapshot must not stop the pass, and the next run retries it.

type HookMetric 

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

HookMetric tracks evaluation metrics for QA pairs

func NewHookMetric 

func NewHookMetric(capacity int) *HookMetric

NewHookMetric creates a new HookMetric with given capacity

func (*HookMetric) MetricResult 

func (h *HookMetric) MetricResult() *types.MetricResult

MetricResult returns the averaged metric results

type HostSandboxManager 

type HostSandboxManager struct {
	Manager sandbox.Manager
	// Desktop is true in the Lite desktop build even when this machine cannot
	// enforce a sandbox. Lite hides remote sandboxes entirely.
	Desktop bool
	// SkillTree and SkillInstaller serve local skill installs. Both are nil
	// unless Manager is.
	SkillTree      HostSkillTree
	SkillInstaller HostSkillInstaller
}

HostSandboxManager is Lite's OS sandbox. A nil Manager means this process cannot enforce a sandbox. Web binaries always inject the zero value, so empty remote configs resolve to disabled rather than host.

func (HostSandboxManager) SkillsAvailable 

func (h HostSandboxManager) SkillsAvailable() bool

SkillsAvailable reports whether local skills can be installed and run.

type HostSessionResolver 

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

HostSessionResolver answers "is this session running on the host backend".

Host is not a named backend, so it never writes a pin. A named ConfigID means the session belongs to a remote backend; an empty or SandboxConfigIDGlobalDefault pin plus a live host manager is the host path. PinnedSessionSandbox must not fall back to host for ExecShellCommand: that would hand WorkspaceCheckpointer a live runner and start committing the user's own repository once a turn.

func NewHostSessionResolver 

func NewHostSessionResolver(sessions sessionConfigReader, host sandbox.Manager, desktop bool) *HostSessionResolver

NewHostSessionResolver wires the shared host-session probe. A nil host manager keeps every caller on the remote path, byte-for-byte.

func (*HostSessionResolver) HostManagerFor 

func (r *HostSessionResolver) HostManagerFor(ctx context.Context, sessionID string) sandbox.Manager

HostManagerFor returns the host manager when this session has no named sandbox config and this process has a host backend; nil otherwise.

type HostSkillInstaller 

type HostSkillInstaller interface {
	sandbox.Manager
	sandbox.SessionInstallCapabilityProvider
	sandbox.SessionFileReader
	StatSessionFile(ctx context.Context, sessionID, filePath string) (*sandbox.RemoteStatEntry, error)
	WriteSessionFile(ctx context.Context, sessionID, filePath string, content []byte) error
	Bind(sessionID, dir string) (release func())
}

HostSkillInstaller is the install-only sandbox used to materialize skill packages. Implemented by adapter.InstallManager.

type HostSkillTree 

type HostSkillTree interface {
	Root() string
	VersionsRoot() string
	NewVersion(name string) (string, error)
	Activate(name, versionDir string) (string, error)
	Discard(versionDir string) error
	Prune(name string, keep ...string) error
	Remove(name string) error
	Installed(name string) bool
	Sweep() error
	Lock(name string) (func(), error)
}

HostSkillTree is the on-disk skill version tree used by Lite local installs. Implemented by internal/localsandbox/skilltree.Tree.

type HousekeepingService 

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

HousekeepingService runs background sweeps to recover stuck rows.

func NewHousekeepingService 

func NewHousekeepingService(
	db *gorm.DB, cfg *config.Config, inspector interfaces.TaskInspector, task interfaces.TaskEnqueuer,
) *HousekeepingService

NewHousekeepingService constructs a HousekeepingService. It does NOT start the cron — call Start in the application bootstrap so a misconfigured cron schedule cannot prevent the rest of the service from coming up.

func (*HousekeepingService) QueuedWork 

func (h *HousekeepingService) QueuedWork(ctx context.Context, ids []string) (map[string]bool, error)

QueuedWork reports which of ids still have work waiting in the queue or in the durable Wiki table, i.e. are backlogged rather than stuck, using the same probes as the sweep. The queue side is one shared scan (see queuedSnapshot), so answering a whole page costs one DB query. An error means the answer is unknown.

func (*HousekeepingService) Start 

func (h *HousekeepingService) Start(ctx context.Context) error

Start registers the sweep schedule and begins the background runner. Idempotent — repeated calls are a no-op so wiring code can call Start without coordinating ordering.

func (*HousekeepingService) Stop 

func (h *HousekeepingService) Stop()

Stop halts the cron and waits for in-flight sweeps to finish.

func (*HousekeepingService) StopWithin 

func (h *HousekeepingService) StopWithin(timeout time.Duration)

StopWithin is Stop with a bound. timeout <= 0 waits for the in-flight sweep.

type ImageMultimodalService 

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

ImageMultimodalService handles image:multimodal asynq tasks. It reads images from storage (via FileService for provider:// URLs), performs OCR and VLM caption, and creates child chunks.

func (*ImageMultimodalService) Handle 

func (s *ImageMultimodalService) Handle(ctx context.Context, task *asynq.Task) (retErr error)

Handle implements asynq handler for TypeImageMultimodal.

type KnowledgeAutoTagService 

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

KnowledgeAutoTagService classifies processed documents against existing knowledge-base tags.

func (*KnowledgeAutoTagService) Handle 

func (s *KnowledgeAutoTagService) Handle(ctx context.Context, task *asynq.Task) error

Handle processes one automatic-tagging task.

type KnowledgeBaseProfileService 

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

KnowledgeBaseProfileService implements interfaces.KnowledgeBaseProfileService.

func NewKnowledgeBaseProfileService 

func NewKnowledgeBaseProfileService(
	cfg *config.Config,
	kbRepo interfaces.KnowledgeBaseRepository,
	knowledgeRepo interfaces.KnowledgeRepository,
	modelService interfaces.ModelService,
	taskEnqueuer interfaces.TaskEnqueuer,
) *KnowledgeBaseProfileService

NewKnowledgeBaseProfileService constructs the service.

func (*KnowledgeBaseProfileService) BuildAggregate 

func (s *KnowledgeBaseProfileService) BuildAggregate(
	ctx context.Context, kb *types.KnowledgeBase,
) (*types.KnowledgeBaseProfileAggregate, error)

BuildAggregate implements interfaces.KnowledgeBaseProfileService.

func (*KnowledgeBaseProfileService) GenerateKnowledgeBaseProfile 

func (s *KnowledgeBaseProfileService) GenerateKnowledgeBaseProfile(
	ctx context.Context, kb *types.KnowledgeBase, force bool,
) (*types.KnowledgeBaseProfile, error)

GenerateKnowledgeBaseProfile implements interfaces.KnowledgeBaseProfileService.

func (*KnowledgeBaseProfileService) Handle 

func (s *KnowledgeBaseProfileService) Handle(ctx context.Context, task *asynq.Task) error

Handle implements the asynq handler for TypeKnowledgeBaseProfile.

func (*KnowledgeBaseProfileService) RequestKnowledgeBaseProfileRefresh 

func (s *KnowledgeBaseProfileService) RequestKnowledgeBaseProfileRefresh(
	ctx context.Context, kb *types.KnowledgeBase, force bool,
) error

RequestKnowledgeBaseProfileRefresh implements interfaces.KnowledgeBaseProfileService.

type KnowledgePostProcessService 

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

KnowledgePostProcessService acts as an orchestrator for all post-processing tasks after a document has been parsed and split into chunks (including multimodal OCR/Caption).

func (*KnowledgePostProcessService) Handle 

func (s *KnowledgePostProcessService) Handle(ctx context.Context, task *asynq.Task) error

Handle implements asynq handler for TypeKnowledgePostProcess.

type ListTenantsParams 

type ListTenantsParams struct {
	Page     int    // Page number for pagination
	PageSize int    // Number of items per page
	Status   string // Filter by tenant status
	Name     string // Filter by tenant name
}

ListTenantsParams defines parameters for listing tenants with filtering and pagination

type LocalShellJobs added in v0.9.0 

type LocalShellJobs interface {
	SubmitShellJob(
		context.Context, interfaces.SessionRef, tools.ShellJobSubmitRequest,
	) (tools.ShellJobSubmission, error)
	Status(context.Context, interfaces.SessionRef, []string, interfaces.StatusOptions) ([]interfaces.WorkResult, error)
	Cancel(context.Context, interfaces.SessionRef, []string) ([]interfaces.WorkResult, error)
	OnDelete(context.Context, interfaces.SessionRef) error
	Shutdown(context.Context) error
	EndShellJobRun(interfaces.SessionRef, string)
}

LocalShellJobs is the Lite host implementation contract. The remote service delegates to it for host sessions.

type MetricList 

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

MetricList stores and aggregates metric results

func (*MetricList) Append 

func (m *MetricList) Append(metricInput *types.MetricInput)

Append calculates and stores metrics for given input

func (*MetricList) Avg 

func (m *MetricList) Avg() *types.MetricResult

Avg calculates average of all stored metric results

type ModelCatalogService 

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

ModelCatalogService persists console overlays independently of the mounted deployment file. Every replica re-reads the singleton every five seconds; this also repairs missed notifications and works without Redis in Lite mode.

func NewModelCatalogService 

func NewModelCatalogService(
	repo *repository.ModelCatalogRepository, audit interfaces.AuditLogService,
) *ModelCatalogService

NewModelCatalogService loads the deployment layer and applies the stored overlay.

func (*ModelCatalogService) Preview 

func (s *ModelCatalogService) Preview(ctx context.Context, req CatalogUpdate) (*CatalogState, error)

Preview compiles an overlay without persisting or applying it.

func (*ModelCatalogService) Publish 

func (s *ModelCatalogService) Publish(ctx context.Context, req CatalogUpdate) (*CatalogState, error)

Publish persists an overlay as a new version and applies it.

func (*ModelCatalogService) Start 

func (s *ModelCatalogService) Start(ctx context.Context)

Start polls for overlays published by other replicas.

func (*ModelCatalogService) State 

func (s *ModelCatalogService) State(ctx context.Context) (*CatalogState, error)

State returns the current catalog state for the console.

func (*ModelCatalogService) Sync 

func (s *ModelCatalogService) Sync(ctx context.Context) error

Sync applies the stored overlay if its version changed.

type NotificationPage added in v0.9.0 

type NotificationPage struct {
	Items      []*types.Notification  `json:"items"`
	Pending    []*PendingNotification `json:"pending"`
	Unread     int64                  `json:"unread"`
	ReadCount  int64                  `json:"read_count"`
	NextCursor string                 `json:"next_cursor"`
}

type NotificationService added in v0.9.0 

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

NotificationService keeps in-app notifications.

func NewNotificationService added in v0.9.0 

func NewNotificationService(db *gorm.DB) *NotificationService

NewNotificationService builds the service.

func (*NotificationService) ClearRead added in v0.9.0 

func (s *NotificationService) ClearRead(ctx context.Context) error

ClearRead removes only read history belonging to the caller. Pending work comes from live runs and prompts and is never affected by this operation.

func (*NotificationService) List added in v0.9.0 

func (s *NotificationService) List(
	ctx context.Context, limit int, unreadOnly bool, cursor string,
) (*NotificationPage, error)

List pages history in stable descending order and always includes current work.

func (*NotificationService) MarkRead added in v0.9.0 

func (s *NotificationService) MarkRead(ctx context.Context, ids []string) error

MarkRead marks the caller's history ids read, or all history when ids is empty. Live pending work is independent of read state.

func (*NotificationService) Notify added in v0.9.0 

func (s *NotificationService) Notify(ctx context.Context, n *types.Notification)

Notify records n. A failure is logged, not returned: a notification never decides whether the work it reports succeeded.

func (*NotificationService) Prune added in v0.9.0 

func (s *NotificationService) Prune(ctx context.Context) error

Prune drops notifications past retention.

type PendingNotification added in v0.9.0 

type PendingNotification struct {
	ID        string    `json:"id"`
	Title     string    `json:"title"`
	Kind      string    `json:"kind"`
	Link      string    `json:"link"`
	CreatedAt time.Time `json:"created_at"`
}

PendingNotification is actionable state, independent of notification read state. Its ID is the current prompt, so successive approvals in one run do not collide.

type PinnedSessionSandbox 

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

PinnedSessionSandbox resolves the session's pinned sandbox manager at request time. The process-wide sandbox.Manager is DisabledManager, so checkpoint and BoundSandboxID must go through the same pin + tenant resolver path ArtifactCollector and SandboxTerminalService already use.

func NewPinnedSessionSandbox 

func NewPinnedSessionSandbox(
	pinner sessionSandboxPinReader,
	resolver sandbox.TenantSandboxResolver,
	fallback sandbox.Manager,
	host *HostSessionResolver,
) *PinnedSessionSandbox

NewPinnedSessionSandbox wires request-time sandbox access. Any dependency may be nil; BoundSandboxID then reports ok=false and ExecShellCommand fails, which WorkspaceCheckpointer treats as "no checkpoint".

host is consulted only by VersionsWorkspace. manager() must not fall back to it: that would hand WorkspaceCheckpointer a live ExecShellCommand and start committing the user's real project.

func (*PinnedSessionSandbox) BackendIsDocker added in v0.9.0 

func (a *PinnedSessionSandbox) BackendIsDocker(ctx context.Context, sessionID string) bool

BackendIsDocker reports whether the session's pinned manager is the Docker backend. The container id is not a backend name, so callers must not infer this from the sandbox id. A missing pin is not Docker.

func (*PinnedSessionSandbox) BeginWorkspaceFork added in v0.9.0 

func (a *PinnedSessionSandbox) BeginWorkspaceFork(
	ctx context.Context, sourceSessionID string,
) (sandbox.WorkspaceFork, error)

BeginWorkspaceFork opens a shared-machine fork through the session's pinned manager. Backends without shared machines cannot.

func (*PinnedSessionSandbox) BoundSandboxID 

func (a *PinnedSessionSandbox) BoundSandboxID(ctx context.Context, sessionID string) (string, bool)

BoundSandboxID reports the sandbox currently bound to sessionID without provisioning one. A session with no pin or no live binding reports ok=false.

func (*PinnedSessionSandbox) CreateForkSnapshot 

func (a *PinnedSessionSandbox) CreateForkSnapshot(
	ctx context.Context, sessionID, name string,
) (string, error)

CreateForkSnapshot snapshots the session's already-bound sandbox through the pinned manager. Managers that do not expose the method return an error so fork can degrade to SNAPSHOT_UNSUPPORTED.

func (*PinnedSessionSandbox) DeleteForkSnapshot 

func (a *PinnedSessionSandbox) DeleteForkSnapshot(
	ctx context.Context, sessionID, snapshotID string,
) error

DeleteForkSnapshot removes a fork snapshot through the session's pinned manager. Managers that do not expose the method return an error so the caller can keep the lease for the reaper.

func (*PinnedSessionSandbox) ExecShellCommand 

func (a *PinnedSessionSandbox) ExecShellCommand(
	ctx context.Context,
	sessionID, command, workDir string,
	timeout time.Duration,
	env map[string]string,
) (*sandbox.ExecuteResult, error)

ExecShellCommand runs a command on the session's pinned sandbox manager.

func (*PinnedSessionSandbox) ExecShellCommandWithOptions 

func (a *PinnedSessionSandbox) ExecShellCommandWithOptions(
	ctx context.Context, sessionID, command string, opts sandbox.ShellExecOptions,
) (*sandbox.ExecuteResult, error)

ExecShellCommandWithOptions forwards maintenance options to the pinned manager, preserving lookup-only execution and the expected sandbox identity.

func (*PinnedSessionSandbox) HasActiveTurn 

func (a *PinnedSessionSandbox) HasActiveTurn(ctx context.Context, sessionID string) (bool, error)

HasActiveTurn reports whether the session's pinned manager currently holds a turn lease. Managers that do not expose the method are treated as not busy.

func (*PinnedSessionSandbox) HasRewindLock 

func (a *PinnedSessionSandbox) HasRewindLock(ctx context.Context, sessionID string) (bool, error)

HasRewindLock reports whether rewind currently holds the session on the pinned manager. Managers that do not expose the method are treated as free.

func (*PinnedSessionSandbox) KeepAliveSession added in v0.9.0 

func (a *PinnedSessionSandbox) KeepAliveSession(ctx context.Context, sessionID string) (time.Duration, error)

KeepAliveSession refreshes the provider idle timer of the session's pinned sandbox. Managers without the method (host, disabled) have nothing to keep alive and report (0, nil).

func (*PinnedSessionSandbox) KeepAliveSessionExpected added in v0.9.0 

func (a *PinnedSessionSandbox) KeepAliveSessionExpected(
	ctx context.Context, sessionID, expectedID string,
) (time.Duration, error)

KeepAliveSessionExpected refreshes the provider timer of the bound sandbox only when it still is expectedID. Managers without the method have nothing to refresh and report (0, nil).

func (*PinnedSessionSandbox) ListSessionFiles added in v0.9.0 

func (a *PinnedSessionSandbox) ListSessionFiles(
	ctx context.Context, sessionID, dir string,
) ([]sandbox.RemoteDirEntry, error)

ListSessionFiles walks dir in the session sandbox. A manager without the method reports an error so callers skip output-file inspection.

func (*PinnedSessionSandbox) SessionWorkspaceLayout added in v0.9.0 

func (a *PinnedSessionSandbox) SessionWorkspaceLayout(
	ctx context.Context, sessionID string,
) (sandbox.WorkspaceLayout, error)

SessionWorkspaceLayout forwards the pinned manager's layout. manager does not return the lookup error (it degrades to nil), so a missing pin keeps the /workspace contract instead of failing the checkpoint closed.

func (*PinnedSessionSandbox) TryLockRewind 

func (a *PinnedSessionSandbox) TryLockRewind(ctx context.Context, sessionID string) (context.Context, func(), error)

TryLockRewind locks rewind through the session's pinned manager. A manager that can take the session exclusively returns the context to reset with; one with only the rewind lock keeps ctx; others succeed as a no-op.

func (*PinnedSessionSandbox) VersionsWorkspace 

func (a *PinnedSessionSandbox) VersionsWorkspace(ctx context.Context, sessionID string) bool

VersionsWorkspace reports whether this session's backend keeps a git history of the sandbox workspace. Host sessions return false: the workspace is a real directory, often shared across sessions, and must not be auto-committed or reset. Sessions without a host backend return true so fork keeps the remote degrade chain rather than pretending the session is on host.

type ProcessChunksOptions 

type ProcessChunksOptions struct {
	EnableQuestionGeneration bool
	QuestionCount            int
	EnableMultimodel         bool
	StoredImages             []docparser.StoredImage
	// ParentChunks holds parent chunk data when parent-child chunking is enabled.
	// When set, the chunks passed to processChunks are child chunks, and each
	// child's ParentIndex references an entry in this slice.
	ParentChunks []types.ParsedParentChunk
	Metadata     map[string]string
}

ProcessChunksOptions contains options for processing chunks

type QaInfo 

type QaInfo struct {
	QID int64 `parquet:"qid"` // Question ID
	AID int64 `parquet:"aid"` // Answer ID
}

QaInfo represents question-answer relations in parquet format

type RelsInfo 

type RelsInfo struct {
	QID int64 `parquet:"qid"` // Question ID
	PID int64 `parquet:"pid"` // Passage ID
}

RelsInfo represents question-passage relations in parquet format

type RewindResult 

type RewindResult struct {
	DeletedMessages int              `json:"deleted_messages"`
	WorkspaceReset  bool             `json:"workspace_reset"`
	Reason          RewindSkipReason `json:"reason"`
}

RewindResult is what the HTTP layer renders. WorkspaceReset is true when the live sandbox was git-reset, or when an unopened fork's ForkBootstrap was retargeted so the first provision lands on the rewind SHA.

type RewindSkipReason 

type RewindSkipReason string

RewindSkipReason explains why rewind left the workspace alone. The conversation is still truncated; the code is so the client can say so. These are not shared with ForkDegradeReason: the two operations skip for overlapping-but-not-identical reasons and the UI copy is different. 

const (
	// RewindSkipNoSandbox means the session has no bound sandbox to reset.
	RewindSkipNoSandbox RewindSkipReason = "NO_SANDBOX"

	// RewindSkipNoCheckpoint means the history being kept has no git
	// checkpoint (the image has no git, or every kept turn failed to commit).
	RewindSkipNoCheckpoint RewindSkipReason = "NO_CHECKPOINT"
)

type SandboxArtifactSource 

type SandboxArtifactSource interface {
	ListSessionFiles(ctx context.Context, sessionID, dir string) ([]sandbox.RemoteDirEntry, error)
	ReadSessionFile(ctx context.Context, sessionID, path string) ([]byte, error)
}

SandboxArtifactSource is the narrow subset of sandbox behaviour the collector needs. SessionBoundManager satisfies it in production; tests use a fake to drive the diff logic without a real sandbox.

type SandboxConfigAgentRepo 

type SandboxConfigAgentRepo interface {
	ListNamesBySandboxConfigID(ctx context.Context, tenantID uint64, configID string) ([]string, error)
}

SandboxConfigAgentRepo is the slice of the agent repository this service needs. Agent references are warnings, never grounds for refusing a change.

type SandboxDesktopLastStore 

type SandboxDesktopLastStore interface {
	Get(ctx context.Context, sessionID string) (string, error)
	Set(ctx context.Context, sessionID, sandboxID string) error
}

SandboxDesktopLastStore remembers which sandbox a session's desktop last attached to. SANDBOX_REBUILT is load-bearing: a skill install can recreate the sandbox under a live session, and a process-local map misses that when the reconnect lands on another replica.

func NewSandboxDesktopLastStore 

func NewSandboxDesktopLastStore(rdb *redis.Client) SandboxDesktopLastStore

NewSandboxDesktopLastStore returns a Redis-backed store, or an in-process one when Redis is not configured (Lite / single-instance).

type SandboxDesktopService 

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

SandboxDesktopService opens desktops on session sandboxes. It shares SandboxTerminalService's dependencies deliberately: the desktop must land on the exact sandbox the chat turn and the terminal use, which is what makes template/snapshot consistency automatic.

func NewSandboxDesktopService 

func NewSandboxDesktopService(terminals *SandboxTerminalService) *SandboxDesktopService

NewSandboxDesktopService wires the desktop service on top of the terminal service's resolution path.

func (*SandboxDesktopService) EnsureSessionDesktop 

func (s *SandboxDesktopService) EnsureSessionDesktop(
	ctx context.Context,
	sessionID string,
	provision SandboxPin,
) (*SessionDesktop, error)

EnsureSessionDesktop is the slot-less path: locate, hold, fix, start. The hold is released when the call returns; the WebSocket handler keeps its own for the relay's lifetime.

A zero provision pin is lookup-only: a running session sandbox is attached (and XFCE is lazily started if needed), a paused one returns sandbox.ErrSandboxPaused, and a missing one returns sandbox.ErrNoLiveSessionSandbox. A non-zero pin is the confirmed-click path and may create or resume, the same contract as EnsureSessionTerminal.

func (*SandboxDesktopService) LocateSessionDesktop added in v0.9.0 

func (s *SandboxDesktopService) LocateSessionDesktop(
	ctx context.Context, sessionID string, provision SandboxPin,
) (*DesktopTarget, error)

LocateSessionDesktop resolves (and, with a provision pin, creates or resumes) the session's machine without starting anything on it.

func (*SandboxDesktopService) StartSessionDesktop added in v0.9.0 

func (s *SandboxDesktopService) StartSessionDesktop(ctx context.Context, t *DesktopTarget) (*SessionDesktop, error)

StartSessionDesktop starts the desktop listeners and dials websockify on t's fixed sandbox.

type SandboxDesktopTicketClaims 

type SandboxDesktopTicketClaims struct {
	UserID    string `json:"user_id"`
	TenantID  uint64 `json:"tenant_id"`
	SessionID string `json:"session_id"`
	TokenID   string `json:"token_id"`
}

SandboxDesktopTicketClaims is the handshake identity bound to one session. TokenID is the auth_tokens.id of the access JWT that minted the ticket, so logout and revocation can tear the relay down.

type SandboxDesktopTicketStore 

type SandboxDesktopTicketStore interface {
	Issue(ctx context.Context, claims SandboxDesktopTicketClaims) (string, error)
	Consume(ctx context.Context, ticket string) (*SandboxDesktopTicketClaims, error)
}

SandboxDesktopTicketStore issues and consumes one-shot desktop tickets.

func NewSandboxDesktopTicketStore 

func NewSandboxDesktopTicketStore(rdb *redis.Client) SandboxDesktopTicketStore

NewSandboxDesktopTicketStore returns a Redis-backed store, or an in-process one when Redis is not configured.

The memory fallback mirrors selectSessionBindingStore (container/sandbox.go): Lite mode has no REDIS_ADDR and runs one process, so a process-local store is correct there. In a multi-replica deployment without Redis the ticket would be minted on one replica and consumed on another, so this logs loudly.

type SandboxInventory 

type SandboxInventory struct {
	SandboxCount int      `json:"sandbox_count"`
	SessionIDs   []string `json:"session_ids,omitempty"`
	AgentNames   []string `json:"agent_names,omitempty"`
	// UserSandboxCount is how many of SandboxCount are shared user machines.
	// They carry no session id, so SessionIDs alone under-reports what a
	// credential change would affect.
	UserSandboxCount int `json:"user_sandbox_count,omitempty"`

	// Unverifiable reports that SandboxCount is unknown rather than zero
	// because the provider could not be reached. The management page must say
	// so instead of rendering a reassuring "0 sandboxes".
	Unverifiable bool `json:"unverifiable,omitempty"`
}

SandboxInventory describes what a config holds and who a change disturbs.

type SandboxOwnerReleaser added in v0.9.0 

type SandboxOwnerReleaser interface {
	ReleaseOwnerAssignments(ctx context.Context, ownerID string) error
	ReclaimOwner(ctx context.Context, ownerID string)
}

SandboxOwnerReleaser frees a deleted account's user machines.

type SandboxPin 

type SandboxPin struct {
	// ConfigID is the sandbox config, or "" when the session has no live
	// sandbox. types.SandboxConfigIDGlobalDefault means the deployment default.
	ConfigID string

	// TenantID owns ConfigID. Zero means the session's own workspace, which
	// covers every sandbox created by an agent that workspace owns as well as
	// pins written before the column existed.
	TenantID uint64
}

SandboxPin addresses the backend a session's live sandbox runs on.

ConfigID alone is not an address. tenant_sandbox_configs is keyed by (tenant_id, id), so the same id resolves in exactly one workspace - and a shared agent runs on ITS OWNER's config while the session belongs to the borrower. Reading the workspace from the ambient request context instead is what used to strand those MicroVMs: teardown, the terminal and desktop panels and fork snapshots all run outside the chat turn, where the ambient tenant is the borrower.

Keeping the pair in one value is the point of the type: an API that can hand out a bare config id is an API where the two can drift apart again.

func (SandboxPin) IsZero 

func (p SandboxPin) IsZero() bool

IsZero reports whether the pin names no sandbox.

func (SandboxPin) TenantOr 

func (p SandboxPin) TenantOr(sessionTenantID uint64) uint64

TenantOr resolves the workspace to look ConfigID up in, falling back to the session's own tenant for pins that carry no owner.

type SandboxShellRunner 

type SandboxShellRunner interface {
	ExecShellCommand(
		ctx context.Context,
		sessionID, command, workDir string,
		timeout time.Duration,
		env map[string]string,
	) (*sandbox.ExecuteResult, error)
}

SandboxShellRunner is the narrow subset of sandbox behaviour the checkpointer needs. *sandbox.SessionBoundManager satisfies it in production; tests use a fake to assert on the emitted script without a real sandbox.

type SandboxTemplateCatalog 

type SandboxTemplateCatalog struct {
	Templates          []sandbox.RemoteTemplate `json:"templates"`
	StandardTemplateID string                   `json:"standard_template_id,omitempty"`
	DesktopTemplateID  string                   `json:"desktop_template_id,omitempty"`
	Provisioned        bool                     `json:"provisioned"`
}

type SandboxTemplateQueryInput 

type SandboxTemplateQueryInput struct {
	Config          *types.TenantSandboxConfig
	ConfigID        string
	EnsureStandard  bool
	ReplaceStandard bool
	EnsureDesktop   bool
	ReplaceDesktop  bool
}

SandboxTemplateQueryInput describes an unsaved connection from the settings drawer. ConfigID is optional and lets masked credentials resolve against the stored row while editing.

type SandboxTerminalService 

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

SandboxTerminalService opens interactive terminals on session sandboxes.

func NewSandboxTerminalService 

func NewSandboxTerminalService(
	pinner *SessionSandboxPinner,
	resolver sandbox.TenantSandboxResolver,
	fallback sandbox.Manager,
	policy WorkspaceSandboxPolicy,
	host HostSandboxManager,
) *SandboxTerminalService

NewSandboxTerminalService wires the terminal service. All dependencies are optional individually; with no pinner the service reports "no live sandbox" for every session.

func (*SandboxTerminalService) EnsureSessionTerminal 

func (s *SandboxTerminalService) EnsureSessionTerminal(
	ctx context.Context,
	sessionID string,
	provision SandboxPin,
	opts sandbox.RemoteTerminalOptions,
) (*SessionTerminal, error)

EnsureSessionTerminal opens a PTY on the session's sandbox, provisioning one when the session has none and the caller supplies a sandbox config ID.

Only call this for a confirmed user action. It creates and bills real infrastructure, so the lookup-only OpenSessionTerminal is what a page load or a background reconnect must use.

Provisioning reuses the exact chat-flow path — the resolved agent's sandbox config, the workspace kill switch, and the pin claim in resolveSandboxForExecution — so a terminal-created sandbox is indistinguishable from one created by a conversation turn. The WebSocket handler must resolve the agent the same way a chat turn does (own agent or shared agent from another workspace) and pass that agent's config AND the workspace that owns it here: a shared agent's config does not exist in the caller's own workspace. A no-op shell command drives the lazy creation, which also seeds the workspace layout.

With a zero provision pin the call stays lookup-only and reports sandbox.ErrNoLiveSessionSandbox, which the WebSocket handler maps onto the SANDBOX_NOT_BOUND guidance frame.

func (*SandboxTerminalService) OpenSessionTerminal 

func (s *SandboxTerminalService) OpenSessionTerminal(
	ctx context.Context,
	sessionID string,
	opts sandbox.RemoteTerminalOptions,
) (*SessionTerminal, error)

OpenSessionTerminal attaches to the session's currently bound sandbox without ever creating or resuming one. This is the path a panel open and an automatic reconnect take: a running sandbox is attached, a bound sandbox that is not confirmed running reports ErrSandboxPaused, and a missing one reports ErrNoLiveSessionSandbox. Opening a panel must not conjure or wake a microVM.

The WebSocket handler turns those errors into SANDBOX_PAUSED / SANDBOX_NOT_BOUND frames the UI answers with an explicit button. That confirmed click is what calls EnsureSessionTerminal.

Error contract (the WebSocket handler maps these onto protocol error frames):

  • sandbox.ErrNoLiveSessionSandbox — the session has no bound sandbox.
  • sandbox.ErrSandboxPaused — the bound sandbox is paused; resume needs an explicit click (EnsureSessionTerminal with AllowResume).
  • ErrTerminalUnsupported — the resolved backend cannot stream PTYs (Docker, disabled manager).
  • any other error — resolution or provider failure.

type SandboxTerminalTicketClaims 

type SandboxTerminalTicketClaims struct {
	UserID    string
	TenantID  uint64
	SessionID string
	TokenID   string
}

SandboxTerminalTicketClaims is the handshake identity bound to one session. TokenID is the auth_tokens.id of the access JWT that minted the ticket; the open PTY rechecks that row so logout / revocation tear the bridge down.

func ParseSandboxTerminalTicket 

func ParseSandboxTerminalTicket(raw string) (*SandboxTerminalTicketClaims, error)

ParseSandboxTerminalTicket validates a handshake ticket. Access and refresh tokens are rejected.

type SandboxesStillLiveError 

type SandboxesStillLiveError struct {
	Inventory SandboxInventory
}

SandboxesStillLiveError carries the provider inventory the handler must show in its 409 response; recomputing it could race and disagree with the refusal.

func (*SandboxesStillLiveError) Error 

func (e *SandboxesStillLiveError) Error() string

func (*SandboxesStillLiveError) Unwrap 

func (e *SandboxesStillLiveError) Unwrap() error

type SessionArtifactStore 

type SessionArtifactStore interface {
	// KnownArtifacts returns every artifact already attached to any prior
	// message of the session. The collector uses SourcePath + ModTime as the
	// cheap identity and ContentHash (or the stored blob) when mtime moved.
	KnownArtifacts(ctx context.Context, sessionID string) ([]types.MessageArtifact, error)
	// RecordRestoredMtime writes the sandbox mtime observed after a same-content
	// restore onto this session's copied artifacts so the next collect can skip
	// by path+mtime. It must not touch any other session. Best-effort.
	RecordRestoredMtime(ctx context.Context, sessionID, sourcePath string, mod time.Time, hash string) error
}

SessionArtifactStore is the minimal repository surface the collector needs to compute the "already recorded" set for a session. In production it is backed by the message repository; in tests it is stubbed with an in-memory map so the collector can be exercised without a database.

func NewMessageRepoArtifactStore 

func NewMessageRepoArtifactStore(repo interfaces.MessageRepository) SessionArtifactStore

NewMessageRepoArtifactStore wraps a MessageRepository so ArtifactCollector can build its de-duplication set from every prior message of the session.

type SessionBusyGate 

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

SessionBusyGate is process-local exclusion between send and rewind when there is no sandbox manager (and a second line when there is). Redis turn leases do not cover a process without that manager; this map does.

func NewSessionBusyGate 

func NewSessionBusyGate() *SessionBusyGate

NewSessionBusyGate returns an empty gate. Container provides one shared instance so send and rewind see each other in-process.

func (*SessionBusyGate) HoldSend 

func (g *SessionBusyGate) HoldSend(sessionID string) (func(), error)

HoldSend marks sessionID as generating. Fails when rewind already holds it.

func (*SessionBusyGate) RewindHeld 

func (g *SessionBusyGate) RewindHeld(sessionID string) bool

RewindHeld reports whether rewind currently owns sessionID.

func (*SessionBusyGate) TryLockRewind 

func (g *SessionBusyGate) TryLockRewind(sessionID string) (func(), error)

TryLockRewind takes exclusive rewind ownership. Fails when a send hold or another rewind already owns the session.

type SessionDesktop 

type SessionDesktop struct {
	Conn    *websocket.Conn
	Backend sandbox.SandboxType

	// SandboxID is what the handler compares against the previous desktop of
	// this session to detect a rebuild. Without it a skill install silently
	// swaps the desktop out from under the user.
	SandboxID string

	// IdleDisconnect is the workspace's terminal idle window, reused for the
	// desktop: both exist so a sandbox can reach its provider TTL once the
	// human walked away. Zero means the handler should use the built-in default.
	IdleDisconnect time.Duration

	// StartTTLRefresh extends the provider idle timeout while the relay
	// ctx is live. Nil when the backend has no timeout to refresh.
	StartTTLRefresh func(ctx context.Context)
}

SessionDesktop is one live desktop relay leg to a sandbox.

type SessionForkSandboxPort 

type SessionForkSandboxPort interface {
	// BoundSandboxID returns the session's currently bound sandbox without
	// provisioning one.
	BoundSandboxID(ctx context.Context, sessionID string) (string, bool)

	// HasActiveTurn reports whether an agent turn currently holds the
	// session's sandbox lease.
	HasActiveTurn(ctx context.Context, sessionID string) (bool, error)

	// CreateForkSnapshot snapshots the session's live sandbox and returns a
	// snapshot ID usable as a create-time template.
	CreateForkSnapshot(ctx context.Context, sessionID, name string) (string, error)

	// DeleteForkSnapshot removes a snapshot created for sessionID. Used when
	// fork persist fails after the snapshot already exists.
	DeleteForkSnapshot(ctx context.Context, sessionID, snapshotID string) error
}

SessionForkSandboxPort is the narrow sandbox surface fork needs. It is an interface so the fork logic is testable without a provider.

type SessionForkService 

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

SessionForkService implements the fork decision chain.

func NewSessionForkService 

func NewSessionForkService(
	sessions forkSessionStore,
	messages forkMessageStore,
	sandboxPort SessionForkSandboxPort,
) *SessionForkService

NewSessionForkService wires the service. A nil sandbox port makes every fork degrade, which is the correct behaviour for a deployment without sandboxes.

func NewSessionForkServiceFromRepos 

func NewSessionForkServiceFromRepos(
	sessions interfaces.SessionRepository,
	messages interfaces.MessageRepository,
	sandboxPort SessionForkSandboxPort,
	activity interfaces.ActivityRegistry,
) *SessionForkService

NewSessionForkServiceFromRepos is the DI-friendly constructor. The repository interfaces satisfy the narrow ports structurally.

func (*SessionForkService) Fork 

func (s *SessionForkService) Fork(
	ctx context.Context,
	tenantID uint64,
	userID string,
	sourceSessionID string,
	messageID string,
	title string,
) (*ForkResult, error)

Fork branches sourceSessionID at messageID.

Errors are reserved for conditions the user must act on: a bad request, a session they do not own, or a source that is busy. Everything about the sandbox degrades instead, and is reported through ForkResult.

func (*SessionForkService) WithAllocations added in v0.9.0 

func (s *SessionForkService) WithAllocations(a forkAllocations) *SessionForkService

WithAllocations enables user-mode forks.

type SessionPlacer added in v0.9.0 

type SessionPlacer interface {
	Place(ctx context.Context, req sandboxalloc.PlaceRequest) (allocation.Assignment, error)
	Lookup(ctx context.Context, sessionID string) (allocation.Assignment, bool, error)
}

SessionPlacer records a session's placement and its pin on first use, and reads it back without placing.

type SessionRewindSandboxPort 

type SessionRewindSandboxPort interface {
	BoundSandboxID(ctx context.Context, sessionID string) (string, bool)
	HasActiveTurn(ctx context.Context, sessionID string) (bool, error)
	// TryLockRewind locks the session for rewind. Run the workspace reset
	// with the returned context: on a shared machine it carries the
	// session's exclusive window.
	TryLockRewind(ctx context.Context, sessionID string) (context.Context, func(), error)
	SandboxShellRunner
}

SessionRewindSandboxPort is the narrow sandbox surface rewind needs.

type SessionRewindService 

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

SessionRewindService implements in-place rewind.

func NewSessionRewindService 

func NewSessionRewindService(
	sessions rewindSessionStore,
	messages rewindMessageStore,
	sandboxPort SessionRewindSandboxPort,
	knowledge rewindKnowledgeCleaner,
	suggestions rewindSuggestionCleaner,
) *SessionRewindService

NewSessionRewindService wires the service. A nil sandbox port skips the workspace reset with NO_SANDBOX, which is the correct behaviour for a deployment without sandboxes.

func NewSessionRewindServiceFromRepos 

func NewSessionRewindServiceFromRepos(
	sessions interfaces.SessionRepository,
	messages interfaces.MessageRepository,
	sandboxPort SessionRewindSandboxPort,
	knowledge interfaces.MessageService,
	suggestions interfaces.MessageSuggestionRepository,
	resolver sandbox.TenantSandboxResolver,
	fallback sandbox.Manager,
	streams interfaces.StreamManager,
	busyGate *SessionBusyGate,
	activity interfaces.ActivityRegistry,
) *SessionRewindService

NewSessionRewindServiceFromRepos is the DI-friendly constructor.

func (*SessionRewindService) Rewind 

func (s *SessionRewindService) Rewind(
	ctx context.Context,
	tenantID uint64,
	userID string,
	sessionID string,
	messageID string,
) (*RewindResult, error)

Rewind truncates sessionID at messageID and resets the live workspace when a reachable checkpoint exists.

Errors are reserved for conditions the user must act on: a bad request, a session they do not own, a busy source, or a git reset that failed. Missing sandbox state degrades to a conversation-only rewind and is reported through RewindResult.Reason.

func (*SessionRewindService) SetSuspendedRuns added in v0.9.0 

func (s *SessionRewindService) SetSuspendedRuns(runs *SuspendedRunService)

SetSuspendedRuns lets rewind close the session's suspended runs.

type SessionSandboxPinner 

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

SessionSandboxPinner is the only place session sandbox pins are read or written. Centralising it is what keeps the four entry points consistent.

func NewSessionSandboxPinner 

func NewSessionSandboxPinner(db *gorm.DB) *SessionSandboxPinner

NewSessionSandboxPinner returns a pinner over the sessions table.

func (*SessionSandboxPinner) Clear 

func (p *SessionSandboxPinner) Clear(ctx context.Context, sessionID string) error

Clear releases the pin once the sandbox is gone, so the session's next sandbox follows its agent's current backend choice.

Best effort by nature: when the reaper reclaims an orphan from the provider side the binding is already gone, so there is nothing to map back to a session and the pin is left stale. Readers must therefore treat a non-zero pin as a hint, not a guarantee that the sandbox still exists.

func (*SessionSandboxPinner) Pin 

func (p *SessionSandboxPinner) Pin(
	ctx context.Context, sessionID string, pin SandboxPin,
) (SandboxPin, error)

Pin claims the session for pin and returns the winning value.

The conditional update is what makes concurrent first-sandbox creations safe: losers adopt the winner's backend instead of building a second sandbox on a second backend.

A missing or soft-deleted session is an error rather than an empty result. Pin is called just after a sandbox was created, so answering the zero pin — which means "no live sandbox" everywhere else — would hide a real sandbox whose backend nobody recorded, leaving the orphan reaper as the only way back.

func (*SessionSandboxPinner) Read 

func (p *SessionSandboxPinner) Read(ctx context.Context, sessionID string) (SandboxPin, error)

Read returns the pinned backend, or the zero pin when the session has no live sandbox. Note the pin is an optimistic hint: the sandbox may have been reclaimed out of band (see Clear).

Once a placer is attached, a recorded assignment is the address (C15). The pin column is only its projection and may still lack a legacy owner tenant. A released assignment is returned too: cleanup and inventory still have to find the machine at its original address.

func (*SessionSandboxPinner) WithPlacer added in v0.9.0 

func (p *SessionSandboxPinner) WithPlacer(placer SessionPlacer) *SessionSandboxPinner

WithPlacer attaches the allocation service. Without one the pinner behaves exactly as before (no assignment rows; every session per-session).

type SessionTerminal 

type SessionTerminal struct {
	Session sandbox.RemoteTerminalSession
	Backend string
	// IdleDisconnect is how long the bridge should wait without PTY
	// activity before closing the socket. Always a positive duration.
	IdleDisconnect time.Duration
}

SessionTerminal bundles an opened PTY with the backend that serves it, so the WebSocket layer can include the backend in its ready frame.

type ShellJobIndex added in v0.9.0 

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

ShellJobIndex is the Redis schedule of shell jobs waiting to be swept.

func NewShellJobIndex added in v0.9.0 

func NewShellJobIndex(client redis.UniversalClient, namespace string) *ShellJobIndex

NewShellJobIndex schedules shell jobs in client. An empty namespace is "weknora".

func (*ShellJobIndex) Add added in v0.9.0 

func (x *ShellJobIndex) Add(ctx context.Context, e ShellJobIndexEntry) error

Add schedules e for a sweep five seconds after its yield window.

func (*ShellJobIndex) Claim added in v0.9.0 

func (x *ShellJobIndex) Claim(ctx context.Context, now time.Time, replica string) (
	map[interfaces.SessionRef][]ShellJobIndexEntry, error,
)

Claim takes the due jobs of every session this replica can lock.

func (*ShellJobIndex) ClaimWork added in v0.9.0 

func (x *ShellJobIndex) ClaimWork(ctx context.Context, s interfaces.SessionRef, workID string) (bool, error)

ClaimWork removes one job from the schedule so another replica will not sweep it.

func (*ShellJobIndex) DeleteSessionCursors added in v0.9.0 

func (x *ShellJobIndex) DeleteSessionCursors(ctx context.Context, s interfaces.SessionRef) error

DeleteSessionCursors removes every cursor key recorded for the session.

func (*ShellJobIndex) Due added in v0.9.0 

func (x *ShellJobIndex) Due(ctx context.Context, now time.Time) ([]ShellJobIndexEntry, error)

Due lists jobs whose sweep time has arrived.

func (*ShellJobIndex) Entries added in v0.9.0 

func (x *ShellJobIndex) Entries(ctx context.Context) ([]ShellJobIndexEntry, error)

Entries includes jobs still in their yield window, which need protection while another session's maintenance exec is blocked.

func (*ShellJobIndex) GetCursor added in v0.9.0 

func (x *ShellJobIndex) GetCursor(ctx context.Context, s interfaces.SessionRef, runID, workID string) (
	sandbox.ShellJobOffsets, bool, error,
)

GetCursor returns a run's stored read offset for one job.

func (*ShellJobIndex) Outcome added in v0.9.0 

func (x *ShellJobIndex) Outcome(ctx context.Context, s interfaces.SessionRef, workID string) (string, bool, error)

Outcome returns the terminal state recorded for one job.

func (*ShellJobIndex) PutCursor added in v0.9.0 

func (x *ShellJobIndex) PutCursor(ctx context.Context, s interfaces.SessionRef, runID, workID string,
	from sandbox.ShellJobOffsets, ttl time.Duration,
) error

PutCursor stores one run's read offset so another replica can continue it.

func (*ShellJobIndex) RecordOutcome added in v0.9.0 

func (x *ShellJobIndex) RecordOutcome(ctx context.Context, s interfaces.SessionRef, workID, state string) error

RecordOutcome keeps a terminal state where every replica can see it. The submitting process is the only one that can run a credential write-back, and an inbox notice can be acknowledged before that process sweeps.

func (*ShellJobIndex) Release added in v0.9.0 

func (x *ShellJobIndex) Release(ctx context.Context, s interfaces.SessionRef, replica string) error

Release drops the session sweep lock when this replica still holds it.

func (*ShellJobIndex) RemoveSession added in v0.9.0 

func (x *ShellJobIndex) RemoveSession(ctx context.Context, s interfaces.SessionRef) error

RemoveSession drops every scheduled job for s.

func (*ShellJobIndex) Requeue added in v0.9.0 

func (x *ShellJobIndex) Requeue(ctx context.Context, e ShellJobIndexEntry) error

Requeue puts e back on the schedule with its current yield time.

func (*ShellJobIndex) ReserveID added in v0.9.0 

func (x *ShellJobIndex) ReserveID(ctx context.Context, s interfaces.SessionRef, id string) (bool, error)

ReserveID arbitrates submissions across replicas before Lifecycle.Begin can touch an existing activity. Keep the handle reserved through notice retention and refresh it for live jobs; even a directory collision must not reuse it.

func (*ShellJobIndex) SandboxIDs added in v0.9.0 

func (x *ShellJobIndex) SandboxIDs(ctx context.Context, s interfaces.SessionRef,
	ids []string,
) map[string]string

SandboxIDs resolves the expected sandbox of each id from this session's index members (D1: the member carries sandbox_id). Ids absent from the index map to "" — the caller falls back to the current binding.

func (*ShellJobIndex) SessionClosed added in v0.9.0 

func (x *ShellJobIndex) SessionClosed(ctx context.Context, s interfaces.SessionRef) (bool, error)

SessionClosed reports whether deletion has closed the session's submission gate.

func (*ShellJobIndex) TouchRunCursors added in v0.9.0 

func (x *ShellJobIndex) TouchRunCursors(
	ctx context.Context, s interfaces.SessionRef, runID string, ttl time.Duration,
) error

TouchRunCursors extends every cursor for one run. EndShellJobRun calls it so the offsets live for an hour after the run, not after the last read.

func (*ShellJobIndex) WithSessionLock added in v0.9.0 

func (x *ShellJobIndex) WithSessionLock(
	ctx context.Context, s interfaces.SessionRef, fn func(context.Context) error,
) error

WithSessionLock prevents deletion racing a launch or a late sweep result. A renewable lock also covers maintenance execs longer than the sweep claim.

type ShellJobIndexEntry added in v0.9.0 

type ShellJobIndexEntry struct {
	SessionRef interfaces.SessionRef
	WorkID     string
	SandboxID  string
	YieldUntil time.Time
}

ShellJobIndexEntry is one scheduled job: which session and sandbox it belongs to, and when its yield window ends.

type ShellJobPayload added in v0.9.0 

type ShellJobPayload struct {
	DisplayType  string                  `json:"display_type"`
	WorkID       string                  `json:"work_id"`
	State        string                  `json:"state"`
	SessionID    string                  `json:"session_id"`
	Command      string                  `json:"command"`
	WorkDir      string                  `json:"work_dir"`
	ExitCode     *int                    `json:"exit_code,omitempty"`
	DurationMS   int64                   `json:"duration_ms"`
	StartedAt    time.Time               `json:"started_at"`
	Stdout       string                  `json:"stdout"`
	Stderr       string                  `json:"stderr"`
	FromBytes    sandbox.ShellJobOffsets `json:"from_bytes"`
	TotalBytes   sandbox.ShellJobOffsets `json:"total_bytes"`
	OmittedBytes sandbox.ShellJobOffsets `json:"omitted_bytes"`
	Truncated    bool                    `json:"truncated"`
	Note         string                  `json:"note,omitempty"`
	OutputFiles  []string                `json:"output_files,omitempty"`
}

ShellJobPayload is the model- and UI-facing body of one shell job result.

type ShellJobService added in v0.9.0 

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

ShellJobService runs remote shell jobs: submit, status, cancel, and the per-replica sweeper.

func NewShellJobService added in v0.9.0 

func NewShellJobService(pinned *PinnedSessionSandbox, bg *BackgroundWork,
	index *ShellJobIndex, events *BackgroundEventEmitter,
	keeper interfaces.SessionKeepAliver,
) *ShellJobService

NewShellJobService wires the remote shell-job service. keeper is the only way a keep-aliver enters: BackgroundWork has no KeepAliver field. events is stored on the lifecycle only when the pointer is non-nil, so a typed nil never sits in the interface.

func (*ShellJobService) Cancel added in v0.9.0 

func (s *ShellJobService) Cancel(ctx context.Context, ref interfaces.SessionRef,
	ids []string,
) ([]interfaces.WorkResult, error)

Cancel asks the watcher of each id to stop. The returned state is whatever status.json said at that moment; the durable terminal state arrives as a notice.

func (*ShellJobService) EndShellJobRun added in v0.9.0 

func (s *ShellJobService) EndShellJobRun(ref interfaces.SessionRef, runID string)

EndShellJobRun keeps the run's output cursors for another hour so a late read can still be incremental.

func (*ShellJobService) HoldSkillEnvCapture added in v0.9.0 

func (s *ShellJobService) HoldSkillEnvCapture(
	ctx context.Context, ref interfaces.SessionRef, jobID string, run func(context.Context),
)

HoldSkillEnvCapture keeps a credential write-back until this process sees the job succeed. The pairs never go into the job files or Redis.

func (*ShellJobService) LocalAvailable added in v0.9.0 

func (s *ShellJobService) LocalAvailable() bool

LocalAvailable reports whether a host implementation is installed.

func (*ShellJobService) OnDelete added in v0.9.0 

func (s *ShellJobService) OnDelete(ctx context.Context, ref interfaces.SessionRef) error

OnDelete stops detached jobs before the session's shared-machine workspace or pin disappears. Failed cancellation leaves tracking intact and aborts deletion.

func (*ShellJobService) OnDeleteAborted added in v0.9.0 

func (s *ShellJobService) OnDeleteAborted(ctx context.Context, ref interfaces.SessionRef) error

OnDeleteAborted allows new jobs after a failed session deletion.

func (*ShellJobService) OnStop added in v0.9.0 

func (s *ShellJobService) OnStop(context.Context, interfaces.SessionRef) error

OnStop does not cancel shell jobs. Stopping the turn only returns the submit tool early; the watcher keeps running.

func (*ShellJobService) RemoteAvailable added in v0.9.0 

func (s *ShellJobService) RemoteAvailable() bool

RemoteAvailable reports whether a remote submit can run. An injected remoteExec counts: tests construct the service with a nil pin and then install the seam.

func (*ShellJobService) StartSweeper added in v0.9.0 

func (s *ShellJobService) StartSweeper(ctx context.Context, interval time.Duration)

StartSweeper polls due index members until ctx is canceled. A service that cannot reach a remote sandbox does not start a goroutine.

func (*ShellJobService) Status added in v0.9.0 

func (s *ShellJobService) Status(ctx context.Context, ref interfaces.SessionRef,
	ids []string, opts interfaces.StatusOptions,
) ([]interfaces.WorkResult, error)

Status returns one result per id. Detailed=false never execs.

func (*ShellJobService) SubmitShellJob added in v0.9.0 

func (s *ShellJobService) SubmitShellJob(ctx context.Context, ref interfaces.SessionRef,
	req tools.ShellJobSubmitRequest,
) (tools.ShellJobSubmission, error)

SubmitShellJob starts a remote shell job, or delegates to the host implementation when this session lives on the user's machine.

func (*ShellJobService) SweepOnce added in v0.9.0 

func (s *ShellJobService) SweepOnce(ctx context.Context) error

SweepOnce claims due jobs, grouped by session, and reports or renews each one.

func (*ShellJobService) WithLocal added in v0.9.0 

func (s *ShellJobService) WithLocal(local LocalShellJobs) *ShellJobService

WithLocal installs the host implementation used for Lite sessions.

type SkillAdminUpdate 

type SkillAdminUpdate struct {
	// Enabled is nil when the request did not mention visibility.
	Enabled *bool
	// EnvValues holds only the names the request sent. A name it does not
	// carry keeps whatever is stored.
	EnvValues map[string]string
}

SkillAdminUpdate is everything one admin request may change about an installed skill. Both fields are optional: a request carrying only values must leave the visibility alone, and one carrying only the toggle must leave every stored value alone.

type SkillBundle 

type SkillBundle struct {
	Name         string
	Version      string
	Description  string
	Instructions string
	// SHA256 is over the uploaded bytes, so re-uploading the same archive is
	// recognisable in the UI and in the ledger, and a ready skill with this
	// digest can skip a billed snapshot rebuild.
	SHA256 string
	// Files maps skill-root-relative paths to contents, SKILL.md included.
	Files map[string][]byte
	// FrontmatterRepaired is true when SKILL.md YAML had to be repaired
	// before it would parse. The original file is unchanged; the install
	// prompt tells the agent to mention this so the user can fix it.
	FrontmatterRepaired bool
}

SkillBundle is a validated, in-memory skill archive.

func ParseSkillBundle 

func ParseSkillBundle(archive []byte) (*SkillBundle, error)

ParseSkillBundle validates an uploaded zip and extracts everything the install flow needs. It accepts both a flat archive and one wrapped in a single top-level directory, because both are what people actually upload.

func ParseSkillBundleWithOptions 

func ParseSkillBundleWithOptions(archive []byte, opts SkillBundleParseOptions) (*SkillBundle, error)

ParseSkillBundleWithOptions is ParseSkillBundle with the extra knobs remote installs need. SHA256 is still over the input bytes, not the re-rooted view.

type SkillBundleParseOptions 

type SkillBundleParseOptions struct {
	// Subdir is a skill-root path inside the archive (after any single wrap
	// directory). Empty means "find SKILL.md by the usual rules".
	Subdir string
	// AllowExtraFiles keeps files under the skill root and drops the rest,
	// instead of rejecting a GitHub zip that also contains the repo README.
	AllowExtraFiles bool
	// AllowNestedSkill accepts a unique SKILL.md nested deeper than one
	// directory, which is how monorepos and GitHub zipballs arrive.
	AllowNestedSkill bool
}

SkillBundleParseOptions relaxes the uploaded-zip rules for archives pulled from a git host or registry, which wrap the skill in a repo directory and ship README/LICENSE next to it.

type SkillCatalogInstallView 

type SkillCatalogInstallView struct {
	SkillID           string `json:"skill_id"`
	SandboxConfigID   string `json:"sandbox_config_id"`
	SandboxConfigName string `json:"sandbox_config_name,omitempty"`
	SandboxType       string `json:"sandbox_type,omitempty"`
	Status            string `json:"status"`
	Enabled           bool   `json:"enabled"`
	Error             string `json:"error,omitempty"`
	Version           string `json:"version,omitempty"`
	BundleSHA256      string `json:"bundle_sha256,omitempty"`
	// Served is set while a newer install of this skill is in flight or has
	// failed and the image still runs the previous version.
	Served    *SkillServedInfo `json:"served,omitempty"`
	UpdatedAt time.Time        `json:"updated_at"`
}

SkillCatalogInstallView is one installation of a catalog skill onto a sandbox.

type SkillCatalogView 

type SkillCatalogView struct {
	ID            string                    `json:"id" extensions:"x-agent=catalog_skill_id"`
	Name          string                    `json:"name" extensions:"x-agent"`
	Version       string                    `json:"version,omitempty" extensions:"x-agent"`
	Description   string                    `json:"description,omitempty" extensions:"x-agent"`
	BundleSHA256  string                    `json:"bundle_sha256,omitempty"`
	CreatedAt     time.Time                 `json:"created_at"`
	UpdatedAt     time.Time                 `json:"updated_at"`
	Installations []SkillCatalogInstallView `json:"installations"`
	// InstalledOn lists the sandbox config ids of Installations.
	InstalledOn []string `json:"installed_on" extensions:"x-agent"`
}

SkillCatalogView is a tenant skill definition plus its sandbox installations.

type SkillEnvGroup 

type SkillEnvGroup struct {
	SkillID     string       `json:"skill_id"`
	SkillName   string       `json:"skill_name"`
	Description string       `json:"description,omitempty"`
	Vars        []EnvVarView `json:"vars"`
}

SkillEnvGroup is one skill's declared credentials. It carries the skill's identity (name and the SKILL.md one-liner) and nothing else about it — not the instructions, not the bundle, not the install state, all of which are Admin+ disclosure.

type SkillFileContent 

type SkillFileContent struct {
	Path      string `json:"path"`
	Size      int64  `json:"size"`
	Encoding  string `json:"encoding"`
	Content   string `json:"content,omitempty"`
	MediaType string `json:"media_type,omitempty"`
	Truncated bool   `json:"truncated,omitempty"`
	Binary    bool   `json:"binary,omitempty"`
}

SkillFileContent is one file the admin browser asked to open.

type SkillFileEntry 

type SkillFileEntry struct {
	Path string `json:"path"`
	Size int64  `json:"size"`
}

SkillFileEntry is one path in an installed skill's stored archive.

type SkillInstallGuidance 

type SkillInstallGuidance struct {
	ID      string `json:"id"`
	Content string `json:"content"`
	Status  string `json:"status"`
}

SkillInstallGuidance describes one instruction and its delivery state.

type SkillInstallGuidanceState 

type SkillInstallGuidanceState struct {
	Accepting bool                   `json:"accepting"`
	Messages  []SkillInstallGuidance `json:"messages"`
}

SkillInstallGuidanceState exposes input availability and the current run's instructions.

type SkillProgress 

type SkillProgress struct {
	Percent int    `json:"percent"`
	Stage   string `json:"stage"`
	Log     string `json:"log,omitempty"`
	Status  string `json:"status,omitempty"`
}

SkillProgress is the transient install/remove progress. It is deliberately NOT persisted: the durable state is tenant_skills.status, and a percentage that outlives the job that produced it is worse than no percentage.

type SkillServedInfo 

type SkillServedInfo struct {
	Version string `json:"version,omitempty"`
}

SkillServedInfo tells the console that a previous version of a skill keeps running while the row describes an install that has not replaced it.

func ServedInfoOf 

func ServedInfoOf(row *types.TenantSkillEntity) *SkillServedInfo

ServedInfoOf reports the previous version a row is still serving, or nil when the row itself is what runs or nothing does.

type SkillSnapshotReleaseFailedError 

type SkillSnapshotReleaseFailedError struct {
	Remaining []string
}

SkillSnapshotReleaseFailedError names the snapshots that are still on the provider so the handler can show them and a retry can skip already-deleted ones.

func (*SkillSnapshotReleaseFailedError) Error 

func (e *SkillSnapshotReleaseFailedError) Error() string

func (*SkillSnapshotReleaseFailedError) Unwrap 

func (e *SkillSnapshotReleaseFailedError) Unwrap() error

type SlugUpdate 

type SlugUpdate struct {
	Slug        string
	Type        string        // "entity", "concept", "summary", "retract", "retractStale"
	Item        extractedItem // For entity/concept
	DocTitle    string
	KnowledgeID string
	SourceRef   string
	// Language is the already-resolved, human-readable language name the
	// Reduce phase interpolates into the editor prompt (e.g. "Chinese
	// (Simplified)"), NOT a locale code. Map resolves it once per document
	// so every page derived from that document shares one language.
	Language          string
	SummaryBody       string // For summary
	SummaryLine       string // For summary
	RetractDocContent string // For retract / retractStale
	// SourceChunks lists the chunk IDs (within KnowledgeID) that substantively
	// support this update. Mirrors Item.SourceChunks for convenience — the
	// Reduce phase reads from here to avoid an extra field hop.
	SourceChunks []string
	// DocSummary is the document-level summary body produced by
	// WikiSummaryPrompt (everything after the SUMMARY: ... headline, falling
	// back to the raw output if no headline could be parsed out). Carried
	// here so the Reduce phase can frame cited chunks with a rich
	// <source_context> block that tells the editor model what the document
	// is about AND what kind of document it is (resume vs announcement vs
	// product page). The one-line headline alone was too terse to keep the
	// editor grounded on longer / multi-topic source documents.
	DocSummary string
}

SlugUpdate represents a single update operation for a specific slug

type Span 

type Span struct {
	KnowledgeID  string
	Attempt      int
	SpanID       string
	ParentSpanID string
	Name         string
	Kind         string
	Status       string
	StartedAt    time.Time
}

Span is the in-memory handle the pipeline holds while a stage / subspan is executing. It carries enough context for End/Fail/Skip to write back without re-querying the DB. Returned (and required) from every Begin*.

type SpanTracker 

type SpanTracker interface {
	// OpenAttempt creates a new root span for (knowledgeID,
	// nextAttempt) and returns its number plus the root *Span. Call
	// at the start of a parse / reparse, before any other Begin*.
	OpenAttempt(ctx context.Context, knowledgeID, langfuseTraceID string) (root *Span, attempt int, err error)

	// LatestAttempt returns the highest attempt number recorded for
	// the knowledge, or 0 if it's never been parsed. Used by the API
	// layer to default to "show me the most recent run".
	LatestAttempt(ctx context.Context, knowledgeID string) int

	// BeginStage starts one of the canonical stages. Looks up the
	// root span for (kid, attempt) — caller passes attempt to make
	// the wiring explicit and let cross-process workers join an
	// existing attempt without new repo lookups.
	BeginStage(ctx context.Context, knowledgeID string, attempt int, stage string, input types.JSONMap) *Span

	// BeginSubSpan creates a child span under parent. parent may be a
	// stage span (for multimodal.image[i] / embedding.batch[i]) or
	// another subspan. kind is "subspan" or "generation" — generations
	// will be stitched to a Langfuse generation by trace_id.
	BeginSubSpan(ctx context.Context, parent *Span, name, kind string, input types.JSONMap) *Span

	// EndSpan marks span as done with optional output. Safe with nil.
	EndSpan(ctx context.Context, span *Span, output types.JSONMap)

	// FailSpan marks span as failed and cascade-cancels its
	// descendants. errorDetail (a Go error) is recorded verbatim in
	// error_detail (truncated to 8 KB) for admin views. Main pipeline stage
	// failures close the root even when unused retry budget remains.
	FailSpan(ctx context.Context, span *Span, errorCode, errorMessage string, errorDetail error)

	// FailSpanRetryable records a failure that the caller will return to the
	// task runner for retry. It leaves the root open only while retries remain.
	FailSpanRetryable(ctx context.Context, span *Span, errorCode, errorMessage string, errorDetail error)

	// SkipSpan marks an intentionally not-run span (e.g. multimodal
	// on a text-only document). Distinct from cancelled — skipped is
	// "we chose not to" while cancelled is "an upstream broke".
	SkipSpan(ctx context.Context, span *Span, reason string)

	// LookupStage returns the stage's *Span for an in-flight attempt
	// — the cross-process bridge that lets an asynq worker (e.g.
	// image_multimodal) attach subspans to the parent stage span
	// created by the upstream pipeline.
	LookupStage(ctx context.Context, knowledgeID string, attempt int, stage string) *Span

	// LookupSpanByName returns the first span of any kind matching name
	// for (knowledgeID, attempt) — the cross-process bridge that lets a
	// fan-out worker (e.g. a question-generation batch) attach its subspan
	// under a grouping span created earlier by the orchestrator. Returns
	// nil when no such span exists (caller should fall back to the stage).
	LookupSpanByName(ctx context.Context, knowledgeID string, attempt int, name string) *Span

	// FinalizeAttempt closes the root span for (knowledgeID, attempt)
	// with the given terminal status (done | failed). Idempotent:
	// re-closing an already-terminal root is a no-op so callers from
	// multiple paths (success orchestrator, dead-letter handler,
	// housekeeping) can fire without coordination. status defaults to
	// done; output/error are written verbatim.
	FinalizeAttempt(ctx context.Context, knowledgeID string, attempt int, status string,
		output types.JSONMap, errorCode, errorMessage string)

	// AbortAttempt cascade-cancels every still-running descendant of the
	// attempt's root span and then closes the root as cancelled. Used by
	// the user-initiated cancel path so the trace viewer doesn't leave
	// stranded subspans (multimodal images, postprocess subtasks)
	// looking like they're still in flight forever after the user
	// stopped the parse. Idempotent.
	AbortAttempt(ctx context.Context, knowledgeID string, attempt int, errorCode, errorMessage, reason string)
}

SpanTracker is the only public surface — kept as an interface so tests can swap in a no-op without spinning up a database.

func NewSpanTracker 

func NewSpanTracker(repo repository.KnowledgeSpanRepository, db *gorm.DB) SpanTracker

NewSpanTracker constructs the GORM-backed tracker. A nil repo collapses to a no-op so test harnesses don't need to spin up a database. The db is optional too: it's used only for the housekeeping heartbeat (see touchKnowledgeHeartbeat) and a nil db just disables that side-channel.

type StorageBackendService 

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

func NewStorageBackendService 

func NewStorageBackendService(
	repo interfaces.StorageBackendRepository,
	db *gorm.DB,
	catalogs ...interfaces.ResourceCatalog,
) *StorageBackendService

NewStorageBackendService creates a storage backend service. The optional catalog keeps focused tests compatible while production uses the explicit constructor below.

func NewStorageBackendServiceWithResources 

func NewStorageBackendServiceWithResources(
	repo interfaces.StorageBackendRepository,
	db *gorm.DB,
	catalog interfaces.ResourceCatalog,
) *StorageBackendService

NewStorageBackendServiceWithResources is the production DI constructor. The variadic constructor above remains convenient for focused tests that do not exercise resource registration.

func (*StorageBackendService) Create 

func (s *StorageBackendService) Create(ctx context.Context, backend *types.StorageBackend) error

func (*StorageBackendService) Delete 

func (s *StorageBackendService) Delete(ctx context.Context, tenantID uint64, id string) error

func (*StorageBackendService) ResolveBackend 

func (s *StorageBackendService) ResolveBackend(ctx context.Context, tenant *types.Tenant, backendID, provider string) (*types.StorageBackend, error)

func (*StorageBackendService) ResolveFileService 

func (s *StorageBackendService) ResolveFileService(ctx context.Context, tenant *types.Tenant, backendID, provider, localBaseDir string) (interfaces.FileService, string, error)

func (*StorageBackendService) SetDefault 

func (s *StorageBackendService) SetDefault(ctx context.Context, tenantID uint64, id string) error

func (*StorageBackendService) Test 

func (s *StorageBackendService) Test(ctx context.Context, backend *types.StorageBackend) error

func (*StorageBackendService) Update 

func (s *StorageBackendService) Update(ctx context.Context, incoming *types.StorageBackend) error

type SuspendedRunService added in v0.9.0 

type SuspendedRunService struct {
	RecoverAnswered func(context.Context, *types.AgentPrompt) (bool, error)
	StopDelegation  func(context.Context, *types.AgentPrompt) error
	// contains filtered or unexported fields
}

SuspendedRunService keeps the assistant message of a suspended run in step with its prompt when the system, not the run itself, moves the prompt on: the reaper taking over, or the user moving on without answering.

func NewSuspendedRunService added in v0.9.0 

func NewSuspendedRunService(broker *hitl.Broker, messages interfaces.MessageRepository) *SuspendedRunService

NewSuspendedRunService builds the service.

func (*SuspendedRunService) Close added in v0.9.0 

func (s *SuspendedRunService) Close(
	ctx context.Context, prompt *types.AgentPrompt, state *types.AgentRunSuspension, reason string,
) error

Close completes a run that will never resume. The blocked call and the calls after it are recorded as failed with reason.

func (*SuspendedRunService) Hooks added in v0.9.0 

func (s *SuspendedRunService) Hooks() hitl.RunHooks

Hooks returns the message updates the broker runs as it takes prompts over.

func (*SuspendedRunService) MarkAwaitingInput added in v0.9.0 

func (s *SuspendedRunService) MarkAwaitingInput(
	ctx context.Context, prompt *types.AgentPrompt, state *types.AgentRunSuspension,
) error

MarkAwaitingInput shows a run suspended on its owner's behalf the way a self-suspended run looks: the saved steps, waiting on the prompt.

func (*SuspendedRunService) Reap added in v0.9.0 

func (s *SuspendedRunService) Reap(ctx context.Context, now time.Time) (int, error)

Reap runs one reaper sweep.

func (*SuspendedRunService) SupersedeSession added in v0.9.0 

func (s *SuspendedRunService) SupersedeSession(ctx context.Context, sessionID string) (int, error)

SupersedeSession closes the session's suspended runs because the user moved on. Call it before the session's history is read or cut.

type TenantSandboxConfigService 

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

TenantSandboxConfigService owns the sandbox config lifecycle.

func NewTenantSandboxConfigService 

func NewTenantSandboxConfigService(
	repo repository.TenantSandboxConfigRepository,
	agents SandboxConfigAgentRepo,
	globalCfg *sandbox.Config,
	skills repository.TenantSkillRepository,
	files sandboxConfigBundleResolver,
) *TenantSandboxConfigService

NewTenantSandboxConfigService wires the config service.

func (*TenantSandboxConfigService) Create 

func (s *TenantSandboxConfigService) Create(
	ctx context.Context, tenantID uint64, in CreateSandboxConfigInput,
) (*types.TenantSandboxConfigEntity, error)

Create stores a new config.

func (*TenantSandboxConfigService) Delete 

func (s *TenantSandboxConfigService) Delete(
	ctx context.Context, tenantID uint64, id string, force bool,
) error

Delete refuses while the config still owns sandboxes, and while skill snapshots on its ledger cannot be destroyed. Agent references are permanent state and are surfaced as warnings by callers, not blockers.

force covers a skill snapshot that could not be destroyed. It does not cover an inventory we could not list: an unreachable provider is not the same as an empty one, and dropping the user-machine registry would make directory cleanup treat those machines as already gone. Sandboxes we can actually see, including user machines, still have to be dealt with through their sessions.

func (*TenantSandboxConfigService) Get 

func (s *TenantSandboxConfigService) Get(
	ctx context.Context, tenantID uint64, id string,
) (*types.TenantSandboxConfigEntity, error)

Get returns one config, or nil when absent.

func (*TenantSandboxConfigService) Inventory 

func (s *TenantSandboxConfigService) Inventory(
	ctx context.Context, tenantID uint64, id string,
) (SandboxInventory, error)

Inventory answers what changing or deleting this config would disturb.

An unreachable provider is reported through Unverifiable rather than as an error: the management page still has to render the card, and the agent names it warns about come from our own database.

func (*TenantSandboxConfigService) List 

func (s *TenantSandboxConfigService) List(
	ctx context.Context, tenantID uint64,
) ([]*types.TenantSandboxConfigEntity, error)

List returns the workspace's user-facing configs (policy row excluded).

func (*TenantSandboxConfigService) ListUserMachines added in v0.9.0 

func (s *TenantSandboxConfigService) ListUserMachines(
	ctx context.Context, tenantID uint64, configID string,
) ([]UserMachineView, error)

ListUserMachines lists the config's shared machines with their retention clock.

func (*TenantSandboxConfigService) QueryTemplates 

func (s *TenantSandboxConfigService) QueryTemplates(
	ctx context.Context, tenantID uint64, in SandboxTemplateQueryInput,
) (*SandboxTemplateCatalog, error)

QueryTemplates reads the provider's template catalog and, when asked, installs the published WeKnora CLI and desktop images if that cluster has none. Credentials come from the workspace connection, not env vars.

func (*TenantSandboxConfigService) ReleaseUserMachine added in v0.9.0 

func (s *TenantSandboxConfigService) ReleaseUserMachine(
	ctx context.Context, tenantID uint64, configID, digest string,
) error

ReleaseUserMachine deletes one shared machine now. Every session on it keeps its assignment and gets a fresh machine on next use; its files and installed packages are gone.

func (*TenantSandboxConfigService) SetWorkspaceScriptsDisabled 

func (s *TenantSandboxConfigService) SetWorkspaceScriptsDisabled(
	ctx context.Context, tenantID uint64, disabled bool,
) error

SetWorkspaceScriptsDisabled toggles script execution for the entire workspace, across all named backend types.

func (*TenantSandboxConfigService) Update 

func (s *TenantSandboxConfigService) Update(
	ctx context.Context, tenantID uint64, id string, in UpdateSandboxConfigInput,
) (*types.TenantSandboxConfigEntity, error)

Update applies an edit. Identity edits are cordoned before inventory and swept afterwards using the old client so credentials are never overwritten while they still own provider resources.

func (*TenantSandboxConfigService) WithAdmission added in v0.9.0 

func (s *TenantSandboxConfigService) WithAdmission(reg admission.Registry) *TenantSandboxConfigService

WithAdmission lets config deletion close and drain in-flight creates.

func (*TenantSandboxConfigService) WithAllocationRepo added in v0.9.0 

func (s *TenantSandboxConfigService) WithAllocationRepo(
	repo interfaces.SandboxAllocationRepository,
) *TenantSandboxConfigService

WithAllocationRepo lets config deletion drop the user machine registry.

func (*TenantSandboxConfigService) WithUserMachineReleaser added in v0.9.0 

func (s *TenantSandboxConfigService) WithUserMachineReleaser(r UserMachineReleaser) *TenantSandboxConfigService

WithUserMachineReleaser enables the admin release action.

func (*TenantSandboxConfigService) WorkspaceScriptsDisabled 

func (s *TenantSandboxConfigService) WorkspaceScriptsDisabled(
	ctx context.Context, tenantID uint64,
) (bool, error)

WorkspaceScriptsDisabled reports whether the workspace-wide kill switch is active, regardless of which named backend an agent selected.

type TenantSkillService 

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

TenantSkillService owns the skill image lifecycle for sandbox configs.

func NewTenantSkillService 

func NewTenantSkillService(
	skillsRepo repository.TenantSkillRepository,
	configsRepo repository.TenantSandboxConfigRepository,
	resolver interfaces.StorageBackendResolver,
	sandboxes sandbox.TenantSandboxResolver,
	sandboxPolicy WorkspaceSandboxPolicy,
	agents interfaces.AgentService,
	customAgents interfaces.CustomAgentService,
	sessions interfaces.SessionService,
	models interfaces.ModelService,
	redisClient *redis.Client,
	streams interfaces.StreamManager,
	messages interfaces.MessageRepository,
	host HostSandboxManager,
) *TenantSkillService

NewTenantSkillService wires the repositories and runtimes the install and remove flows share. Redis may be nil; the local lock then serialises one process only.

func (*TenantSkillService) DeleteCatalog 

func (s *TenantSkillService) DeleteCatalog(
	ctx context.Context, tenantID uint64, catalogID string,
) error

DeleteCatalog removes a definition that has no remaining installations and drops the stored zip. Sandbox uninstalls never delete that zip.

func (*TenantSkillService) GetSkill 

func (s *TenantSkillService) GetSkill(
	ctx context.Context, tenantID uint64, configID, skillID string,
) (*types.TenantSkillEntity, error)

GetSkill returns one installed skill, or nil when this workspace's config does not carry it. The repository scopes the lookup by workspace and config, so a skill ID from another workspace is indistinguishable from a missing one.

func (*TenantSkillService) InstallCatalogToConfigs 

func (s *TenantSkillService) InstallCatalogToConfigs(
	ctx context.Context, tenantID uint64, catalogID string, configIDs []string,
) (*CatalogInstallResult, error)

InstallCatalogToConfigs copies a catalog skill onto each named sandbox using the existing image-install pipeline. Missing configs are skipped with an error; partial success is returned so the caller can show per-config status.

func (*TenantSkillService) InstallGuidance 

func (s *TenantSkillService) InstallGuidance(
	ctx context.Context,
	tenantID uint64,
	configID, skillID string,
) (*SkillInstallGuidanceState, error)

InstallGuidance reads the instructions for the scoped skill's current install.

func (*TenantSkillService) InstallSkill 

func (s *TenantSkillService) InstallSkill(
	ctx context.Context, tenantID uint64, configID string, archive []byte,
) (string, error)

InstallSkill validates an uploaded archive, records it, and kicks off the install in the background. It returns the skill ID so the caller can answer 202 and let the UI subscribe to progress.

func (*TenantSkillService) InstallSkillFromSource 

func (s *TenantSkillService) InstallSkillFromSource(
	ctx context.Context, tenantID uint64, configID, source string,
) (string, error)

InstallSkillFromSource resolves a ClawHub / SkillHub / skills.sh / git / direct-zip locator to a skill bundle and runs the same install as an upload.

Every fetch is anonymous. Private registries are deliberately out of scope: carrying a credential here would mean deciding, per hop, which of a registry's handoff targets may see it, and there is no private registry to validate that against yet.

func (*TenantSkillService) LastProgress 

func (s *TenantSkillService) LastProgress(
	ctx context.Context, tenantID uint64, configID, skillID string,
) (SkillProgress, bool)

LastProgress returns the last known value so a fresh SSE connection can paint immediately instead of waiting for the next tick. tenantID is part of the key so a caller that only has config/skill IDs cannot read another workspace's progress.

func (*TenantSkillService) ListCatalog 

func (s *TenantSkillService) ListCatalog(
	ctx context.Context, tenantID uint64,
) ([]SkillCatalogView, error)

ListCatalog returns every workspace skill definition and which sandbox configs currently carry an installation of it.

func (*TenantSkillService) ListCatalogFiles 

func (s *TenantSkillService) ListCatalogFiles(
	ctx context.Context, tenantID uint64, catalogID string,
) ([]SkillFileEntry, error)

func (*TenantSkillService) ListSkillFiles 

func (s *TenantSkillService) ListSkillFiles(
	ctx context.Context, tenantID uint64, configID, skillID string,
) ([]SkillFileEntry, error)

ListSkillFiles lists the stored archive of one installed skill. The files come from the uploaded bundle rather than the live image: browsing must work while the skill is still installing, and without booting a sandbox.

func (*TenantSkillService) ListSkills 

func (s *TenantSkillService) ListSkills(
	ctx context.Context, tenantID uint64, configID string,
) ([]*types.TenantSkillEntity, error)

ListSkills returns the skills installed onto one sandbox config.

The config is read first so a config that does not belong to this workspace is reported as missing rather than as a config that simply has no skills.

func (*TenantSkillService) ListUsableSkills 

func (s *TenantSkillService) ListUsableSkills(
	ctx context.Context, tenantID uint64, configID string,
) []*types.TenantSkillEntity

ListUsableSkills returns the installed skills a chat turn can actually invoke on this config: ready, enabled, and carried by the live snapshot. Failures and a missing config yield an empty list so @ mention never 500s.

Lite has no sandbox config and no snapshot. The host target uses the same on-disk set hostSkillsForRun offers the model. The standard edition never takes that branch.

func (*TenantSkillService) PruneSupersededSnapshots 

func (s *TenantSkillService) PruneSupersededSnapshots(ctx context.Context) (int, error)

PruneSupersededSnapshots deletes provider snapshots the ledger has already retired, once they are older than this config's retention. The current image is never touched, nor is anything the ledger does not name: extras belong to other environments on a shared provider account.

Retention is a lower bound, not a guarantee the provider will accept the delete. Session sandboxes pause instead of dying when idle, and a paused sandbox still pins its template; a Conflict is left on the ledger for the next sweep.

func (*TenantSkillService) ReadCatalogFile 

func (s *TenantSkillService) ReadCatalogFile(
	ctx context.Context, tenantID uint64, catalogID, relativePath string,
) (*SkillFileContent, error)

func (*TenantSkillService) ReadSkillFile 

func (s *TenantSkillService) ReadSkillFile(
	ctx context.Context, tenantID uint64, configID, skillID, relativePath string,
) (*SkillFileContent, error)

ReadSkillFile returns one file from the stored archive. Binary files are either inlined as base64 (images small enough to preview) or reported without a body so the UI can say they cannot be opened.

func (*TenantSkillService) ReapStuckRuns 

func (s *TenantSkillService) ReapStuckRuns(ctx context.Context) (int, error)

ReapStuckRuns recovers skill rows whose install or remove process died.

Both branches turn on one question — does the image every new session boots still carry this skill's files — and skillFilesInLiveImage answers it from the snapshot ledger rather than from the row.

An installing row whose heartbeat has been silent for skillInstallStuckTTL is healed to ready when the files are there: a re-install that died before the pointer moved, or a terminal ready write that never landed. Leaving it at installing would hide a skill the image still carries. Otherwise it becomes failed so the UI stops spinning. A live install keeps stamping InstallingSince, so a run that is merely slow is never swept.

A removing row is restored to ready while the files are still there, so the operator can retry. Once they are gone the leftover row is deleted, so the agent cannot be told to invoke a skill no image carries.

func (*TenantSkillService) ReconcileSnapshots 

func (s *TenantSkillService) ReconcileSnapshots(
	ctx context.Context, tenantID uint64, configID string,
) (int, error)

ReconcileSnapshots compares provider ListSnapshots against the ledger for one config. Snapshots that exist on the provider but not in the ledger are logged as warnings and never deleted: the same provider account may be shared across environments, and an extra here is often another environment's live image.

func (*TenantSkillService) RegisterCatalogFromArchive 

func (s *TenantSkillService) RegisterCatalogFromArchive(
	ctx context.Context, tenantID uint64, archive []byte,
) (*types.TenantSkillCatalogEntity, error)

RegisterCatalogFromArchive records a skill definition without installing it onto any sandbox. Re-uploading the same name updates the stored bundle.

func (*TenantSkillService) RegisterCatalogFromSource 

func (s *TenantSkillService) RegisterCatalogFromSource(
	ctx context.Context, tenantID uint64, source string,
) (*types.TenantSkillCatalogEntity, error)

RegisterCatalogFromSource fetches a public skill and records it in the catalog.

func (*TenantSkillService) ReinstallSkill 

func (s *TenantSkillService) ReinstallSkill(
	ctx context.Context, tenantID uint64, configID, skillID string, instructions ...string,
) (string, error)

ReinstallSkill runs the install again from the archive already stored for this skill. Most failed installs have nothing to do with the archive — an unreachable sandbox, a package index that timed out, a checker that has since been corrected — and making the operator find the original zip again (or the registry URL it came from) is a poor answer to any of them.

It deliberately shares the upload path rather than jumping to runInstall. That path owns the in-flight check, the row ownership handover and the per-config lock, and a retry is exactly the moment two installs of one config are most likely to overlap.

func (*TenantSkillService) RemoveSkill 

func (s *TenantSkillService) RemoveSkill(
	ctx context.Context, tenantID uint64, configID, skillID string,
) error

RemoveSkill deletes a skill from the image. It is not a DB-only operation: the files really do leave the image, via a new snapshot taken after the directory is gone. Removal therefore shares the install's per-config lock — both grow a new snapshot from the current one, and interleaving them would discard whichever wrote the pointer first.

func (*TenantSkillService) SetSkillEnabled 

func (s *TenantSkillService) SetSkillEnabled(
	ctx context.Context, tenantID uint64, configID, skillID string, enabled bool,
) (*types.TenantSkillEntity, error)

SetSkillEnabled hides or reveals an installed skill.

func (*TenantSkillService) SetSkillEnvValues 

func (s *TenantSkillService) SetSkillEnvValues(
	ctx context.Context, tenantID uint64, configID, skillID string, values map[string]string,
) (*types.TenantSkillEntity, error)

SetSkillEnvValues stores the workspace-wide values for one skill.

func (*TenantSkillService) Start 

func (s *TenantSkillService) Start(ctx context.Context) error

Start registers the five-minute stuck-run sweep and begins the background runner. Idempotent — repeated calls are a no-op so wiring code can call Start without coordinating ordering.

func (*TenantSkillService) SteerInstall 

func (s *TenantSkillService) SteerInstall(
	ctx context.Context,
	tenantID uint64,
	configID, skillID, expectedMessageID, steerID, content string,
) error

SteerInstall queues an administrator instruction for exactly the expected install run.

func (*TenantSkillService) Stop 

func (s *TenantSkillService) Stop()

Stop halts the cron and waits for in-flight sweeps to finish.

func (*TenantSkillService) StopSkill 

func (s *TenantSkillService) StopSkill(
	ctx context.Context, tenantID uint64, configID, skillID string,
) (*types.TenantSkillEntity, error)

StopSkill aborts an in-flight install so the operator can retry or uninstall. After a process restart there is no goroutine to cancel; the row is still rewritten, which is what unblocks the UI. Removal is left alone.

func (*TenantSkillService) StopWithin 

func (s *TenantSkillService) StopWithin(timeout time.Duration)

StopWithin is Stop with a bound. timeout <= 0 waits for the in-flight sweep.

func (*TenantSkillService) SubscribeProgress 

func (s *TenantSkillService) SubscribeProgress(
	ctx context.Context, tenantID uint64, configID, skillID string,
) (<-chan SkillProgress, func(), error)

SubscribeProgress follows one install or removal.

A nil channel means there is no live stream to follow — without Redis nothing is published at all — and the caller is expected to fall back to the durable status. The returned closer must always be called: it releases the Redis subscription, which would otherwise outlive the request.

func (*TenantSkillService) UpdateSkillAdmin 

func (s *TenantSkillService) UpdateSkillAdmin(
	ctx context.Context, tenantID uint64, configID, skillID string, update SkillAdminUpdate,
) (*types.TenantSkillEntity, error)

UpdateSkillAdmin applies every admin-owned change of one request in a single read-modify-write.

Doing both in one cycle is what makes a request all-or-nothing: applying the toggle and the values as two updates can persist the first and then fail the second, leaving a credential rotation (disable → clear → re-enable → set) half applied while the admin is told it failed.

Visibility is metadata only: the files stay in the image either way, which is what makes it safe to toggle while the image is otherwise untouched. Removing the files is RemoveSkill's job and needs a new snapshot.

Of the values, only declared names are written, and a name outside the declaration is ignored rather than refused: a stale settings tab must not fail an otherwise valid save, and an invented variable would store a value nothing reads while making the declaration — the record of what this skill consumes — untrustworthy. An empty string clears the value and keeps the declaration, because "nobody filled this in" and "this is not needed" are different states.

It returns nil when the skill is not reachable for this workspace and config, so the handler renders the usual 404.

func (*TenantSkillService) WithPlacer added in v0.9.0 

func (s *TenantSkillService) WithPlacer(p SessionPlacer) *TenantSkillService

WithPlacer attaches the allocation service. Without one, maintenance sessions keep today's behaviour.

type TextInfo 

type TextInfo struct {
	ID   int64  `parquet:"id"`   // Unique identifier
	Text string `parquet:"text"` // Text content
}

TextInfo represents text data with ID in parquet format

type UpdateSandboxConfigInput 

type UpdateSandboxConfigInput struct {
	Name        string
	Description string
	Config      *types.TenantSandboxConfig
}

UpdateSandboxConfigInput is the update payload. Config may carry redacted placeholders that resolve against the stored row.

type UserEnvService 

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

UserEnvService serves the values one identity keeps for itself, in both scopes: config-wide variables and the credentials a skill declared.

It is separate from TenantSkillService because its authority is different in kind. Every method derives the workspace and the identity from the context alone and touches only rows keyed by that identity, which is what makes the endpoints safe for any logged-in member while install and listing stay Admin+.

func NewUserEnvService 

func NewUserEnvService(
	skills repository.TenantSkillRepository,
	configs repository.TenantSandboxConfigRepository,
	host HostSandboxManager,
) *UserEnvService

func (*UserEnvService) CaptureSkillEnv 

func (s *UserEnvService) CaptureSkillEnv(
	ctx context.Context, configID, skillName string, pairs map[string]string,
) error

CaptureSkillEnv stores values a successful skill run already used, for the current principal only. It writes only names that skill declared, and only names nothing has filled in yet: a value already stored by this caller or by the workspace is left alone.

Filling blanks is the whole contract. Rotation belongs to the settings page, where a person chose the new value; a value read out of a command the model composed is not that, and letting it overwrite a working credential would turn one hallucinated `export KEY=test` into a permanently broken skill. Unknown or unusable skills are a no-op.

func (*UserEnvService) DeleteMineSandbox 

func (s *UserEnvService) DeleteMineSandbox(ctx context.Context, configID, name string) error

DeleteMineSandbox removes one of the caller's own config-wide variables.

func (*UserEnvService) DeleteMineSkill 

func (s *UserEnvService) DeleteMineSkill(ctx context.Context, skillID, name string) error

DeleteMineSkill removes one of the caller's own skill values.

It deliberately does not require the skill to be visible: the delete is already scoped to rows this identity owns, and revoking a credential must keep working after an admin disables the skill it was entered for — which is precisely when somebody is most likely to want it gone.

func (*UserEnvService) ListMine 

func (s *UserEnvService) ListMine(ctx context.Context) ([]ConfigEnvGroup, error)

ListMine returns every sandbox config of the workspace with the caller's own config-wide variables and the credentials its skills declared.

A config is listed even when it carries nothing, because the config-wide editor has to be reachable. A skill is listed only when it declared something: an empty declaration has nothing to ask for.

Only enabled, ready skills are listed. A disabled or half-installed skill is never handed to the agent, so asking for its credential asks for something nothing reads.

func (*UserEnvService) SetMineSandbox 

func (s *UserEnvService) SetMineSandbox(ctx context.Context, configID, name, value string) error

SetMineSandbox stores the caller's own value for one config-wide variable. The name is free-form: this is the member-side counterpart of the admin's sandbox env_vars, not something a skill declared.

func (*UserEnvService) SetMineSkill 

func (s *UserEnvService) SetMineSkill(ctx context.Context, skillID, name, value string) error

SetMineSkill stores the caller's own value for one variable a skill declared. A name outside the declaration is refused: free-form names belong to the config-wide scope, which is not tied to any one skill.

type UserMachineReleaser added in v0.9.0 

type UserMachineReleaser interface {
	ReleaseMachine(ctx context.Context, m allocation.UserMachine) (bool, error)
}

UserMachineReleaser deletes one shared machine. It returns false when a live session or registered work still holds the machine, and must not destroy it in that case.

type UserMachineView added in v0.9.0 

type UserMachineView struct {
	Digest        string     `json:"digest"`
	OwnerTenantID uint64     `json:"owner_tenant_id"`
	OwnerID       string     `json:"owner_id,omitempty"`
	LiveSessions  int64      `json:"live_sessions"`
	EmptySince    *time.Time `json:"empty_since,omitempty"`
	ReclaimAfter  *time.Time `json:"reclaim_after,omitempty"`
}

UserMachineView is one shared machine as the config admin sees it. Owners from other workspaces (borrowers of a shared agent) are not named.

type WebSearchService 

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

WebSearchService provides web search functionality. It resolves provider configurations from the database and creates provider instances on-demand via the infrastructure registry.

func (*WebSearchService) CompressWithRAG 

func (s *WebSearchService) CompressWithRAG(
	ctx context.Context, sessionID string, tempKBID string, questions []string,
	webSearchResults []*types.WebSearchResult, cfg *types.WebSearchConfig,
	kbSvc interfaces.KnowledgeBaseService, knowSvc interfaces.KnowledgeService,
	seenURLs map[string]bool, knowledgeIDs []string,
) (compressed []*types.WebSearchResult, kbID string, newSeen map[string]bool, newIDs []string, err error)

CompressWithRAG performs RAG-based compression using a temporary, hidden knowledge base. The temporary knowledge base is deleted after use. The UI will not list it due to repo filtering.

func (*WebSearchService) Search 

func (s *WebSearchService) Search(
	ctx context.Context,
	providerID string,
	config *types.WebSearchConfig,
	query string,
) ([]*types.WebSearchResult, error)

Search performs web search using the provider entity identified by providerID. If providerID is empty, it falls back to the deprecated config.Provider field for backward compatibility.

type WikiBatchContext 

type WikiBatchContext struct {
	// SlugTitle resolves a slug to its current title (or "" if missing).
	// Backed by ListBySlugs; cache is populated as callers ask, so we
	// only pay for the slugs we actually look at.
	SlugTitle func(ctx context.Context, slug string) string

	// SlugTitleMany batches a slug-set into a single ListBySlugs query
	// and returns the resolved titles map. Convenient when a caller
	// already has the full slug list; results are still cached.
	SlugTitleMany func(ctx context.Context, slugs []string) map[string]string

	// SummaryContentByKnowledgeID returns the surviving summary page's
	// content for the given knowledge id (or "" if no summary page
	// exists / was archived). Backed by ListSummariesByKnowledgeIDs;
	// cache is populated lazily as well.
	SummaryContentByKnowledgeID func(ctx context.Context, kid string) string

	// ExtractionGranularity drives Pass 0 (candidate slug extraction)
	// aggressiveness. Resolved once per batch from the KnowledgeBase's
	// WikiConfig so every doc in the batch sees the same scope rules.
	// Already Normalize()'d — consumers can assume it is one of the
	// three valid values.
	ExtractionGranularity types.WikiExtractionGranularity

	// ContentInstructions and ExtractionInstructions are KB-scoped business
	// guidance. Stable citation, merge, taxonomy and JSON rules remain in the
	// system templates and cannot be replaced by these fields.
	ContentInstructions    string
	ExtractionInstructions string

	// PlannedFolderID holds the per-slug wiki_folders.id assigned by the batch
	// taxonomy planning pass (planBatchTaxonomy + folder resolution), keyed by
	// page slug. Reduce applies it only to pages that aren't already filed
	// (FolderID == ""), so the whole batch lands on one coherent tree without
	// churning user-curated placements. The folders themselves are created
	// sequentially before reduce, so the parallel reduce phase only assigns
	// pre-resolved ids and never races on folder creation. Read-only during
	// reduce.
	PlannedFolderID map[string]string
	// contains filtered or unexported fields
}

WikiBatchContext holds shared data across Map and Reduce phases.

Historically this carried a fully materialized `AllPages` slice plus pre-built SlugTitleMap / SummaryContentByKnowledgeID lookup tables. At 4w-document scale that meant the very first thing every batch did was load 100K+ wiki_pages rows (content TEXT included) into Go memory — and then walk them several more times for cleanDeadLinks / injectCrossLinks / getExistingPageSlugsForKnowledge.

We now lazy-load via fetchers backed by lightweight projections (ListBySlugs / ListSummariesByKnowledgeIDs). Each fetcher caches results keyed by its input so repeat lookups within a batch are free; the cache is per-batch and goroutine-local-via-mutex (sync.Map would also work but mutex keeps the surface small).

type WikiIngestPayload 

type WikiIngestPayload struct {
	types.TracingContext
	TenantID        uint64 `json:"tenant_id"`
	KnowledgeBaseID string `json:"knowledge_base_id"`
	Language        string `json:"language,omitempty"`
}

WikiIngestPayload is the asynq task payload for wiki ingest batch trigger. The actual document IDs are stored in the task_pending_ops table; this payload only carries the trigger metadata so the worker can resolve the queue tuple (task_type, scope, scope_id) and process whatever rows are queued under it.

type WikiLintIssue 

type WikiLintIssue struct {
	Type     WikiLintIssueType     `json:"type" extensions:"x-agent"`
	Severity WikiLintIssueSeverity `json:"severity" extensions:"x-agent"`
	PageSlug string                `json:"page_slug" extensions:"x-agent=slug"`
	// TargetSlug identifies the other page involved in the issue (e.g. the
	// broken link target, or the entity slug for a missing cross-ref). It is
	// the structured field used by AutoFix instead of parsing Description.
	TargetSlug  string `json:"target_slug,omitempty" extensions:"x-agent"`
	Description string `json:"description" extensions:"x-agent"`
	AutoFixable bool   `json:"auto_fixable" extensions:"x-agent"`
}

WikiLintIssue represents a single lint finding

type WikiLintIssueSeverity 

type WikiLintIssueSeverity string

WikiLintIssueSeverity defines the severity of a lint issue 

const (
	SeverityInfo    WikiLintIssueSeverity = "info"
	SeverityWarning WikiLintIssueSeverity = "warning"
	SeverityError   WikiLintIssueSeverity = "error"
)

type WikiLintIssueType 

type WikiLintIssueType string

WikiLintIssueType defines the type of lint issue 

const (
	LintIssueOrphanPage      WikiLintIssueType = "orphan_page"
	LintIssueBrokenLink      WikiLintIssueType = "broken_link"
	LintIssueStaleRef        WikiLintIssueType = "stale_ref"
	LintIssueMissingCrossRef WikiLintIssueType = "missing_cross_ref"
	LintIssueEmptyContent    WikiLintIssueType = "empty_content"
	LintIssueDuplicateSlug   WikiLintIssueType = "duplicate_slug"
)

type WikiLintReport 

type WikiLintReport struct {
	KnowledgeBaseID string           `json:"knowledge_base_id"`
	Issues          []WikiLintIssue  `json:"issues"`
	HealthScore     int              `json:"health_score" extensions:"x-agent"` // 0-100
	Stats           *types.WikiStats `json:"stats"`
	Summary         string           `json:"summary" extensions:"x-agent"`
}

WikiLintReport is the complete lint report for a wiki KB

type WikiLintService 

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

WikiLintService provides wiki health checking capabilities

func NewWikiLintService 

func NewWikiLintService(
	wikiService interfaces.WikiPageService,
	kbService interfaces.KnowledgeBaseService,
	knowledgeService interfaces.KnowledgeService,
) *WikiLintService

NewWikiLintService creates a new wiki lint service

func (*WikiLintService) AutoFix 

func (s *WikiLintService) AutoFix(ctx context.Context, kbID string) (int, error)

AutoFix attempts to automatically fix fixable issues

func (*WikiLintService) RunLint 

func (s *WikiLintService) RunLint(ctx context.Context, kbID string) (*WikiLintReport, error)

RunLint performs a comprehensive health check on a wiki knowledge base.

At 4w-document scale the legacy "load every page in one shot" approach was the dominant tail in this method (and intermittently caused OOM in production). We now walk the page set via ListPagesCursor in lintCursorBatch-sized windows, accumulating issues incrementally — memory stays bounded regardless of KB size.

We also drop the GetGraph(Limit:0) call that the legacy path used to compute the live-slug set. ListAllSlugs is a one-column projection over the same predicate (kbID + status<>archived), so it gives the same answer at a fraction of the cost.

type WikiPendingOp 

type WikiPendingOp struct {
	Op          string `json:"op"`
	KnowledgeID string `json:"knowledge_id"`
	// Ingest fields
	Language string `json:"language,omitempty"`
	// Attempt is the parse attempt whose finalizing counter owns this op's
	// slot. An op that outlived a reparse must not release a slot of the
	// newer attempt. Zero on ops queued before this field existed.
	Attempt int `json:"attempt,omitempty"`
	// Retract fields
	DocTitle   string   `json:"doc_title,omitempty"`
	DocSummary string   `json:"doc_summary,omitempty"`
	PageSlugs  []string `json:"page_slugs,omitempty"`
	FolderIDs  []string `json:"folder_ids,omitempty"`
	// contains filtered or unexported fields
}

WikiPendingOp represents a single operation queued in task_pending_ops under task_type="wiki:ingest". The struct is the JSON payload of the task_pending_ops row; the surrounding (task_type, scope, scope_id, dedup_key) fields live as separate columns and are not serialized here.

dbID is the auto-increment primary key of the task_pending_ops row the op was loaded from. PeekBatch fills it; consumers carry it through Map/Reduce so DeleteByIDs (after consume) and IncrFailCount (after failure) can address the right row. It is intentionally unexported and excluded from JSON so the persisted payload does not duplicate the column.

type WikiRetractPayload 

type WikiRetractPayload struct {
	types.TracingContext
	TenantID        uint64   `json:"tenant_id"`
	KnowledgeBaseID string   `json:"knowledge_base_id"`
	KnowledgeID     string   `json:"knowledge_id"`
	DocTitle        string   `json:"doc_title"`
	DocSummary      string   `json:"doc_summary,omitempty"` // one-line summary of the deleted document
	Language        string   `json:"language,omitempty"`
	PageSlugs       []string `json:"page_slugs"`
	FolderIDs       []string `json:"folder_ids,omitempty"`
}

WikiRetractPayload is the asynq task payload for wiki content retraction

type WorkspaceCheckpointer 

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

WorkspaceCheckpointer commits /workspace after each agent turn.

func NewWorkspaceCheckpointer 

func NewWorkspaceCheckpointer(runner SandboxShellRunner) *WorkspaceCheckpointer

NewWorkspaceCheckpointer wires a checkpointer. A nil runner yields a checkpointer that degrades to "no checkpoint", matching the graceful degradation contract used by ArtifactCollector.

func (*WorkspaceCheckpointer) Checkpoint 

func (c *WorkspaceCheckpointer) Checkpoint(
	ctx context.Context, sessionID, sandboxID, messageID string,
) *types.SandboxCheckpoint

Checkpoint commits /workspace and returns the resulting checkpoint, or nil when anything at all went wrong. It never returns an error by design: the caller runs on the turn-completion path and must not be blocked here.

func (*WorkspaceCheckpointer) WithActivityRegistry added in v0.9.0 

func (c *WorkspaceCheckpointer) WithActivityRegistry(activity interfaces.ActivityRegistry) *WorkspaceCheckpointer

WithActivityRegistry makes Checkpoint flag commits taken while background work is live. Without a registry every checkpoint is unflagged.

type WorkspaceSandboxPolicy 

type WorkspaceSandboxPolicy interface {
	WorkspaceScriptsDisabled(ctx context.Context, tenantID uint64) (bool, error)
}

WorkspaceSandboxPolicy reports whether sandbox execution is disabled for the entire workspace, including agents bound to any named backend config.

type WorkspaceVersioning 

type WorkspaceVersioning interface {
	VersionsWorkspace(ctx context.Context, sessionID string) bool
}

WorkspaceVersioning is implemented by backends that keep a git history of the sandbox workspace. Host backends must return false: the workspace is a real directory, often shared across sessions, and must not be reset or auto-committed.

Source Files

View all Source files

Directories

Path Synopsis
Package memory implements cross-session long-term memory: what the system remembers about one principal inside one workspace, independently of any single chat session.
 Package memory implements cross-session long-term memory: what the system remembers about one principal inside one workspace, independently of any single chat session.
Package sandboxalloc decides and records how sessions are placed on remote sandboxes.
 Package sandboxalloc decides and records how sessions are placed on remote sandboxes.

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.