Watch Streams, Read APIs, Webhooks and Proto Conversion

• internal/transport/watch.go
• internal/transport/reads.go
• internal/transport/webhook.go
• internal/transport/convert.go
• SessionReader
• TurnSubscriber
• DownloadSigner
• incidentToProto
• sessionToProto
• reportToProto
• artifactToProto
• incidentStatusFromProto
• sessionKindFromProto

The internal/transport package is Korvus's edge: it turns in-process domain objects into Connect RPC responses, streams a running agent's transcript to a browser, and accepts Alertmanager's webhook deliveries. This page covers the four files that make up the read-heavy half of that edge — watch.go (the live session stream), reads.go (the request/response reads), webhook.go (the Alertmanager ingress path) and convert.go (the row-to-proto mapping). The write side of the session lifecycle lives in lifecycle.go, which is described in Connect API Server and Service Handlers.

The design rule that shows up everywhere in this package is that the transport owns no domain logic. It declares its own narrow interfaces onto the layers below — SessionReader for the store, TurnSubscriber for the agent's broadcaster, DownloadSigner for blob storage — so that a partially wired process can still serve some RPCs and answer honestly for the rest, and so the package's own tests need neither a database nor a running agent. That contract, and how the mux is assembled from it, is covered in Connect API Server and Service Handlers.

How it works

The watch stream

WatchSession is the interesting RPC: it must deliver a session's transcript with no gaps and no lost turns, to a client that may attach at any moment during a run that is still producing output. The handler lives in watch.go and it is deliberately written so that the ordering of two operations is load-bearing.

The handler validates the request first: session_id is required, after_seq must not be negative, and the session must exist or the RPC is NotFound. It then decides whether the session is live — that is, whether its status is non-terminal and a broadcaster is wired in at all (watch.go). A terminal session never subscribes: the database is already the complete transcript, so the handler sends a snapshot and closes.

For a live session, the handler subscribes to the broadcaster before it drains the database. This is the whole trick. A turn is persisted before it is published, so a turn that lands during the drain is either read from the database or delivered on the channel — possibly both, never neither. Draining first and subscribing second would open a window in which a turn is written (too late for the drain) and published (before there is a subscriber), and that turn would vanish silently. Duplicates are cheap for a client to remove; gaps are not even detectable, so the design trades the first for the second (watch.go).

The snapshot itself is built by the snapshot helper, which lists the turns after after_seq and the session's context refs, and returns the highest sequence number it carries as a de-duplication watermark. That watermark starts at after_seq rather than zero, so a client resuming from seq 12 does not get turn 12 re-sent if it arrives late on the channel (watch.go).

The live loop then selects over three things: the request context, the broadcaster channel, and a 15-second heartbeat ticker (watch.go). The heartbeat exists because a tool call can block for a minute and idle proxies drop connections well before that, so the stream must emit something even when the session produces nothing (watch.go). Turn events are filtered against the watermark before being sent, which is what makes subscribe-then-drain safe (watch.go). Tool-start events carry no sequence number and are always forwarded.

When the agent closes the session's channels, the run is over — but the report and the terminal status may not exist yet, because the agent closes before the manager persists. So finishLive waits for the manager's settle signal first, then re-reads the row and, if it is terminal, emits the status change followed by the report (watch.go). The wait itself runs on its own goroutine so that every Send stays on the handler goroutine — a Connect ServerStream is not safe for concurrent sends — and it keeps the heartbeat going on the same ticker, so the 15-second count is not restarted at the agent's close (watch.go). A row that is still not terminal after the wait gets no status event at all: the stream reports what the row says, never a transition that did not happen.

The settle capability is optional. SessionSettler is a one-method interface that *session.Manager satisfies, but a test fake or a process wired without a manager may not, in which case the stream simply ends on the broadcaster's close as it always did (settle.go). The AwaitSettled contract itself is described in Session Manager and Lifecycle.

The read RPCs

reads.go holds the plain request/response half of SessionService: GetSession, ListSessions, ListTurns, GetReport, ListReportRevisions, ListArtifacts and GetArtifactDownload. They all follow the same shape — validate the request, call the store through SessionReader, convert the rows, and map errors through sessionError, which translates store and manager sentinels into Connect codes (lifecycle.go).

Two of these reads carry an extra defensive step. ListTurns and ListReportRevisions read the session row first, so an unknown session id is NotFound rather than an empty list (reads.go, reads.go). The list queries alone would answer an unknown id with an empty result, which a replaying client would render as a session that simply has not spoken yet. ListSessions deliberately does the opposite: it answers an empty list for an incident with no sessions and for one that does not exist, because distinguishing the two would need an incident read the session service does not otherwise make, and an empty list is a true answer to both (reads.go).

ListArtifacts accepts exactly one of session_id or incident_id, and rejects a revision filter combined with incident_id — an incident listing holds the incident's own artifacts, which no revision produced, so the filter would have no meaning (reads.go).

GetArtifactDownload is the one read that leaves the database entirely. It resolves the artifact first, so an unknown id is NotFound on every deployment, and only then asks whether this deployment can serve bytes at all. A filesystem blob backend cannot issue a URL a client could fetch, so with no signer wired the answer is FailedPrecondition — the request is valid, the deployment is what refuses it (reads.go). The URL's lifetime is bounded at 15 minutes by downloadURLTTL, long enough for a browser to start a download after a page render and short enough that a URL pasted into a ticket stops working before it matters, since there is no authentication anywhere else to fall back on (reads.go). The blob-key rules behind the signed URL live in Blob Storage Backend and Key Rules.

The Alertmanager webhook

The webhook is a plain net/http handler mounted on the same mux as the Connect services, not a Connect RPC (transport.go). It accepts a JSON body, decodes the subset of Alertmanager's version-4 format that the alert path reads, and hands the result to the AlertIngestor as an incident.AlertGroup (webhook.go, webhook.go). Unknown fields — groupKey, externalURL, truncatedAlerts, the per-alert fingerprint — are ignored rather than rejected, because Alertmanager adds fields between releases.

The status code is the contract with Alertmanager, which retries a 5xx and drops a 4xx. A payload that can never succeed is a 400; a store failure or a process shutting down is a 5xx so the delivery is retried against the next process. The body is bounded at 1 MiB by maxWebhookBody, since Alertmanager's own max_alerts defaults to unlimited and the bound has to be ours (webhook.go).

Ingestion is deliberately detached from the request: once it starts it must finish, or a sender that hangs up mid-write leaves alerts recorded but never attached to the incident they open. That is why the handler passes context.WithoutCancel(r.Context()) into Ingest (webhook.go). Alertmanager's zero endsAt — sent for a firing alert instead of omitting the field — is mapped to nil, which is what the alerts table means by "not ended" (webhook.go). What the ingestor does with the group — fingerprinting, debounce, opening incidents — is covered in Alert Ingestion, Fingerprinting and Debounce.

Proto conversion

convert.go is the boundary between the domain's string enums and the wire's proto enums. The store carries statuses like "open" and "running" as plain strings; the proto carries INCIDENT_STATUS_OPEN and SESSION_STATUS_RUNNING. The conversions are exhaustive switches whose default arm is UNSPECIFIED rather than a panic, because the database is the older half of a rolling deployment and may hold a value this binary predates (convert.go).

The row-to-proto direction is lossless in one place and lossy in another, and the file says so. Fingerprint, StartedAt and ClosedAt are nullable in the row but proto3 has no nullable scalar, so NULL collapses to the empty string on the wire — the reverse mapping must turn "" back into nil, never into a pointer to "" (convert.go).

The proto-to-row direction is where the interesting policy lives. incidentStatusFromProto maps UNSPECIFIED — and any enum value this binary does not know — to the empty string, which every caller reads as "the client did not say": an unfiltered list, an unchanged status, the default terminal status. Treating a future value as "no opinion" is safer than guessing which known status it resembles (convert.go). sessionKindFromProto does the same, and the empty string is what session.Manager reads as "decide from the incident's existing sessions" (convert.go). contextRefRoleFromProto follows the pattern, with "" meaning "derive the role from the artifact's kind" (convert.go).

Two conversions swallow errors on purpose. A turn's stored tool-call JSON that will not parse drops the tool-call list rather than failing the turn — the content is what the investigation is about, and the same detail is on the span (convert.go). The same applies to a report's structured Data blob and to a tool call's arguments, which are advisory display data: a value that will not round-trip through structpb must not cost the client the notice that the call started (convert.go).

Relationships

flowchart TD
  client[Connect client] --> mux[NewMux]
  mux --> svc[sessionService]
  mux --> wh[webhookHandler]
  svc --> sr[SessionReader]
  svc --> ts[TurnSubscriber]
  svc --> dl[DownloadSigner]
  svc --> life[SessionLifecycle]
  life -.optional.-> settle[SessionSettler]
  svc --> conv[convert.go converters]
  sr --> store[store package]
  ts --> bc[agent.TurnBroadcaster]
  dl --> blob[blob.Presigner]
  wh --> ing[incident.Ingestor]
  conv --> proto[korvusv1 protos]

Error handling

Reference

For the store rows these converters read, see Storage Domain Model and Store Interfaces and PostgreSQL Persistence, Migrations and Vector Search. For the broadcaster on the other side of TurnSubscriber, see Agent Session State, Resume and Event Broadcasting. For the schema of the report that ends a stream, see Incident Report Schema, and for the wire definitions themselves see Generated Korvus v1 Protobuf and Connect API.

Generated by astrolabe from commit 0c2dca73 on branch main.