Documentation
¶
Overview ¶
Package chatsync holds the contract probe for the CLI side of chat session sync: pushing local session events to the mivia API and long-polling it for remote input.
The sync client does not exist yet. The API half shipped first, and the client design is still being planned. What this package holds today is only the live contract probe in live_contract_test.go, behind the `livechat` build tag, which checks the deployed /v1/chat-sessions surface end to end.
The probe is deliberately not a client. It speaks raw HTTP and keeps its own wire structs, so it pins what the SERVER does without freezing decisions about how the CLI should be built. When the client lands, promote those structs out of the test file rather than writing new ones from the docs: the probe's versions are the ones proven against a running deployment.
The probe found four API defects on its first run: an oversized payload returned 500 with the failing SQL, a batch could carry its own sequence gap, a second consume of the same input reported success, and events still appended to an ended session. All four were fixed in apps/api and verified green against the deployment on 2026-08-31, so a red run now means a real regression rather than known debt.
It also found the cross-replica fan-out failure that a single-replica run cannot see: with two replicas, an SSE stream only received an append served by its own instance. See TestLiveChatSessionFanOutReachesEveryStream, and run it against more than one replica or it proves nothing.
Index ¶
- Constants
- Variables
- func DefaultBaseURL(configured string) string
- func FlushOutbox(ctx context.Context, client *Client, outbox *Outbox, sessionID string) (int, error)
- func FlushOutboxWithTrace(ctx context.Context, client *Client, outbox *Outbox, ...) (int, error)
- func IdentityDir(storeDir string) string
- func IdentityKey(chatSessionID string) string
- func OutboxDirFor(storeDir string, handle LocalHandle) string
- func ProbeEndpoint(ctx context.Context, baseURL string) (reachable bool, detail string)
- func SaveIdentity(dir, key string, id SyncIdentity) error
- type AgentOrigin
- type AppendResult
- type AssistantDeltaPayload
- type AssistantMessagePayload
- type AssistantResetPayload
- type AuthorUserIDProvider
- type BadRequestError
- type BlockGrammar
- type Client
- func (c *Client) AppendEvents(ctx context.Context, sessionID string, events []EventItem) (*AppendResult, error)
- func (c *Client) AppendEventsWithTrace(ctx context.Context, sessionID string, events []EventItem, ...) (*AppendResult, error)
- func (c *Client) AuthLost() bool
- func (c *Client) ConsumeInput(ctx context.Context, sessionID, inputID string) (*SessionInput, error)
- func (c *Client) CreateSession(ctx context.Context, params CreateSessionParams) (*Session, error)
- func (c *Client) EndSession(ctx context.Context, sessionID string) (*Session, error)
- func (c *Client) GetEvents(ctx context.Context, sessionID string, afterSeq int64, limit int) ([]StoredEvent, error)
- func (c *Client) GetSession(ctx context.Context, sessionID string) (*Session, error)
- func (c *Client) Heartbeat(ctx context.Context, sessionID, status string) (*Session, error)
- func (c *Client) NextInput(ctx context.Context, sessionID string, waitSeconds int) (*NextInput, error)
- type ClientOptions
- type ConflictError
- type ContextCompactedPayload
- type CreateSessionParams
- type Cursor
- type Endpoint
- type Envelope
- type ErrorEnvelope
- type EventItem
- type HeartbeatRunner
- type HookRanPayload
- type IdentityRef
- type InputPoller
- func (p *InputPoller) Inputs() <-chan RemoteInput
- func (p *InputPoller) MarkReceived(id string)
- func (p *InputPoller) SetOnRejected(fn func(id, sessionID, reason string))
- func (p *InputPoller) SetSessionID(sessionID string)
- func (p *InputPoller) Start(ctx context.Context)
- func (p *InputPoller) Stop(ctx context.Context)
- type LocalHandle
- type NextInput
- type Outbox
- func (ob *Outbox) AdvanceCursor(seq int64) error
- func (ob *Outbox) Append(events ...WireEvent) error
- func (ob *Outbox) Close() error
- func (ob *Outbox) Cursor() Cursor
- func (ob *Outbox) Dead() bool
- func (ob *Outbox) MaxSeq() int64
- func (ob *Outbox) Rebase(base int64) (int, error)
- func (ob *Outbox) ResetForFork() (int, error)
- func (ob *Outbox) UnflushedCount() int
- func (ob *Outbox) UnflushedEvents() ([]StoredEvent, error)
- type Projector
- func (p *Projector) Flush(currentDrops uint64) []WireEvent
- func (p *Projector) LastSeq() int64
- func (p *Projector) Project(ev events.Event) []WireEvent
- func (p *Projector) ProjectWithDrops(ev events.Event, currentDrops uint64) []WireEvent
- func (p *Projector) ResetSeq(seq int64)
- func (p *Projector) RollbackDrops(delta uint64)
- func (p *Projector) RollbackSeq(n int)
- func (p *Projector) RollbackStreaming(wireEvents []WireEvent)
- func (p *Projector) SetSyncOpts(includeThinking, includeToolIO, streamAssistant bool)
- type ProjectorOptions
- type RemoteInput
- type Session
- type SessionAttachment
- type SessionInput
- type SessionOptions
- type StoredEvent
- type SubagentAssistantDeltaPayload
- type SubagentAssistantMessagePayload
- type SubagentEndedPayload
- type SubagentProgressPayload
- type SubagentStartedPayload
- type SubagentThinkingDeltaPayload
- type SubagentThinkingMessagePayload
- type SubagentToolEndedPayload
- type SubagentToolStartedPayload
- type SyncDroppedPayload
- type SyncForkedPayload
- type SyncIdentity
- type SyncSession
- func (s *SyncSession) HandleEvent(ctx context.Context, ev events.Event)
- func (s *SyncSession) Inputs() <-chan RemoteInput
- func (s *SyncSession) LastSeq() int64
- func (s *SyncSession) MarkInputReceived(id string)
- func (s *SyncSession) ProjectorOptsForTest() ProjectorOptions
- func (s *SyncSession) SessionID() string
- func (s *SyncSession) SetStatus(ctx context.Context, status string)
- func (s *SyncSession) SetSyncOpts(includeThinking, includeToolIO, streamAssistant bool)
- func (s *SyncSession) Stop(ctx context.Context) error
- func (s *SyncSession) StopReason() string
- func (s *SyncSession) Stopped() bool
- func (s *SyncSession) StoppedForTest()
- type SyncStatus
- type SyncTelemetry
- type SyncTelemetrySnapshot
- type ThinkingDeltaPayload
- type ThinkingMessagePayload
- type TokenEnsurer
- type TokenProvider
- type ToolEndedPayload
- type ToolStartedPayload
- type TruncField
- type Truncation
- type TurnEndedPayload
- type TurnFailedPayload
- type TurnStartedPayload
- type WireEvent
- type WireEventSpec
Constants ¶
const ( SyncStateHealthy = "healthy" SyncStateDegraded = "degraded" SyncStateRecovered = "recovered" SyncStateStopped = "stopped" )
Sync health states, as written to status.json.
const ( BudgetPromptText = 32 * 1024 BudgetAssistantText = 32 * 1024 BudgetToolOutput = 16 * 1024 BudgetDeltaText = 8 * 1024 BudgetToolInput = 4 * 1024 BudgetErrorMessage = 2 * 1024 BudgetShortField = 200 )
const ( TypeTurnStarted = "mivia.chat.v1.turn.started" TypeAssistantDelta = "mivia.chat.v1.assistant.delta" TypeAssistantMessage = "mivia.chat.v1.assistant.message" // A turn that is being re-driven from the beginning. Everything already // sent for its assistant block belongs to an attempt that no longer // exists, so a viewer that accumulates deltas must drop them. TypeAssistantReset = "mivia.chat.v1.assistant.reset" TypeThinkingDelta = "mivia.chat.v1.thinking.delta" // The settled form of one thinking block, emitted when the block closes. // It exists so reasoning survives a redaction policy: a policy suppresses // the per-fragment text (a secret split across two deltas matches // neither pattern), and without a whole-block form to redact as ONE // string the reasoning was lost outright. Same INV-1 shape as // assistant.message - Text is empty exactly when Fragments is non-zero. TypeThinkingMessage = "mivia.chat.v1.thinking.message" TypeToolStarted = "mivia.chat.v1.tool.started" TypeToolEnded = "mivia.chat.v1.tool.ended" TypeHookRan = "mivia.chat.v1.hook.ran" TypeSubagentToolStarted = "mivia.chat.v1.subagent.tool.started" TypeSubagentToolEnded = "mivia.chat.v1.subagent.tool.ended" TypeSubagentStarted = "mivia.chat.v1.subagent.started" TypeSubagentProgress = "mivia.chat.v1.subagent.progress" TypeSubagentEnded = "mivia.chat.v1.subagent.ended" // A subagent's own prose. These are DELIBERATELY distinct types rather // than the root assistant/thinking types with Envelope.Agent set: a // viewer that predates them must be able to keep a subagent's output out // of the main transcript, and the only thing such a viewer can key on is // the type string. Every subagent type shares the // "mivia.chat.v1.subagent." prefix for exactly that reason, so one // prefix rule covers types minted after the client shipped. TypeSubagentAssistantDelta = "mivia.chat.v1.subagent.assistant.delta" TypeSubagentAssistantMessage = "mivia.chat.v1.subagent.assistant.message" TypeSubagentThinkingDelta = "mivia.chat.v1.subagent.thinking.delta" TypeSubagentThinkingMessage = "mivia.chat.v1.subagent.thinking.message" TypeTurnEnded = "mivia.chat.v1.turn.ended" TypeContextCompacted = "mivia.chat.v1.context.compacted" TypeTurnFailed = "mivia.chat.v1.turn.failed" TypeSyncDropped = "mivia.chat.v1.sync.dropped" TypeSyncForked = "mivia.chat.v1.sync.forked" )
const (
DefaultHeartbeatInterval = 30 * time.Second
)
const (
DefaultMaxUnflushed = 5000
)
const RecommendedStopTimeout = 15 * time.Second
RecommendedStopTimeout is the ctx budget a caller should give Stop for its FINAL flush - one real network round trip carrying whatever backlog is still unflushed, in a single request (FlushOutbox sends the whole backlog as one batch, uncapped). 2 seconds - the previous ad-hoc value at both call sites - is not that budget: it is sized for a small, mostly- idle turn's tail, and a real remote server plus a genuinely accumulated backlog (periodic mid-turn flushes cannot always keep pace with local event generation, especially once sync.stream_assistant multiplies event count 5-10x) can legitimately take several seconds. A ctx that expires first makes Stop return early via its timedOut path, which hands the outbox close to a background goroutine the caller's own process exit then kills - so the accumulated backlog is not merely delayed, it is permanently lost: nothing re-opens a one-shot run's outbox directory on a later invocation. 15 seconds is a real chance at that one request without leaving process exit hanging indefinitely on a truly dead network - Stop still returns ctx.Err() if that budget runs out too.
const RepairedAtIngestKey = "repaired_at_ingest"
RepairedAtIngestKey marks a payload the SERVER rewrote on the way in, so this client can recognise its own body in a form it did not send.
The server repairs a payload it cannot otherwise store at all: a NUL is rejected by Postgres outright, and the rejection takes the whole batch with it. Repairing is better than losing a hundred events - but it makes stored and sent differ, and this file exists to treat that difference as corruption.
Without the flag the two rules collide and the strict one wins: a repaired batch that is later re-sent (an ambiguous ack, where insertedCount comes back short because ON CONFLICT skipped rows already stored) reads back a body this client did not send, raises ErrTranscriptConflict, and stops the session's sync permanently. The repair turned a batch-sized loss into a session-sized one.
Variables ¶
var ( ErrConflict = errors.New("chatsync: session conflict (409)") ErrNotFound = errors.New("chatsync: session not found (404)") // ErrBadRequest reports a 400. It is a distinct sentinel because the // settled policy splits on it: a sequence-gap 400 rebases and continues, // while every other 400 is poison a retry cannot fix. ErrBadRequest = errors.New("chatsync: bad request (400)") // ErrTranscriptConflict reports that the server already holds events at // the sequences this client just sent, with bodies that are not the ones // it sent. Terminal for sync. ErrTranscriptConflict = errors.New("chatsync: the server holds a different transcript at these sequences") // ErrNoTokenProvider reports a construction attempt with no way to // authenticate. ErrNoTokenProvider = errors.New("chatsync: a token provider is required") // ErrEmptyToken reports a token provider that returned no error and no // token. Sending the request anyway would omit the Authorization header, // so this fails closed instead. ErrEmptyToken = errors.New("chatsync: the token provider returned an empty token") // ErrInvalidPathID reports an id this client refused to interpolate into // a request path. See validatePathID. ErrInvalidPathID = errors.New("chatsync: invalid id for request path") )
var ( ErrOutboxLocked = errors.New("chatsync: outbox is locked by another process") ErrOutboxOverflow = errors.New("chatsync: outbox capacity exceeded") )
var ErrAuthStop = errors.New("chatsync: sync stopped; run `mivia login`")
ErrAuthStop reports an authentication failure that no retry can fix: the local CLI session is gone and only `mivia login` restores it. Sync stops on this error instead of retrying, because the alternative is a background loop that either prompts in the middle of a chat or hammers the API with a dead session.
var KnownWireTypes = derivedWireTypes()
KnownWireTypes lists all recognized mivia.chat.v1.* type constants. It is derived from wireEventSpecs and is never hand-maintained.
var ProseBlockPattern = regexp.MustCompile(`^(.+:(?:assistant|thinking)):(\d+)$`)
ProseBlockPattern is the grammar of a prose block id: a STREAM id - one of `<turn>:assistant`, `<turn>:thinking`, `<turn>:<task>:assistant`, `<turn>:<task>:thinking` - followed by `:<step>`. It is anchored on the stream suffix, not on `.+`: a looser `.+:\d+` would also accept a tool_call_id, which ends in digits under any id scheme, and a consumer keying on a parsed step must never mistake a tool row for prose.
A reset names the bare stream with no step, so it deliberately does not match; nor does the legacy stepless id older producers wrote.
Functions ¶
func DefaultBaseURL ¶
DefaultBaseURL returns configured when it names something, else the mivia API root internal/miviaauth already resolves.
Nothing asks a user to configure sync, so an empty api_url is the normal case rather than an error. The fallback reads through miviaauth on purpose: that package owns where the API lives (DefaultServerURL, overridable with MIVIA_API_BASE_URL), and a second copy of the URL here would be a second place to change when it moves, and a second place to get a staging or local override wrong.
func FlushOutbox ¶
func FlushOutbox(ctx context.Context, client *Client, outbox *Outbox, sessionID string) (int, error)
FlushOutbox sends all unflushed events from the outbox to the remote server, in batches of at most maxAppendBatch events.
The cursor advances only over events the server actually took for THAT batch, never to the session's high-water mark - see resolveAck. Each batch gets its own ack and cursor advance, so a failure partway through a large backlog leaves the cursor at the last successfully acked batch, not at 0: the next flush attempt (this call, retried, or the periodic ticker) resumes from there rather than resending what already landed. If a 409 Conflict is returned it returns ErrConflict so the session can stop.
func FlushOutboxWithTrace ¶
func IdentityDir ¶
IdentityDir is where the sync identity records for a store directory live. One file per session key - REVIEW CHANGE 7 - so two processes sharing a store directory never read-modify-write one shared file. An empty store dir yields an empty identity dir, not a RELATIVE one: filepath.Join("", "chat-sync", "identity") is "chat-sync/identity", which would put a durable record wherever the process happens to be running.
func IdentityKey ¶
IdentityKey derives the lookup key for a chat session's sync identity file.
The key must be stable across runs, or a restart cannot find the identity it wrote and every resume mints a new handle, a new outbox and a new remote session. The chat session id is the only session-scoped value with that property: chat.Session.Load retargets it at the resumed session (internal/chat/context_catalog.go reclaimContextSession, pinned by TestResumeAcrossProcessesUpdatesSameSessionInPlace), and /clear and an agent switch both keep it (internal/chat/session_id_stability_test.go). The session store directory is NOT usable: it is workspace-scoped and is copied between sessions by the TUI pool, so keying on it would merge unrelated conversations into one transcript.
The chat session id is also the principal, so it is never stored. A SHA-256 digest is stable exactly when the id is, is one-way, and grants nothing. This mirrors provider.sessionUserKey, which hashes the same value for the same reason.
A session whose id is NOT restored on load (a file-backed session with no context store, whose meta.json carries no session id) simply gets a fresh key and a fresh identity, which is the correct answer: nothing else about that session is rebound to its previous run either.
func OutboxDirFor ¶
func OutboxDirFor(storeDir string, handle LocalHandle) string
OutboxDirFor is the outbox directory for a local handle. The layout lives here rather than in each host because two hosts (the plain CLI surface and the TUI session pool) build it, and a layout that drifts between them silently orphans one surface's cursor.
An empty store dir yields an empty outbox dir, not a RELATIVE one - same reasoning as IdentityDir above. Without this guard, a caller that failed to resolve a real store dir silently got an outbox at "chat-sync/sessions/<handle>" relative to wherever the process's cwd happened to be, which for `mivia chat` run from inside a project checkout means real conversation transcripts land inside the project tree itself. OpenOutbox refuses an empty dir (MkdirAll fails), so the caller's sync attach fails closed instead.
func ProbeEndpoint ¶
ProbeEndpoint makes one bounded, unauthenticated request to the API's version-neutral /health route and reports what came back. Any HTTP answer means the host is reachable; the status code is reported so a URL that points at something other than the mivia API (a web app answering 200 to everything, a proxy answering 404) can be told apart from the real one.
func SaveIdentity ¶
func SaveIdentity(dir, key string, id SyncIdentity) error
SaveIdentity writes id durably: temp file, fsync, rename, matching the cursor's write discipline. A half-written identity file is indistinguishable from a corrupt one, and that costs a fresh remote session.
Types ¶
type AgentOrigin ¶
type AgentOrigin struct {
Task string `json:"task,omitempty"`
Name string `json:"name,omitempty"`
// Depth carries NO omitempty, deliberately. omitempty omits the zero
// value, so a depth of 0 - the root agent - vanished from the wire and an
// attributed root event became indistinguishable from a subagent event
// whose depth simply had not been stamped. Consumers split the main
// transcript from the subagent lanes on exactly this field, so the web
// viewer filed the root agent's own prose and reasoning under a lane and
// showed a transcript of tool cards with nothing between them. When an
// origin is present at all, its depth is now always stated.
Depth int `json:"depth"`
// ParentTask is the Task of the subagent that dispatched this one, empty
// when the root loop did. Depth reports how deep a run sits; this reports
// under WHICH run, which two runs at the same depth do not share.
ParentTask string `json:"parent_task,omitempty"`
}
AgentOrigin identifies a subagent when an event was emitted by one.
type AppendResult ¶
type AppendResult struct {
LastSeq int64 `json:"lastSeq"`
InsertedCount int `json:"insertedCount"`
}
AppendResult is the response from appending events.
type AssistantDeltaPayload ¶
AssistantDeltaPayload is the payload of mivia.chat.v1.assistant.delta.
type AssistantMessagePayload ¶
type AssistantMessagePayload struct {
Envelope
Fragments int `json:"fragments"`
Bytes int `json:"bytes"`
Status string `json:"status"`
Text string `json:"text,omitempty"`
}
AssistantMessagePayload is the payload of mivia.chat.v1.assistant.message.
type AssistantResetPayload ¶
AssistantResetPayload is the payload of mivia.chat.v1.assistant.reset.
It carries only the envelope. The block it applies to is the envelope's own Block, so one reset can never be mistaken for another turn's or another subagent run's. Reason is a short, content-free classification of why the turn restarted.
type AuthorUserIDProvider ¶
AuthorUserIDProvider resolves the local CLI session's own authenticated principal id, so a remote input can be checked against it before it is ever handed to the UI as something to execute. nil (or an error return) means the id cannot be verified, which fails closed: see InputPoller.resolveAuthorUserID.
func DefaultAuthorUserIDProvider ¶
func DefaultAuthorUserIDProvider() AuthorUserIDProvider
DefaultAuthorUserIDProvider resolves the logged-in CLI session's own authenticated principal id into an AuthorUserIDProvider, or nil when there is no local session.
Unlike DefaultTokenProvider, the closure this returns is NOT free to call: every invocation hits POST-free but network-bound /v1/auth/me (miviaauth.Service.Whoami), with no local cache of the identity to fall back on. InputPoller.resolveAuthorUserID calls it at most once per poller lifetime and caches the result, specifically so this cost is paid once, lazily, only when a remote input actually needs verifying - never unconditionally at session-attach time, which would put a real network call on every chat-sync session this process ever opens.
type BadRequestError ¶
BadRequestError conveys a 400 from the API.
The distinction it carries is load-bearing. Retrying an identical body against a 400 resubmits a request the server has already judged malformed, for as long as the process lives; treating a sequence gap as fatal "guarantees the failure it is trying to avoid" (chat-sync-cli-slice.md:163). A caller therefore needs to tell the two apart, which a bare error string cannot express.
func (*BadRequestError) Error ¶
func (e *BadRequestError) Error() string
func (*BadRequestError) Is ¶
func (e *BadRequestError) Is(target error) bool
func (*BadRequestError) IsSequenceComplaint ¶
func (e *BadRequestError) IsSequenceComplaint() bool
IsSequenceComplaint reports whether the server named a sequence problem.
The check is on the message because the API returns no machine-readable code for it - the live guard probe pins only that the message names the sequence (internal/chatsync/live_guards_test.go, "a forward sequence gap is rejected"). It matches both recoverable shapes: the gap wording ("sequence gap: ...") and the in-batch wording ("Non-contiguous batch: events[1] has seq 2494, expected 2493"). Schema-validation messages name no sequence and stay poison.
It is deliberately a NECESSARY, not a sufficient, condition: the caller must still re-read the session and confirm the server's mark actually moves the batch, because a message match alone would rebase on a client bug. The bounded-fork tradeoff stands: one rebase at an unmoved mark is followed by the loop guard, so a complaint that no rebase can fix forks at most maxNoProgressRecoveries times before sync stops.
type BlockGrammar ¶
type BlockGrammar struct {
Prose string `json:"prose"`
Streams []string `json:"streams"`
ToolBlockField string `json:"toolBlockField"`
ResetBlockIsStreamOnly bool `json:"resetBlockIsStreamOnly"`
StepIsDense bool `json:"stepIsDense"`
StepReuse string `json:"stepReuse"`
}
BlockGrammar is the recorded grammar of the envelope's block field, the values api/contracts/chat-sessions.v1.json publishes under blockGrammar. TestBlockGrammarMatchesContractSnapshot holds the two together, so the consumer's vendored copy cannot drift looser than what this producer asserts.
func RecordedBlockGrammar ¶
func RecordedBlockGrammar() BlockGrammar
RecordedBlockGrammar returns the grammar this producer implements.
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client executes authenticated requests against /v1/chat-sessions.
func NewClient ¶
func NewClient(tokens TokenProvider, opts ClientOptions) (*Client, error)
NewClient constructs a Client. tokens is required. A nil provider is refused rather than tolerated: this client uploads conversation content, and an unauthenticated upload is the fail-open case this argument exists to make impossible.
func (*Client) AppendEvents ¶
func (c *Client) AppendEvents(ctx context.Context, sessionID string, events []EventItem) (*AppendResult, error)
AppendEvents appends a batch of events to a session stream.
func (*Client) AppendEventsWithTrace ¶
func (c *Client) AppendEventsWithTrace(ctx context.Context, sessionID string, events []EventItem, batchID, writerID string) (*AppendResult, error)
AppendEventsWithTrace appends events and carries correlation headers for server-side delivery diagnostics without changing the event payload schema.
func (*Client) AuthLost ¶
AuthLost reports whether this client has latched a fatal authentication failure. The session loop reads it to stop the pusher, poller and heartbeat instead of retrying a session that cannot come back.
func (*Client) ConsumeInput ¶
func (c *Client) ConsumeInput(ctx context.Context, sessionID, inputID string) (*SessionInput, error)
ConsumeInput acknowledges receipt of a remote input via POST /v1/chat-sessions/{id}/inputs/{inputID}/consume.
inputID is server-supplied (the id NextInput's response just handed back), so it is validated exactly like sessionID before it is re-embedded into this second request's path - see validatePathID's doc comment.
func (*Client) CreateSession ¶
CreateSession registers a new session via POST /v1/chat-sessions.
func (*Client) EndSession ¶
EndSession ends a session via POST /v1/chat-sessions/{id}/end.
func (*Client) GetEvents ¶
func (c *Client) GetEvents(ctx context.Context, sessionID string, afterSeq int64, limit int) ([]StoredEvent, error)
GetEvents reads back events by cursor via GET /v1/chat-sessions/{id}/events.
func (*Client) GetSession ¶
GetSession fetches session metadata via GET /v1/chat-sessions/{id}.
type ClientOptions ¶
ClientOptions configures Client. It deliberately carries no token provider: the provider is a positional argument to NewClient so that a missing one is a compile error, not a zero value that silently sends conversation content with no Authorization header.
type ConflictError ¶
ConflictError conveys detailed 409 conflict payload from server.
func (*ConflictError) Error ¶
func (e *ConflictError) Error() string
func (*ConflictError) Is ¶
func (e *ConflictError) Is(target error) bool
type ContextCompactedPayload ¶
type ContextCompactedPayload struct {
Envelope
Compaction any `json:"compaction,omitempty"`
Message string `json:"message,omitempty"`
BetweenTurns bool `json:"between_turns,omitempty"`
}
ContextCompactedPayload is the payload of mivia.chat.v1.context.compacted.
type CreateSessionParams ¶
type CreateSessionParams struct {
Title string `json:"title"`
CwdLabel string `json:"cwdLabel,omitempty"`
HostLabel string `json:"hostLabel,omitempty"`
// ForkedFromSessionID names the session this one continues, set only by
// recovery. omitempty: an API that predates the field runs
// forbidNonWhitelisted and would 400 a create that sent it empty.
ForkedFromSessionID string `json:"forkedFromSessionId,omitempty"`
}
CreateSessionParams defines the request payload for registering a new chat session.
type Cursor ¶
type Cursor struct {
FlushedSeq int64 `json:"flushed_seq"`
FlushedAt time.Time `json:"flushed_at"`
}
Cursor represents the durable flushed sequence marker.
type Endpoint ¶
type Endpoint struct {
URL string
// Source names what supplied URL: "[sync] api_url", the env var with the
// file or process that set it, or "default".
Source string
}
Endpoint is where sync would upload and what decided that.
func ResolveEndpoint ¶
ResolveEndpoint resolves the sync API root the same way OpenSession will use it, and says where the value came from. It is the ONE resolver: a diagnostic that re-derived the chain would be a second place to get it wrong, which is the failure the diagnostic exists to catch.
type Envelope ¶
type Envelope struct {
V int `json:"v"`
At time.Time `json:"at"`
Turn string `json:"turn"`
Block string `json:"block,omitempty"`
Agent *AgentOrigin `json:"agent,omitempty"`
Trunc *Truncation `json:"trunc,omitempty"`
Redacted []string `json:"redacted,omitempty"`
SourceTurnID string `json:"source_turn_id,omitempty"`
WriterID string `json:"writer_id,omitempty"`
}
Envelope is the common envelope embedded in every wire event payload.
type ErrorEnvelope ¶
type ErrorEnvelope struct {
StatusCode int `json:"statusCode"`
Error string `json:"error"`
Message json.RawMessage `json:"message"`
Path string `json:"path"`
}
ErrorEnvelope is the shape the API's global exception filter emits. Message is a string for thrown exceptions and an array for validation failures, so it stays raw.
type EventItem ¶
type EventItem struct {
Seq int64 `json:"seq"`
Type string `json:"type"`
Payload json.RawMessage `json:"payload"`
}
EventItem is one event in an append batch.
type HeartbeatRunner ¶
type HeartbeatRunner struct {
// contains filtered or unexported fields
}
HeartbeatRunner maintains the periodic and event-driven heartbeat for an active session.
func NewHeartbeatRunner ¶
func NewHeartbeatRunner(client *Client, sessionID string, interval time.Duration) *HeartbeatRunner
NewHeartbeatRunner creates a HeartbeatRunner with default status "waiting_input".
func NewHeartbeatRunnerWithStatus ¶
func NewHeartbeatRunnerWithStatus(client *Client, sessionID string, interval time.Duration, status string) *HeartbeatRunner
NewHeartbeatRunnerWithStatus creates a HeartbeatRunner with the specified initial status.
func (*HeartbeatRunner) SetSessionID ¶
func (h *HeartbeatRunner) SetSessionID(sessionID string)
SetSessionID updates the remote session ID for heartbeat requests.
func (*HeartbeatRunner) SetStatus ¶
func (h *HeartbeatRunner) SetStatus(ctx context.Context, status string)
SetStatus updates the current status and triggers an immediate heartbeat send.
func (*HeartbeatRunner) Start ¶
func (h *HeartbeatRunner) Start(ctx context.Context)
Start begins the heartbeat background goroutine.
func (*HeartbeatRunner) Stop ¶
func (h *HeartbeatRunner) Stop(ctx context.Context)
Stop terminates the heartbeat runner.
type HookRanPayload ¶
type HookRanPayload struct {
Envelope
// Phase is PreToolUse, PostToolUse or Stop.
Phase string `json:"phase"`
// Program is the script's NAME, never its path. A path describes the
// operator's machine and this payload leaves it.
Program string `json:"program"`
// Tool is the call the hook ran for; empty for a Stop hook.
Tool string `json:"tool,omitempty"`
ToolCallID string `json:"tool_call_id,omitempty"`
// Blocked is true for the run that refused the call. A reader must not
// have to infer this from the absence of a later event.
Blocked bool `json:"blocked"`
// OutputBytes always reports the real size, even when Output is withheld,
// so a reader can tell "the hook said nothing" from "the hook said
// something you are not being shown".
OutputBytes int `json:"output_bytes"`
// Output is what the hook printed - for a blocking hook, the reason. It
// rides the same include-tool-io gate as tool output, because it is the
// same class of thing: text a local program produced.
Output string `json:"output,omitempty"`
}
HookRanPayload is the payload of mivia.chat.v1.hook.ran: one lifecycle hook run on the operator's machine.
It exists for one case above all. A hook can BLOCK a tool call, and a blocked call never produces a tool.ended - so without this event a remote reader watches a tool.started that simply never finishes, with no way to learn that a local policy stopped it.
ToolCallID ties the row to the call it gated. It is empty for a Stop hook, which runs for the turn rather than for one call.
type IdentityRef ¶
IdentityRef locates a stored identity so a running session can write back the remote session id it resolved. A zero IdentityRef disables the write-back, which is what every test that does not care about resume uses.
func (IdentityRef) IsZero ¶
func (r IdentityRef) IsZero() bool
IsZero reports whether the ref names no identity file.
type InputPoller ¶
type InputPoller struct {
// contains filtered or unexported fields
}
InputPoller polls for remote user inputs and executes the exactly-once handshake.
func NewInputPoller ¶
func NewInputPoller(client *Client, sessionID string, waitSeconds int, authorUserID AuthorUserIDProvider, stateDir ...string) *InputPoller
NewInputPoller creates a new InputPoller. authorUserID resolves the CLI's own authenticated principal for verifying who queued each input; nil verifies nothing and so refuses every input (fail closed).
func (*InputPoller) Inputs ¶
func (p *InputPoller) Inputs() <-chan RemoteInput
Inputs returns the channel receiving consumed remote inputs.
func (*InputPoller) MarkReceived ¶
func (p *InputPoller) MarkReceived(id string)
MarkReceived records that id reached UI custody. Idempotent: a second call for the same id is a no-op, since AckReceived may be invoked more than once by design (see its doc comment). Uses the same synchronous, fsync-before-return durable write delivered_ledger.go uses (writeFileDurably) rather than an async/best-effort one, because the guarantee this ledger is FOR - "this id definitely reached custody, even across a crash" - requires it. Callers must never invoke this from the bubbletea Update goroutine directly; internal/tui/view/screen/conversation's ackCmd wraps it in a tea.Cmd instead.
Serialized on p.mu: internal/tui/view/screen/conversation's mount.go batches one ackCmd per buffered remote-input event for the SAME session into one tea.Batch, and bubbletea's runtime executes every Cmd in a batch concurrently (execBatchMsg). Without a lock, two concurrent calls on this same poller instance can both read the ledger before either write lands, and the loser's write silently drops the id the winner already recorded - see TestInputPoller_MarkReceived_ConcurrentCallsDoNotLoseIDs.
func (*InputPoller) SetOnRejected ¶
func (p *InputPoller) SetOnRejected(fn func(id, sessionID, reason string))
SetOnRejected installs the callback pollOnce and recoverPendingInput invoke for every input validateRemoteInput refuses. Must be called before Start; nil (the default) means refusals stay silent to the host.
func (*InputPoller) SetSessionID ¶
func (p *InputPoller) SetSessionID(sessionID string)
SetSessionID updates the session ID polled by this runner.
func (*InputPoller) Start ¶
func (p *InputPoller) Start(ctx context.Context)
Start launches the polling loop.
func (*InputPoller) Stop ¶
func (p *InputPoller) Stop(ctx context.Context)
Stop terminates the poller and waits for loop exit, bounded by ctx.
Closing stopCh also cancels the in-flight long poll, so the loop normally exits at once. The ctx arm is the backstop for a request the transport is slow to abandon: a shutdown deadline the caller set must bound this call, never the poll deadline.
type LocalHandle ¶
type LocalHandle string
LocalHandle is the local-only sync handle: the outbox directory name and the cursor key. It is a distinct type so a chat principal cannot be assigned into it by accident.
Be honest about what this type does and does not buy. It is NOT a barrier: LocalHandle(sess.SessionID) compiles. Two things enforce the separation, and the type only makes them easier to see:
- mivia.go.no-chat-principal-as-sync-handle rejects the conversion form.
- TestChatPrincipalNeverReachesTheWireOrDisk proves at runtime that the principal is in no request target, no request body and no file this package writes. That test is the real gate; it survives every loophole the type and the rule miss.
type NextInput ¶
type NextInput struct {
Input *SessionInput `json:"input"`
}
NextInput is the envelope returned by long-polling /inputs/next.
type Outbox ¶
type Outbox struct {
// contains filtered or unexported fields
}
Outbox owns local per-session append-only event files, mutual exclusion locking, and durable cursor markers.
func OpenOutbox ¶
OpenOutbox opens or creates the outbox directory, acquires the process lock, and loads the current cursor.
func (*Outbox) AdvanceCursor ¶
AdvanceCursor updates cursor.json atomically via rename after server ack.
func (*Outbox) Append ¶
Append persists WireEvents into events.jsonl with fsync before returning.
The batch is all-or-nothing. A failure part way through used to leave the earlier records on disk while `unflushed` stayed un-incremented, so the batch read as never stored: the caller rolled the seq counter back and reissued the same seqs, and the file then held two records for one seq. The server's contiguity check rejects that, which wedges the stream for good. On any failure the file is therefore truncated back to the offset it held before the batch started, and no counter moves.
func (*Outbox) Dead ¶
Dead reports whether a failed rebase left this outbox permanently unwritable (see rewriteEventsFileLocked). Recovery checks it before minting a session it could never move the backlog into.
func (*Outbox) Rebase ¶
Rebase re-indexes every unflushed event to start at base+1 and sets the flushed cursor to base. It returns the number of re-indexed events.
This is the runtime answer to a sequence-gap 400 when the server is BEHIND the outbox: the events between the server's mark and the outbox's first unflushed seq are gone, and no resend can produce them, so the only way back to a contiguous stream is to renumber onto the server's mark. Forking is the same operation with base 0.
func (*Outbox) ResetForFork ¶
ResetForFork rewrites the outbox with only the currently unflushed events re-indexed starting at sequence 1, and resets the flushed cursor to 0. It returns the number of re-indexed unflushed events.
func (*Outbox) UnflushedCount ¶
UnflushedCount reports how many appended events the server has not acked.
func (*Outbox) UnflushedEvents ¶
func (ob *Outbox) UnflushedEvents() ([]StoredEvent, error)
UnflushedEvents reads all stored events with seq > cursor.flushed_seq.
type Projector ¶
type Projector struct {
// contains filtered or unexported fields
}
Projector performs pure synchronous projection of events.Event streams into WireEvent sequences for a single chat session.
func NewProjector ¶
func NewProjector(sessionID string, initialSeq int64, opts ProjectorOptions) *Projector
NewProjector constructs a Projector for sessionID starting at initialSeq.
func (*Projector) ProjectWithDrops ¶
ProjectWithDrops checks the subscription drop counter and projects ev. If drops advanced, a sync.dropped event is emitted immediately before ev.
func (*Projector) ResetSeq ¶
ResetSeq resets the sequence counter to a specified sequence (e.g. on fork).
func (*Projector) RollbackDrops ¶
RollbackDrops un-advances the drop watermark by delta, so a sync.dropped marker that was built but never stored does not consume the loss it reported.
checkDrops moves the watermark when it CONSTRUCTS the marker. If the append that would have made it durable fails, the marker never reaches the wire while the watermark has already moved, so the next marker under-reports and the hole settled decision 6 exists to expose becomes invisible again. The watermark must therefore track what was STORED, exactly as the seq counter does.
func (*Projector) RollbackSeq ¶
RollbackSeq rolls back the sequence counter by n (e.g. on outbox append failure).
func (*Projector) RollbackStreaming ¶
RollbackStreaming undoes the streaming bookkeeping for a batch of wire events that was projected but never stored.
Without it a failed append is worse than a dropped event: the counters say the text was streamed, so INV-1 empties the settled message, and a viewer ends up holding neither the fragments (never stored) nor the whole answer (deliberately omitted). The turn's entire reply disappears while the transcript still looks contiguous.
The counters must therefore track what was STORED, exactly as the sequence number and the drop watermark already do.
func (*Projector) SetSyncOpts ¶
SetSyncOpts flips the three content-gate flags - IncludeThinking, IncludeToolIO, StreamAssistant - on a live Projector. It is the live re-arm seam: a runtime toggle from the General settings screen calls this, and the next projection round honors the new values.
Non-flag fields (WriterID, RedactToolArgs, ErrorMessage) are preserved; they are part of the session's identity and contract (WriterID is the per-run identity the server uses to detect foreign writes; ErrorMessage and RedactToolArgs are policy), and silently rewriting them would break the durable-transcript chain or the redaction contract.
SetSyncOpts is safe to call from any goroutine: opts is a value field and Go's memory model guarantees a non-pointer write is atomic for single-word bools on every supported platform; for the three we touch, a torn read would at worst mean one stale event before the new flag is visible, which is the same behavior a restart would have.
type ProjectorOptions ¶
type ProjectorOptions struct {
StreamAssistant bool
IncludeToolIO bool
IncludeThinking bool
WriterID string
// RedactToolArgs is the host's tool-argument redaction decision, passed
// in as a value. It composes with IncludeToolIO as an AND: tool IO ships
// only when IncludeToolIO is true AND RedactToolArgs is false.
RedactToolArgs bool
// ErrorMessage classifies a turn error into redaction-safe text for the
// wire. Nil selects defaultErrorMessage, which is content-free by
// construction. Provider and tool error text can quote the request that
// produced it (DC-14), so err.Error() must never reach this wire.
ErrorMessage func(error) string
}
ProjectorOptions configures a Projector.
Settled decision 7 keeps this package a LEAF: it takes values and functions rather than importing the app's own packages. ErrorMessage and RedactToolArgs exist for that reason - the alternatives were an import of internal/chat (which drags in ~47 internal packages) and a read of internal/tools' process-global atomic.Bool (which makes any test of gate composition mutate a package global).
type RemoteInput ¶
RemoteInput represents a received and consumed user input from the remote viewer.
type Session ¶
type Session struct {
ID string `json:"id"`
OrganizationID string `json:"organizationId"`
UserID string `json:"userId"`
Title string `json:"title"`
CwdLabel *string `json:"cwdLabel"`
HostLabel *string `json:"hostLabel"`
Status string `json:"status"`
LastSeq int64 `json:"lastSeq"`
LastEventAt string `json:"lastEventAt"`
LastHeartbeatAt string `json:"lastHeartbeatAt"`
EndedAt *string `json:"endedAt"`
CreatedAt string `json:"createdAt"`
}
Session models a chat session record on the API.
type SessionAttachment ¶
type SessionAttachment struct {
SessionID string
ServerSeq int64
FlushedSeq int64
ForkedFrom string
}
SessionAttachment represents the established sync binding for a session.
func AttachSession ¶
func AttachSession(ctx context.Context, client *Client, outbox *Outbox, params CreateSessionParams, existingSessionID string, writerID string) (*SessionAttachment, error)
AttachSession connects to an existing remote session or creates a new remote session. When serverLastSeq > cursor.FlushedSeq: - If all events past cursor match writerID (or writerID is empty) -> ADOPT: ServerSeq = sess.LastSeq, FlushedSeq = sess.LastSeq - If any event has a different writerID -> FORK: end old session, create new session, ForkedFrom = oldID
type SessionInput ¶
type SessionInput struct {
ID string `json:"id"`
SessionID string `json:"sessionId"`
AuthorUserID string `json:"authorUserId"`
Kind string `json:"kind"`
Body string `json:"body"`
CreatedAt string `json:"createdAt"`
ConsumedAt *string `json:"consumedAt"`
}
SessionInput is a remote input queued for this session.
type SessionOptions ¶
type SessionOptions struct {
// TokenProvider is required. OpenSession refuses to start without it.
TokenProvider TokenProvider
ClientOptions ClientOptions
ProjectorOptions ProjectorOptions
OutboxDir string
// RemoteSessionID is the server-assigned session to re-attach to, read
// from the persisted identity. Empty creates a fresh remote session. It is
// the ONLY value that may appear in a request URL, and it is deliberately
// NOT the chat session id OpenSession filters the bus on.
RemoteSessionID string
// LocalHandle is the local-only handle that names OutboxDir. It is carried
// here so the identity write-back can rewrite the record without re-reading
// (and possibly re-minting) it.
LocalHandle LocalHandle
// Identity names the file the resolved remote session id is written back
// to. Zero disables the write-back: the session then works exactly as
// before, but the next run cannot find the remote session it created.
Identity IdentityRef
MaxUnflushed int
PollWaitSeconds int
HeartbeatPeriod time.Duration
Telemetry *SyncTelemetry
CreateTitle string
CwdLabel string
HostLabel string
EnablePolling bool
// AuthorUserIDProvider resolves the CLI's own authenticated principal id
// for verifying who queued a remote input. Required for EnablePolling to
// deliver anything: nil verifies nothing, so InputPoller refuses every
// input closed. See chatsync.AuthorUserIDProvider.
AuthorUserIDProvider AuthorUserIDProvider
// OnInputRejected, when set, is called for every remote input
// InputPoller refuses (session id mismatch, unsupported kind, malformed
// body, unverifiable or mismatched author). Nil means refusals stay
// silent to the host.
OnInputRejected func(id, sessionID, reason string)
// EventBufSize bounds the handler-to-worker channel. Zero means
// defaultEventBufSize. Overflow here is real loss and is counted into the
// sync.dropped marker, exactly like loss on the bus.
EventBufSize int
// OnStop is called exactly once, with the same string StopReason returns,
// when sync latches a terminal stop: a fatal auth failure, a poison 400,
// or a recovery bound. It is never called for an orderly Stop, and never
// for a 409 or 404 on their own - those recover onto a new session.
//
// It exists because the contract's poison rule is "stop syncing and SAY
// SO", and a reason only a getter can reach says nothing: no host polls
// SyncSession, so a silent stop looks exactly like a healthy idle
// session. Nil is valid and means the host does not surface stops.
//
// It runs on a detached goroutine, NOT on the worker: the worker still has
// a drain and a final flush to finish, and a host callback that blocks
// (writing to a terminal, or sending on a full UI channel) must not be
// able to hold that up.
OnStop func(reason string)
// OnDegraded is called once when pushes stop landing - three consecutive
// failures, or a failure a minute after the last success - and
// OnRecovered once when they land again. Neither is a stop: the outbox
// keeps the backlog and the retry schedule keeps trying. They exist
// because a transient failure that never ends is otherwise silent for the
// whole session (flushNow's default branch retries forever and says
// nothing), and the same fact is written to status.json in OutboxDir so
// it outlives the process. Both run on a detached goroutine, like OnStop.
OnDegraded func(reason string)
OnRecovered func()
}
SessionOptions configures a SyncSession.
type StoredEvent ¶
type StoredEvent struct {
SessionID string `json:"sessionId"`
Seq int64 `json:"seq"`
Type string `json:"type"`
Payload json.RawMessage `json:"payload"`
CreatedAt string `json:"createdAt"`
}
StoredEvent is an event read back from the cursor endpoint.
type SubagentAssistantDeltaPayload ¶
type SubagentAssistantDeltaPayload struct {
Envelope
Text string `json:"text"`
Index int `json:"index"`
}
SubagentAssistantDeltaPayload is the payload of mivia.chat.v1.subagent.assistant.delta. It mirrors AssistantDeltaPayload, but Index counts fragments within ONE subagent run: two subagents streaming at once each start at 0, and Envelope.Agent.Task is what tells them apart.
type SubagentAssistantMessagePayload ¶
type SubagentAssistantMessagePayload struct {
Envelope
Fragments int `json:"fragments"`
Bytes int `json:"bytes"`
Status string `json:"status"`
Text string `json:"text,omitempty"`
}
SubagentAssistantMessagePayload is the payload of mivia.chat.v1.subagent.assistant.message. Fragments and Text obey the same rule as the root aggregate (INV-1: Text is empty exactly when Fragments is non-zero), counted per subagent run rather than per turn.
type SubagentEndedPayload ¶
type SubagentEndedPayload struct {
Envelope
Name string `json:"name,omitempty"`
Status string `json:"status,omitempty"`
}
SubagentEndedPayload is the payload of mivia.chat.v1.subagent.ended.
type SubagentProgressPayload ¶
SubagentProgressPayload is the payload of mivia.chat.v1.subagent.progress.
Detail is the only field. This type once also declared elapsed_seconds, steps and tool_calls, which no projection ever set: the producer (MultiStepHandler.stepOnEvent) formats those three numbers into the heartbeat's Detail text, and agent.Event has no numeric fields to carry them structurally. Three permanently-absent optional fields are worse than none - a client cannot tell "this build never sends it" from "nothing to report" - so they are gone until something actually populates them.
type SubagentStartedPayload ¶
type SubagentStartedPayload struct {
Envelope
Name string `json:"name,omitempty"`
Task string `json:"task,omitempty"`
}
SubagentStartedPayload is the payload of mivia.chat.v1.subagent.started: the run-level opening signal for one subagent.
The envelope's own timestamp is the run's start time, so this carries no separate started_at.
Task is a bounded PREVIEW of what the run was asked to do, not the prompt it received: the producer caps it at 200 bytes (subagents.maxTaskDescriptionBytes) and, for a dispatch whose input is not a bare string, it can be raw JSON rather than natural language. It is named and budgeted accordingly - a 32 KiB prompt budget could never bind against a 200-byte producer bound, so the real limit would never appear in trunc.fields.
type SubagentThinkingDeltaPayload ¶
type SubagentThinkingDeltaPayload struct {
Envelope
Bytes int `json:"bytes"`
Index int `json:"index"`
Text string `json:"text,omitempty"`
}
SubagentThinkingDeltaPayload is the payload of mivia.chat.v1.subagent.thinking.delta. Text is present only when the host enabled thinking; Bytes always reports the real size, so a viewer can show that a subagent is reasoning without the content.
type SubagentThinkingMessagePayload ¶
type SubagentThinkingMessagePayload struct {
Envelope
Fragments int `json:"fragments"`
Bytes int `json:"bytes"`
Status string `json:"status"`
Text string `json:"text,omitempty"`
}
SubagentThinkingMessagePayload is the payload of mivia.chat.v1.subagent.thinking.message. Same settled form and same INV-1 rule as the root aggregate, counted per subagent run: a lane's reasoning is withheld by a redaction policy for the same reason the root's is, so it needs the same whole-block form to fall back to.
type SubagentToolEndedPayload ¶
type SubagentToolEndedPayload struct {
Envelope
ToolCallID string `json:"tool_call_id"`
Name string `json:"name"`
Status string `json:"status"`
OutputBytes int `json:"output_bytes"`
Detail string `json:"detail,omitempty"`
Output string `json:"output,omitempty"`
}
SubagentToolEndedPayload is the payload of mivia.chat.v1.subagent.tool.ended.
type SubagentToolStartedPayload ¶
type SubagentToolStartedPayload struct {
Envelope
ToolCallID string `json:"tool_call_id"`
Name string `json:"name"`
InputBytes int `json:"input_bytes"`
Phase string `json:"phase,omitempty"`
Input string `json:"input,omitempty"`
}
SubagentToolStartedPayload is the payload of mivia.chat.v1.subagent.tool.started.
type SyncDroppedPayload ¶
type SyncDroppedPayload struct {
Envelope
Dropped uint64 `json:"dropped"`
TotalDropped uint64 `json:"total_dropped"`
}
SyncDroppedPayload is the payload of mivia.chat.v1.sync.dropped.
type SyncForkedPayload ¶
type SyncForkedPayload struct {
Envelope
NewSessionID string `json:"new_session_id"`
// ForkedFrom names the session that was abandoned, so the new session's
// stream self-describes its ancestor. Empty on markers older producers
// wrote.
ForkedFrom string `json:"forked_from,omitempty"`
}
SyncForkedPayload is the payload of mivia.chat.v1.sync.forked.
type SyncIdentity ¶
type SyncIdentity struct {
LocalHandle LocalHandle `json:"local_handle"`
RemoteSessionID string `json:"remote_session_id,omitempty"`
WriterID string `json:"writer_id"`
}
SyncIdentity is the durable sync identity of ONE chat session. It is the record that makes cross-run resume work, and it is deliberately separate from the chat session's own principal.
Three identities exist on this path and conflating any two of them is a defect:
- LocalHandle - local only. It names the outbox directory and is the cursor key. It never reaches the wire.
- RemoteSessionID - the server-assigned id, the ONLY one that goes in a URL. Persisting it is what lets a restart re-attach to the SAME remote session instead of 404ing into a fresh one at lastSeq 0 while the local cursor survives - the permanent sequence-gap 400.
- The chat session id - a live capability (the contextstate authorization subject). It appears NOWHERE in this struct, nowhere in a request body, and nowhere in a URL. It is used only as the input to IdentityKey and as the local event-bus filter.
func LoadIdentityReadOnly ¶
func LoadIdentityReadOnly(dir, key string) (SyncIdentity, bool)
LoadIdentityReadOnly reads the persisted sync identity for key WITHOUT minting or creating anything on disk. If the file is missing, invalid, or unreadable, it returns (SyncIdentity{}, false).
func LoadOrCreateIdentity ¶
func LoadOrCreateIdentity(dir, key string) (SyncIdentity, error)
LoadOrCreateIdentity returns the persisted sync identity for key, minting and storing a fresh one when none is readable.
A missing, unreadable, corrupt or incomplete file mints fresh rather than failing: losing the mapping costs one orphaned remote session, while refusing to start costs the user their sync entirely. The returned identity is always usable even when the error is non-nil - the error then reports only that the mint could not be made durable, so this run holds a handle the next run will not find.
type SyncSession ¶
type SyncSession struct {
// contains filtered or unexported fields
}
SyncSession coordinates real-time synchronization between a local chat session, its outbox, and the remote API.
func NewSyncSessionForTest ¶
func NewSyncSessionForTest(opts ProjectorOptions) *SyncSession
NewSyncSessionForTest constructs a SyncSession whose projector is already attached, without going through OpenSession (which requires a real remote server).
func OpenSession ¶
func OpenSession(ctx context.Context, bus *events.Bus, chatSessionID string, opts SessionOptions) (*SyncSession, error)
OpenSession arms chat sync for a local chat session: it opens the outbox, subscribes to the event bus and starts the local workers. It puts NOTHING on the wire. The remote attach - the create or re-attach request, the identity write-back, the heartbeat and the input poller - is deferred to the FIRST event, in ensureAttached. Events only exist once a turn starts, so a user who opens the CLI and sends no message creates no remote session and no request of any kind; a session closed before its first event is a purely local object. The first event attaches exactly once and is then delivered like any other.
chatSessionID is the LOCAL chat session id and is used for exactly one thing: filtering the event bus, which stamps that id on every event. It is the contextstate authorization subject, so it never reaches a URL, a request body or a file. The session to attach to comes from opts.RemoteSessionID and the outbox directory from opts.OutboxDir, both resolved by the caller from the persisted identity - the caller must resolve them, because the outbox is opened here BEFORE the attach, so OutboxDir has to already carry the handle.
func (*SyncSession) HandleEvent ¶
func (s *SyncSession) HandleEvent(ctx context.Context, ev events.Event)
HandleEvent processes incoming events via non-blocking send to an internal channel.
func (*SyncSession) Inputs ¶
func (s *SyncSession) Inputs() <-chan RemoteInput
Inputs returns the channel of remote inputs.
func (*SyncSession) LastSeq ¶
func (s *SyncSession) LastSeq() int64
LastSeq returns the current sequence number assigned by the projector.
func (*SyncSession) MarkInputReceived ¶
func (s *SyncSession) MarkInputReceived(id string)
MarkInputReceived records that id reached UI custody. Thin delegation to the underlying poller's ledger, mirroring Inputs() above. A no-op if this session has no poller.
func (*SyncSession) ProjectorOptsForTest ¶
func (s *SyncSession) ProjectorOptsForTest() ProjectorOptions
ProjectorOptsForTest returns the projector's ProjectorOptions by value, so an external test can assert on it without reaching into SyncSession's unexported fields.
func (*SyncSession) SessionID ¶
func (s *SyncSession) SessionID() string
SessionID returns the currently active remote session ID, or the empty string until the first event attaches - an id exists only once the server has assigned one, and the server is not asked until a message is sent.
func (*SyncSession) SetStatus ¶
func (s *SyncSession) SetStatus(ctx context.Context, status string)
SetStatus forwards status updates to the heartbeat runner. Before the first event attaches there is no runner and nothing to send to: a status is API traffic, and a session with no message produces none.
func (*SyncSession) SetSyncOpts ¶
func (s *SyncSession) SetSyncOpts(includeThinking, includeToolIO, streamAssistant bool)
SetSyncOpts flips the three content-gate flags on the live projector without rebuilding the outbox, identity, or writer id. It is the per-session live re-arm seam: the pool calls this for every attached session when the operator toggles a sync opt-out in General settings, and the next projection round honors the new values.
Two no-op cases, both silent by design:
- projector nil: the session is pre-attach (OpenSession has not yet finished the deferred remote attach that mints the projector). Record the values on s.opts so the next attach that consults opts sees the desired state.
- remoteEnded true: the session has stopped (poison 400, fatal auth, recovery bound). The toggle cannot reach the wire, and the live session is going down anyway.
Otherwise, flip the projector's opts in place under s.mu. Locking matches LastSeq's; the projector read sites also lock s.mu, so a concurrent HandleEvent either observes the old flags and projects one more event under them, or observes the new flags - either way the durable transcript is consistent with the operator's last successful persist.
func (*SyncSession) Stop ¶
func (s *SyncSession) Stop(ctx context.Context) error
Stop terminates the session sync loop, flushes pending events, and closes resources. Every wait it performs is bounded by ctx: a caller that gives Stop a 200ms deadline gets control back inside 200ms even when a long poll is parked and the final append is stalled on a dead network.
When the deadline arrives first, Stop returns ctx.Err() and hands the outbox close to a goroutine that waits for the worker - which itself waits for the uploader - because closing the outbox under a goroutine still using it would race its file handle.
func (*SyncSession) StopReason ¶
func (s *SyncSession) StopReason() string
StopReason returns why sync stopped, or the empty string while it runs.
func (*SyncSession) Stopped ¶
func (s *SyncSession) Stopped() bool
Stopped reports whether sync has latched a terminal stop: a fatal auth failure, a poison 400, or a recovery bound. The local chat is unaffected either way.
func (*SyncSession) StoppedForTest ¶
func (s *SyncSession) StoppedForTest()
StoppedForTest latches remoteEnded=true so the next SetSyncOpts call hits the stopped-session no-op branch. Real stop paths (Stop, classifyFlushError -> recoverRemoteSession) own the same atomic in production.
type SyncStatus ¶
type SyncStatus struct {
State string `json:"state"`
Reason string `json:"reason"`
Unflushed int `json:"unflushed"`
LastSuccessAt *time.Time `json:"last_success_at"`
ConsecutiveFailures int `json:"consecutive_failures"`
Recoveries int `json:"recoveries"`
CreateFailures int `json:"create_failures"`
CreateThrottledUntil *time.Time `json:"create_throttled_until"`
At time.Time `json:"at"`
}
SyncStatus is the record status.json holds. A reader tells a throttled create stall from a plain push stall by create_throttled_until, which is null whenever no throttle is engaged.
type SyncTelemetry ¶
type SyncTelemetry struct {
// contains filtered or unexported fields
}
SyncTelemetry records local synchronization stages. Counters are process local; log records carry identifiers for one event and upload attempt.
func NewSyncTelemetry ¶
func NewSyncTelemetry(logger *slog.Logger) *SyncTelemetry
NewSyncTelemetry creates telemetry using logger. A nil logger disables logs but keeps counters active.
func (*SyncTelemetry) Snapshot ¶
func (t *SyncTelemetry) Snapshot() SyncTelemetrySnapshot
type SyncTelemetrySnapshot ¶
type SyncTelemetrySnapshot struct {
Produced uint64
Projected uint64
Appended uint64
Uploaded uint64
UploadFailures uint64
LastAckSeq int64
LastSuccessAt time.Time
OutboxDepth int64
}
SyncTelemetrySnapshot is a point-in-time metrics view for diagnostics.
type ThinkingDeltaPayload ¶
type ThinkingDeltaPayload struct {
Envelope
Bytes int `json:"bytes"`
Index int `json:"index"`
Text string `json:"text,omitempty"`
}
ThinkingDeltaPayload is the payload of mivia.chat.v1.thinking.delta.
type ThinkingMessagePayload ¶
type ThinkingMessagePayload struct {
Envelope
Fragments int `json:"fragments"`
Bytes int `json:"bytes"`
Status string `json:"status"`
Text string `json:"text,omitempty"`
}
ThinkingMessagePayload is the payload of mivia.chat.v1.thinking.message: one thinking block, settled at the moment the block closed.
Fragments and Text obey INV-1 exactly as the assistant aggregate does - Text is empty exactly when Fragments is non-zero - counted per BLOCK rather than per turn, because a turn reasons once per step and each step is its own block. When the deltas shipped their text the viewer already has it and this event only completes the block; when they did not (a redaction policy withheld every fragment, or thinking never streamed at all) Text carries the whole reasoning, redacted as ONE string. That is the entire point: a pattern spanning two fragments is invisible to a per-fragment redactor and visible to this one.
type TokenEnsurer ¶
TokenEnsurer is the narrow view of *miviaauth.Service this package needs.
It is an interface rather than the concrete Service so tests can drive the busy and reauth branches without a token file, but it is deliberately NOT a second token cache: the refresh token is one-time use and internal/miviaauth/lock.go serializes the refresh window across processes. Every caller must hand over the one real Service so that lock stays the only refresher.
type TokenProvider ¶
TokenProvider retrieves a bearer token for authentication.
func DefaultTokenProvider ¶
func DefaultTokenProvider() TokenProvider
DefaultTokenProvider resolves the logged-in CLI session into the TokenProvider this package requires, or nil when there is no local session.
A nil result means "do not sync", and every wiring site must treat it that way: uploading conversation content anonymously is the failure this returns nil to avoid. It is also the activation rule's other half - authentication is what turns sync on, so the absence of a session is what turns it off, silently and with no error.
It lives here, next to NewTokenProvider, because the plain-CLI surface and the TUI session pool are separate wiring sites, and a rule honoured by one of two sibling call sites is the defect class this repo hits most often.
func NewTokenProvider ¶
func NewTokenProvider(ensurer TokenEnsurer) TokenProvider
NewTokenProvider adapts a TokenEnsurer to a TokenProvider, and classifies its errors per the settled 401 policy:
- ErrRefreshBusy is TRANSIENT. Another mivia process holds the refresh lock; the stored token is untouched and the next flush retries.
- ErrReauthRequired and ErrSessionLost STOP sync. There is no local session left to refresh, and this path must never prompt mid-chat.
forceRefresh is ignored on purpose. Ensure already owns the refresh decision, under the cross-process lock; refreshing again from here would be exactly the double-refresh that revokes the session for both processes.
type ToolEndedPayload ¶
type ToolEndedPayload struct {
Envelope
ToolCallID string `json:"tool_call_id"`
Name string `json:"name"`
Status string `json:"status"`
OutputBytes int `json:"output_bytes"`
Detail string `json:"detail,omitempty"`
Output string `json:"output,omitempty"`
}
ToolEndedPayload is the payload of mivia.chat.v1.tool.ended.
type ToolStartedPayload ¶
type ToolStartedPayload struct {
Envelope
ToolCallID string `json:"tool_call_id"`
Name string `json:"name"`
InputBytes int `json:"input_bytes"`
Phase string `json:"phase,omitempty"`
Input string `json:"input,omitempty"`
}
ToolStartedPayload is the payload of mivia.chat.v1.tool.started.
type TruncField ¶
TruncField records the kept and total byte lengths of a truncated string field.
type Truncation ¶
type Truncation struct {
Fields map[string]TruncField `json:"fields"`
}
Truncation contains field-level truncation records.
type TurnEndedPayload ¶
TurnEndedPayload is the payload of mivia.chat.v1.turn.ended (F4: blocks dropped).
type TurnFailedPayload ¶
TurnFailedPayload is the payload of mivia.chat.v1.turn.failed.
type TurnStartedPayload ¶
type TurnStartedPayload struct {
Envelope
Text string `json:"text,omitempty"`
SessionRef string `json:"session_ref,omitempty"`
Synthetic bool `json:"synthetic,omitempty"`
}
TurnStartedPayload is the payload of mivia.chat.v1.turn.started.
type WireEvent ¶
type WireEvent struct {
Seq int64 `json:"seq"`
Type string `json:"type"`
Payload any `json:"payload"`
}
WireEvent is one projected event ready for the outbox or upload batch.
type WireEventSpec ¶
type WireEventSpec struct {
// Type is the mivia.chat.v1.* type string. The API names each SSE frame
// after this exact string, so a browser client must register a listener
// per entry: EventSource.onmessage never fires for these events. This is
// confirmed live by live_sse_test.go's "names the frame after the event
// type" subtest (build tag livechat) and is intentional, documented
// behavior per api/contracts/chat-sessions.v1.json's notes array; see
// docs/development/agent-workflow.md's "Live chat-session probe" section
// for the forward-compatibility trade-off this implies.
Type string
// Payload is a zero value of the struct that models this type's payload.
Payload any
}
WireEventSpec binds one wire type string to the Go struct that models its payload.
func WireEventSpecs ¶
func WireEventSpecs() []WireEventSpec
WireEventSpecs returns the event vocabulary in wire order. The returned slice is a copy, so a caller cannot edit the definition site.
Source Files
¶
- attach.go
- auth.go
- block_grammar.go
- client.go
- client_errorbody.go
- delivered_ledger.go
- doc.go
- flush_ack.go
- flush_outcome.go
- gates.go
- heartbeat.go
- identity.go
- outbox.go
- outbox_lock_unix.go
- outbox_rebase.go
- outbox_recover.go
- outbox_truncate_unix.go
- poller.go
- projector.go
- projector_assistant_settle.go
- projector_rollback.go
- projector_state.go
- projector_stream_flush.go
- projector_thinking.go
- projector_tools.go
- received_ledger.go
- reconcile.go
- remote_input_validate.go
- session.go
- session_attach.go
- session_badrequest.go
- session_flush.go
- session_process.go
- session_recover.go
- session_shutdown.go
- sync_status.go
- telemetry.go
- test_helpers.go
- truncate.go
- wire.go
- wiring.go