httpsrv

Documentation

Overview

Package httpsrv wires Poe HTTP requests into the router.

Index

• Constants
• func DebugHandler(r *router.Router) http.Handler
• type CommandHandler
• type Config
• type Handler
  • func New(cfg Config) *Handler
  • func (h *Handler) InFlight() []StreamInfo
  • func (h *Handler) SealInFlight(text string, timeout time.Duration) int
  • func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request)
• type StreamInfo

Constants

const AbandonMessage = "relay restarted — retrying"

AbandonMessage is the user-visible text emitted on an abandoned stream. Short and legible: Poe renders it as an error bubble and, because the event carries allow_retry, may redrive the same message_id against the new worker.

const DefaultSealTimeout = 3 * time.Second

DefaultSealTimeout bounds the whole graceful-abandon step on a force-cut drain. The process is exiting either way, and each seal is itself bounded by the SSE write deadline, so this is only the outer cap that stops one wedged writer delaying the force-close.

Functions

func DebugHandler

func DebugHandler(r *router.Router) http.Handler

DebugHandler returns an http.Handler that dumps router state as JSON.

Types

type CommandHandler

type CommandHandler interface {
	HasPending(convID string) bool
	Handle(ctx context.Context, convID, text string) (*command.Outcome, error)
	// Passthrough reports whether text is an allowlisted agent command;
	// if ok, rewritten is the prompt text to forward to the agent.
	Passthrough(text string) (rewritten string, ok bool)
}

CommandHandler is the surface httpsrv depends on; *command.Broker implements it. Extracted so tests can inject handlers that return odd combinations the real one can't produce.

type Config

type Config struct {
	Router *router.Router
	// Settings is the static response for `settings` requests. Parameter
	// controls may be overridden per-request by ParameterControlsProvider.
	Settings poeproto.SettingsResponse
	// HeartbeatInterval is the SSE heartbeat tick. The heartbeat
	// emits an animated `> _Thinking._` spinner via replace_response
	// until the first user-visible write closes the gate. <=0 disables
	// the heartbeat. Doubles as the spinner animation rate, so values
	// in the 1–2s range read well to humans.
	HeartbeatInterval time.Duration
	// ParameterControlsProvider, if set, is called on each `settings`
	// request to populate SettingsResponse.ParameterControls. If nil,
	// Settings.ParameterControls is used as-is.
	ParameterControlsProvider func() *poeproto.ParameterControls
	// Commands, if set, intercepts relay chat-commands (login family,
	// !help, !status, !models, !model, !new — any of the sigils /, !, .)
	// and pasted redirect URLs from in-flight logins before they reach
	// the router. Optional; nil disables the command surface.
	Commands CommandHandler
	// TurnTimeout is an OPTIONAL absolute wall-clock ceiling on a prompt
	// turn run on a context DECOUPLED from the request ctx. Poe tears
	// down the bot-facing HTTP connection pre-output on a transport drop;
	// decoupling lets the in-flight turn finish so its answer can be
	// buffered and served on the redrive, instead of aborting and losing
	// the work.
	//
	// <=0 (the default) means NO absolute ceiling: a turn is bounded
	// SOLELY by the progress-resetting IdleWriteTimeout backstop. While
	// the agent keeps producing user-visible output the turn runs for as
	// long as it needs — a long, actively-working turn is never cut. The
	// absolute cap is opt-in for operators who deliberately want a hard
	// upper bound regardless of progress; it is NOT a wedge guard (that
	// is IdleWriteTimeout's job).
	TurnTimeout time.Duration
	// AnswerTTL bounds how long a buffered (absorbed) turn answer is held
	// for a redrive before it is discarded. <=0 falls back to
	// defaultAnswerTTL (2m).
	AnswerTTL time.Duration
	// IdleWriteTimeout is the per-stream backstop for a WEDGED turn: the
	// agent has hung, no SSE content byte has been written, and the
	// client never disconnected. If no user-visible chunk lands within
	// this window the single wedged turn is cancelled so it cannot block
	// a graceful drain forever; every other in-flight stream keeps
	// draining. Heartbeat keepalive frames do NOT reset it — only real
	// agent output does — so a genuinely wedged turn is detected even
	// though its spinner keeps ticking. A tool_call session/update DOES
	// reset it: a legitimately long-running tool is genuine progress, so
	// it must not be cut. <=0 falls back to defaultIdleWriteTimeout (2m).
	IdleWriteTimeout time.Duration
	// StallThreshold is how long the SSE stream may go without a
	// user-visible content write before the heartbeat re-arms the
	// mid-turn keepalive spinner. Poe drops the bot-facing connection on
	// content-starvation, so during a long tool-heavy turn (no tokens
	// for minutes) the spinner must resume to keep the stream alive.
	// It re-arms via `replace_response` (a content event that renders in
	// place), preserving the text so far and animating a transient
	// status line below it; the moment real output resumes the spinner
	// is stripped. Before the first output there is no stall gate — the
	// cold-start spinner fires immediately (see heartbeat). <=0 falls
	// back to defaultStallThreshold (8s), conservatively under Poe's
	// drop tolerance. Reuses HeartbeatInterval for the animation cadence.
	StallThreshold time.Duration
	// SSEWriteTimeout bounds every SSE wire write via a per-write
	// connection deadline. A client that has stopped reading (Poe pressed
	// Stop, or a dead transport) applies TCP backpressure; without a
	// deadline a blocked write — including the terminal `done` — never
	// returns and the turn never finalizes (the "Stop button stays"
	// hang). With it, a wedged write fails after this window, the
	// heartbeat swallows the error, and the terminal `done` still seals
	// the stream so the HTTP response closes. <=0 falls back to
	// defaultSSEWriteTimeout (30s): generous enough never to cut a
	// slow-but-live reader mid-frame, tight enough to bound finalization.
	SSEWriteTimeout time.Duration
	// CoalesceInterval, when >0, buffers outbound `text` events and
	// flushes them periodically instead of emitting one SSE frame per
	// agent chunk. Frames-per-turn is a direct proxy for the phone's
	// radio wakeups, and coalescing upstream is monotone — Poe cannot
	// relay frames it was never given. 0 (the default) preserves the
	// 1:1 chunk→frame behaviour exactly. See coalesce.go.
	CoalesceInterval time.Duration
	// CoalesceGrid aligns coalesced flushes to absolute wall-clock
	// instants so several bots on several hosts wake the modem in the
	// SAME window instead of interleaving. Only meaningful when
	// CoalesceInterval > 0. Callers resolve the default (see
	// config.Defaults.Stream); this struct takes the explicit value.
	CoalesceGrid bool
	// SpinnerStatic freezes the keepalive spinner's dots so identical
	// keepalive frames can be deduped (see orderedWriter.hbFrame). Zero
	// value = animated = today's behaviour.
	SpinnerStatic bool
}

Config configures a Handler.

type Handler

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

Handler serves the /poe endpoint.

func New

func New(cfg Config) *Handler

New creates a Handler. HeartbeatInterval <=0 disables heartbeat; otherwise no defaulting is applied — pass an explicit value. AnswerTTL and IdleWriteTimeout default when <=0 (see their docs). TurnTimeout is NOT defaulted: <=0 means no absolute turn ceiling, so the progress-resetting IdleWriteTimeout backstop is the sole guard.

func (*Handler) InFlight

func (h *Handler) InFlight() []StreamInfo

InFlight returns the query streams currently being served, oldest first. Used by the worker's bounded-drain path to report exactly which streams a force-cut drain abandoned.

func (*Handler) SealInFlight

func (h *Handler) SealInFlight(text string, timeout time.Duration) int

SealInFlight terminates every in-flight stream at the Poe protocol level: an `error` event (which carries allow_retry) followed by the terminal `done`, flushed. It returns the number of streams it sealed.

It exists for the force-cut drain path: without it, srv.Close() cuts the SSE connection mid-stream, Poe sees a truncated response and renders a red transport error whose retry is at its discretion. With it Poe sees a well-formed terminal error with explicit retry permission and may redrive the same message_id against the new worker, which resumes the same fir session by conversation cwd — so the CHAT stays continuous even though the turn itself is re-run.

Sealing is idempotent with the handler's own finalization backstop: both go through orderedWriter, whose `closed` flag makes the second `done` a no-op, so no stream can ever emit `done` twice. Each seal runs on its own goroutine (streams live on other goroutines and a wire write is bounded only by the SSE write deadline) and the whole step is capped by timeout — a stalled reader must not wedge the drain. timeout <= 0 means DefaultSealTimeout.

func (*Handler) ServeHTTP

func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request)

ServeHTTP implements http.Handler.

type StreamInfo

type StreamInfo struct {
	// ConvID is the Poe conversation_id the stream belongs to.
	ConvID string
	// UserID is the Poe user_id that opened it.
	UserID string
	// MessageID is the Poe message_id of the request being served.
	MessageID string
	// Age is how long the stream had been running at snapshot time.
	Age time.Duration
}

StreamInfo describes one in-flight query stream at snapshot time.