Connect API Server and Service Handlers

• internal/transport/transport.go
• internal/transport/incident.go
• internal/transport/session.go
• internal/transport/system.go
• internal/transport/lifecycle.go
• internal/transport/settle.go
• internal/transport/health.go
• Options
• Server
• ServerOptions
• IncidentManager
• AlertIngestor
• SessionLifecycle
• SessionSettler
• ToolRegistry
• Pinger
• WorkspaceSyncer
• SessionDefaults

internal/transport is the edge of Korvus. It owns the single port that speaks three protocols at once — gRPC, gRPC-Web and the Connect protocol — and it owns the /healthz probe and the Alertmanager webhook that live on that same port. Nothing else in the repository opens a socket: cmd/korvus builds the subsystems and then hands them to this package, which turns them into handlers korvus CLI and Process Wiring.

The package is deliberately thin. It does not own a database, an MCP connection or a session run; it owns the views of those things — the smallest interface each RPC needs — and the code that maps domain errors onto Connect status codes. The interfaces live here rather than in internal/store, internal/incident or internal/session because an interface belongs to the package that consumes it, a rule the package states repeatedly on the declarations themselves ToolRegistry, SessionLifecycle, SessionReader.

How it works

NewMux is the single entry point. It takes an Options value, and returns an *http.ServeMux with one Connect handler per service mounted, plus /healthz and, when configured, the webhook path NewMux.

Options carries the process-level values and the already-started subsystems the transport reports on: the build version, the process start time, the MCP registry and the authoritative configured-server map, the skill catalog, the database handle, the session store, the session manager, the incident manager, the alert ingestor, the workspace syncer, the turn broadcaster and the artifact download signer Options. cmd/korvus fills it in one literal at startup, passing the live registry, pool, dataStore, sessions, incident.NewManager(dataStore), broadcaster, workspaceSyncer and alerts main.go.

Before any handler is mounted, NewMux builds an otelconnect interceptor and passes it as the single Connect interceptor to all three generated handler constructors NewMux. The interceptor takes its tracer from the global provider at construction, which is why cmd/korvus installs telemetry before anything instrumented runs — the transport's interceptor included main.go.

The nil-subsystem pattern

Each subsystem is optional, but "optional" does not mean "silently empty" everywhere. What varies is where the absence shows up, and the code is explicit about which shape it takes:

The rule the code states at the top of Options is that a nil field "reports its honest empty state rather than failing, so a partially wired process (and the transport package's own tests) still serves status instead of panicking" Options. In practice that means the status RPCs degrade gracefully while the action RPCs refuse loudly — a nil Lifecycle is an Unimplemented, not an empty session.

Incident service

incidentService wraps an IncidentManager — the five-method view of the manual incident lifecycle: Open, Get, List, Update and Close IncidentManager. CreateIncident rejects any origin other than INCIDENT_ORIGIN_MANUAL and INCIDENT_ORIGIN_UNSPECIFIED, because the alert path owns fingerprinted incidents and its own debounce window; it also refuses a non-empty fingerprint with incident.ErrFingerprintNotManual CreateIncident.

The alert-path RPCs (ListAlerts, WatchIncidents) are not implemented by incidentService — it embeds unimplementedIncidentService, so adding an RPC to the proto becomes a compile error rather than a silent gap incidentService. That embedded type carries two distinct sentinels, because two different conditions reach it unimplementedIncidentService:

• errNoIncidentManager — "this process was started without an incident manager; no incident RPC is served". It answers the five lifecycle RPCs, which are only reachable through the nil-manager fallback since incidentService overrides all five errNoIncidentManager.
• errAlertPathNotImplemented — "ListAlerts is not served yet: alerts are recorded, but no RPC reads them back; incident event streaming lands with AJX-329". It answers ListAlerts and WatchIncidents, and it is the one message that names the ticket that will implement the path errAlertPathNotImplemented.

Both are codes.Unimplemented; only the second names a ticket. The distinction matters because the message is a wire contract that reaches clients and operator logs incident.go.

Session service

The session handlers are split across three files by concern. reads.go holds the store-backed reads: GetSession, ListSessions, ListTurns, GetReport, ListReportRevisions, ListArtifacts and GetArtifactDownload reads.go. lifecycle.go holds the write side — CreateSession, CancelSession, SendMessage — and the sessionError mapper that turns store and session sentinels into Connect codes: NotFound for a missing row, InvalidArgument for a malformed request, FailedPrecondition for a closed incident or an unfinished session, Unavailable for a shutting-down process, Internal for anything unrecognised sessionError. The incident service has its own equivalent mapper with the same "sentinel, never message text" contract incidentError.

CreateSession pins the new session to the workspace revision currently being served, or to an empty commit when no workspace is configured workspaceCommit. CancelSession treats session.ErrAlreadyEnded as a no-op and returns the settled row, because the proto says cancelling a terminal session "has no effect" and a retried cancel is the ordinary outcome of a client that timed out CancelSession.

watch.go holds WatchSession, the streaming RPC. Its ordering is the load-bearing part: the broadcaster subscription is taken before the database drain, because a turn is persisted before it is published, so any turn that lands during the drain is read from the database or delivered on the channel — possibly both, never neither WatchSession. A heartbeat every 15 seconds keeps idle streams alive while a tool call blocks heartbeatInterval.

The optional settler

SessionSettler is the one method WatchSession needs that a plain lifecycle cannot answer: AwaitSettled(ctx, id) SessionSettler. The agent closes the broadcast channels before the manager writes the report and the terminal status, so a stream that ended on the channel close alone would carry neither. finishLive therefore performs a type assertion on the lifecycle — if settler, ok := s.lifecycle.(SessionSettler); ok — and, when it succeeds, waits for the settle signal while continuing to send heartbeats on the original ticker finishLive, awaitSettledBeating. The assertion is what keeps the production manager from silently becoming a lifecycle that cannot settle; a compile-time assertion pins that the session manager does satisfy it settle.go.

System service

systemService reports what startup discovered systemService. GetSystemStatus returns the version, the start time, the MCP server list, the loaded skill count and the database connectivity flag, and the database ping is bounded at two seconds so a wedged database reports "not connected" instead of hanging the RPC GetSystemStatus, databaseConnected.

mcpServers reports every configured server, not merely the connected ones: a server that failed to dial is absent from the registry, and reporting only the registry's view would silently drop exactly the server an operator is trying to diagnose mcpServers. SyncWorkspace reports a failed sync as Unavailable rather than Internal, because the previous checkout is still being served and a retry may succeed SyncWorkspace. ListTemplates and TriggerAlert answer Unimplemented — the first needs the workspace, the second because alerts are accepted on the webhook instead ListTemplates, TriggerAlert.

The webhook and the probe

The Alertmanager webhook is a plain net/http handler mounted at Options.WebhookPath, which is required whenever Alerts is set — NewMux returns an error otherwise NewMux. It accepts only POST, bounds the body at 1 MiB, and decodes the version-4 payload while ignoring unknown fields, because Alertmanager adds fields between releases webhookHandler. The status code is the contract with Alertmanager: a payload that can never succeed is a 400, a store failure or a shutdown is a 5xx so the delivery is retried. The ingest call is detached from the request context, so a sender that hangs up mid-write cannot leave alerts recorded but never attached to their incident webhook.go.

/healthz is the liveness probe: it answers 200 with ok handleHealthz.

Serving and shutdown

NewServer binds the address — the port may be 0 — and returns a Server ready to Run NewServer. Binding happens there so a bad address fails before any goroutine starts and so a test can read the real port from Addr Server.Addr. The listener enables HTTP/1.1 and unencrypted HTTP/2 through the stdlib http.Protocols, so the same port accepts cleartext prior-knowledge HTTP/2 (gRPC) alongside HTTP/1.1 (Connect, gRPC-Web, /healthz) with no proxy and no second server NewServer.

Run serves until the context is cancelled, then drains in-flight requests and streams within the shutdown timeout before returning Server.Run. Zero timeouts on ServerOptions fall back to 10 seconds for the read header and 30 seconds for shutdown rather than to "no timeout at all" ServerOptions.

Relationships

graph TD
  main["cmd/korvus startup"] --> NewMux
  NewMux --> systemService
  NewMux --> sessionService
  NewMux --> incidentService
  NewMux --> webhookHandler
  NewMux --> handleHealthz
  NewMux --> Options
  NewMux --> NewServer
  NewServer --> Server
  Server --> Run
  incidentService --> IncidentManager
  incidentService --> AlertIngestor
  sessionService --> SessionReader
  sessionService --> SessionLifecycle
  sessionService --> SessionDefaults
  sessionService --> WorkspaceSyncer
  sessionService --> TurnSubscriber
  sessionService --> DownloadSigner
  systemService --> ToolRegistry
  systemService --> Pinger
  systemService --> WorkspaceSyncer
  webhookHandler --> AlertIngestor

The transport is the boundary between the generated Connect stubs and the rest of the codebase. Its proto-facing types come from Generated Korvus v1 Protobuf and Connect API; the subsystems it reports on are covered by Incident Manager: Open, List, Update, Close, Alert Ingestion, Fingerprinting and Debounce, Session Manager and Lifecycle, Storage Domain Model and Store Interfaces, MCP Client, Registry and Server Connection and Workspace Sync, Checkout and Materialization. The streaming and read APIs it serves are detailed in Watch Streams, Read APIs, Webhooks and Proto Conversion, and the tracing it installs is described in OpenTelemetry Setup and GenAI Semantic Attributes. The full picture of how the process comes up around this package is in Korvus Overview.

Reference

Generated by astrolabe from commit 0c2dca73 on branch main.