inject

package
v0.3.0 Latest Latest
Warning

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

Go to latest
Published: Aug 14, 2026 License: Apache-2.0 Imports: 13 Imported by: 0

Documentation

Overview

Package inject implements the HTTP endpoint that receives edge-trigger payloads (from k8s-event-watcher and any other source speaking the envelope.InjectPayload shape) and dispatches them into the mast runtime.

For the v0.1 spike this endpoint is single-session and single-bearer. Multi-session substrate and the X-Asserted-Caller proxy-identity mechanism from core-agent's recipe are deferred.

Index

Constants

View Source
const SharedCredentialIdentity = "shared-bearer-token"

SharedCredentialIdentity is the Caller identity recorded when no Authenticator is configured. It is deliberately not "anonymous" and deliberately not a person: the request did present a credential — the daemon's shared bearer token — and the honest audit record says that a holder of that token approved, not that nobody did and not that somebody in particular did.

Variables

View Source
var ErrBadPayload = errors.New("invalid inject payload")

ErrBadPayload, returned (or wrapped) by a Handler, marks a request the daemon refuses on its content (e.g. a payload UID deriving a reserved session ID). Mapped to 400 instead of the generic 500 so emitters don't retry a request that can never succeed.

View Source
var ErrConflict = errors.New("session state refuses this request")

ErrConflict, returned (or wrapped) by a handler, marks a request the daemon refuses because of the session's current state — aborted (terminal), gate-paused, or an expired resume token. Mapped to 409: not the emitter's payload (400), not a transient daemon condition (503) — the session's state has to change first (resume, ack, extend-token), so blind retries are wrong.

View Source
var ErrUnavailable = errors.New("daemon is shutting down; not accepting new work")

ErrUnavailable, returned (or wrapped) by a Handler or ResumeHandler, tells the server the daemon is refusing new work — a shutdown drain is underway. The server maps it to 503 + Retry-After instead of the generic 500, so emitters retry against the replacement pod rather than treating a rolling restart as a crash.

Functions

This section is empty.

Types

type AbortHandler

type AbortHandler func(ctx context.Context, req AbortRequest) error

AbortHandler applies an abort request. Optional; when nil the /abort route responds 404.

type AbortRequest

type AbortRequest struct {
	// SessionID identifies the session to mark aborted.
	SessionID string `json:"session_id"`

	// Reason is the operator-supplied reason, recorded in the abort
	// marker and surfaced by `mast sessions list/show`.
	Reason string `json:"reason,omitempty"`
}

AbortRequest asks the daemon to mark a session aborted.

Semantics are those of pkg/transcript's Store.Abort — a durable operator-abort marker appended to the session's event log, not preemption of in-flight work. See that method's doc for the full contract (docs/durable-execution-design.md, "Operator-facing surface"; engine-level terminal abort is v0.2).

type AckEffectsHandler added in v0.2.0

type AckEffectsHandler func(ctx context.Context, req AckEffectsRequest) error

AckEffectsHandler applies an effects acknowledgement. Optional; when nil the /ack-effects route responds 404.

type AckEffectsRequest added in v0.2.0

type AckEffectsRequest struct {
	// SessionID identifies the session being acknowledged.
	SessionID string `json:"session_id"`

	// Reason is the operator-supplied note, recorded in the marker.
	Reason string `json:"reason,omitempty"`
}

AckEffectsRequest records the operator's acknowledgement of ambiguous prior effects on a session — the standalone twin of ResumeRequest.AckEffects, for the outbox's primary scenario: an interrupted turn leaves a dangling mutating tool call but NO pending interrupt, so there is nothing to resume. Semantics are those of pkg/transcript's Store.AckEffects (a durable watermark on the companion ops row; covers only intents persisted at or before it).

type Config

type Config struct {
	// Listen is the bind address, e.g. ":7777".
	Listen string

	// BearerToken is the shared secret required in the Authorization
	// header. Empty disables auth (intended only for local development;
	// production deploys must set it).
	BearerToken string

	// Handler is called for each valid inject. Required.
	Handler Handler

	// ResumeHandler is called for each valid resume POST. Optional.
	ResumeHandler ResumeHandler

	// Authenticator, when set, resolves the caller identity behind a
	// /resume request and puts it on the handler's context
	// (auth.CallerFromContext). It runs IN ADDITION to BearerToken, not
	// instead of it: the shared token still admits the request, and
	// this says who presented it.
	//
	// Only /resume consults it, deliberately. An approval is the one
	// place in v0.3 where identity is load-bearing — "who authorized
	// this change" is the audit question the release exists to answer,
	// and it is the half that cannot be backfilled (docs/v0.3-plan.md
	// W2.1). Extending authentication to the other routes is a
	// behaviour change for every existing emitter and belongs with the
	// multi-session substrate, not here.
	//
	// Nil means unauthenticated: verdicts record the shared-credential
	// identity, which is honest about what a single bearer token can
	// prove.
	Authenticator auth.Authenticator

	// AssertedCallerHeader is the header a proxy Caller uses to assert
	// the effective identity. Empty means auth.HeaderAssertedCaller.
	// Ignored unless Authenticator implements auth.AuthenticatorWithProxy.
	AssertedCallerHeader string

	// AbortHandler is called for each valid abort POST. Optional.
	AbortHandler AbortHandler

	// AckEffectsHandler is called for each valid ack-effects POST.
	// Optional.
	AckEffectsHandler AckEffectsHandler

	// PauseHandler is called for each valid pause POST. Optional.
	PauseHandler PauseHandler

	// ExtendTokenHandler is called for each valid extend-token POST.
	// Optional.
	ExtendTokenHandler ExtendTokenHandler

	// StopHandler is called for each valid stop POST. Optional.
	StopHandler StopHandler

	// Logger is the structured logger. Defaults to slog.Default().
	Logger *slog.Logger

	// Metrics, when non-nil, is served at GET /metrics (Prometheus
	// scrape). Unauthenticated by design — scrape configs don't carry
	// the inject bearer token, and the payload is aggregate counters
	// only. Nil leaves the route unregistered.
	Metrics http.Handler

	// BaseContext, when non-nil, is the context every request context
	// derives from. The daemon passes its turn-lifetime context so
	// that when the shutdown drain window elapses, in-flight handler
	// turns are cancelled and unwind instead of dying at process exit.
	BaseContext context.Context
}

Config configures the inject server.

type ExtendTokenHandler added in v0.2.0

type ExtendTokenHandler func(ctx context.Context, req ExtendTokenRequest) (ExtendTokenResult, error)

ExtendTokenHandler applies a token extension. Optional; when nil the /extend-token route responds 404.

type ExtendTokenRequest added in v0.2.0

type ExtendTokenRequest struct {
	// Token is the mast resume token (mrt_...).
	Token string `json:"token"`

	// TTL (Go duration) sets the new lifetime from now.
	TTL string `json:"ttl"`
}

ExtendTokenRequest lengthens a resume token's lifetime — the audited recovery for an expired (or expiring) token; the pause itself is untouched.

type ExtendTokenResult added in v0.2.0

type ExtendTokenResult struct {
	Token     string `json:"token"`
	ExpiresAt string `json:"expires_at"`
}

ExtendTokenResult reports the token's new expiry.

type Handler

type Handler func(ctx context.Context, payload envelope.InjectPayload) error

Handler receives a validated inject payload and drives the mast runtime. It returns an error if dispatch fails; the server maps that to a 5xx response (503 + Retry-After for ErrUnavailable).

type PauseHandler added in v0.2.0

type PauseHandler func(ctx context.Context, req PauseRequest) (PauseResult, error)

PauseHandler applies a gate pause. Optional; when nil the /pause route responds 404.

type PauseRequest added in v0.2.0

type PauseRequest struct {
	// SessionID identifies the session to pause.
	SessionID string `json:"session_id"`

	// Reason is the pause-reason enum value (transcript.ValidReasons).
	Reason string `json:"reason"`

	// Message is the human-readable context, surfaced by list/show.
	Message string `json:"message,omitempty"`

	// Metadata is free-form context recorded on the pause record.
	Metadata map[string]any `json:"metadata,omitempty"`

	// ResumeAt (RFC3339), when set, arms the timed-pause scheduler.
	ResumeAt string `json:"resume_at,omitempty"`

	// Interrupt additionally cancels the session's in-flight turn (hard
	// pause). The cancellation leaves no engine record — the pause
	// record is the durable truth — and may strand dangling mutating
	// intents for the effects outbox to guard.
	Interrupt bool `json:"interrupt,omitempty"`

	// TTL (Go duration, e.g. "48h") shortens the resume token's default
	// 7-day lifetime. Lengthening at mint is not offered — extend-token
	// is the audited operator path.
	TTL string `json:"ttl,omitempty"`
}

PauseRequest asks the daemon to gate-pause a session (plane B of the v0.2 pause/abort surface, docs/durable-execution-design.md "The v0.2 pause/abort mechanics"): the daemon's turn chokepoint refuses every subsequent turn on the session until the pause is resumed by token.

type PauseResult added in v0.2.0

type PauseResult struct {
	Token     string `json:"token"`
	SessionID string `json:"session_id"`
	ExpiresAt string `json:"expires_at"`
}

PauseResult carries the minted pause handle back to the caller.

type ResumeHandler

type ResumeHandler func(ctx context.Context, req ResumeRequest) error

ResumeHandler feeds a resume payload into the runtime. Optional; when nil the /resume route responds 404.

type ResumeRequest

type ResumeRequest struct {
	// SessionID identifies the paused session (e.g. "incident-<uid>").
	// Not required when Token is set.
	SessionID string `json:"session_id,omitempty"`

	// InterruptID matches the pending RequestInput's InterruptID. Not
	// required when Token is set.
	InterruptID string `json:"interrupt_id,omitempty"`

	// Token is a mast resume token (mrt_...) minted at pause time —
	// the v0.2 programmatic-pause keying. Mutually exclusive with
	// SessionID/InterruptID.
	Token string `json:"token,omitempty"`

	// Response is the reply payload; validated against the interrupt's
	// ResponseSchema by the workflow engine on resume.
	Response any `json:"response"`

	// AckEffects acknowledges ambiguous prior effects before the resume
	// turn runs: dangling mutating tool calls from an interrupted turn
	// stop tripping the recorded-effect outbox's fail-closed refusal
	// (docs/durable-execution-design.md, "Recorded-effect outbox"). The
	// operator asserts they have checked whether those calls took
	// effect externally.
	AckEffects bool `json:"ack_effects,omitempty"`
}

ResumeRequest is the operator's answer to a pending HITL interrupt (keyed by session + interrupt ID), or a token-keyed resume of a v0.2 pause (Token alone suffices; the daemon resolves it to the session and, for interrupt pauses, the pending call ID).

type Server

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

Server is the HTTP inject endpoint.

func New

func New(cfg Config) (*Server, error)

New constructs a Server. It does not start listening; call ListenAndServe.

func (*Server) ListenAndServe

func (s *Server) ListenAndServe() error

ListenAndServe blocks serving requests. Returns http.ErrServerClosed on graceful shutdown.

func (*Server) Shutdown

func (s *Server) Shutdown(ctx context.Context) error

Shutdown attempts a graceful stop.

type StopHandler added in v0.2.0

type StopHandler func(ctx context.Context, req StopRequest) (StopResult, error)

StopHandler initiates a planned stop. Optional; when nil the /stop route responds 404.

type StopRequest added in v0.2.0

type StopRequest struct {
	// Reason is appended to the interruption markers' "operator stop"
	// classification.
	Reason string `json:"reason,omitempty"`

	// PauseSessions gate-pauses every session that receives an
	// interruption marker during the drain, so boot-time auto-resume
	// hands them back to the operator instead of continuing them.
	PauseSessions bool `json:"pause_sessions,omitempty"`
}

StopRequest asks the daemon for a planned stop (issue #42): the same drain the SIGTERM path runs, classified in the interruption markers as an operator stop.

type StopResult added in v0.2.0

type StopResult struct {
	DrainBound string `json:"drain_bound"`
}

StopResult reports the drain bound the daemon will honor.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL