textmachine/platform/internal/ingest/events.go

// Package ingest is the platform side of the engine seam (D39.85): it reads the engine's NDJSON
// event stream and hands each event to a sink that materializes it into the reporting database. It
// never opens the engine's SQLite and never parses human output.
//
// The vocabulary below started as the platform's PROPOSAL, written as code so the engine zone could
// answer it with a diff. It did (D39.131, engine commit 9cfe080): the emitter EXISTS, it writes
// stream version 1.1, and the diff it answered with is carried here — `Ceiling.Scope` and two more
// `Finished` outcomes. Where this file and `backend/internal/runevents/runevents.go` describe the
// same field they must say the same thing; the engine's file is the one that writes the bytes.
//
// The TRANSPORT, by contrast, is not a proposal and not open: it was ratified as D39.106 §2 after
// four independent panels (research/25 §Форма). Its form, so that nothing here reads as a brief to
// build something else:
//
//   - The engine is a TRANSIENT SYSTEMD UNIT per run (Restart=no, stopped by SIGTERM). The platform
//     is NOT its parent — a run has to outlive a platform deploy or restart, and a child cannot.
//   - Events are events.jsonl in the BOOK's directory, append-only: an outbox projection of rows
//     the engine has already committed to its SQLite, written in the same transaction as the
//     checkpoint. The platform TAILS that file; its cursor (engine_run_id, seq) is committed in one
//     Postgres transaction with the effect the event had.
//   - Re-reading lines after a reader crash is therefore NORMAL, not an anomaly: delivery is
//     at-least-once and a duplicate is not an error (PD-105).
//
// Rejected there with zero votes out of fifteen, so that none of them comes back as a fresh idea:
// platform-as-parent with a stdout pipe, a dedicated fd 3, and stdout/journald as the source of
// events. What the pipe path in supervisor.go is today is the DEV mode, and only that (D39.106 §3).
package ingest

import (
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"time"
)

// StreamVersion is the version this decoder implements. The rule is terraform's, ratified by
// D39.85: a MINOR bump adds fields and event types — unknown ones are ignored; a MAJOR bump is
// refused, because a stream whose meaning changed must not be materialized as if it had not.
//
// 1.1, matching the emitter that landed (D39.131). Only the MAJOR is compared, so the reader would
// have gone on working at "1.0" — the bump is here because the constant is a claim about what this
// build UNDERSTANDS, and since `Ceiling.Scope` and the two new outcomes are read below, leaving it
// at 1.0 would have been a false one. A reader that lies about its own vocabulary is how a later
// minor gets shipped against the wrong assumption.
const StreamVersion = "1.1"

// Type is the event name. Unknown values are legal under the minor rule and are dropped by the
// sink, not by the decoder.
type Type string

const (
	// TypeHello is always the first line: the version handshake.
	TypeHello Type = "hello"
	// TypeProgress carries the per-phase counters. Per phase because a unit is done only once its
	// edit resolved, so one end-to-end counter reads zero for the whole draft wave.
	TypeProgress Type = "progress"
	// TypeUnitDone is one shipped (or withheld) edit unit.
	TypeUnitDone Type = "unit_done"
	// TypeBankStop is the book-wide signing stop before the edit wave.
	TypeBankStop Type = "bank_stop"
	// TypeCeiling is the resumable halt on the spend ceiling: the fact only, no figures.
	TypeCeiling Type = "ceiling"
	// TypeSpend is the cumulative spend counter (owner 05.08, PLATFORM_DIRECTION §2).
	TypeSpend Type = "spend"
	// TypeFinished is the last line of a clean stream.
	TypeFinished Type = "finished"
)

// Envelope is one line of the stream.
type Envelope struct {
	// Seq is per PROCESS and starts at 1. It is the second half of the ratified idempotency key
	// (run_id, seq) — where run_id is Hello.EngineRunID, not the platform's run: a resumed run is
	// a new process whose seq restarts, so keying on the platform run would drop its whole stream.
	Seq  int64           `json:"seq"`
	Type Type            `json:"type"`
	Time time.Time       `json:"time"`
	Data json.RawMessage `json:"data"`
}

// Hello is the handshake payload.
type Hello struct {
	StreamVersion string `json:"stream_version"`
	// EngineRunID is the id the engine mints per invocation (today: the trace id, main.go:67-70).
	// Once the engine accepts an external trace context (row 102) the platform can supply it, and
	// the idempotency namespace becomes ours by construction.
	EngineRunID string `json:"engine_run_id"`
	BookID      string `json:"book_id"`
	// ChunkerVersion lets the platform notice that the chapter manifest it persisted was produced
	// by a different chunker — the case that silently re-numbers chapters (row 100).
	ChunkerVersion string `json:"chunker_version"`
}

// Counter is a phase counter, in units.
type Counter struct {
	Done  int `json:"done"`
	Total int `json:"total"`
}

// Progress is the run's counters. ETASeconds mirrors what status already computes
// (pipeline/status.go:112) — the owner asked for a visible ETA (K-5).
type Progress struct {
	Draft      Counter `json:"draft"`
	Edit       Counter `json:"edit"`
	ETASeconds int     `json:"eta_seconds,omitempty"`
}

// UnitDone is one resolved edit unit. Chapter and Unit are the engine's own numbering; mapping
// them onto the platform's opaque ids is the materializer's job.
//
// Shipped and Flagged are BOTH needed and neither implies the other: a flagged unit legally ships
// with text (sanitizer cleanup, c-lite member drop), and the pair of them is exactly the contract's
// derivation of unit state — shipped → translated, flagged without text → withheld.
type UnitDone struct {
	// Chapter is the engine's dense 1-based ordinal; Unit is the unit's LEADER chunk index — together
	// the join key the engine's manifest publishes as `first_chunk_idx`. The pair is what makes the
	// fold an assignment: it is stable across processes, so re-delivery is idempotent by identity
	// rather than by luck (D39.131 п.2г).
	Chapter int    `json:"chapter"`
	Unit    int    `json:"unit"`
	Wave    string `json:"wave"` // draft | edit
	Shipped bool   `json:"shipped"`
	Flagged bool   `json:"flagged"`
	// Reason is the ENGINE's flag reason (glossary_miss, sanitizer_stripped, …). It is stored and
	// never projected: the product phrase is applied at read time from the contract's map, so a
	// reason this platform has never heard of still gets a neutral phrase instead of a hole.
	Reason string `json:"reason,omitempty"`
}

// The waves a unit can be resolved by (backend/internal/runevents: WaveDraft, WaveEdit). A closed
// vocabulary on both sides of the seam.
const (
	WaveDraft = "draft"
	WaveEdit  = "edit"
)

// BankStop is the signing stop. The full table travels as an artifact (engine backlog row 101),
// not through the stream: a thousand rows are not an event.
type BankStop struct {
	TermsProposed int `json:"terms_proposed"`
}

// Ceiling is the ceiling halt. It carries the FACT and nothing else: money never reaches the
// platform's wire or its INFO logs (D39.84), and the stop is resumable, so it is not a failure.
//
// Scope names WHICH ceiling stopped the run and arrived with the engine's diff (D39.131). It is the
// diagnosis half of PD-157: the platform sets the BOOK ceiling flush against the hold it took, and
// has no way at all to see `ceilings.day_usd`, which lives in a book.yaml the operator owns. Before
// this field a run stopped by a daily limit the platform never chose was indistinguishable from one
// that spent exactly the money bought for it — and the two need opposite answers to "may I
// continue".
type Ceiling struct {
	Halted bool `json:"halted"`
	// Scope is `book` or `day`. Anything else — absent, or a word a later minor invents — is a
	// ceiling this build cannot place, and is recorded as exactly that (pgstore.CeilingPause).
	Scope string `json:"scope,omitempty"`
}

// The ceiling scopes (backend/internal/runevents: ScopeBook, ScopeDay).
const (
	ScopeBook = "book"
	ScopeDay  = "day"
)

// Spend is the freshness channel for money, and ONLY that: the balance is protected by the hold
// taken before the process is spawned and by the per-book ceiling the engine enforces itself, so a
// lost tail costs an indicator its accuracy and never costs the account its correctness. Building
// enforcement on this event is forbidden — the stream is at-least-once and a crash truncates it.
//
// CUMULATIVE, not a delta: a redelivered or duplicated line is then harmless, because the
// materializer keeps the maximum seen for the run instead of adding anything up. Integer
// micro-USD: money never travels as a float, and the engine's ledger is a lower bound, so the
// conversion at the seam rounds up (this is an internal channel into a private table — D39.84
// governs the USER's wire, screen and INFO logs, and none of them see this).
type Spend struct {
	CommittedMicroUSD int64 `json:"committed_micro_usd"`
}

// Finished is the terminal line of a run that ended on purpose. Its ABSENCE is meaningful too: a
// stream without it was truncated by a crash.
//
// Outcome mirrors the engine's exit contract so a stream that ends cleanly needs no exit-code
// archaeology. `ceiling` and `stopped` arrived with the engine's diff (D39.131) for the two endings
// the original four could not name: a resumable ceiling halt, which `failed` is exactly what the
// contract forbids calling it, and a caught SIGTERM.
// Its values are the Outcome vocabulary of exit.go, minus the refusal — an invocation that was
// turned down did no work, so it writes no verdict about work at all.
//
// The platform does not ACT on this line: a run is closed from its exit marker, because money is
// settled from what the process did rather than from a line it managed to write before dying.
type Finished struct {
	Outcome Outcome `json:"outcome"`
}

// checkVersion applies the semver rule to a handshake.
func checkVersion(got string) error {
	gotMajor, err := major(got)
	if err != nil {
		return err
	}
	wantMajor, err := major(StreamVersion)
	if err != nil {
		return err
	}
	if gotMajor != wantMajor {
		return fmt.Errorf("%w: stream is %s, this build speaks %s", ErrUnsupportedVersion, got, StreamVersion)
	}
	return nil
}

func major(v string) (int, error) {
	head, _, _ := strings.Cut(v, ".")
	n, err := strconv.Atoi(head)
	if err != nil {
		return 0, fmt.Errorf("ingest: malformed stream version %q", v)
	}
	return n, nil
}