textmachine/backend/internal/pipeline/events.go

package pipeline

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"path/filepath"
	"sync"
	"sync/atomic"
	"time"

	"textmachine/backend/internal/chunk"
	"textmachine/backend/internal/obs"
	"textmachine/backend/internal/runevents"
	"textmachine/backend/internal/store"
)

// events.go: WHEN the run-event seam emits (row 103). The form lives in internal/runevents, the durable
// sequencing in internal/store (events_outbox); here is the driver's half — the call sites, the counters
// and the policy for a journal that cannot be written.
//
// ORDER, and it is the invariant: a fact is committed to SQLite, THEN its line is enqueued in the
// outbox, THEN the outbox is projected onto the file. Nothing is ever written to the journal that the
// database does not already hold, which is the direction that would cost money and lie to a user; the
// opposite direction — a fact committed whose line was lost to a crash — is bounded and named per event
// below, and its ratified repair channel is the `status --json` resync (D39.106 §2).
//
// The two events a lost line cannot hurt are `spend` (cumulative: the next line carries the total again)
// and `progress` (an assignment: the next line carries the counters again). The one it can is
// `unit_done`, which a reader FOLDS BY INCREMENT — see unitResolved.

// emitter writes one process's region of a book's event journal.
//
// Everything is under one mutex, including the projection: the waves emit from N workers, and the reader
// cannot survive seq N+1 appearing above seq N. Under the mutex the projection always reads the outbox
// in sequence order, so it appends a dense prefix even when two workers commit concurrently (the write
// pool is single-connection, so a transaction sees MAX(seq)=N only after N's transaction committed).
type emitter struct {
	mu       sync.Mutex
	store    *store.Store
	journal  *runevents.Journal
	log      *slog.Logger
	runID    string
	bookID   string
	now      func() time.Time
	lastSeen int64 // the highest seq already on the file
	degraded bool  // the journal refused a write; the run continues (PD-60)
	// marks are the announce-once keys whose line has been enqueued but not yet confirmed on the file.
	marks map[int64]string

	// waves is the live per-phase counter. nil until the driver knows the book's cut (beginWaves).
	waves *waveCounters
}

// openEmitter opens the book's journal, drops the previous processes' outbox rows and writes this
// process's handshake as its first line. runID is the process identity (the trace id) and is the half of
// the ratified idempotency key (engine_run_id, seq) that makes a resumed run a new stream in the same
// file rather than a corruption of the old one.
func openEmitter(st *store.Store, dir, runID, bookID string, log *slog.Logger) (*emitter, error) {
	j, err := runevents.OpenJournal(dir)
	if err != nil {
		return nil, err
	}
	// The stream identity must be FRESH, and since row 102 the caller can supply it — so this is the one
	// place that checks. Reusing an id that already wrote for this book is not a small mistake: the
	// sequence continues from the previous process's numbers instead of restarting at 1, and the
	// projection, whose cursor starts at zero, re-appends that entire previous stream before writing a
	// handshake at some seq far past 1. A reader then re-applies every counting event it already applied
	// and finally refuses the handshake. Measured on the real binary before this guard: two runs under
	// one TM_TRACE_ID produced 33 lines, 15 of them byte-identical duplicates, and a hello at seq 16.
	//
	// The run is NOT stopped for it — a paid book must not die because a caller mislabelled it — so the
	// stream takes a fresh identity and says so at ERROR. The caller's id keeps naming the run in the
	// logs; only the seam's key changes.
	if used, err := st.EventsUsed(runID); err != nil {
		j.Close()
		return nil, err
	} else if used {
		fresh := obs.NewTraceID()
		log.Error("this run was given an id that has already written run events for this book; a stream identity must be unique per process, so the event stream announces itself under a fresh one (fix the caller: TM_TRACE_ID must be unique per invocation)",
			"given", runID, "stream_id", fresh)
		runID = fresh
	}
	if err := st.ForgetEvents(runID); err != nil {
		j.Close()
		return nil, err
	}
	e := &emitter{store: st, journal: j, log: log, runID: runID, bookID: bookID, now: time.Now, marks: map[int64]string{}}
	e.emit(runevents.TypeHello, runevents.Hello{
		StreamVersion:  runevents.StreamVersion,
		EngineRunID:    runID,
		BookID:         bookID,
		ChunkerVersion: chunkerVersion,
	})
	return e, nil
}

func (e *emitter) close() error {
	if e == nil {
		return nil
	}
	return e.journal.Close()
}

// emit enqueues one event and projects the journal. It never returns an error, and that is the answer to
// PD-60 ("what does the engine do when the journal cannot be written"), chosen explicitly because the
// row says both positions are legitimate and a silent third is not:
//
// the engine DEGRADES LOUDLY and keeps running. A paid book must not die because a freshness side-channel
// cannot be written — the money protection is the platform's hold and the engine's own ceiling, neither
// of which depends on this file (D39.106 §2) — and the two facts a reader most needs, the ceiling halt
// and the graceful stop, also travel as distinct exit codes, so they survive a journal that never opened.
// Nothing is dropped: the lines stay in the outbox, every later event retries the whole pending prefix,
// so a transient ENOSPC or EIO heals itself and the journal ends up complete and gap-free. If it never
// heals, the reader's ratified fallback is the `status --json` resync.
//
// The rejected alternative is to BLOCK or abort the run on a journal failure. It is legitimate — it keeps
// the projection exact — but it makes the availability of a status channel a precondition for translating
// a book that has already been paid for, and on the realistic failure (a full disk) the ledger's own
// writes are failing at the same moment and the run stops anyway, one layer lower and with a better error.
func (e *emitter) emit(t runevents.Type, data any) {
	if e == nil {
		return
	}
	e.mu.Lock()
	defer e.mu.Unlock()
	e.enqueue(t, data)
	e.project()
}

func (e *emitter) enqueue(t runevents.Type, data any) {
	at := e.now()
	if err := e.store.EnqueueEvent(e.runID, func(seq int64) ([]byte, error) {
		return runevents.Line(seq, t, at, data)
	}); err != nil {
		e.log.Error("run-event outbox write failed; this event will not reach the platform's stream (the run continues; the resync channel is `tmctl status --json`)",
			"event", string(t), "err", err)
	}
}

// project appends every stored line the file has not got yet, in sequence order, and only THEN records
// which announce-once events a reader has now seen. The order is the point: a line still in the outbox
// when a process dies has no ledger row, so the next process announces its unit again instead of
// assuming a reader was told.
func (e *emitter) project() {
	caught := 0
	for {
		lines, err := e.store.PendingEvents(e.runID, e.lastSeen, projectBatch)
		if err != nil {
			e.degrade("read the pending run events", err)
			return
		}
		for _, l := range lines {
			if err := e.journal.Append(l.Line); err != nil {
				e.degrade("append to the run-event journal", err)
				e.markAnnounced() // whatever DID land is still announced
				return
			}
			e.lastSeen = l.Seq
			// Per line, not per batch: what is left unmarked is what a crash makes a later process announce
			// a second time, so the window is kept to a single transaction rather than the rest of the batch.
			e.markAnnounced()
		}
		caught += len(lines)
		if len(lines) < projectBatch {
			break // drained: a short batch means there was nothing more to read
		}
	}
	if e.degraded && caught > 0 {
		e.degraded = false
		e.log.Warn("the run-event journal is writable again; the pending events were caught up",
			"caught_up", caught, "last_seq", e.lastSeen)
	}
	e.markAnnounced()
}

// projectBatch bounds one read of the outbox. It matters only on the failure path: while the journal
// refuses writes the cursor cannot advance, so the unprojected prefix grows with every event and an
// unbounded read would re-materialize all of it on each one, under this mutex. Large enough that a
// healthy run always drains in one read, small enough that a degraded one costs a constant.
const projectBatch = 256

// markAnnounced records the announce-once events whose line is now on the file.
func (e *emitter) markAnnounced() {
	if len(e.marks) == 0 {
		return
	}
	landed := map[int64]string{}
	for seq, key := range e.marks {
		if seq <= e.lastSeen {
			landed[seq] = key
		}
	}
	if len(landed) == 0 {
		return
	}
	if err := e.store.MarkAnnounced(e.runID, landed); err != nil {
		// Not fatal and not silent: the cost of losing this write is that a later process announces those
		// units a second time, which an at-least-once reader is required to absorb.
		e.log.Error("could not record which run events have been delivered; a later run may announce them again",
			"events", len(landed), "err", err)
		return
	}
	for seq := range landed {
		delete(e.marks, seq)
	}
}

// degrade reports the FIRST failure of a run of them and stays quiet afterwards: a full disk fails on
// every event, and a per-event ERROR line would bury the run's real output.
func (e *emitter) degrade(what string, err error) {
	if e.degraded {
		return
	}
	e.degraded = true
	e.log.Error("could not "+what+"; the platform's live stream stalls here and resumes if this heals (the run continues — money is protected by the hold and the ceiling, not by this file; resync channel: `tmctl status --json`)",
		"last_seq_on_file", e.lastSeen, "err", err)
}

// spendLine is the money event, handed to the ledger so it rides the SAME transaction that settles the
// money (D39.106 §2 "the outbox row is written in the transaction of the checkpoint"). It is the one
// event for which that matters: a journal line claiming spend the ledger never booked is a divergence
// about money, not about a progress bar.
//
// It never fails the settle. A render failure returns "no event" rather than an error, because the
// alternative is a rolled-back settle for a call the provider has already billed — losing a paid
// checkpoint to protect an indicator is the wrong way round.
func (e *emitter) spendLine() *store.SpendLine {
	if e == nil {
		return nil
	}
	at := e.now()
	return &store.SpendLine{RunID: e.runID, Line: func(seq int64, committedUSD float64) ([]byte, error) {
		line, err := runevents.Line(seq, runevents.TypeSpend, at,
			runevents.Spend{CommittedMicroUSD: runevents.MicroUSD(committedUSD)})
		if err != nil {
			e.log.Error("could not render the spend event; the settle proceeds without it (the counter is cumulative — the next one carries the total again)", "err", err)
			return nil, nil
		}
		return line, nil
	}}
}

// flush projects whatever the ledger enqueued inside its own transactions (the spend event). Called
// after a settle, so money freshness does not wait for the unit to finish.
func (e *emitter) flush() {
	if e == nil {
		return
	}
	e.mu.Lock()
	defer e.mu.Unlock()
	e.project()
}

// waveCounters is the live per-phase progress, in OUTPUT UNITS — the granularity every engine read model
// counts in, so the stream and the `status --json` resync fold into one column on the same scale.
//
// `counted` starts as the units ALREADY resolved when this process opened the book, which is what makes
// the counter book-state rather than process-state: a resumed run walks every unit again at $0, and a
// counter that started at zero would walk a reader's progress bar backwards on every resume.
type waveCounters struct {
	draft, edit runevents.Counter
	counted     map[unitWave]bool
	// since/resolved are this PROCESS's throughput, which is what the stream's ETA is derived from: the
	// wall clock since the run began working, and how many units it has resolved in it. A resumed run
	// that replays everything at $0 resolves nothing and therefore offers no estimate, which is correct
	// — it has measured nothing.
	since    time.Time
	resolved int
}

// eta is the seconds remaining at the rate this run has actually achieved, or 0 when nothing has been
// measured yet (the reader renders no estimate rather than "0 s left").
func (w *waveCounters) eta(now time.Time) int {
	left := (w.draft.Total - w.draft.Done) + (w.edit.Total - w.edit.Done)
	if w.resolved == 0 || left <= 0 {
		return 0
	}
	eta := int(now.Sub(w.since).Seconds() / float64(w.resolved) * float64(left))
	if eta < 1 {
		// Work remains, so the honest floor is a second: the field is optional in the wire form, and an
		// omitted one does not leave the reader's column alone — it NULLS it (see runevents.Progress).
		return 1
	}
	return eta
}

type unitWave struct {
	wave    string
	chapter int
	unit    int
}

// beginWaves seeds the counters from the store and announces them. Called once, when the driver knows
// the book's cut, before the first wave.
func (e *emitter) beginWaves(units []editUnit, shape waveShape, stored map[chunkKey][]store.ChunkStatus) {
	if e == nil {
		return
	}
	w := &waveCounters{counted: map[unitWave]bool{}, since: e.now()}
	if shape.nDraft > 0 {
		w.draft.Total = len(units)
	}
	if shape.nEdit > 0 {
		w.edit.Total = len(units)
	}
	for _, u := range units {
		draft, edit := shape.resolved(u, unitRows(u, stored))
		if draft {
			w.counted[unitWave{runevents.WaveDraft, u.Chapter, u.FirstChunkIdx}] = true
			w.draft.Done++
		}
		if edit {
			w.counted[unitWave{runevents.WaveEdit, u.Chapter, u.FirstChunkIdx}] = true
			w.edit.Done++
		}
	}
	e.mu.Lock()
	e.waves = w
	e.mu.Unlock()
	// No estimate yet: nothing has been measured, and inventing one from a previous run would be a
	// number about work this process has not done.
	e.emit(runevents.TypeProgress, runevents.Progress{Draft: w.draft, Edit: w.edit})
}

// unitResolved announces one output unit a WAVE has just finished with, and the counters it moved.
//
// Two different questions are asked here, and conflating them is what a first version got wrong:
//
//   - HAS A READER BEEN TOLD? — answered by the announce-once ledger in the outbox, because a reader
//     folds `unit_done` by INCREMENT and this is the vocabulary's only counting event. Re-announcing a
//     unit a resumed run merely replayed at $0 would add a whole book to a chapter counter on every
//     resume; NOT announcing one whose process died between the disposition and the insert would leave
//     that chapter short forever. Only a durable ledger answers both, and it is why the once-rows
//     outlive their run.
//   - DOES THE COUNTER MOVE? — answered by what was already resolved when this process opened the book.
//     The counters are BOOK state, not process state: a resumed run re-walks every finished unit, and a
//     counter that started at zero would walk a reader's progress bar backwards.
//
// The two disagree exactly in the interesting cases, and each is then right on its own terms: a unit
// whose announcement was lost to a crash is re-announced but not re-counted, and a book translated
// before this ledger existed is announced for the first time without its counters jumping.
//
// ⚠ Residual, and it is the irreducible one: between the outbox commit below and the line reaching the
// file there is no transaction, so a process killed between those two adjacent statements loses that
// announcement for good. The class ENDS at the fold, not here — `unit_done` carries a stable
// (chapter, unit, wave) identity, so a reader that assigns instead of incrementing is exact under any
// delivery, which is what at-least-once delivery (D39.119 п.3) has always asked of a consumer. Raised
// as a diff with the pack.
func (e *emitter) unitResolved(wave string, chapter, unit int, shipped, flagged bool, reason string) {
	if e == nil {
		return
	}
	e.mu.Lock()
	defer e.mu.Unlock()
	at := e.now()
	key := unitWave{wave, chapter, unit}
	// The counters may be unavailable (their baseline read failed, beginRunEvents) — the ANNOUNCEMENT is
	// not: it needs no baseline, a reader folds it per chapter, and dropping it would leave that chapter
	// short forever. Only the progress line is skipped in that case.
	w := e.waves
	counts := false
	if w != nil {
		if counts = !w.counted[key]; counts {
			w.counted[key] = true
			if wave == runevents.WaveEdit {
				w.edit.Done++
			} else {
				w.draft.Done++
			}
			w.resolved++
		}
	}
	onceKey := e.unitOnceKey(key)
	announcing, seq, err := e.store.EnqueueOnce(e.runID, onceKey, func(seq int64) ([]byte, error) {
		return runevents.Line(seq, runevents.TypeUnitDone, at, runevents.UnitDone{
			Chapter: chapter, Unit: unit, Wave: wave, Shipped: shipped, Flagged: flagged, Reason: reason,
		})
	})
	if err != nil {
		e.log.Error("run-event outbox write failed; this unit will not reach the platform's stream (the run continues; the resync channel is `tmctl status --json`)",
			"event", string(runevents.TypeUnitDone), "wave", wave, "chapter", chapter, "unit", unit, "err", err)
	}
	if announcing {
		e.marks[seq] = onceKey
	}
	if counts {
		e.enqueue(runevents.TypeProgress, runevents.Progress{Draft: w.draft, Edit: w.edit, ETASeconds: w.eta(at)})
	}
	if announcing || counts {
		e.project()
	}
}

// unitOnceKey is the identity of one announcement. It carries the BOOK because the ledger lives in a
// project database and `project_db` may be shared by two books: without it the second book's units would
// collide with the first's keys and never be announced at all.
func (e *emitter) unitOnceKey(k unitWave) string {
	return fmt.Sprintf("unit:%s:%s:%d:%d", e.bookID, k.wave, k.chapter, k.unit)
}

// terminal writes the stream's last line — and, on a ceiling halt, the `ceiling` event before it.
//
// The ceiling is the whole of PD-113: the engine returns the stop as an ERROR, every unrecognised error
// maps to exit 1, and the platform then records `failed` for a stop its own contract says is `paused`.
// It travels twice by design — as this event and as its own exit code — because the two channels fail
// independently: a journal that could not be written still leaves the code, and a process killed before
// it exited still leaves the event.
func (e *emitter) terminal(res *BookResult, err error) {
	if e == nil {
		return
	}
	var halt *CeilingHalt
	var sigStop *WaveSignatureStop
	var refusal *Refusal
	switch {
	case errors.As(err, &refusal):
		// A refusal did no work, so the stream gets no verdict about work. Writing `failed` here would
		// have the exit code say "refused, nothing happened" while the stream said the run failed.
		return
	case errors.As(err, &halt):
		e.emit(runevents.TypeCeiling, runevents.Ceiling{Halted: true, Scope: halt.Scope})
		e.emit(runevents.TypeFinished, runevents.Finished{Outcome: runevents.OutcomeCeiling})
	case errors.As(err, &sigStop):
		// The bank_stop event itself was written where the stop happened, with its term count.
		e.emit(runevents.TypeFinished, runevents.Finished{Outcome: runevents.OutcomeBankStop})
	case errors.Is(err, context.Canceled):
		// In tmctl the run context is cancelled by SIGINT/SIGTERM and by nothing else (main.go's
		// signal.NotifyContext), so this is the graceful stop, not a crash.
		e.emit(runevents.TypeFinished, runevents.Finished{Outcome: runevents.OutcomeStopped})
	case err != nil:
		e.emit(runevents.TypeFinished, runevents.Finished{Outcome: runevents.OutcomeFailed})
	case res != nil && res.Flagged > 0:
		e.emit(runevents.TypeFinished, runevents.Finished{Outcome: runevents.OutcomeFlagged})
	default:
		e.emit(runevents.TypeFinished, runevents.Finished{Outcome: runevents.OutcomeClean})
	}
}

// openEvents starts this process's region of the book's journal, once per runner.
//
// A run without a trace id on its context is not an invocation the seam describes — the id IS the
// engine_run_id, half of the ratified idempotency key, and a reader refuses an empty one rather than
// collapsing every run into one namespace. tmctl always opens a traced context (main.go); an in-process
// caller that does not is driving the pipeline, not running a job, and writes no journal.
func (r *Runner) openEvents(ctx context.Context) {
	if r.events != nil {
		return
	}
	ri, ok := obs.ReqInfoFromContext(ctx)
	if !ok || ri.TraceID == "" {
		return
	}
	e, err := openEmitter(r.Store, r.journalDir(), ri.TraceID, r.Book.BookID, r.Log)
	if err != nil {
		// A journal that cannot even be OPENED is reported and not fatal, by the same policy emit() is:
		// the run is paid for, the money is protected by the hold and the ceiling, and the reader's
		// fallback channel is the `status --json` resync. Nothing here returns an error, on purpose — an
		// error return would be a way for observability to stop a run, and there must not be one.
		r.Log.ErrorContext(ctx, "could not open the run-event journal; the platform's live stream will be silent for this run (the run continues; resync channel: `tmctl status --json`)",
			"dir", r.journalDir(), "err", err)
		return
	}
	r.events = e
}

// beginRunEvents seeds the live per-phase counters from what the store already holds. It reads the
// stored rows once — the same read `status` makes — so the baseline is the BOOK's state and not this
// process's: a resumed run walks every finished unit again at $0, and counters that started at zero
// would walk a reader's progress bar backwards on every resume.
// It degrades rather than aborting, by the same policy emit() follows: a read taken for a progress
// indicator must not be what kills a paid run. Without the baseline the counters are simply not
// published — every other event still is.
func (r *Runner) beginRunEvents(ctx context.Context, chunks []chunk.Chunk) {
	if r.events == nil {
		return
	}
	statuses, err := r.Store.ChunkStatusesForBook(r.Book.BookID)
	if err != nil {
		r.Log.ErrorContext(ctx, "could not read the stored dispositions for the run-event counters; this run publishes no progress (the run continues; resync channel: `tmctl status --json`)", "err", err)
		return
	}
	byChunk := map[chunkKey][]store.ChunkStatus{}
	for _, cs := range statuses {
		byChunk[chunkKey{cs.Chapter, cs.ChunkIdx}] = append(byChunk[chunkKey{cs.Chapter, cs.ChunkIdx}], cs)
	}
	r.events.beginWaves(r.outputUnits(chunks), r.waveShape(), byChunk)
}

// draftUnitTracker answers "was that the LAST member of its unit?" for the draft wave, which fans out
// over chunks while the seam counts output units.
type draftUnitTracker struct {
	units []editUnit
	of    map[chunkKey]int // member chunk → its unit's index
	at    map[chunkKey]int // member chunk → its index in the wave's result slice
	left  []atomic.Int32
}

func newDraftUnitTracker(units []editUnit, chunks []chunk.Chunk) *draftUnitTracker {
	t := &draftUnitTracker{
		units: units,
		of:    make(map[chunkKey]int, len(chunks)),
		at:    make(map[chunkKey]int, len(chunks)),
		left:  make([]atomic.Int32, len(units)),
	}
	for i, ch := range chunks {
		t.at[chunkKey{ch.Chapter, ch.ChunkIdx}] = i
	}
	for i, u := range units {
		t.left[i].Store(int32(len(u.Members)))
		for _, m := range u.Members {
			t.of[chunkKey{m.Chapter, m.ChunkIdx}] = i
		}
	}
	return t
}

// memberDone records one member's completion and reports the unit when it was the last one. The atomic
// is also the synchronisation point for the results slice the caller then reads: each worker writes its
// own slot BEFORE decrementing, so the worker that sees zero sees every sibling's write.
func (t *draftUnitTracker) memberDone(ch chunk.Chunk) (editUnit, bool) {
	i, ok := t.of[chunkKey{ch.Chapter, ch.ChunkIdx}]
	if !ok {
		return editUnit{}, false
	}
	if t.left[i].Add(-1) != 0 {
		return editUnit{}, false
	}
	return t.units[i], true
}

// outcome folds a unit's member drafts into the pair the stream carries. The member→result index is the
// tracker's own and is built ONCE: rebuilding it per completed unit is quadratic in the book. SHIPPED means the wave
// left something the next one can work on (or, in a draft-only pipeline, something that exports);
// FLAGGED means at least one member needs a human, which is the c-lite rule the read models already use
// — a unit can legally be both.
func (t *draftUnitTracker) outcome(u editUnit, results []stageSeqResult) (shipped, flagged bool, reason string) {
	for _, m := range u.Members {
		i, ok := t.at[chunkKey{m.Chapter, m.ChunkIdx}]
		if !ok {
			continue
		}
		d := results[i]
		if d.flagged {
			if !flagged {
				flagged, reason = true, string(d.flagReason)
			}
			shipped = shipped || d.recovered != "" // a cosmetic strip still exports its cleaned remainder
			continue
		}
		shipped = shipped || d.finalText != ""
	}
	return shipped, flagged, reason
}

// journalDir is where this book's journal lives: the directory of book.yaml — the BOOK's directory
// (D39.106 §2), which is also the working directory the platform spawns the run in and tails. It is
// deliberately NOT the project DB's directory: `project_db` may point anywhere, and the seam is anchored
// to the config the caller passed.
func (r *Runner) journalDir() string {
	if r.Book.Dir != "" {
		return r.Book.Dir
	}
	return filepath.Dir(r.Book.ProjectDB)
}

// ingestSource reads and normalizes the book's source, classifying a failure as the refusal it is: this
// is the one place where "the text we were handed is unusable" is distinguishable from "the operator's
// config or host is broken", and an automated intake acts on the user's upload from that distinction.
func (r *Runner) ingestSource() (*chunk.Document, error) {
	doc, err := chunk.IngestEncoded(r.Book.SourceFile, r.Book.Encoding, r.Book.SourceLang)
	if err != nil {
		return nil, refuseSource(err)
	}
	return doc, nil
}