processor

package
v0.2.45 Latest Latest
Warning

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

 Go to latest Published: Oct 7, 2026 License: MPL-2.0 Imports: 70 Imported by: 0

Documentation

Index

Constants

View Source 
const (

	// FuelCostBlobPerMiB charges txco://blob/put and blob/get per MiB of
	// bytes moved (rounded up), on top of the flat dispatch fuel every
	// EXEC pays. A 1 MiB artifact pays 100 ≈ 10 ms of chassis work —
	// decode + hash + CAS write at the §2 calibration. Exported because
	// the handlers live in package server and charge via AddFuel.
	FuelCostBlobPerMiB int64 = 100
	// FuelCostNotebookPerMiB charges txco://notebook/read and notebook/export
	// per MiB of entry bytes returned (rounded up), on top of the flat
	// dispatch fuel. Appends pay only the dispatch: the per-entry cap
	// (--notebook-max-entry-bytes) bounds bytes in. Exported because the
	// handlers live in package server and charge via AddFuel.
	FuelCostNotebookPerMiB int64 = 100
	// FuelCostOutletPerMiB charges outlet://<name>/query and /exec per MiB
	// of rows returned (rounded up), on top of the flat dispatch fuel — the
	// notebook rate. Exported for symmetry with the other per-MiB rates.
	FuelCostOutletPerMiB int64 = 100
	// FuelCostKVPerMiB charges txco://kv/mget and kv/list with values per
	// MiB of stored values returned, and kv/mset per MiB written (rounded
	// up), on top of the flat dispatch fuel. They are the only KV ops that
	// move many values at once; kv/get and kv/set move one value, bounded by
	// --kv-max-value-bytes, and pay only the dispatch. Exported because the
	// handlers live in package server and charge via AddFuel.
	FuelCostKVPerMiB int64 = 100
	// FuelCostDrivePerMiB charges txco://drive/put per MiB written and
	// drive/get per MiB returned (rounded up), on top of the flat dispatch
	// fuel — the blob rate: decode + hash + object write. The streaming
	// WebDAV head is not fuel-metered (it runs no stack); its bytes are
	// bounded by the drive limits instead. Exported because the handlers
	// live in package server and charge via AddFuel.
	FuelCostDrivePerMiB int64 = 100
)

Cost table — v1 baked-in constants. v2 may promote to config keys if real deployments need per-tenant ratios. The numbers anchor the §3 worked examples in the fuel doc; final defaults need traffic measurement.

View Source 
const AllowanceDenyReason = "allowance_exhausted"

AllowanceDenyReason is the admission reason a spent allowance refuses with (`x-txc-deny-reason` on the web head).

View Source 
const RunGrantHeader = "Txco-Run-Grant"

RunGrantHeader carries a run grant's token on a request to a node.

Variables

View Source 
var ErrAllowanceEntered = errors.New("request already entered another allowance")

ErrAllowanceEntered refuses a second allowance on a request: once a request is in one, nothing downstream can move it to another.

View Source 
var StagePartsRE = regexp.MustCompile(`(.*)\/+(\d+)$`)

StagePartsRE Set compile limit for stage

Functions

func AddFuel added in v0.2.24 

func AddFuel(ctx context.Context, cost int64, stage string) error

AddFuel is the exported charge point for core op handlers that do metered work outside the processor package (txco://blob/* per-MiB bytes). `stage` labels the charge for the exhaustion error — handlers pass the envelope's `_txc.op` stamp. Handlers charge and continue, like Exec's flat dispatch charge: the next scope entry reports the overshoot.

func AllowanceFromEnvelope added in v0.2.45 

func AllowanceFromEnvelope(raw string) string

AllowanceFromEnvelope reads the entered allowance's name from an envelope's `_txc.allowance`, for the usage lines built from a payload.

func AllowanceScope added in v0.2.45 

func AllowanceScope(ctx context.Context) string

AllowanceScope is the allowance the request entered, or "".

func EnterAllowance added in v0.2.45 

func EnterAllowance(ctx context.Context, c AllowanceCheck) (denied bool, err error)

EnterAllowance pins the request to an allowance. Entering is write-once: a request already in another allowance gets ErrAllowanceEntered, while entering the same one again refreshes the check. For a defined allowance the request's fuel ceiling is lowered to what the window has left, so one request cannot run far past it; a window already spent marks the request denied, and its next budget check (the next scope entry, or the next LOOP pass) ends it with AllowanceExhaustedError. denied reports that outcome.

func FuelUsedFromEnvelope added in v0.2.4 

func FuelUsedFromEnvelope(raw string) int64

FuelUsedFromEnvelope reads the live fuel value from an envelope. Used at the UsageEvent construction site (server.go:1010) where the merged envelope is available but the per-request context is not.

func NoteIssuedSecret added in v0.2.31 

func NoteIssuedSecret(ctx context.Context, v string)

NoteIssuedSecret records a secret an op just issued, so ScrubIssued keeps it out of this request's trace. A no-op outside a request.

func PrincipalScope added in v0.2.31 

func PrincipalScope(ctx context.Context) string

PrincipalScope returns the id of the principal the request acts as (`pony:paris`, `user:usr_…`), or "" when no one signed in. It is the trusted answer: the head that verified the password pinned it on the dispatch context (authn.WithAuthenticated). `_txc.principal` is a read-only copy for stacks; access decisions read this.

func RunGrantFrom added in v0.2.33 

func RunGrantFrom(ctx context.Context) string

RunGrantFrom is the token the run was dispatched with, or "".

func RunScope added in v0.2.31 

func RunScope(ctx context.Context) string

RunScope exposes the id of the continuation run this request executes as part of — a resumed run, or the run a deferred-join op created — or "" on an ordinary request. It is the trusted answer to "which run is calling": the identity the processor pinned, never an envelope field. The state ops record it as a transition's cause.

func ScrubIssued added in v0.2.31 

func ScrubIssued(ctx context.Context, b []byte) []byte

ScrubIssued returns b with every secret this request issued removed — the value itself replaced by [REDACTED], and its base64 forms by `A`s of the same length (so the text still decodes, to zero bytes where the secret was). b is not modified; the result is a copy when anything was found.

func ScrubbingTracer added in v0.2.31 

func ScrubbingTracer(ctx context.Context, t trace.RequestTracer) trace.RequestTracer

ScrubbingTracer wraps a request's tracer so every step it records, from any emitter, and the final payload are scrubbed of the secrets the request issued (ScrubIssued). Install the list on ctx first (WithIssuedSecrets): the wrapper reads the one the run writes to.

func SetLiveRunEntry added in v0.2.38 

func SetLiveRunEntry(ctx context.Context, stack string)

SetLiveRunEntry records the stack the run on ctx was routed into — what the boot handoff records itself (maybeRetenant). Exposed for a caller that routes a run without the handoff, and for tests in other packages.

func SourceScope added in v0.2.19 

func SourceScope(ctx context.Context) string

SourceScope exposes the pinned originating inlet to bundled ops outside this package. Like TenantScope it is captured at first Run from the ingress stamp, so it is the trusted answer to "which inlet started this request" — NOT the mutable `_txc.src` envelope field, which a rule could rewrite via `SET @src`. Privileged egress ops (e.g. txco://relay) gate on it.

func StackScope added in v0.2.31 

func StackScope(ctx context.Context) string

StackScope exposes the stack whose rule dispatched the running op, or "" if unknown. It is the trusted answer to "which stack is calling": it comes from the deployed rule (`op.Stack`), NOT from `_txc.op`, which is an envelope field — the chassis overwrites that one right before dispatch, but parsing it back is fragile (names contain slashes) and a handler should never take identity from the envelope. A slot is baked into the name ("web/canary"); callers that mean "the deployed stack" reduce it (see authn.BaseStack). The identity ops use it to stamp and check which stack owns a principal.

func StripBudgetFromOutbound added in v0.2.4 

func StripBudgetFromOutbound(raw string) string

StripBudgetFromOutbound deletes the chassis-internal budget fields from the response delivered to the inlet. Called by server.go just before forwarding the payload to the inlet client. Mirrors the existing `_txc.halt` / `_txc.goto` strip; exported so the server package (which owns the convergence between chassis emit and inlet write) can apply it after capturing the fuel value for UsageEvent.

func TenantFromEnvelope added in v0.2.5 

func TenantFromEnvelope(raw string) string

TenantFromEnvelope reads the resolved tenant slug from an envelope's `_txc.tenant`. Mirrors FuelUsedFromEnvelope; used at the trace-usage emit sites (main + resume paths) to attribute a trace to its tenant — the slug admin tenant-scoping filters on.

func TenantScope added in v0.2.6 

func TenantScope(ctx context.Context) string

TenantScope exposes the pinned request tenant slug to bundled ops outside this package (e.g. txco://sendmail's From-domain anti-spoof check). It is the trusted, immutable tenant pinned at first Run — NOT the mutable `_txc.tenant` envelope field, which a mid-pipeline op could overwrite.

func WebHostFromEnvelope added in v0.2.28 

func WebHostFromEnvelope(raw string) string

WebHostFromEnvelope reads the client's Host header from an envelope's `_txc.web.req.host`, which the web personality stamps at ingress. Read it from the INGRESS envelope, never from a finished payload: the stack can EMIT over `_txc.web.req.*`, and a usage line attributing a request to a hostname of the stack's choosing is worse than no field at all. Empty for every non-HTTP source, which the sinks omit rather than log blank.

func WithIssuedSecrets added in v0.2.31 

func WithIssuedSecrets(ctx context.Context) context.Context

WithIssuedSecrets installs the request's list of issued secrets, unless one is there already. Run installs it; a caller that records the request's final payload (runWithTrace) installs it first, so it sees what the run noted.

func WithRunGrant added in v0.2.33 

func WithRunGrant(ctx context.Context, token string) context.Context

WithRunGrant puts a run grant's token in a run's context. Only the web head calls it, from the request header; the token is never in the envelope, so no rule and no trace can see it.

func WithSource added in v0.2.19 

func WithSource(ctx context.Context, src string) context.Context

WithSource pins the request's originating inlet (`_txc.src`) into context. Exposed for tests and out-of-band callers; the data plane pins it itself on the first Run (see Run).

func WithStack added in v0.2.31 

func WithStack(ctx context.Context, stack string) context.Context

WithStack pins the dispatching rule's stack into context. Exposed for tests and out-of-band callers; the data plane pins it itself in dispatch.

func WithTenant 

func WithTenant(ctx context.Context, slug string) context.Context

WithTenant pins a tenant slug onto ctx for op resolution. Exposed for callers that drive Run/OpsForStage outside the normal ingress path (and for tests). Within the normal data plane, Run pins this itself from the envelope's `_txc.tenant`.

func WithTenantObserver added in v0.2.4 

func WithTenantObserver(ctx context.Context, o *TenantObserver) context.Context

WithTenantObserver attaches a tenant observer to ctx for the processor's pin sites to record into.

Types

type AbortError added in v0.2.38 

type AbortError struct {
	By     string
	Reason string
}

AbortError is the cause an aborted run's context carries: who asked, and why. The scope loop reads it to record the run's reason.

func AbortCause added in v0.2.38 

func AbortCause(ctx context.Context) (*AbortError, bool)

AbortCause reports the AbortError a context was cancelled with, when it was an abort (and not a deadline, a client disconnect or shutdown). An op's own timeout context inherits its parent's cause, so this reads true from inside an op as well as from the scope loop.

func (*AbortError) Error added in v0.2.38 

func (e *AbortError) Error() string

type AllowanceCheck added in v0.2.45 

type AllowanceCheck struct {
	Name     string
	Defined  bool // false: metered, never refused
	Fuel     int64
	Used     int64
	ResetsAt time.Time
}

AllowanceCheck is the state of the allowance being entered, as the caller read it from the allowance store.

type AllowanceExhaustedError added in v0.2.45 

type AllowanceExhaustedError struct {
	Allowance string
	Fuel      int64
	ResetsAt  time.Time
	FuelUsed  int64
	LastStage string
}

AllowanceExhaustedError ends a request whose allowance is spent: either the window was spent when it entered, or the request reached the ceiling the allowance set for it.

func (*AllowanceExhaustedError) AsJSON added in v0.2.45 

func (e *AllowanceExhaustedError) AsJSON() string

AsJSON is the response body: the error code, the allowance and when its window resets, and what this request burned.

func (*AllowanceExhaustedError) Error added in v0.2.45 

func (e *AllowanceExhaustedError) Error() string

func (*AllowanceExhaustedError) RetryAfter added in v0.2.45 

func (e *AllowanceExhaustedError) RetryAfter(now time.Time) time.Duration

RetryAfter is the wait until the window resets, at least a second.

type AsyncEnvelope 

type AsyncEnvelope struct {
	OpContinuationID  string `json:"op_continuation_id"`
	CallbackURL       string `json:"callback_url"`
	RunID             string `json:"run_id"`
	RunContinuationID string `json:"run_continuation_id"`
	Stack             string `json:"stack"`
	Stage             string `json:"stage"`
	Op                string `json:"op"`
	ExpiresAt         string `json:"expires_at,omitempty"`
}

AsyncEnvelope is the `_txc` block handed to an async worker so it can post one op result back. The single-use bearer token is delivered out-of-band of the body (request header), not in the envelope.

type FuelExhaustedError added in v0.2.4 

type FuelExhaustedError struct {
	MaxFuel         int64    `json:"max_fuel"`
	FuelUsed        int64    `json:"fuel_used"`
	LastStage       string   `json:"last_stage"`
	LastTransitions []string `json:"last_transitions"`
}

FuelExhaustedError is the structured terminal error for fuel exhaustion. Serialized through the same final-response path as a wall-clock cancel.

func (*FuelExhaustedError) AsJSON added in v0.2.4 

func (e *FuelExhaustedError) AsJSON() string

func (*FuelExhaustedError) Error added in v0.2.4 

func (e *FuelExhaustedError) Error() string

type GrantHandoff added in v0.2.31 

type GrantHandoff interface {
	// ForExec opens each named sandbox of run grant grantID and returns the
	// variables the command starts with — what the sandboxes set and, where
	// the command can reach this chassis, what it needs to open more itself
	// — with the values that must never appear in its output. tenant and
	// stack are the dispatching rule's own; workspaceID is the workspace the
	// command is about to run in.
	ForExec(ctx context.Context, tenant, stack, workspaceID, grantID string, sandboxes []string) (env map[string]string, scrub [][]byte, err error)
}

GrantHandoff hands a run grant to one command. The grant gateway implements it (chassis/server/grantgw); it is an interface here so the processor does not import the server.

type LiveRun added in v0.2.38 

type LiveRun struct {
	RID     string    `json:"rid"`
	Tenant  string    `json:"tenant"`
	Src     string    `json:"src"`
	Entry   string    `json:"entry,omitempty"`
	Stack   string    `json:"stack,omitempty"`
	Stage   string    `json:"stage,omitempty"`
	Started time.Time `json:"started"`
	// AbortedBy names who asked for the abort when one is in progress — the
	// run is still listed until its goroutine returns.
	AbortedBy string `json:"aborted_by,omitempty"`
}

LiveRun is one run as List reports it.

type LiveRuns added in v0.2.38 

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

LiveRuns is the registry of the runs in flight in this process, so a person (or a stack) can name one and end it without ending the chassis. The server's bus loop registers each request goroutine's run on entry and removes it on return; the processor's pin sites keep the entry current — the tenant (WithTenant), the stack the run was routed into (maybeRetenant) and the scope running now (Run, at dispatch).

An abort is a context cancel with an AbortError as its cause. The scope loop's existing cancel branch takes it exactly as it takes a client disconnect or a deadline — every op still in flight is flushed as a step and nothing later in the run executes — and records the run as `aborted`, with who asked, rather than `cancelled`: a client going away is a different fact. A hard stop: there is no cleanup hook. A stack that must tidy up does it before it asks.

Nil-safe, so a Unit built without New (unit tests) tracks nothing.

func NewLiveRuns added in v0.2.38 

func NewLiveRuns() *LiveRuns

NewLiveRuns returns an empty registry.

func (*LiveRuns) Abort added in v0.2.38 

func (r *LiveRuns) Abort(tenant, rid, by, reason string) bool

Abort ends one run: its context is cancelled with the AbortError as the cause. False when no run has the rid, or it belongs to another tenant (an empty tenant matches any). Aborting a run twice is a no-op that still answers true.

func (*LiveRuns) AbortStack added in v0.2.38 

func (r *LiveRuns) AbortStack(tenant, stack, by, reason string) int

AbortStack ends every live run of a tenant that was routed into the stack or is running one of its scopes now, and reports how many.

func (*LiveRuns) Attach added in v0.2.38 

func (r *LiveRuns) Attach(ctx context.Context, rid, src string) (context.Context, func())

Attach joins work that outlives its request goroutine — a continuable op's detached exec, which runs on a context of its own — to the run's entry under the same rid, so an abort of the run reaches it after the request has answered its client. The entry is the request's when it is still registered, or a new one (a resumed run) pinned to the tenant on ctx. The returned context is cancelled when the run is aborted; the func releases the attachment. Work attached to a run already aborted is cancelled at once.

func (*LiveRuns) Count added in v0.2.38 

func (r *LiveRuns) Count() int

Count is the number of runs registered now.

func (*LiveRuns) List added in v0.2.38 

func (r *LiveRuns) List(tenant string) []LiveRun

List reports the live runs of a tenant, oldest first; an empty tenant reports every tenant's (the super-admin's view).

func (*LiveRuns) Register added in v0.2.38 

func (r *LiveRuns) Register(ctx context.Context, rid, src string) (context.Context, func())

Register enters a run under its rid and returns the context the run must execute on — cancellable by Abort — and the func that releases the entry when the run returns (call it once; more is a no-op). A second registration of the same rid replaces the first in the registry (an inlet that takes the rid from its client cannot promise uniqueness); the first run keeps running and releases only itself.

type TTLExhaustedError added in v0.2.4 

type TTLExhaustedError struct {
	MaxTTL          int64    `json:"max_ttl"`
	Consumed        int64    `json:"consumed"`
	LastStage       string   `json:"last_stage"`
	LastTransitions []string `json:"last_transitions"`
}

TTLExhaustedError is the structured terminal error for hop-budget exhaustion. Distinct code from fuel exhaustion so operators can see "this is a loop" vs "this is expensive work" at a glance.

func (*TTLExhaustedError) AsJSON added in v0.2.4 

func (e *TTLExhaustedError) AsJSON() string

func (*TTLExhaustedError) Error added in v0.2.4 

func (e *TTLExhaustedError) Error() string

type TenantObserver added in v0.2.4 

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

TenantObserver records the tenant a request is pinned to, so the server can attribute usage/billing from immutable pipeline state rather than the mutable response envelope. The envelope's `_txc.tenant` is display/debug/routing metadata an author-controlled stack can rewrite; billing must not trust it.

It mirrors the admission.Lease holder pattern: created in the server bus-loop goroutine that owns a request's lifetime, attached to the request context, and read back after the pipeline returns. The processor's tenant-pin sites (first pin in Run; the one-way `_sys`->concrete retenant) record the resolved slug via WithTenant. Last write wins — for a routed request that is the concrete tenant; for an unrouted one it stays `_sys`.

Nil-safe so non-server callers (tests, CLI) can ignore it.

func NewTenantObserver added in v0.2.4 

func NewTenantObserver() *TenantObserver

NewTenantObserver returns a fresh observer with no tenant recorded yet.

func (*TenantObserver) NotifyAccepted added in v0.2.31 

func (o *TenantObserver) NotifyAccepted(ch chan<- event.Acceptance)

NotifyAccepted registers the channel the processor's acceptance point sends on (Envelope.Accepted). A nil channel leaves acceptance silent. Nil-safe.

func (*TenantObserver) Route added in v0.2.31 

func (o *TenantObserver) Route() (ingress string, hostnameVerified bool)

Route returns the ingress key and verified flag recorded beside Stack at the _sys->tenant handoff; zero values when no handoff happened.

func (*TenantObserver) Stack added in v0.2.31 

func (o *TenantObserver) Stack() (stack string, ok bool)

Stack returns the stack recorded at the _sys->tenant handoff (routeBody stamps `_txc.stack` from the route proposal; maybeRetenant records it here as it rebinds the pin). ok is false when no handoff happened.

func (*TenantObserver) Tenant added in v0.2.4 

func (o *TenantObserver) Tenant() (slug string, ok bool)

Tenant returns the last-recorded tenant slug and whether any pin was observed. ok is false if the pipeline never pinned a tenant (e.g. a request that errored before Run pinned), letting the caller fall back.

type Tracker added in v0.2.30 

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

Tracker counts pipeline work in flight so shutdown can let it finish before cancelling it. The server's bus loop tracks its request goroutines with one; Unit.Work tracks the work that outlives its request — a continuation's detached tail, a local async op, a worker callback's resume — which no request goroutine waits for.

Unlike sync.WaitGroup, Begin may race a waiter at a zero count: a draining node keeps admitting internal work while shutdown waits. Nil-safe, so a Unit built without New (unit tests) tracks nothing.

func NewTracker added in v0.2.30 

func NewTracker() *Tracker

NewTracker returns an idle tracker.

func (*Tracker) Begin added in v0.2.30 

func (t *Tracker) Begin() (end func())

Begin counts one unit of work until the returned func is called. Calling it more than once is a no-op.

func (*Tracker) Count added in v0.2.30 

func (t *Tracker) Count() int

Count is the work in flight now.

func (*Tracker) Idle added in v0.2.30 

func (t *Tracker) Idle() <-chan struct{}

Idle returns a channel that is closed once nothing is in flight: already closed when the count is zero, otherwise closed when it next reaches zero.

type TreeBundleSource added in v0.2.39 

type TreeBundleSource interface {
	Get(ctx context.Context, hash string) ([]byte, error)
}

TreeBundleSource reads a stack tree's bundle by its sha256: the chassis file store (filecas.Store satisfies it).

type Unit 

type Unit struct {
	Conf      config.Config
	Logger    *zap.Logger
	Kv        store.Store
	RuntimeDB *sql.DB
	// RuntimeDialect is the SQL dialect of RuntimeDB, derived once from the
	// runtime DSN. Today always SQLite (the runtime DSN is file:); it exists
	// so the runtime write path is dialect-portable ahead of the cloud
	// service moving the runtime store to Postgres. Admin direct-tx writers
	// and the control-publish pump read it via pu.RuntimeDialect; the stores
	// carry their own copy via their constructors. Note: dbcache snapshot
	// reads are always SQLite regardless of this value (the mirror is a
	// :memory: SQLite dump), so mirror/snapshot readers pass registry.SQLite,
	// not this field. See docs/todo-runtime-postgres.md.
	RuntimeDialect authregistry.Dialect
	AuthDB         *sql.DB
	Dbc            *dbcache.DbCache
	Mc             *metrics.Metrics

	Bus        chan<- *event.Envelope
	Reg        registry.Registry
	Mux        *radix.Tree
	HTTPClient *http.Client

	// Runs is the continuation store wrapper. Used only on the barrier
	// (async) path; nil-safe — the sync fast path never touches it.
	Runs            *continuation.Runs
	CallbackBaseURL string

	// Sink spawns resume traces from background goroutines (local
	// async; future similar work). The same sink the web personality
	// uses, exposed here so chassis-internal goroutines that outlive
	// their originating request can write traces too. Nil-safe — when
	// unset, local async still functions but the resume work is
	// untraced (used in unit tests + non-server contexts).
	Sink trace.Sink

	// Work tracks pipeline work that outlives its request goroutine — a
	// continuation's detached tail, a local async op, a worker callback's
	// resume — so shutdown can let it finish before cancelling. Nil-safe.
	Work *Tracker

	// Live is the registry of the runs in flight in this process, by rid,
	// so one can be named and aborted (the admin plane's run endpoints,
	// txco://run/abort). The server registers each request goroutine's
	// run; the pin sites here keep the entry current. Nil-safe.
	Live *LiveRuns

	// MCPSessions caches server-minted `Mcp-Session-Id` values per
	// (tenant, endpoint) so hot MCP paths skip the init lifecycle
	// (3 HTTPS round-trips → 1). Nil-safe — when unset, ExecMCPHTTP
	// runs the full lifecycle every call (test default; backwards-
	// compatible).
	MCPSessions *mcpSessionCache

	// Computes runs op:// computes resolved to "compute://<alg>/<digest>".
	// nil-safe: ExecCompute fails loudly if a compute:// op fires while
	// this is unset (no engine wired). Set when the chassis is built with
	// a compute runtime.
	Computes compute.Runner

	// Workspaces runs workspace://<name>/<verb> ops: an owned, stateful
	// execution environment per (tenant, stack, name), backed by the
	// provider named by --workspace-provider. nil-safe: ExecWorkspace
	// fails loudly if a workspace:// op fires while this is unset (no
	// provider configured, or the local provider refused without
	// --workspace-allow-local).
	Workspaces *workspace.Manager

	// TreeBundles reads a stack tree's bundle by its sha256 (chassis/stackdir)
	// for a workspace exec that asks for it with `cwd = "$TXCO_STACK_DIR"`:
	// the chassis file store. nil-safe: such an exec fails in-band with
	// stack_dir_unavailable.
	TreeBundles TreeBundleSource

	// Grants hands a run grant to a command: `workspace://<name>/exec WITH
	// grant = <id>`. nil-safe: the op fails in-band when unset (this node
	// opened no identity store).
	Grants GrantHandoff

	// Attachments is the registry of attached transports — a live resource
	// in a workspace (a PTY, or a workspace-local service) bound to one
	// WebSocket session for the life of a lease. The `workspace://<name>/attach`
	// and `/connect` ops create bindings here; the websocket personality
	// pumps them. nil-safe: both fail in-band when unset.
	Attachments *attach.Registry

	// Usage is the usage sink. nil-safe. When set, each compute or
	// workspace invocation emits a usage event (src="compute" /
	// src="workspace") alongside the per-request one.
	Usage usage.Sink

	// Allowances is the tenant-defined fuel budget store (txco://allowance/*
	// and the admin API share it, so a definition set on this node is seen
	// at once by its charges). nil when no KV store is configured.
	Allowances *allowance.Store

	// Secrets is the per-tenant secret-store Resolver. Non-nil when
	// --secret-master-key is configured AND its file loads cleanly at
	// boot; nil otherwise. PR 3 will consult this from the processor
	// splice that populates op.Secrets between WHEN/SET/SELECT/WITH
	// decoration and Exec. While Secrets is nil, any op with
	// `secrets` in its WITH clause must fail loud with
	// `secret_store_unavailable` rather than silently skip.
	Secrets *secrets.Resolver

	// Outlets runs outlet://<name>/<op>: pooled, bounded calls to external
	// services a stack declared under OUTLETS/, credentials held by the
	// chassis. nil-safe — an outlet op then reports txco_outlet_unavailable
	// as data. Set by the server (chassis/server) once the egress guard and
	// the secret resolver exist.
	Outlets *outlet.Runtime

	// Admission gates the one-time _sys -> concrete-tenant handoff: a
	// suspended/disabled tenant is denied (402/403) before its stack
	// runs. nil-safe — when unset every tenant is admitted (the test and
	// non-server default). Set by the server from a dbcache-backed
	// provider rebuilt on every reload.
	Admission admission.Provider
	// contains filtered or unexported fields
}

Unit is the chassis's per-process processor handle. RuntimeDB holds content (ops, stacks, versions, files, tenants) and is always present. AuthDB holds identity (actors, keys, memberships, invitations, browser sessions) and is nil on data-plane-only chassis where the admin personality isn't active.

func New 

func New(conf config.Config, logger *zap.Logger, reg registry.Registry, mc *metrics.Metrics, bus chan<- *event.Envelope, kv store.Store, runtimeDB, authDB *sql.DB, dbc *dbcache.DbCache, runs *continuation.Runs, guard egress.Guard, secretsResolver *secrets.Resolver) *Unit

New Processor. authDB may be nil on data-plane-only chassis. guard is the outbound op-dial policy consulted at the dial step (see egress).

func (*Unit) DecorateInput 

func (pu *Unit) DecorateInput(input string, overrides []resonator.BranchValue) (string, error)

DecorateInput overrides input with preset values, but only if the target path doesn't already exist (set-if-absent semantics — used by SET PRE / SET POST).

Each override.Value flows through runtime.Resolve so future Value shapes (PathRef, FunctionCall) plumb through without touching this function. On a resolution error the call returns the partially-updated input plus the error; per the strict-by-default semantics in internal docs/todo-txcl-expressions.md §5, callers MUST treat the error as a rule-halt signal (silently dropping the SET would hide protocol errors).

func (*Unit) DriveDeferredResume 

func (pu *Unit) DriveDeferredResume(runID, stage string)

DriveDeferredResume claims and resumes a deferred-join suspend on a fresh background context (the request ctx is gone). Self-contained and safe for both the suspend-time race guard and the worker-completion path: it re-checks resumability and lets ClaimResume pick the single winner.

func (*Unit) EnableMCPSessionCache 

func (pu *Unit) EnableMCPSessionCache(ttl time.Duration)

EnableMCPSessionCache turns on session caching for hot MCP paths. The cache type is unexported (it's an implementation detail of the processor); this method is the construction surface that server.go uses at boot. Idempotent — calling twice replaces the underlying cache (dropping any in-flight entries; safe at boot time before any traffic flows).

func (*Unit) Exec 

func (pu *Unit) Exec(ctx context.Context, op operation.Operation) (event.Payload, string, error)

Exec Execute an operation at this step Exec dispatches an operation to its transport and returns the produced payload plus the transport string the dispatch switch actually took ("txco", "https", "ai", "mock", …). Callers derive the trust decision via transportAuthorControlled(transport) and MUST sanitizeAuthorOutput the output before merging it into the envelope when that reports true, so an untrusted producer (remote HTTP, sandboxed compute, an MCP tool, or a rule-author mock) cannot forge reserved `_txc.*` control fields (tenant, computed auth, budget, …). Callers that STORE the output (continuation terminals) persist the transport with it so the later resume merge can make the same decision.

func (*Unit) ExecAI added in v0.2.4 

func (pu *Unit) ExecAI(ctx context.Context, op operation.Operation) (event.Payload, error)

ExecAI dispatches ai://<sub_op>. v1 supports only ai://chat.

The handler:

  1. Parses the sub-op (chat is the only one wired today).
  2. Decodes op.Meta (WITH-clause materialization) into a chat.Request.
  3. Resolves a backend via chat.Resolve (provider override or first-registered default; capability matching is its own follow-up).
  4. Materializes the backend's RequiredSecrets() through the per-tenant store with optional env-var fallback (AIChatEnvFallback config), reusing the existing chassis pattern: SecretBag, RecordSecretMaterialize, fuelCostSecretMaterialize fuel charge.
  5. Renders {{@path}} prompt templates over op.Input. NOTE: on a rule with SELECT assignments, op.Input is the PROJECTED view, so a {{@path}} of an unselected path renders empty — select what the prompt needs.
  6. Calls backend.Run exactly once (no tool loop in v1).
  7. Validates the response against WITH schema if present (binary ok / failed; no repair semantics).
  8. Emits a chat.completion trace event with provider, model, token counts, routing decision, retries, latency, and schema validation outcome.
  9. Builds the response envelope: top-level text + schema_validated_payload + chat.error (when set) + _txc.chat.* metadata. Token counts are recorded HERE for downstream observability/billing; they are NOT charged to the chassis fuel meter — provider compute is a separate dimension.

Error policy: chat-level failures (auth, provider HTTP/net, schema validation, missing secret, no backend) surface as a structured `chat.error` field on the response envelope so rule authors can handle uniformly via `WHEN @chat.error EXEC ...`. ExecAI returns a non-nil error only for malformed-shape conditions a rule author cannot work around (unrecognized sub-op, op.Meta failing JSON parse, an unset runtime dependency). This gives rule authors one error-handling pattern; structured codes (txco_chat_*) live on the envelope.

func (*Unit) ExecCap added in v0.2.33 

func (pu *Unit) ExecCap(ctx context.Context, op operation.Operation) (event.Payload, error)

ExecCap runs a `cap://<name>` op: one call up to the parent.

func (*Unit) ExecCompute 

func (pu *Unit) ExecCompute(ctx context.Context, op operation.Operation) (event.Payload, error)

ExecCompute runs a `compute://<alg>/<digest>` op: it resolves the content-addressed artifact and invokes it on its engine, returning the output as a JSON payload. The output is handed back to Exec's shared tail, so it flows the identical post-EXEC processing (EMIT overlay, per-scope merge, _txc.goto/halt) as http:// and txco:// — the compute transport is not a special case downstream.

func (*Unit) ExecCore 

func (pu *Unit) ExecCore(ctx context.Context, op operation.Operation) (event.Payload, error)

ExecCore on core execution (vs remote)

func (*Unit) ExecHTTP 

func (pu *Unit) ExecHTTP(ctx context.Context, op operation.Operation) (event.Payload, error)

ExecHTTP Handles execution of http, https operations.

When op.Meta declares `secrets.*` refs (per internal docs/todo-secret-store.md §4), this is the only place cleartext crosses from op.Secrets (the in-process bag) into the outbound wire — applied as request headers or as JSON body field overlays. op.Input is NOT mutated, so the trace event still records the request body the operator authored; the wire body is a separate buffer that lives only for the duration of the HTTP call.

func (*Unit) ExecHTTPAsync 

func (pu *Unit) ExecHTTPAsync(ctx context.Context, op operation.Operation, env AsyncEnvelope, token string) (jobID string, err error)

ExecHTTPAsync invokes an async worker. Unlike ExecHTTP it wraps the op input as `{ "input": <op.Input>, "_txc": {…} }` (async-only — ExecHTTP is untouched) and carries the single-use callback token in the X-Txco-Continuation-Token request header. The worker is expected to answer 202 Accepted (optionally `{"job_id":"…"}`); the real result arrives later via the callback endpoint. Returns the worker-supplied job id (optional, may be "").

func (*Unit) ExecMCPHTTP 

func (pu *Unit) ExecMCPHTTP(ctx context.Context, op operation.Operation) (payload event.Payload, err error)

ExecMCPHTTP dispatches an MCP-over-HTTP `tools/call` op. Mirrors ExecHTTP structurally and shares pu.HTTPClient so the egress guard, per-op context, and trace pipeline all apply for free.

v0 is correct but not optimized: every op pays the full session lifecycle (initialize → notifications/initialized → tools/call). A session cache per (tenant, endpoint) is a v0.5 optimization that doesn't change this signature or the rule syntax.

func (*Unit) ExecOutlet added in v0.2.31 

func (pu *Unit) ExecOutlet(ctx context.Context, op operation.Operation) (event.Payload, error)

ExecOutlet dispatches outlet://<name>/<op>: a statement written in the op against an external service the stack declared under OUTLETS/, with every value bound through `args`. The chassis holds the credential, owns the pool and bounds the call; the op sees rows.

The result lands under `WITH into` (default `_outlet`): 

<into>.ok             true | false
<into>.outlet         the outlet name
<into>.columns        SELECT order (query, or exec with RETURNING)
<into>.rows           [{col: val}, ...]
<into>.count          rows returned
<into>.rows_affected  exec only
<into>.error          {code, message[, sqlstate]} (only when not ok)

Every failure — an undeclared outlet, a missing secret, a refused dial, a database error, a crossed ceiling, a timeout — is data: `ok = false` plus a coded error, and a nil Go error so the op still merges. The next op gates on `<into>.ok == true`; a missing path compares as false.

outlet is a trusted transport (the handler builds the output), so the author-chosen `into` goes through txcguard.AuthorTarget: it may not name a reserved `_txc` path. The outlet is looked up for the dispatching op's own stack (op.Stack reduced to its app stack, as txco://kv does), never for a stack named in the envelope.

func (*Unit) ExecWorkspace added in v0.2.28 

func (pu *Unit) ExecWorkspace(ctx context.Context, op operation.Operation) (event.Payload, error)

ExecWorkspace runs a `workspace://<name>/<verb>` op against the configured workspace provider. The output is handed back to Exec's shared tail, so it flows the identical post-EXEC processing (EMIT overlay, per-scope merge, _txc.goto/halt) as every other transport.

Error policy — exit codes are data. A command that exits 3 is a successful dispatch whose result says `exit: 3`. A transport or provider failure (timeout, capacity, the workspace would not wake) is ALSO returned as a merged payload — `<into>: {}` plus a top-level `workspace.error {code, message}` — so a rule can gate on it (`WHEN .workspace.error.code == "timeout"`). Only authoring errors return a Go error and drop the op: a malformed ref, a bad name, no provider configured, an untenanted request, or a malformed `secrets` block.

Secrets: `WITH secrets.env.<NAME>.secret = "STORED"` (plus the optional `.format`) materializes a stored secret into the command's environment — the only place a workspace op accepts one. The processor splice has already put the cleartext in op.Secrets; here it is copied into ExecRequest.Env, and every materialized value is scrubbed from stdout, stderr and error text before anything is built into the payload (and so before the trace step sees it). See scrubSecrets for the caveat.

func (*Unit) Handle 

func (pu *Unit) Handle(prefix []byte, handler event.OpsHandler)

Handle Setup the internal routing table, passing off opnames starting with prefix to this handler

func (*Unit) MakeMockResponse 

func (pu *Unit) MakeMockResponse(op operation.Operation, errMsg string) event.Payload

func (*Unit) MergeJSON 

func (pu *Unit) MergeJSON(src string, dst string) (string, error)

MergeJSON folds dst's top-level keys into src: scalars overwrite, objects deep-merge, arrays append (dst elements after src — this is the accumulator design). The fast path builds the result in a single buffer; it bails to the sjson-based slow path on any input it cannot prove it reproduces byte-for-byte (odd keys, duplicate keys, whitespace-led docs, merge errors). Both paths MUST stay byte-identical: envelopes are golden-matched in tests and traces. See merge_oracle_test.go.

func (*Unit) NodeForOp 

func (pu *Unit) NodeForOp(ctx context.Context, op operation.Operation) (*registry.Node, error)

NodeForOp Given an operation description, find matching node.

Now that gRPC is removed and HTTP rules dispatch directly to op.Resonator.Exec, this is only consulted for txco:// (local) operations that route by the fixed registry. Kept around because the ExecCore path may still want logical-name resolution in the future; today it's effectively a passthrough.

func (*Unit) OpsForStage 

func (pu *Unit) OpsForStage(ctx context.Context, stage string) ([]operation.Operation, error)

OpsForStage lookup operations for given stage. If no rows match the requested stack, the lookup falls back along stack-prefix boundaries: a request at `website/canary/100` that finds nothing tries `website/100`, then “. This implements the overlay model — a sparse `website/canary` tree inherits any scope it doesn't explicitly override from the parent stack. Wildcard stacks (containing `%`, the one SQL LIKE metacharacter a pattern may carry) skip the fallback walk; they already match across stacks at one level.

The walk stops at a `_`-prefixed segment: `<stack>/_mail` is a CHANNEL, not a specialization, and must not inherit its parent's HTTP rules. See peelParent.

func (*Unit) OverlayResponse 

func (pu *Unit) OverlayResponse(env, output string, overrides []resonator.BranchValue) (string, error)

OverlayResponse writes branch values onto a JSON document with overwrite semantics — the EMIT clause's runtime. Sibling to DecorateInput, which is set-if-absent.

Two distinct documents flow through here:

  • env: the envelope to RESOLVE PathRef / FunctionCall args against. Typically op.Input — the envelope the op saw at dispatch, so EMIT's PathRefs like `@web.req.body` read live envelope state.
  • output: the WRITE accumulator. EMIT's resolved values are sjson-set onto this; the merger picks it up after the op returns. Empty/missing becomes "{}" so callers don't need to special-case EMIT-only rules.

Splitting env from output is load-bearing: if both were the same document (the EMIT accumulator), `EMIT @x = @env.field` couldn't see the envelope at all — `@env.field` would resolve against the initially-empty accumulator and come back nil. The two-arg form is a fix for that.

Each override.Value flows through runtime.Resolve. Resolution errors halt and propagate up (strict-by-default); sjson write errors (path unwritable) keep the existing log-and-continue behavior because the path syntax is a programming bug rather than a value-computation failure.

This exported form applies AUTHOR provenance (fail-closed): reserved `_txc.*` writes outside the author allowlist are dropped. Pipeline call sites use OverlayResponseFor, which derives the provenance from the run's pinned tenant so system-authored boot rules keep their route-proposal capability.

func (*Unit) OverlayResponseFor added in v0.2.24 

func (pu *Unit) OverlayResponseFor(ctx context.Context, env, output string, overrides []resonator.BranchValue) (string, error)

OverlayResponseFor is OverlayResponse with the writer's provenance derived from ctx: a run pinned to the `_sys` tenant (the boot pipeline — see systemMayWriteTxc for why the pin is unforgeable) is SYSTEM-authored and may additionally EMIT `_txc.route.*`. Callers on detached paths must pass the tenant-preserving work ctx (detachedOpContext), not a bare Background ctx — getting it wrong fails closed (drops to author provenance).

func (*Unit) ResonatingOps 

func (pu *Unit) ResonatingOps(input string, ops []operation.Operation, hashSeed string) ([]operation.Operation, error)

ResonatingOps returns the subset of ops whose WHEN clause matches the input. If any matching op has Priority > 0, only the highest-priority op is returned; otherwise all matching ops are returned (parallel execution at this stage).

Clause evaluation order on a single rule: WHEN match, PRIORITY filter, PRE-SET (set-if-absent input decoration), WITH (resolved against the FULL pre-projection input — chassis directives may reference anything), then SELECT (copy + input projection), then dispatch. POST-SET is NOT applied here — it overlays the merged envelope after the scope completes (advanceAfterScope). WITH carries per-call metadata; PRIORITY picks the highest-priority rule set; EXEC is the dispatch target.

func (*Unit) Resume 

func (pu *Unit) Resume(ctx context.Context, runID, stage string) error

Resume merges all op-terminal outputs for (runID, stage) in ascending ordinal onto the stage's scope-entry envelope, then drives the opstack forward via advanceAfterScope (the same engine the sync path uses). Called by the winning continuation callback. Writes result.json on terminal completion; if the pipeline re-suspends at a later async barrier nothing is written here — that stage's callbacks drive it. A failed sibling op fails the stage (design: a failed sibling fails the scope). Runs in the caller's ctx, not the dead request ctx.

func (*Unit) Run 

func (pu *Unit) Run(ctx context.Context, raw string, stage string, resCh chan event.Payload) error

Run Execute a request.

Snapshot semantics: the first call to Run captures the current dbcache in-memory *sql.DB and attaches it to ctx. All subsequent recursive Run calls (for goto / next-stage advancement within the same request) reuse that snapshot. A `txco apply` that completes mid-request swaps pu.Dbc.Db on the chassis side but leaves the captured pointer untouched — Go's GC keeps the old in-memory DB alive while any request holds a reference. So the opstack a request sees at stage 0 is the same opstack it sees at every later stage in the same request.

Tests that drive Run directly without a snapshot in ctx still work: OpsForStage falls back to pu.Dbc.Db when the ctx key is absent.

func (*Unit) StageParse 

func (pu *Unit) StageParse(stage string) (stack string, scope int, err error)

Source Files

View all Source files

Jump to

Keyboard shortcuts

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