comm

package
v1.7.0 Latest Latest
Warning

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

 Go to latest Published: Aug 10, 2026 License: AGPL-3.0 Imports: 22 Imported by: 0

Documentation

Overview

Package comm owns the inter-session communication subsystem: authenticated message passing between AI sessions on the same or different machines (design decision D9; full contract in docs/COMM.md).

It is OPT-IN and OFF BY DEFAULT. Nothing in this package runs, and no table is created, unless the operator enables COMM — a default Ken install stays exactly the curated knowledge base it advertises.

Boundaries

This package owns its OWN SQLite file (data/comm/comm.db) and never touches ken.db. That separation is the point: message traffic is high-churn and EXPENDABLE, knowledge is low-churn and DURABLE, so keeping the files apart keeps ephemeral WAL churn out of the replicated database and out of the KB's single writer. Losing this file costs an in-flight conversation, never knowledge, and it is deliberately outside both backup tiers.

Ownership columns (actor, space, token) identify rows in ken.db and are plain values here — SQLite foreign keys cannot span database files, so the CALLER (which holds both handles) is responsible for supplying identities it has already authenticated. See migrations/0001_init.sql.

What this package does not do

It does not decide whether a receiving session should act on a message. COMM authenticates WHO may talk to whom — structurally, via a human-minted pairing code — but the handling of message content is the receiving harness's responsibility, and instruction text is not a control. docs/COMM.md §8 states that boundary rather than implying a guarantee this code cannot make.

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrNotFound covers an unknown endpoint, channel, or pairing code.
	ErrNotFound = errors.New("not found")
	// ErrDenied means authenticated but not entitled: a wrong endpoint secret, a
	// revoked endpoint, or an endpoint acting on a channel it does not belong to.
	ErrDenied = errors.New("denied")
	// ErrBackpressure means the channel's un-acked depth is at its cap. Callers
	// must surface this as "stop and wait", never as a retryable transport error:
	// full-duplex has no turn-taking, so two auto-processing sessions can
	// otherwise enter a reply loop that grows the database without bound.
	ErrBackpressure = errors.New("channel backpressure: too many unacknowledged messages")
	// ErrTooLarge means the body exceeds the configured cap.
	ErrTooLarge = errors.New("message body too large")
	// ErrChannelClosed means the channel is not open (still pending a second join,
	// or revoked).
	ErrChannelClosed = errors.New("channel is not open")
)

Sentinel errors. Callers map these to their own surface (an MCP tool error, an HTTP status); they are matched with errors.Is, never by string.

View Source 
var (
	// ErrFilesDisabled means the operator has not enabled file exchange (or has
	// live-disabled it — the check doubles as a kill switch).
	ErrFilesDisabled = errors.New("file exchange is disabled")
	// ErrQuota means the global storage budget or the free-space floor would be
	// violated. Fails CLOSED on purpose: a refused upload is a retry; a fail-open
	// quota is a disk-full outage that takes the knowledge base's writes with it.
	ErrQuota = errors.New("file storage quota exceeded")
	// ErrBadName means the offered name failed validation.
	ErrBadName = errors.New("invalid attachment name")
)

Additional sentinel errors for the file surface.

View Source 
var ErrSequenceCollision = errors.New("sequence collision")

ErrSequenceCollision reports that a message could not be numbered because this endpoint's per-channel counter has fallen behind its own history on that channel.

It exists because the failure it names is otherwise a bare "internal error", and an operator who has just adopted a station has no path from that string to a sequence counter — they will suspect the network, the token, the peer, or the restart they happened to do. A production operator hit exactly this and said so: they only knew where to look because a report arrived first.

The condition is repaired by comm_unbind, which is the remediation path for anyone who bound on a release before the counter was carried across.

View Source 
var ErrShortWrite = fmt.Errorf("upload ended before the declared size")

ErrShortWrite reports a stream that ended before the declared size.

Functions

func ValidateAttachmentName 

func ValidateAttachmentName(name string) error

ValidateAttachmentName enforces the C9 name contract: a bare basename, nothing that could steer a receiving session toward an arbitrary local path. The server cannot see the client filesystems, but it CAN reject the string — which turns a convention into a validated contract.

Types

type AdminChannel 

type AdminChannel struct {
	ChannelID string
	// Label is the human name the operator gave when minting the pairing code
	// (e.g. "Ken dev <-> prod"); empty when the code had no label. This is the
	// identifier a human recognizes — the opaque ChannelID is for machines.
	Label        string
	State        string
	SpaceID      int64
	OwnerActorID int64
	EndpointA    string // endpoint_id of the first party
	LabelA       string
	EndpointB    string // empty until the second party joins
	LabelB       string
	Unacked      int // messages queued or delivered but not yet acked
	CreatedAt    string
	OpenedAt     string
}

AdminChannel is one row of the operator's channel view: the channel plus the human-readable labels of both ends and its current queue depth.

This exists separately from ListChannels (which answers "what can THIS endpoint talk on") because the operator's question is different: "what is talking to what, and is anything piling up". A security model whose enforcement point is the human needs the human to be able to see, and stop, what is happening.

type Attachment 

type Attachment struct {
	AttachmentID string
	ChannelID    string
	Name         string
	SizeBytes    int64
	SHA256       string
	Transfer     string
	NonceSHA256  string
	State        string
	ExpiresAt    string
	MessageID    string // public id of the enqueued message ("" until one exists)
	// contains filtered or unexported fields
}

Attachment is the durable record of a file offer/transfer.

type Channel 

type Channel struct {
	ID           int64 // internal rowid
	ChannelID    string
	SpaceID      int64
	OwnerActorID int64 // the human who authorized the pairing
	EndpointA    int64
	EndpointB    int64 // 0 until the second endpoint joins
	State        string
	CreatedAt    string
	OpenedAt     string
}

Channel joins exactly two distinct endpoints, full-duplex.

There is deliberately no turn or "waiting" state here: channel-level turn-taking is a distributed state machine that wedges when a session dies mid-turn. Request/response is a property of a message instead (see message.go).

func (*Channel) Open 

func (c *Channel) Open() bool

Open reports whether the channel may carry traffic.

type Endpoint 

type Endpoint struct {
	ID         int64 // internal rowid
	EndpointID string
	Owner      Owner
	Label      string
	HostHint   string
	CreatedAt  string
	LastSeenAt string
	// RotatedAt and RotateCount are DISPLAY state for the console ("did I already
	// rotate this one?"). comm.db is expendable and not backed up, so the
	// authoritative audit record is the server log, not these columns.
	RotatedAt   string
	RotateCount int

	// StationID is the durable station this endpoint reads for, or "" when unbound
	// (docs/STATIONS.md S4). An opaque id into ken.db with no foreign key: S7's rule
	// is that cross-database pointers run expendable -> durable, and under restore
	// skew an id that no longer resolves is treated as UNBOUND rather than as an
	// error. Bound endpoints share their station's inbox with claim-once delivery.
	StationID string
	// BoundByStationKeyID is the station key that authorised the binding. Revoking
	// that key severs every endpoint it bound (S6) — without this column, revocation
	// would stop future bindings and leave the leaked capability running.
	BoundByStationKeyID string
	BoundAt             string
}

Endpoint is one AI session's communication point.

Sender identity derived from an Endpoint is honestly described as "token-authenticated, endpoint-scoped": trustworthy across machines and users, advisory between two sessions that share one token (the operating convention is one token per MACHINE). The endpoint secret is what stops two sessions on one box from polling and acking each other's messages by accident.

type FileInfo 

type FileInfo struct {
	AttachmentID string
	Name         string
	SizeBytes    int64
	SHA256       string
	Transfer     string
	NonceSHA256  string
}

FileInfo is the attachment descriptor carried on a delivered message.

type FileOffer 

type FileOffer struct {
	Name        string
	SizeBytes   int64
	SHA256      string
	Transfer    string // "path" | "upload"
	NonceSHA256 string // required for "path": the C9 rendezvous hash
	Note        string // optional message body accompanying the offer
	// IdempotencyKey mirrors message idempotency: a re-offer with the same key
	// returns the ORIGINAL attachment (with a fresh upload grant while it is still
	// awaiting bytes) instead of minting a second one.
	IdempotencyKey string
	TTLSeconds     int
}

FileOffer is one sender's declaration of a file to move.

type GrantInfo 

type GrantInfo struct {
	AttachmentRow int64
	AttachmentID  string
	EndpointToken string // token_id owning the grant's endpoint — the HTTP caller must match
	RecipientRow  int64  // the attachment's recipient (for the upload-completion notify)
	Name          string
	SizeBytes     int64
	SHA256        string
	State         string
}

GrantInfo is a resolved, consumed transfer grant.

type Limits 

type Limits struct {
	// MaxBodyBytes caps one message body. Kept modest on purpose: tool arguments
	// are generated token-by-token by a model, so even a 64 KiB message is a
	// five-figure output-token count.
	MaxBodyBytes int
	// MaxUnackedPerChannel caps un-acked depth per channel (backpressure).
	MaxUnackedPerChannel int
	// MessageTTLSeconds is how long a DELIVERED but un-acked message survives
	// before expiring. The clock starts at FIRST DELIVERY, not at send.
	//
	// It used to start at send, and that was wrong in a way no measurement inside
	// the system could reveal: a human works ~8 h a day, so a session goes 16 h
	// between pulls on a weeknight, 64 h over a weekend and weeks over annual
	// leave. Against the shipped 24 h default that made every message sent during
	// a Friday shift dead before Monday — 2.67x the TTL — and it is what killed a
	// real 4 661-byte message sent on Sunday 2026-08-02. The clock was running
	// during exactly the window in which nobody could possibly poll.
	//
	// Anchoring at delivery asks the right question: not "how long has this
	// existed" but "how long has the recipient had it and done nothing".
	MessageTTLSeconds int
	// UndeliveredTTLSeconds bounds a message NOBODY HAS EVER SEEN. It exists only
	// as a backstop against a permanently dead endpoint, so it is generous by
	// design: undelivered mail is bounded primarily by MaxUnackedPerChannel, which
	// is a volume bound and therefore immune to how long a human is away.
	//
	// Must be comfortably longer than the longest absence an operator expects. A
	// value shorter than MessageTTLSeconds is nonsense — it would kill mail before
	// the delivered clock could even start — and is treated as "use the default".
	UndeliveredTTLSeconds int
	// BodyRetentionSeconds is how long a body survives AFTER the message settles.
	//
	// Zero restores the historical behaviour: blank on ack. That behaviour
	// destroyed 97% of one live deployment's message bodies (153 of 159) through
	// the ordinary, instructed path — poll, act, ack — because ack blanked the
	// body unless the message happened to require a response. The un-acked inbox
	// was not a safety net; it was the archive, and acking was the instruction to
	// destroy the only copy.
	BodyRetentionSeconds int
	// MetadataTTLSeconds is how long a settled (acked/expired) message's row
	// survives after creation.
	//
	// It is no longer only the audit shell. Bodies survive acknowledgement now, and
	// a body that was NEVER DELIVERED cannot be reclaimed by BodyRetentionSeconds —
	// retention measures from the settle time and an unread message has none — so
	// this is the sole bound on that population. An operator raising it for a longer
	// audit trail is also raising retained bytes.
	MetadataTTLSeconds int
	// ReplyDeadlineSeconds is the default deadline applied to a message that
	// requires a response.
	ReplyDeadlineSeconds int
	// PairingCodeTTLSeconds is how long a human-minted pairing code stays valid.
	PairingCodeTTLSeconds int
	// ClaimLeaseSeconds is how long a station-bound reader holds a claimed message
	// before it returns to the unclaimed tail (docs/STATIONS.md S4).
	//
	// It bounds how long a session that claimed a message and then DIED can strand
	// it from its station's other readers. Too short and a reader still working is
	// undercut by a second reader picking the message up; too long and a dead
	// session's mail sits invisible. Sized against a turn rather than a request,
	// because "claimed" means "a model is reasoning about it", not "a query ran".
	//
	// Applies ONLY to bound endpoints. An unbound endpoint is the sole reader of
	// its own mail, so there is nothing to claim it against.
	ClaimLeaseSeconds int

	// FilesEnabled gates file exchange INDEPENDENTLY of COMM itself, and defaults
	// off: the relay is the bulk of the subsystem's risk (disk, quotas, orphan
	// sweeping), so an operator opts into it separately. Live-togglable — checked
	// per operation, so it doubles as a kill switch.
	FilesEnabled bool
	// FileMaxBytes caps one attachment.
	FileMaxBytes int64
	// FileTTLSeconds bounds how long an offered/undelivered attachment survives.
	FileTTLSeconds int
	// FileBudgetBytes is the GLOBAL cap on bytes held in the relay at once. This is
	// the rule that makes C1's isolation honest: the relay shares a volume with the
	// knowledge base, and filling it would fail durable KB writes over chat traffic.
	FileBudgetBytes int64
	// FileMinFreeBytes is the free-space floor: even under budget, an upload is
	// refused when the volume has less than this left, so the KB writer always has
	// headroom.
	FileMinFreeBytes int64
	// GrantTTLSeconds bounds a transfer grant. Short on purpose: it only has to
	// survive being handed to curl.
	GrantTTLSeconds int
	// EndpointIdleTTLSeconds is how long an endpoint with no traffic and no live
	// attachment survives before the sweeper removes it (its channels cascade).
	// Sessions register once and never unregister, so without this the row set
	// grows forever under ordinary use.
	EndpointIdleTTLSeconds int
}

Limits are the operator-tunable bounds enforced by this package. They are enforced in SQL inside the writing transaction rather than by keying the shared rate-limiter bucket: that bucket fails OPEN when saturated, which is correct for IP and token keys (an attacker cannot mint those cheaply) and wrong for identifiers a caller can create in a loop. A fail-open quota is a disk-full outage; a refused message is a retry.

func DefaultLimits 

func DefaultLimits() Limits

DefaultLimits are deliberately conservative: COMM shares a disk with the knowledge base, and the failure this guards against is ephemeral traffic filling the volume and failing durable KB writes.

type Message 

type Message struct {
	MessageID        string
	ChannelID        string
	Seq              int64
	SenderEndpointID string
	Body             string
	RequiresResponse bool
	ReplyToMessageID string
	State            string
	DeliveryCount    int
	BodyBytes        int
	CreatedAt        string
	ExpiresAt        string
	ReplyDeadlineAt  string
	// File is the attachment descriptor when this message carries a file offer
	// (docs/COMM.md §11); nil for ordinary messages.
	File *FileInfo
	// Kind is "message" for peer traffic and "status" for a server-authored notice
	// about an earlier message's fate (expired, reply_overdue). A peer cannot author
	// a status row, so a receiver may trust the distinction.
	Kind string

	// TTLClampedFrom is the ttl_seconds the caller ASKED for when the server gave it
	// something shorter, and zero when nothing was overridden.
	//
	// Silent clamping is how a sender ends up believing a message will outlive an
	// absence it will not: the result reported the resulting expires_at and never
	// mentioned that the request had been overruled, so noticing required diffing a
	// timestamp against a number you had to remember passing.
	TTLClampedFrom int

	// WaitingForYou is how many messages were already queued or delivered FOR THE
	// SENDER on this channel at the moment of sending.
	//
	// A session that sends without reading what is already waiting answers a question
	// its peer has often moved past — measured on this project: a reply that
	// re-argued a point the peer had already conceded. The value of checking is not
	// the read, it is the pause before sending; a non-zero count here is the prompt
	// to take it. Send already computed this number for backpressure and discarded
	// it, so it costs one extra aggregate over a scan that was happening anyway.
	WaitingForYou int
}

Message is one atomic transfer over a channel.

Body is empty once the message has been acked (or expired) — the content is deleted while the metadata row survives. See the schema comment for why that split is load-bearing rather than an optimization.

func (*Message) Redelivered 

func (m *Message) Redelivered() bool

Redelivered reports whether the receiver has seen this message before. At-least- once delivery makes this normal, not exceptional: a receiver should treat a redelivery as "you may not have finished processing this", not as a duplicate to discard blindly.

type OfferResult 

type OfferResult struct {
	Attachment *Attachment
	// Message is non-nil for "path" offers, which enqueue immediately.
	Message *Message
	// UploadGrant is the one-time grant plaintext for "upload" offers, shown once.
	UploadGrant string
	// RecipientRow is the peer endpoint's rowid, for the caller's wakeup notify.
	RecipientRow int64
}

OfferResult is what an accepted offer produced.

type Owner 

type Owner struct {
	TokenID string
	ActorID int64
	SpaceID int64
}

Owner identifies who a COMM object belongs to. All three fields name rows in ken.db and are supplied by the authenticated caller.

ActorID alone is NOT an ownership key and must never be used as one: actors resolve by (kind, display_name), so every token minted with the same actor name collapses to one actor row across machines and humans. Ownership is SpaceID plus the authorizing human recorded on the channel.

type PendingCode 

type PendingCode struct {
	Label     string
	Joined    int // 0 = nobody has redeemed it yet, 1 = waiting for the second session
	ExpiresAt string
	CreatedAt string
}

PendingCode is an unredeemed pairing code shown to the operator so they can see what they minted. The code itself is NOT recoverable — only its hash is stored — so this shows metadata only; a lost code is re-minted, never recovered.

type SendOpts 

type SendOpts struct {
	// IdempotencyKey makes a resend safe. A repeat with the same key returns the
	// ORIGINAL message instead of delivering a second copy. This matters because a
	// response lost after the server committed is the ordinary failure here — a
	// harness timeout, a reset connection, a restart inside the shutdown grace.
	IdempotencyKey string
	// RequiresResponse marks the message as owing a reply, and arms a server-clock
	// reply deadline. Without a deadline, full-duplex would move the hang from the
	// channel to the requester: a responder that dies leaves the sender waiting
	// forever with no signal.
	RequiresResponse bool
	// ReplyToMessageID correlates this message with an earlier request on the same
	// channel.
	ReplyToMessageID string
	// TTLSeconds overrides the default un-acked lifetime. Relative on purpose:
	// clients never supply absolute timestamps, so clock skew between agent
	// machines cannot silently shorten or extend a lifetime.
	TTLSeconds int
}

SendOpts carries the optional parts of a send.

type Staffing added in v1.6.0 

type Staffing struct {
	Endpoints  int    // live (non-revoked) endpoints bound to the station
	LastSeenAt string // freshest last_seen_at across them; empty when Endpoints is 0
}

Staffing is what COMM knows about whether anyone is actually at a station: how many live endpoints are reading for it, and when the freshest of them was last seen.

Deliberately two facts rather than one boolean. "Staffed" is a judgement about freshness and the right threshold depends on how the reader intends to use it — a directory shown to a human and a routing decision made by an agent do not want the same cutoff. Reporting the inputs and letting the caller judge is the same choice the console makes everywhere else: never fake a number, never hide one.

type Stats 

type Stats struct {
	Endpoints    int
	OpenChannels int
	Unacked      int
	BodyBytes    int64 // retained message bodies; the thing that grows a disk
	Files        int   // live attachments (offered or awaiting delivery)
	FileBytes    int64 // relay bytes currently held on disk
}

Stats is the operator's at-a-glance view, and the source for metrics.

type Store 

type Store struct {
	W *sql.DB // single-writer pool (MaxOpenConns == 1)
	R *sql.DB // reader pool
	// contains filtered or unexported fields
}

Store holds the writer and reader pools over comm.db.

The single-writer discipline mirrors the knowledge base (D6): one writer connection with BEGIN IMMEDIATE turns contention into an in-process queue instead of SQLITE_BUSY races, and avoids the upgrade-mid-transaction deadlock. It also makes the per-(channel,sender) sequence assignment in Send a plain MAX+1 rather than a contended counter.

func Open 

func Open(path string, limits Limits) (*Store, error)

Open opens (creating if needed) the COMM database at path.

fts5.Register is passed on both pools even though no table here uses FTS5. In this driver FTS5 is a PER-CONNECTION extension, not part of the default WASM build, so a pool opened without it fails with "no such module: fts5" the moment anything touches an FTS table. Registering costs nothing measurable and means a future migration that adds message search cannot reintroduce that trap.

func (*Store) Ack 

func (s *Store) Ack(ctx context.Context, ep *Endpoint, messageID string) error

Ack marks a message PROCESSED — not merely received — and drops its body.

The distinction is deliberate and belongs in the instruction text too: a model should ack after acting, so that a turn truncated mid-processing leaves the message to be redelivered rather than silently lost.

Acking an unknown or already-acked message succeeds. Idempotency is required because the transport is at-least-once: a retried ack after a lost response must not surface as an error the model then tries to "fix".

A message that requires a response keeps its body until the reply arrives or its deadline passes: a responder that crashed and recovered plausibly needs to re-read what it owes.

func (*Store) AckUpTo 

func (s *Store) AckUpTo(ctx context.Context, ep *Endpoint, channelID string, seq int64) error

AckUpTo cumulatively acks every message from one sender on a channel up to and including seq. Cumulative acking collapses ack chatter and is idempotent by construction, which is why it exists alongside per-message Ack.

func (*Store) AuthenticateEndpoint 

func (s *Store) AuthenticateEndpoint(ctx context.Context, endpointID, secret string) (*Endpoint, error)

AuthenticateEndpoint resolves an endpoint id + secret to an Endpoint, and refreshes last_seen_at.

The secret is compared in constant time. A revoked endpoint authenticates as ErrDenied rather than ErrNotFound so a caller cannot use the distinction to probe which endpoint ids exist.

func (*Store) BindEndpointToStation added in v1.5.0 

func (s *Store) BindEndpointToStation(ctx context.Context, endpointID, stationID, keyID string) error

BindEndpointToStation attaches an endpoint to a station, making it a reader of that station's inbox rather than the sole owner of its own (docs/STATIONS.md S4).

Called from comm_register AFTER the caller's binding voucher has been redeemed on the durable side: this function trusts stationID because RedeemBindingVoucher is what established it, and there is deliberately no path that lets a caller name a station directly. Binding is set once at registration and never changed — an endpoint that could move between stations would let a session carry another station's unread mail across, which is the shared-inbox failure in a new costume.

func (*Store) ChannelFor 

func (s *Store) ChannelFor(ctx context.Context, ep *Endpoint, channelID string) (*Channel, int64, error)

ChannelFor resolves an open channel by its public id and verifies the endpoint belongs to it, returning the peer's rowid.

Membership is re-checked on every operation rather than trusted from an earlier call: a channel can be revoked, and an endpoint that was a member a moment ago must not keep acting on one.

func (*Store) CheckFileQuota 

func (s *Store) CheckFileQuota(ctx context.Context, incoming int64) error

CheckFileQuota enforces the global budget and the free-space floor for an incoming size. Exported because the HTTP handler re-checks at PUT time: the offer-time check bounds grants, this bounds bytes.

func (*Store) ClearStoredBytes 

func (s *Store) ClearStoredBytes(ctx context.Context, attachmentID string) error

ClearStoredBytes zeroes a settled attachment's byte accounting after its file has actually been removed. Separate from the sweep transaction on purpose: the budget must free only for bytes that are really gone.

func (*Store) Close 

func (s *Store) Close() error

Close closes both pools.

func (*Store) CompleteUpload 

func (s *Store) CompleteUpload(ctx context.Context, attachmentRow int64, storedBytes int64) (*Message, int64, error)

CompleteUpload marks an upload's bytes verified and enqueues the message the receiver will poll. Called by the HTTP handler after the streamed sha256 matched the offer — which is why the receiver can never observe partial state.

func (*Store) ConsoleFingerprint added in v1.3.0 

func (s *Store) ConsoleFingerprint(ctx context.Context, spaceID int64) (int64, error)

ConsoleFingerprint returns a single number that changes whenever the console's view of a space would look different: an endpoint registered or revoked, a channel created/opened/revoked, a pairing code minted or consumed, or messages flowing. It backs the /comm page's live auto-refresh — the page reloads when this diverges from the value it was rendered with, and updates its "last checked" stamp on every poll.

Distinct prime weights make an accidental collision (two offsetting changes summing to the same number) unlikely; this is a change detector, not a checksum, so unlikely is enough.

func (*Store) ConsumeGrant 

func (s *Store) ConsumeGrant(ctx context.Context, plaintext, kind string) (*GrantInfo, error)

ConsumeGrant resolves and consumes a grant in one step. Single-use even when the transfer then fails — the agent mints a fresh grant rather than retrying a credential that has been on the wire. An unknown, expired, consumed, or wrong-kind grant are all ErrNotFound: indistinguishable on purpose.

func (*Store) CountEndpointsBoundBy added in v1.5.0 

func (s *Store) CountEndpointsBoundBy(ctx context.Context, keyID string) (int, error)

CountEndpointsBoundBy reports how many LIVE endpoints a station key bound, so the console can say "this will disconnect N live sessions" before the operator clicks (S6). A destructive action whose blast radius is only visible afterwards is one an operator learns to fear rather than use.

func (*Store) CountOpenChannelsBetweenStations added in v1.6.0 

func (s *Store) CountOpenChannelsBetweenStations(ctx context.Context, stationA, stationB string) (int, error)

CountOpenChannelsBetweenStations reports how much live traffic revoking a link would end. It exists to be shown BEFORE the click: S6 asks for the blast radius in front of the human, and "revoke" with no number attached is a button people either avoid or press twice.

Returns 0 rather than an error when the pair has never spoken, which is the common case and is not a failure.

func (*Store) EnsureFilesDir 

func (s *Store) EnsureFilesDir() error

EnsureFilesDir creates the relay directory (0700) on first use.

func (*Store) FailUpload 

func (s *Store) FailUpload(ctx context.Context, attachmentRow int64) error

FailUpload marks an upload failed (checksum mismatch, overrun, aborted stream). The sender recovers by re-offering; the failed row survives as audit.

func (*Store) FilePath 

func (s *Store) FilePath(attachmentID string) string

FilePath returns the on-disk location for an attachment id. The id is server-minted base62, so it is safe to splice into a path by construction.

func (*Store) GrantDownload 

func (s *Store) GrantDownload(ctx context.Context, ep *Endpoint, attachmentID string) (string, *Attachment, error)

GrantDownload mints a one-time download grant for an attachment addressed to ep. Callable repeatedly — redelivered messages and failed curls need fresh grants — but only while the bytes exist.

func (*Store) JoinChannel 

func (s *Store) JoinChannel(ctx context.Context, ep *Endpoint, code string) (*Channel, error)

JoinChannel redeems a pairing code for an endpoint.

Establishment is two-sided from day 1: the first redeem creates a pending channel, the second opens it. Both sides call this even though both currently share one owner — turning a unilateral "A opens a channel to B" into an accept flow later would tighten an already-shipped tool, which is a breaking change.

Re-redeeming the same code from an endpoint already on the channel is idempotent and returns the channel unchanged, so a retried call after a lost response cannot consume the code twice or wedge the pairing.

func (*Store) Limits 

func (s *Store) Limits() Limits

Limits returns the bounds this store currently enforces.

func (*Store) ListChannels 

func (s *Store) ListChannels(ctx context.Context, ep *Endpoint) ([]Channel, error)

ListChannels returns the channels an endpoint belongs to.

func (*Store) ListChannelsForSpace 

func (s *Store) ListChannelsForSpace(ctx context.Context, spaceID int64) ([]AdminChannel, error)

ListChannelsForSpace returns every channel owned by one space, newest first.

Scoped by space even though only one exists today, for the same reason ListEndpoints is: an unscoped listing becomes the enumeration surface the moment a second human exists, and narrowing it later would be a behavioural break.

func (*Store) ListEndpoints 

func (s *Store) ListEndpoints(ctx context.Context, spaceID int64) ([]Endpoint, error)

ListEndpoints returns the endpoints owned by one space, newest first. Scoped by space from day 1 even though only one exists today: an unscoped listing would be the enumeration surface in a multi-human future, and scoping it later would be a behavioural break for anything that relied on the full list.

func (*Store) ListPendingCodes 

func (s *Store) ListPendingCodes(ctx context.Context, spaceID int64) ([]PendingCode, error)

ListPendingCodes returns codes that are minted, unexpired, and not yet fully consumed.

func (*Store) LiveEndpointForStation added in v1.5.0 

func (s *Store) LiveEndpointForStation(ctx context.Context, stationID string) (*Endpoint, error)

LiveEndpointForStation returns the most recently seen live endpoint reading for a station, or nil when nobody is staffing it.

"Most recent" rather than "the only one" because a station may legitimately have several readers (S4). Picking the freshest is a heuristic for "who is actually here", and it does not need to be exact: whichever endpoint is chosen, the message lands in the STATION's inbox and any reader can claim it.

func (*Store) MessageByID 

func (s *Store) MessageByID(ctx context.Context, messageID string) (*Message, error)

MessageByID loads one message by its public id.

func (*Store) Migrate 

func (s *Store) Migrate() error

Migrate applies embedded COMM migrations in lexical order, skipping versions already recorded. Idempotent, forward-only, and independent of the knowledge base's migration state — the two databases version separately on purpose, so a COMM schema change never touches ken.db.

func (*Store) MintPairingCode 

func (s *Store) MintPairingCode(ctx context.Context, spaceID, humanActorID int64, label string) (string, error)

MintPairingCode creates a human-authorized pairing code and returns the plaintext exactly once; only its SHA-256 is stored.

This is COMM's structural gate: an agent cannot conjure a channel, because channel creation requires a value only the human web UI can produce. It is the same move that makes the curation gate trustworthy — withhold the capability rather than instruct the model not to use it — applied at the one place in COMM where it is available.

func (*Store) OfferFile 

func (s *Store) OfferFile(ctx context.Context, ep *Endpoint, channelID string, in FileOffer) (*OfferResult, error)

OfferFile records a file offer from ep on channelID.

"path" offers enqueue their message immediately (there is nothing to wait for); "upload" offers return a one-time grant and enqueue only at CompleteUpload.

func (*Store) OpenLinkedChannel added in v1.5.0 

func (s *Store) OpenLinkedChannel(ctx context.Context, a, b *Endpoint, ownerActorID int64, label string) (*Channel, error)

OpenLinkedChannel materializes a channel between two station-bound endpoints whose stations a human has already linked (docs/STATIONS.md S9).

This is what a link is FOR. Without it a link records the human's decision without ever spending it, and every conversation still costs a pairing code — which is the step the link exists to remove. The decision itself is not removed: the channel can only come into existence because a human approved the relationship, which is the same gate one level up.

The caller MUST have verified the link in the knowledge-base store first; this package cannot see it. That is S7's boundary, not laziness — comm.db holds no durable authorization and must not start.

Opened directly rather than left pending: the pairing flow is pending-until-both-join because a CODE is a rendezvous between two sessions that have not met. A link has already established that both stations may talk, so there is nothing to wait for. Idempotent — asking twice returns the existing channel rather than a second one, because a session that retries after a lost response must not fragment the conversation into two.

func (*Store) PartPath 

func (s *Store) PartPath(attachmentID string) string

PartPath is the temporary upload target; renamed onto FilePath only after the checksum matches, so a final file is complete by construction.

func (*Store) Path 

func (s *Store) Path() string

Path reports the database file this store was opened from (startup logging).

func (*Store) PendingReplies 

func (s *Store) PendingReplies(ctx context.Context, ep *Endpoint) ([]Message, error)

PendingReplies lists this endpoint's sent messages that still owe a response. Exposed as a query so a sender can ask what is outstanding rather than inferring it from a reference message that may already have been superseded.

func (*Store) Poll 

func (s *Store) Poll(ctx context.Context, ep *Endpoint, limit int) ([]Message, error)

Poll returns the un-acknowledged messages this endpoint may read, oldest first, and counts a delivery attempt for each.

TWO REGIMES, and which one applies depends on whether the endpoint is bound to a station (docs/STATIONS.md S4).

UNBOUND — the shipped behaviour, unchanged. The endpoint is the sole reader of its own mail. Poll is a pure read of DELIVERABILITY: being polled never hides a message from the next poll, only Ack advances state. That is what makes a lost poll response harmless — the messages simply come back — and it is why "delivered" is an informational timestamp rather than a gate.

BOUND — the STATION owns the inbox and this endpoint is one of possibly several credentialed readers, so delivery becomes CLAIM-ONCE. The first reader to poll a message claims it, and while the claim holds, that message is hidden from the station's other readers. This deliberately weakens the "polling never hides anything" property above, and it has to: without it, two sessions staffing one station would both act on the same message, which is precisely the shared-inbox accident the per-endpoint secret exists to prevent.

The claim is a LEASE, not a transfer of ownership. When it expires unacknowledged the message returns to the unclaimed tail and may reach a DIFFERENT reader than first saw it. Without the lease, a session that claims and then dies strands its messages permanently and COMM's C6 promise — a message delivered but never acted upon comes back — would be false.

The ordering promise weakens accordingly, and the tool description says so: from "per channel and direction" to "per channel and direction, across the station's readers". Two sessions polling one station see a PARTITIONED stream and neither sees the whole order. That is the price of letting a second session help without severing the first.

func (*Store) ReceivedSince 

func (s *Store) ReceivedSince(ctx context.Context, actorID int64, windowSeconds int) (bool, error)

ReceivedSince reports whether any endpoint owned by actorID has had a message DELIVERED to it within the last windowSeconds.

This exists for one reason: a message is a side channel into curation. A session told "entry X is verified, propose a revision at high confidence" will author a proposal that is indistinguishable from first-hand knowledge — the invariant survives literally (an AI authored it, a human promotes it) while the curator's signal quality has quietly degraded to hearsay with no chain of custody. Marking the authored version lets the curator ask for a first-hand citation before promoting. See docs/COMM.md §7.

It keys on delivery, not on arrival: a message sitting un-polled in the queue has influenced nothing. It considers the LAST delivery, not the first: under at-least-once semantics a message is redelivered until acked, so keying on first_delivered_at alone produced a systematic false negative — a message first delivered before the window but re-read inside it left no mark, in the system's normal operating mode. Acked messages fall back to their acknowledgement time, which is when the receiver acted on them.

It keys on the ACTOR, not the token, and that is forced rather than chosen: a COMM token must be DEDICATED (it may not also carry knowledge-base scopes), so the token that receives messages is never the token that authors an entry. Keying on the token would make this function always return false. The actor is the identity the two tokens legitimately share — mint both with the same `--actor` and the link holds.

Consequence the operator must know: if the two tokens are minted under DIFFERENT actor names, nothing is ever marked. That is a silent false negative, so it is called out in docs/COMM.md §7 rather than left to be discovered.

Actors resolve by display name and therefore collapse across machines, which would be wrong for an ownership check — but here over-matching is the SAFE direction, for the same reason the whole marker is biased toward over-reporting.

Deliberately conservative in one direction: metadata rows outlive acknowledgement (bodies do not), so a message that was read and acted upon still answers true for the whole window. Deliberately imprecise in another: it cannot know whether the message had anything to do with what is being saved. A false positive costs the curator one extra glance; a false negative would silently launder hearsay into the knowledge base, so the marker is biased toward over-reporting.

Callers must treat any error as "unknown" and NOT as "no": failing to mark is the direction that loses information.

func (*Store) RegisterEndpoint 

func (s *Store) RegisterEndpoint(ctx context.Context, owner Owner, label, hostHint string) (*Endpoint, string, error)

RegisterEndpoint mints a new endpoint for an authenticated session and returns it together with its one-time secret, which is never recoverable afterwards.

A repeat registration under the same token and label deliberately creates a NEW endpoint rather than attaching to the existing one: silently handing a second session the first session's inbox is the failure this avoids, and it is far more likely by accident (two sessions with the same label) than by malice.

hostHint is stored opaquely and is never consulted for authorization — see the schema comment and docs/COMM.md C9 for why a self-reported machine identity cannot prove a shared filesystem.

func (*Store) RevokeChannel 

func (s *Store) RevokeChannel(ctx context.Context, channelID string) error

RevokeChannel closes a channel permanently. This is the operator's brake: a security model whose enforcement point is the human needs the human to have one.

func (*Store) RevokeChannelsBetweenStations added in v1.6.0 

func (s *Store) RevokeChannelsBetweenStations(ctx context.Context, stationA, stationB string) (int, error)

RevokeChannelsBetweenStations closes every open channel between two stations and returns how many it closed.

This is the caller RevokeStationLink's doc comment asks for: revoking the LINK withdraws the permission, but a channel opened while the permission held keeps working, because the channel row carries its own state. Ending the relationship without ending its live traffic is a revocation that revokes nothing observable — the same shape as a flag with no reader.

Idempotent: revoking twice closes nothing the second time and returns 0. A pair that never spoke is not an error.

func (*Store) RevokeEndpoint 

func (s *Store) RevokeEndpoint(ctx context.Context, endpointID string) error

RevokeEndpoint soft-revokes an endpoint, immediately denying further use. Its channels stay queryable for the operator; its messages age out normally.

func (*Store) RotateEndpointSecret added in v1.5.0 

func (s *Store) RotateEndpointSecret(ctx context.Context, endpointID string) (string, error)

RotateEndpointSecret replaces an endpoint's secret and returns the new one, shown once. The endpoint keeps its id, its owner and — the point of the whole operation — every channel it belongs to, so its peers are unaffected and nothing needs re-pairing.

THIS IS DELIBERATELY NOT REACHABLE FROM ANY TOOL, and that placement is the entire security argument. One bearer token covers a machine, so the endpoint pair is the only thing separating two sessions sharing it; a reissue any SESSION could trigger would let any session on that machine seize any endpoint on it. That is why deriving a new secret from token material was rejected. The defect there is the AUTOMATION, not the reissuing — so rotation lives behind curator authentication, which is a credential no session holds or can obtain from the machine, and a neighbouring session with the COMM token gains nothing.

Two callers in mind, and the second is the stronger reason to have it:

  • A session lost its secret (context compaction destroys it, and it is unrecoverable by construction). Today that costs one fresh pairing code PER CHANNEL plus coordinated re-joins with every peer.
  • A secret LEAKED — into a transcript, a log, a file something else could read. Until now the only remedy was revoking the endpoint and rebuilding every channel from scratch, which is why containing a leak was expensive enough to hesitate over. Rotation is the missing incident-response primitive.

A revoked endpoint is refused: rotating one would quietly resurrect a capability an operator deliberately destroyed, and the revoke path is what a leak response escalates TO, never back from.

func (*Store) Send 

func (s *Store) Send(ctx context.Context, ep *Endpoint, channelID, body string, opts SendOpts) (*Message, error)

Send enqueues one message from ep to its peer on the named channel.

Enforced inside the writing transaction: channel membership and openness, the body cap, the per-channel un-acked cap (backpressure), and sequence assignment. Quotas are checked here rather than in the shared rate-limiter bucket because that bucket fails OPEN when saturated — correct for keys an attacker cannot mint cheaply, wrong for identifiers a caller creates in a loop.

func (*Store) SetLimits 

func (s *Store) SetLimits(l Limits)

SetLimits replaces the enforced bounds. Safe to call at any time, including while requests are in flight: the settings page applies changes live, so an operator can tighten a limit during a runaway rather than after a restart.

An operation that already read the old limits completes under them; there is no attempt to make a single request see a consistent snapshot across several reads, because every enforcement point reads once.

func (*Store) SeverEndpointsBoundBy added in v1.5.0 

func (s *Store) SeverEndpointsBoundBy(ctx context.Context, keyID string) (int, error)

SeverEndpointsBoundBy revokes every endpoint a given station key bound, and releases their claims. It reports how many were severed so the console can state the count BEFORE the click, as S6 requires.

This is what makes revoking a station key mean something. You revoke because the key leaked; a revocation that stops future bindings but leaves the already-bound sessions running until an idle sweep notices is theatre — and traffic keeps an endpoint alive indefinitely, so the sweep may never come.

Claims are released in the same statement rather than left to expire: a severed reader is never coming back to ack, so holding its messages for the rest of the lease would hide them from the station's remaining readers for no reason.

func (*Store) StaffingByStation added in v1.6.0 

func (s *Store) StaffingByStation(ctx context.Context) (map[string]Staffing, error)

StaffingByStation reports staffing for every station COMM has ever seen an endpoint for, in ONE query.

Batch rather than per-station on purpose: a directory listing N stations must not cost N round trips, and the per-station form already exists for the single-target case (LiveEndpointForStation). Stations with no endpoint are simply absent from the map — a missing key means "nobody has ever staffed this", which is what the caller wants to render, and materialising a zero row for every station in the space would make the map lie about which ones COMM knows.

func (*Store) StatsFor 

func (s *Store) StatsFor(ctx context.Context, spaceID int64) (Stats, error)

StatsFor reports counters for one space.

func (*Store) Sweep 

func (s *Store) Sweep(ctx context.Context) (expired, purged int64, err error)

Sweep enforces every time-based rule in one pass and returns what it changed.

Runs on a cadence of a minute or less, deliberately NOT folded into the hourly housekeeping loop: at a sustained send rate a single sender writes hundreds of megabytes before an hourly sweep first runs, which is why a TTL is not a quota and must not be mistaken for one.

func (*Store) UnbindEndpointFromStation added in v1.5.2 

func (s *Store) UnbindEndpointFromStation(ctx context.Context, endpointID string) error

UnbindEndpointFromStation returns a bound endpoint to standing alone. It keeps its id, its secret and every channel it is in; only the station association goes.

This exists because binding was a ONE-WAY DOOR and nobody should have to walk through one to try a feature. An operator weighing adoption asked the right question — "is it reversible?" — and the honest answer was no, which is a bad answer for a step whose whole purpose is to make things cheaper.

What unbinding means for mail is the reason it is safe. Messages are addressed to an ENDPOINT rowid; the station merely widens which endpoint may read them. So unbinding narrows this endpoint back to its own mail and strands nothing: anything addressed to it is still addressed to it, and anything addressed to a sibling was never its to begin with. Claims it currently holds ARE released, because after unbinding it will not be polling for them and leaving them held would hide those messages from the station's remaining readers for the rest of the lease.

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.